amtenc.py

#!/usr/bin/env python3

from serial import Serial
from struct import unpack

class AMT_12bit:
    """
    Reads the position from a 12-bit single-turn AMT series absolute encoder

    This was developed and tested using an AMT213A-V, which is a fixed 2Mbaud device
    that communicates over RS-485. It should work with other 12-bit single-turn AMT
    series absolute encoders, assuming they are accessible via a serial port/device.
    """

    # Constants
    _DEFAULT_PORT         = '/dev/ttyUSB0'  # Typical for FTDI USB-to-RS485 adapter cable
    _DEFAULT_BAUDRATE     = 2000000         # 2Mbaud
    _DEFAULT_TIMEOUT      = 0.1             # 100ms
    _DEFAULT_NODE_ADDRESS = 0x54            # 0x54 is the default for these encoders
    _DEFAULT_REVERSE      = False           # Normal rotation direction.
    _DEFAULT_CONNECT      = True            # Connect serial port when initializing instance

    # Private
    _port = None
    _baud = None
    _timeout = None
    _nodeaddress = None
    _reverse = _DEFAULT_REVERSE
    _serial = None

    def connect(self):
        """
        Attempts to connect the serial port.
        """
        self._serial = Serial(port=self._port,
                              baudrate=self._baud, # type: ignore
                              timeout=self._timeout)
        self._serial.read_all()  # Empty the read buffer

    def disconnect(self):
        """
        Disconnects the serial port.
        """
        if self._serial and self._serial.is_open:  # Verify that we're open so we can close
            self._serial.close()
        self._serial = None

    def __init__(self, port: str = _DEFAULT_PORT,
                       baudrate: int = _DEFAULT_BAUDRATE,
                       timeout: float = _DEFAULT_TIMEOUT,
                       nodeaddress: int = _DEFAULT_NODE_ADDRESS,
                       reverse: bool = _DEFAULT_REVERSE,
                       connect: bool = _DEFAULT_CONNECT):
        """
        Initializes the object.

        :param port: Serial port (ex: '/dev/ttyUSB0')
        :param baudrate: Baud rate (ex: 2Mbaud)
        :param timeout: Read timeout (in seconds) (ex: 1ms)
        :param nodeaddress: Encoder's node address (ex: 0x54)
        :param reverse: Set to True so read() returns as if the encoder was rotating the opposite direction
        :param connect: Set to False to not connect when initializing
        """
        self._port = port
        self._baud = baudrate
        self._timeout = timeout
        self._reverse = reverse

        if ((nodeaddress & 0xFF) >> 2 << 2) == nodeaddress:  # Check that this is a valid node address
            self._nodeaddress = nodeaddress
        else:
            raise Exception("Node address is not valid, must be evenly divisible by 4")

        if connect:
            self.connect()

    def __del__(self):
        """
        Cleans up the object.
        """
        self.disconnect()

    def _checksum(self, value: int) -> int:
        """
        Calculates the checksum/ECC bits. Simplified from the manufacturer's
        documentation, masking into 2-bit chunks to handle even and odd bits
        simultaneously. Once all the chunks are XOR'd, you invert the bits
        and mask to 2 bits to get the checksum/ECC bits. Since RS-485 is quite
        robust and resistant to noise, I have yet to see a single mismatch.

        This will work for 14-bit encoders without modification.

        :param value: The unpacked response with the bytes in the correct order
        :return: A byte with both calculated checksum bits, in order.
        """
        return ~(((value >> 12) & 3) ^  # Calculates both even and odd checksums simultaneously
                 ((value >> 10) & 3) ^
                 ((value >>  8) & 3) ^
                 ((value >>  6) & 3) ^
                 ((value >>  4) & 3) ^
                 ((value >>  2) & 3) ^
                 ( value        & 3)) & 3

    def _parse_response(self, response: bytes) -> tuple[int, int, int]:
        """
        Parses the 2-byte response from the encoder to extract the position
        and verifies the checksum/status bits.

        :param response: The 2-byte response received from the encoder.
        :return: A tuple containing (int: data, int: position, int: checksum).
        """
        data = unpack('<H', response)[0] & 0xFFFF  # Unpack Little-Endian unsigned short and ensure we end up with 16 bits
        position = (data >> 2) & 0xFFF  # Shift to the right to drop the noise and mask to 12 bits
        checksum = data >> 14

        return data, position, checksum

    def read(self) -> tuple[int, bool]:
        """
        Reads the absolute position from the encoder

        Sends a read command, which is just the node address.
        The encoder responds with a 2-byte word [L, H, K0, K1]

        :returns: A tuple containing the 12-bit absolute position (int) and whether the checksum matched (bool).
        """
        self._serial.write(bytes([self._nodeaddress]))  # type: ignore  # Write the node address to request a position
        response = self._serial.read(2)  # type: ignore  # Read two bytes, subject to timeout
        if len(response) == 2:
            data, pos, checksum = self._parse_response(response)
            return (pos if not self._reverse else ~pos & 0xFFF), (checksum == self._checksum(data))
        return 0, False


if __name__ == '__main__':
    """
    Example to show functionality.
    """
    enc = AMT_12bit()
    old_val = None
    while (True):
        val, valid = enc.read()
        if valid:
            if val != old_val: # type: ignore
                print(f"Encoder returned: {val:012b} 0x{val:04x} {val:04d}")
                old_val = val
        else:
            print("Bad checksum")