tinance2-doorlock-controller/python/example-tinance2.py

103 lines
3.4 KiB
Python
Raw Normal View History

2023-10-31 17:54:21 +00:00
import serial
import requests
import binascii
2023-10-31 21:09:34 +00:00
import threading
import time
2023-10-31 17:54:21 +00:00
2023-10-31 21:09:34 +00:00
ser = serial.Serial('/dev/tty.usbserial-0001', 9600)
2023-10-31 17:54:21 +00:00
tinance2_reader_identifier = "TEST-READER-1"
tinance2_reader_accesskey = "4f95166a-8a15-4963-bad8-35c7047b4269"
2023-10-31 21:09:34 +00:00
def handle_log_msg(prefix, value):
2023-10-31 17:54:21 +00:00
print(f"Received LOG_MSG: {value}")
2023-10-31 21:09:34 +00:00
def send_http_post(prefix, value):
headers = {
"Content-Type": "application/json",
"X-Identifier": tinance2_reader_identifier,
"X-Access-Key": tinance2_reader_accesskey
}
2023-10-31 17:54:21 +00:00
2023-10-31 21:09:34 +00:00
url = 'http://localhost:8000/accesscontrol/api/check-card-id'
2023-10-31 17:54:21 +00:00
data_to_post = {"full_card_id": value}
if value == "0:0":
return
response = requests.post(url, json=data_to_post, headers=headers)
if response.status_code == 200:
print(f"Sent FULL_CARD_ID: {value} via HTTP POST successfully")
ser.write("unlock".encode('utf-8'))
else:
print(f"Failed to send FULL_CARD_ID via HTTP POST. Status code: {response.status_code}")
print(response.text)
2023-10-31 21:09:34 +00:00
def reader_config_check():
previous_enabled = None # Initialize to None
while True:
print("Checking HTTP endpoint for new reader config...")
# Replace with your HTTP GET endpoint URL
url = 'http://localhost:8000/accesscontrol/api/readerinfo'
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
enabled = data.get("enabled", False)
if enabled != previous_enabled:
print("New reader enabled status: ", enabled)
if enabled:
print("Enabling reader...")
ser.write("enable".encode('utf-8'))
else:
print("Disabling reader...")
ser.write("disable".encode('utf-8'))
previous_enabled = enabled # Update the previous_enabled status
time.sleep(5) # Check every 15 seconds
def card_auth():
while True:
data = ser.readline()
2023-10-31 17:54:21 +00:00
try:
data = data.decode('utf-8')
if '=' in data:
prefix, value = data.split('=')
prefix = prefix.strip()
value = value.strip()
actions = {
"LOG_MSG": handle_log_msg,
"FULL_CARD_ID": send_http_post,
"DOOR_DISABLED_STATUS": handle_log_msg,
"DOOR_LOCK_STATUS": handle_log_msg
}
if prefix in actions:
2023-10-31 21:09:34 +00:00
actions[prefix](prefix, value)
2023-10-31 17:54:21 +00:00
else:
print(f"Unknown prefix: {prefix}")
except UnicodeDecodeError:
hex_data = binascii.hexlify(data).decode('utf-8')
print(f"Received non-UTF-8 data: {hex_data}")
2023-10-31 21:09:34 +00:00
if __name__ == "__main__":
headers = {
"Content-Type": "application/json",
"X-Identifier": tinance2_reader_identifier,
"X-Access-Key": tinance2_reader_accesskey
}
# Create and start the two threads
card_auth_thread = threading.Thread(target=card_auth)
reader_config_check_thread = threading.Thread(target=reader_config_check)
card_auth_thread.start()
reader_config_check_thread.start()
# Wait for both threads to finish (which is never in this case since they run indefinitely)
card_auth_thread.join()
reader_config_check_thread.join()