yourovo/mac/chafon.py

#!/usr/bin/env python3
"""
Chafon H-102 UHF reader client (macOS) -- read-only verification path.

ASSESSMENT.md sec.4 found the gun was HID-keyboard-only on Windows and parked it. On
macOS it enumerates a CH340 serial bridge at /dev/cu.usbserial-*, so the documented
serial protocol works fine and it can be used as an INDEPENDENT check that a chip the
Urovo programmed really holds what we think it does.

Protocol (Chafon H100/H102/H103/H104 manual, Appendix B), matching the existing Node
rfid-daemon byte-for-byte:

  frame     CF [ADDR] [CMD_H] [CMD_L] [LEN] [DATA...] [CRC_H] [CRC_L]
  response  CF [ADDR] [CMD_H] [CMD_L] [LEN] [STATUS] [DATA...] [CRC_H] [CRC_L]
            LEN counts STATUS + DATA. CRC covers CF through the last data byte.
  crc16     poly 0x8408, init 0xFFFF, LSB-first (CRC-16/IBM)
  115200 8N1

Deliberately read-only: inventory and read. Writing is left to the Urovo (and to the
Node daemon, which already owns the write path).
"""
import glob
import time

import serial

BAUD = 115200
ADDR = 0x00
CMD_INVENTORY = (0x00, 0x01)
CMD_STOP = (0x00, 0x02)
CMD_READ = (0x00, 0x03)
MEM_EPC = 0x01
DEFAULT_ACCESS_PWD = b'\x00\x00\x00\x00'

STATUS = {0x00: 'Tag found', 0x01: 'Parameter error', 0x12: 'Inventory complete',
          0x13: 'No tags found', 0x14: 'Tag timeout'}


class ChafonError(RuntimeError):
    pass


def crc16(buf):
    crc = 0xFFFF
    for byte in buf:
        crc ^= byte
        for _ in range(8):
            crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1
    return crc


def frame(cmd, data=b''):
    body = bytes([0xCF, ADDR, cmd[0], cmd[1], len(data)]) + data
    c = crc16(body)
    return body + bytes([(c >> 8) & 0xFF, c & 0xFF])


def find_port():
    ports = sorted(glob.glob('/dev/cu.usbserial*'))
    if not ports:
        raise ChafonError('no CH340 serial port found (/dev/cu.usbserial*) -- '
                          'is the Chafon plugged in and awake?')
    return ports[0]


class Chafon(object):
    def __init__(self, port=None, timeout=0.3):
        self.port = port or find_port()
        self.ser = serial.Serial(self.port, BAUD, timeout=timeout)

    def close(self):
        try:
            self.ser.close()
        except Exception:
            pass

    def __enter__(self):
        return self

    def __exit__(self, *a):
        self.close()

    def command(self, cmd, data=b'', timeout=6.0):
        """Send a frame and return the first CRC-valid response matching cmd.

        Frames with a bad CRC are resynced past rather than trusted -- a corrupt LEN
        or a 0xCF inside a payload can otherwise fabricate a false frame boundary and
        swallow the real response queued behind it.
        """
        self.ser.reset_input_buffer()
        self.ser.write(frame(cmd, data))
        self.ser.flush()
        buf, end = bytearray(), time.time() + timeout
        while time.time() < end:
            chunk = self.ser.read(256)
            if chunk:
                buf += chunk
            while len(buf) >= 7:
                start = buf.find(0xCF)
                if start < 0:
                    buf.clear()
                    break
                if start:
                    del buf[:start]
                if len(buf) < 7:
                    break
                total = 5 + buf[4] + 2
                if len(buf) < total:
                    break
                f = bytes(buf[:total])
                if ((f[-2] << 8) | f[-1]) != crc16(f[:-2]):
                    del buf[0]
                    continue
                del buf[:total]
                if (f[2], f[3]) != cmd:
                    continue                       # streaming frame for another command
                return {'status': f[5], 'data': f[6:-2]}
        raise ChafonError('no response from reader within %.1fs -- pull the trigger to '
                          'wake it, or it may be in HID mode' % timeout)

    def stop(self):
        try:
            self.command(CMD_STOP, timeout=1.0)
        except ChafonError:
            pass

    def inventory(self, seconds=1):
        """Scan the field. Returns {status, statusText, epc, rssi, antenna}."""
        r = self.command(CMD_INVENTORY, bytes([0x00, 0, 0, 0, seconds & 0xFF]), timeout=6.0)
        self.stop()
        d = r['data']
        found = r['status'] == 0x00
        # inventory payload: RSSI(1) ANT(1) PC(2) EPC_LEN(1) EPC(12)
        epc = d[5:17] if found and len(d) >= 17 else None
        return {'status': r['status'],
                'statusText': STATUS.get(r['status'], '0x%02X' % r['status']),
                'found': found,
                'epc': epc.hex().upper() if epc else None,
                'rssi': (d[0] - 256 if d[0] >= 128 else d[0]) if found and len(d) >= 2 else None,
                'antenna': d[1] if found and len(d) >= 2 else None,
                'rawHex': d.hex().upper()}

    def read_epc(self, words=6):
        """Read the EPC bank directly (6 words = 12 bytes)."""
        data = b'\x00' + DEFAULT_ACCESS_PWD + bytes([MEM_EPC, 0x00, 0x02, words])
        r = self.command(CMD_READ, data, timeout=12.0)
        self.stop()
        if r['status'] != 0x00:
            return {'ok': False, 'status': r['status'],
                    'statusText': STATUS.get(r['status'], '0x%02X' % r['status'])}
        d = r['data']
        return {'ok': True, 'epc': d[-words * 2:].hex().upper(),
                'rssi': d[0] - 256 if d[0] >= 128 else d[0],
                'antenna': d[1], 'rawHex': d.hex().upper()}


if __name__ == '__main__':
    with Chafon() as c:
        print('port     :', c.port)
        inv = c.inventory()
        print('inventory:', inv)