"""Plays the controller from the Mac, over the RS-485 adapter. It polls the reader, decides on every badge and PIN, and tells the reader what to light, beep and show. python3 controller.py [port] """ import os import sys import termios import time import osdp PORT = sys.argv[1] if len(sys.argv) > 1 else "/dev/cu.usbmodem5ACC0227761" READER = 1 REPLY_TIMEOUT_S = 0.2 POLL_INTERVAL_S = 0.2 PIN_CODE = "1234" KNOWN_BADGES = ["6B AD 30 8E"] SHOWN_DS = 20 IDLE_TEXT = "Badge or PIN" def open_port(path): port = os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) settings = termios.tcgetattr(port) settings[0] = 0 settings[1] = 0 settings[2] = termios.CS8 | termios.CREAD | termios.CLOCAL settings[3] = 0 settings[4] = termios.B9600 settings[5] = termios.B9600 termios.tcsetattr(port, termios.TCSANOW, settings) termios.tcflush(port, termios.TCIOFLUSH) return port def read_available(port): try: return os.read(port, 256) except BlockingIOError: time.sleep(0.005) return b"" def ask(port, sequence, code, data=b"", address=READER): """Sends one command and returns it with the reader's reply, or None after the timeout. A late reply to an earlier command is dropped, as the real controller does. """ command = osdp.build(address, sequence, code, data) termios.tcflush(port, termios.TCIFLUSH) os.write(port, command) buffer = b"" answer = None deadline = time.monotonic() + REPLY_TIMEOUT_S while answer is None and time.monotonic() < deadline: frame, buffer = osdp.parse(buffer + read_available(port)) if frame is not None and frame.address == address | osdp.REPLY and frame.sequence == sequence: answer = frame return command, answer NAMES = { osdp.POLL: "POLL", osdp.ID: "ID", osdp.CAP: "CAP", osdp.ISTAT: "ISTAT", osdp.OSTAT: "OSTAT", osdp.OUT: "OUT", osdp.ISTATR: "ISTATR", osdp.OSTATR: "OSTATR", osdp.LED: "LED", osdp.BUZ: "BUZ", osdp.TEXT: "TEXT", osdp.ACK: "ACK", osdp.NAK: "NAK", osdp.PDID: "PDID", osdp.PDCAP: "PDCAP", osdp.RAW: "RAW", osdp.KEYPAD: "KEYPAD", } CAPABILITY_NAMES = { 1: "contact status monitoring", 2: "output control", 3: "card data format", 4: "reader LED control", 5: "reader buzzer", 6: "reader text output", 7: "time keeping", 8: "check character support", 9: "communication security", 10: "receive buffer size", 11: "largest combined message", 12: "smart card support", 13: "readers", 14: "biometrics", } def describe_identity(data): return "vendor {}, model {}, version {}, serial {}, firmware {}.{}.{}".format( data[0:3].hex(" "), data[3], data[4], data[5:9][::-1].hex(), data[9], data[10], data[11] ) def describe_capabilities(data): records = [data[start : start + 3] for start in range(0, len(data) - 2, 3)] return "".join( "\n {}: compliance {}, count {}".format(CAPABILITY_NAMES.get(record[0], record[0]), record[1], record[2]) for record in records ) def describe_badge(data): bits = data[2] | (data[3] << 8) return "reader {}, format {}, {} bits: {}".format(data[0], data[1], bits, data[4:].hex(" ").upper()) def describe_keys(data): return "reader {}, keys {}".format(data[0], data[2 : 2 + data[1]].decode()) def describe_states(data): return " ".join("#{} {}".format(number, "on" if state else "off") for number, state in enumerate(data)) def describe(answer): if answer is None: return "no reply" details = { osdp.PDID: describe_identity, osdp.PDCAP: describe_capabilities, osdp.RAW: describe_badge, osdp.KEYPAD: describe_keys, osdp.ISTATR: describe_states, osdp.OSTATR: describe_states, }.get( answer.code, lambda data: data.hex(" ") )(answer.data) return "{}, sequence {}{}".format(NAMES.get(answer.code, hex(answer.code)), answer.sequence, details and ": " + details) def sequence_numbers(): """0 first, then 1, 2, 3 and round again. 0 tells the reader the controller just started.""" number = 0 while True: yield number number = number % 3 + 1 def talk(port, sequence, code, data=b"", address=READER, label=""): """Asks, and prints the exchange unless it is a POLL answered by a plain ACK.""" command, answer = ask(port, sequence, code, data, address) quiet = code == osdp.POLL and answer is not None and answer.code == osdp.ACK if not quiet: print(label + "->", NAMES.get(code, hex(code)), command.hex(" ")) print(label + "<-", describe(answer)) return answer def decide(granted): """The controller's decision, as the three commands that show it at the reader.""" color, beeps, message = (osdp.GREEN, (1, 1, 1), "Access granted") if granted else (osdp.RED, (2, 2, 3), "Access denied") print(" decision:", message) return [ (osdp.LED, osdp.led_temporary(color, SHOWN_DS)), (osdp.BUZ, osdp.buzzer(*beeps)), (osdp.TEXT, osdp.text(message)), ] def on_keys(keys, typed): """Returns the PIN typed so far and the commands these keys call for.""" commands = [] for key in keys: if key == "#": commands, typed = decide(typed == PIN_CODE), "" elif key == "*": commands, typed = [(osdp.TEXT, osdp.text(IDLE_TEXT))], "" else: typed += key commands = [(osdp.TEXT, osdp.text("PIN: " + "*" * len(typed)))] return typed, commands def on_answer(answer, typed): if answer is not None and answer.code == osdp.KEYPAD: return on_keys(answer.data[2 : 2 + answer.data[1]].decode(), typed) if answer is not None and answer.code == osdp.RAW: return "", decide(answer.data[4:].hex(" ").upper() in KNOWN_BADGES) return typed, [] def main(): port = open_port(PORT) sequence = sequence_numbers() talk(port, next(sequence), osdp.ID, b"\x00") talk(port, next(sequence), osdp.CAP, b"\x00") typed = "" outbox = [(osdp.TEXT, osdp.text(IDLE_TEXT))] idle_at = None while True: code, data = outbox.pop(0) if outbox else (osdp.POLL, b"") typed, commands = on_answer(talk(port, next(sequence), code, data), typed) outbox += commands if any(command == osdp.LED for command, _ in commands): idle_at = time.monotonic() + SHOWN_DS / 10 if idle_at is not None and time.monotonic() >= idle_at: outbox.append((osdp.TEXT, osdp.text(IDLE_TEXT))) idle_at = None time.sleep(0 if outbox else POLL_INTERVAL_S) if __name__ == "__main__": main()