ADP120 / ADP102 Isolated DIO Module

''' Example usage of the ADP120 DIO expansion card '''

import sys
from time import sleep
import functools

from serial import Serial  # python -m pip install pyserial

# Detecting serial port
import serial.tools.list_ports as system_ports


def get_device_port() -> str:
    ''' Scans system to detect device CDC ACM port '''

    all_ports = system_ports.comports()

    for port, _, hwid in sorted(all_ports):
        if "1FC9:0094" in hwid:
            # Replace with "15A2:0300" if using ADP102
            # Fix for windows COM ports above 10
            if 'win' in sys.platform:
                return "\\\\.\\" + port
            else:
                return port

    return None


class PinConfig:
    ''' Pin configuration; used to set or report a pin config

    PARAMETERS:
        state_change:            Report to host when pin state has changed
        counter_overflow:        Report to host when pin counter has overflown
        counter_polarity:
        latch_polarity:
        starting_state:          Initial pin state
        enable:                  Enable/disable pin

    '''

    def __init__(self, state_change=False, counter_overflow=False, counter_polarity=0, latch_polarity=0,
                 starting_state=0, enabled=True):
        self.state_change = state_change
        self.counter_overflow = counter_overflow
        self.counter_polarity = counter_polarity
        self.latch_polarity = latch_polarity
        self.starting_state = starting_state
        self.enabled = enabled

    def bytes(self):
        nbytes = [0x00, 0x00, 0x00, 0x00]

        if self.state_change:
            nbytes[2] |= 0x02

        if self.counter_overflow:
            nbytes[2] |= 0x01

        if self.enabled:
            nbytes[3] |= 0x01

        if self.starting_state:
            nbytes[3] |= 0x04

        if self.latch_polarity:
            nbytes[3] |= 0x40

        if self.counter_polarity:
            nbytes[3] |= 0x80

        return bytes(nbytes)

    @staticmethod
    def from_bytes(nbytes):
        nbytes = list(nbytes)

        config = {
            'state_change': True if nbytes[2] & 0x02 else False,
            'counter_overflow': True if nbytes[2] & 0x01 else False,
            'enabled': True if nbytes[3] & 0x01 else False,
            'starting_state': True if nbytes[3] & 0x04 else False,
            'latch_polarity': True if nbytes[3] & 0x40 else False,
            'counter_polarity': True if nbytes[3] & 0x80 else False,
        }

        return PinConfig(**config)


class ADP120(Serial):
    ''' Subclass serial with ADP120 specific commands '''

    START = b'\x24'
    END = b'\x00\x80\x01'
    COMMANDS = {
        'model': b'\x00\x00\x01',
        'serial': b'\x00\x00\x03',
        'read_state': b'\x01\x01',
        'read_latch': b'\x01\x02',
        'read_count': b'\x01\x03',
        'clear_latch': b'\x01\x04',
        'clear_count': b'\x01\x05',
        'toggle_output': b'\x01\x09',
        'save_config': b'\x00\x06',
    }

    def __init__(self, *args, **kwargs):
        ''' Initialize serial device; set port timeout if missing '''
        if 'timeout' not in kwargs:
            kwargs['timeout'] = 5
        if 'write_timeout' not in kwargs:
            kwargs['write_timeout'] = 0

        super(ADP120, self).__init__(*args, **kwargs)

    def command(self, cmd: bytes, adr=None) -> bytes:
        ''' Send and ADP120.COMMAND to the hardware device '''
        if adr is None:
            return self.write_command_raw(cmd)
        else:
            return self.write_command_raw(bytes([adr]) + cmd)

    def read_response(self) -> bytes:
        ''' Read the response to a command '''
        r = self.read(1)
        if r == self.START:
            rlen = self.read(1)
            return self.read(ord(rlen) - 1)[3:]
        else:
            sleep(0.01)
            return self.read(self.in_waiting)

    def write_command_raw(self, cmd: bytes) -> bytes:
        ''' Write the raw bytes to the serial device '''
        # Raw command
        raw = self.START + bytes([len(cmd) + 1]) + cmd

        # Write the command
        count = self.write(raw)

        # Check whole command was written
        if count != len(raw):
            return None

        # Get the response
        reply = self.read_response()

        return reply

    def write_config(self, address, config):
        ''' Configure a pin '''
        return self.write_command_raw(bytes([address]) + b'\x00\x05' + config.bytes())

    def read_config(self, address):
        config = self.write_command_raw(bytes([address]) + b'\x00\x04')

        return PinConfig.from_bytes(config)

    def write_output(self, address, state):
        return self.write_command_raw(bytes([address]) + b'\x01\x08' + bytes([state]))

    # Support ADP120.model() syntax
    def __getattr__(self, name):
        cmd = self.COMMANDS.get(name)

        if cmd is None:
            raise AttributeError(name)
        else:
            result = functools.partial(self.command, cmd)

        return result


if __name__ == "__main__":
    port_name = get_device_port()

    # Detect the ADP120 module
    if port_name is None:
        print("Failed to detect device!")
        sys.exit(-1)

    adp = ADP120(port_name)

    # Report model and firmware version
    print(f"Model: {adp.model()}")

    # Read input and output states
    for i in range(0, 8):
        print(f"{'Input' if i < 4 else 'Output'} {i if i < 4 else i - 4} State: {adp.read_state(i)}")

    # Read the config of output 0
    cfg = adp.read_config(4)
    print(f"Output 0:\n  Starting state: {cfg.starting_state}\n  Enabled: {cfg.enabled}")

    # Toggle the starting state, and enable the port
    cfg.starting_state = False if cfg.starting_state else True
    cfg.enabled = True
    adp.write_config(4, cfg)

    cfg = adp.read_config(4)
    print(f"Output 0:\n  Starting state: {cfg.starting_state}\n  Enabled: {cfg.enabled}")

    # Write an output and confirm it worked
    adp.write_output(4, 1)
    print(f"Current State: {adp.read_state(4)}")