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

import serial
import requests
import binascii
import threading
import time

ser = serial.Serial('/dev/tty.usbserial-0001', 9600)
tinance2_reader_identifier = "TEST-READER-1"
tinance2_reader_accesskey = "4f95166a-8a15-4963-bad8-35c7047b4269"
reader_info = {
    "enabled": False,
    "mode": None
}

def handle_log_msg(prefix, value):
    print(f"Received LOG_MSG: {value}")

def check_valid_card(prefix, value):
    headers = {
        "Content-Type": "application/json",
        "X-Identifier": tinance2_reader_identifier,
        "X-Access-Key": tinance2_reader_accesskey
    }

    url = 'http://localhost:8000/accesscontrol/api/check-card-id'
    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")
        if reader_info.get("mode") == "LATCH":
            ser.write("unlock".encode('utf-8'))
        elif reader_info.get("mode") == "TOGGLE":
            ser.write("toggle".encode('utf-8'))
    else:
        print(f"Failed to send FULL_CARD_ID via HTTP POST. Status code: {response.status_code}")
        print(response.text)

def fetch_initial_reader_config():
    global reader_info
    try:
        url = 'http://localhost:8000/accesscontrol/api/readerinfo'
        response = requests.get(url, headers=headers)

        if response.status_code == 200:
            data = response.json()
            reader_info["enabled"] = data.get("enabled", False)
            reader_info["mode"] = data.get("mode")
    except Exception as e:
        print(f"Error fetching initial reader config: {e}")

def reader_config_check():
    global reader_info
    while True:
        print("Checking HTTP endpoint for new reader config...")
        url = 'http://localhost:8000/accesscontrol/api/readerinfo'
        response = requests.get(url, headers=headers)

        if response.status_code == 200:
            data = response.json()
            new_enabled = data.get("enabled", False)
            new_mode = data.get("mode")

            if new_enabled != reader_info["enabled"]:
                print("New reader enabled status: ", new_enabled)
                if new_enabled:
                    print("Enabling reader...")
                    ser.write("enable".encode('utf-8'))
                else:
                    print("Disabling reader...")
                    ser.write("disable".encode('utf-8'))
                reader_info["enabled"] = new_enabled

            if new_mode != reader_info["mode"]:
                print("Reader mode has changed: ", new_mode)
                if new_mode == "LATCH" and reader_info["mode"] == "TOGGLE":
                    print("Transitioned from TOGGLE to LATCH, sending 'lock' over serial...")
                    ser.write("lock".encode("utf-8"))
                reader_info["mode"] = new_mode

                if new_mode == "LATCH" and reader_info["mode"] == "TOGGLE":
                    print("Transitioned from LATCH to TOGGLE, sending 'unlock' over serial...")
        time.sleep(15)  # Check every 15 seconds

def serial_router():
    while True:
        data = ser.readline()
        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": check_valid_card,
                    "DOOR_DISABLED_STATUS": handle_log_msg,
                    "DOOR_LOCK_STATUS": handle_log_msg
                }

                if prefix in actions:
                    actions[prefix](prefix, value)
                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}")

if __name__ == "__main__":
    headers = {
        "Content-Type": "application/json",
        "X-Identifier": tinance2_reader_identifier,
        "X-Access-Key": tinance2_reader_accesskey
    }

    # Fetch the initial reader configuration
    fetch_initial_reader_config()

    # Create and start the two threads
    serial_router_thread = threading.Thread(target=serial_router)
    reader_config_check_thread = threading.Thread(target=reader_config_check)

    serial_router_thread.start()
    reader_config_check_thread.start()

    # Wait for both threads to finish (which is never in this case since they run indefinitely)
    serial_router_thread.join()
    reader_config_check_thread.join()