DoorAdmin/imaginaerraum_door_admin/door_handle.py

import paho.mqtt.client as mqtt
import socket
from pathlib import Path
import logging

class DoorHandle:
    def __init__(self, token_file, mqtt_host, mqtt_port=1883, nfc_socket='/tmp/nfc.sock', logger=None):
        self.state = None
        self.encoder_position = None

        if not Path(token_file).exists():
            raise FileNotFoundError(f"File with door tokens could not be found at {Path(token_file).absolute()}")

        self.token_file = token_file
        self.last_invalid = {}

        self.mqtt_client = mqtt.Client()
        self.mqtt_client.on_connect = self.on_connect
        self.mqtt_client.on_message = self.on_message
        self.mqtt_client.connect_async(host=mqtt_host, port=mqtt_port)
        self.mqtt_client.loop_start()

        if logger:
            self.logger = logger
        else:
            self.logger = logging

        self.nfc_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            self.nfc_sock.connect(nfc_socket)
            self.logger.info(f"Connected to NFC socket at {nfc_socket}.")
        except Exception as e:
            print(f"Could not connect to NFC socket at {nfc_socket}. Exception: {e}")
            self.nfc_sock = None
            #raise

        self.data_fields = ['name', 'organization', 'email', 'valid_thru']

    # The callback for when the client receives a CONNACK response from the server.
    def on_connect(self, client, userdata, flags, rc):
        self.logger.info("Connected to MQTT broker with result code " + str(rc))

        # Subscribing in on_connect() means that if we lose the connection and
        # reconnect then subscriptions will be renewed.
        client.subscribe("#")

    # The callback for when a PUBLISH message is received from the server.
    def on_message(self, client, userdata, msg):
        #print(msg.topic + " " + str(msg.payload))
        if msg.topic == 'door/state/value':
            self.state = msg.payload.decode()
        elif msg.topic == 'door/position/value':
            self.encoder_position = int(msg.payload)
        elif msg.topic == 'door/token/last_invalid':
            timestamp, token = msg.payload.decode().split(";")
            self.last_invalid = {'timestamp': timestamp, 'token': token}

    def get_tokens(self):
        tokens = {}
        with open(self.token_file) as f:
            token_lines = f.readlines()
            for line in token_lines[1:]:
                data = line.split('|')
                inactive = '#' in data[0]
                token = data[0].strip().strip('#')
                name = data[1].strip() if len(data) > 1 else None
                organization = data[2].strip() if len(data) > 2 else None
                email = data[3].strip() if len(data) > 3 else None
                valid_thru = data[4].strip() if len(data) > 4 else None

                tokens[token] = {'name': name,
                                 'organization': organization,
                                 'email': email,
                                 'valid_thru': valid_thru,
                                 'inactive': inactive}
        return tokens

    def store_tokens(self, tokens):
        output = '# token | ' + ' | '.join(self.data_fields) + '\n'

        for t, data in tokens.items():
            output += '#' if data['inactive'] else ''
            output += t
            for key in self.data_fields:
                output += '|'
                output += data[key] if data[key] else ''
            output += '\n'
        # write new tokens to file and trigger reload
        with open(self.token_file, 'w') as f:
            f.write(output)

        if self.nfc_sock is not None:
            self.nfc_sock.send(b'rld\n')

    def open_door(self):
        if self.nfc_sock is not None:
            self.nfc_sock.send(b'open\n')
        else:
            raise Exception("No connection to NFC socket. Cannot close door!")

    def close_door(self):
        if self.nfc_sock is not None:
            self.nfc_sock.send(b'close\n')
        else:
            raise Exception("No connection to NFC socket. Cannot close door!")

    def get_most_recent_token(self):
        # read last invalid token from logfile
        return self.last_invalid