EL-UHF-RMT01 Menggunakan Python §

HEADER: bytes = b'\xBB'
END: bytes = b'\x7E'
CMD_INVENTORY_SINGLE: bytes = b'\x22'
CMD_INVENTORY_MULTI_START: bytes = b'\x27'
CMD_INVENTORY_MULTI_STOP: bytes = b'\x28'
CMD_SET_BAUD_RATE: bytes = b'\x11'
CMD_GET_POWER: bytes = b'\xB7'
CMD_SET_POWER: bytes = b'\xB6'
CMD_SAVE_CONFIG: bytes = b'\x09'
 
 
class Command:
    def __init__(self, command: bytes, data: bytes | int | None = None):
        self.command: bytes = command
        self.data: bytes | int | None = data
        if isinstance(data, int):
            self.data = bytearray([data])
        if data is None:
            self.data = bytearray()
 
    def serialize(self) -> bytes:
        # Calculate CRC
        crc_sum: int = sum((bytearray([0x00]) + self.command
                            + len(self.data).to_bytes(2, byteorder="big") + self.data))
        crc: bytes = (crc_sum & 0xFF).to_bytes(1, byteorder="big")
        serialize: bytes = (HEADER + b'\x00' + self.command
                            + len(self.data).to_bytes(2, byteorder="big") + self.data + crc + END)
        return serialize
 

class Response:
    def __init__(self, response_bytes: bytes):
        self.response_bytes: bytes = response_bytes
        self.header: bytes = int.to_bytes(response_bytes[0], byteorder="big")
        self.frame_type: bytes = int.to_bytes(response_bytes[1], byteorder="big")
        self.command: bytes = int.to_bytes(response_bytes[2], byteorder="big")
        self.data_length: int = int.from_bytes(response_bytes[3:5], "big")
        self.data: bytes = response_bytes[5:-2]
        self.checksum: bytes = int.to_bytes(response_bytes[-2], byteorder="big")
        self.end: bytes = int.to_bytes(response_bytes[-1], byteorder="big")
 
    def __str__(self) -> str:
        return_value = ''
        value = '>>> START RESPONSE ================================'
        return_value = f'{return_value}\n{value}'
        value = f'RESPONSE       >> {hex_readable(self.response_bytes)}'  # Response
        return_value = f'{return_value}\n{value}'
        value = f'HEADER         >> {hex_readable(self.header)}'  # Header
        return_value = f'{return_value}\n{value}'
        value = f'TYPE           >> {hex_readable(self.frame_type)}'  # Type
        return_value = f'{return_value}\n{value}'
        value = f'COMMAND        >> {hex_readable(self.command)}'  # Command
        return_value = f'{return_value}\n{value}'
        if self.data:
            value = f'DATA           >> {hex_readable(self.data)}'  # Data
            return_value = f'{return_value}\n{value}'
        value = f'CHECKSUM (CRC) >> {hex_readable(self.checksum)}'  # Checksum (CRC)
        return_value = f'{return_value}\n{value}'
        value = f'END            >> {hex_readable(self.end)}'  # End
        return_value = f'{return_value}\n{value}'
        value = '>>> END RESPONSE   ================================'
        return_value = f'{return_value}\n{value}'
        return return_value.strip()
 
 
def hex_readable(data: bytes | int, bytes_separator: str = " ") -> str:
    if isinstance(data, int):
        return "{:02X}".format(data)
    return bytes_separator.join("{:02X}".format(x) for x in data)

from typing import Iterator
from command import *
from response import *
import serial
 
 
class Reader:
    def __init__(self, serial_port: str, baud_rate: int) -> None:
        self.serial_port: str = serial_port
        self.baud_rate: int = baud_rate
        self.serial = serial.Serial(serial_port, baud_rate,
                                    timeout=0.5, write_timeout=0.5)
 
    def close(self) -> None:
        self.serial.close()
 
    def __send_request(self, command: Command) -> None:
        self.serial.write(command.serialize())
 
    def __get_response(self) -> bytes:
        header: bytes = self.serial.read(1)
        assert header == HEADER  # Must equal to default header (0xBB)
        header_frame: bytes = self.serial.read(4)
        data_length: int = int.from_bytes(header_frame[-2:], "big")
        data: bytes = self.serial.read(data_length)
        crc_end: bytes = self.serial.read(2)
        end: bytes = int.to_bytes(crc_end[-1], byteorder="big")
        assert end == END
        complete_frame: bytes = header + header_frame + data + crc_end
        return complete_frame
 
    def inventory_single(self) -> bytes | None:
        """
        0x22 Single Inventory
        :return: Data or none (no tag)
        """
        command: Command = Command(CMD_INVENTORY_SINGLE)
        self.__send_request(command)
        response: Response = Response(self.__get_response())
        if response.frame_type == b'\x01' and response.data == b'\x15':
            return
        assert response.frame_type == b'\x02'  # Frame type ➜ 0x01: Response from the Interrogator to the Host Computer
        return response.data
 
    def inventory_multiple_start(self, count: int) -> Iterator[bytes]:
        """
        0x27 Multiple Inventory
        :param count: 0~65535
        :return: yield data
        """
        assert 0 <= count <= 65535, "Value must be between 0 and 65535, inclusive."
        data: bytes = b'\x22' + count.to_bytes(2, byteorder="big")
        command: Command = Command(CMD_INVENTORY_MULTI_START, data=data)
        self.__send_request(command)
        for _ in range(count):
            response: Response = Response(self.__get_response())
            if response.frame_type == b'\x01' and response.data == b'\x15':
                continue
            assert response.frame_type == b'\x02'
            yield response.data
 
    def inventory_multiple_stop(self) -> None:
        """
        0x28 Multiple Inventory
        :return:
        """
        command: Command = Command(CMD_INVENTORY_MULTI_STOP)
        self.__send_request(command)
        self.__get_response()
 
    def save_config(self, violate: bool) -> Response:
        data: int = 0x00 if violate else 0x01
        command: Command = Command(CMD_SAVE_CONFIG, data=data)
        self.__send_request(command)
        return Response(self.__get_response())
 
    def get_power(self) -> int:
        command: Command = Command(CMD_GET_POWER)
        self.__send_request(command)
        response: Response = Response(self.__get_response())
        return int(int.from_bytes(response.data, byteorder='big') / 100)
 
    def set_power(self, power: int) -> Response:
        power:int = power * 100
        data: bytes = power.to_bytes(2, byteorder="big")
        command: Command = Command(CMD_SET_POWER, data=data)
        self.__send_request(command)
        return Response(self.__get_response())
 
    def set_baud_rate(self, baud_rate: int) -> None:
        baud_rate: int = int(baud_rate / 100)
        data: bytes = baud_rate.to_bytes(2, byteorder="big")
        command: Command = Command(CMD_SET_BAUD_RATE, data=data)
        self.__send_request(command)

from typing import Iterator
 
from response import hex_readable, Response
from reader import Reader
 
# Windows: Replace '/dev/ttyUSB0' to 'COM1' (check Device Manager)
reader = Reader('/dev/ttyUSB0', 115200)
 
# # 1. Inventory - Single
# tag: bytes | None = reader.inventory_single()
# if isinstance(tag, bytes):
#     print(hex_readable(tag))
 
# # 2. Inventory - Multi
count: int = 100
try:
    tags: Iterator[bytes] = reader.inventory_multiple_start(count)
    for tag in tags:
        print(hex_readable(tag))
except KeyboardInterrupt:
    reader.inventory_multiple_stop()
 
# # 3. Power
# response: Response = reader.set_power(21) # Must call save_config() for stored config in the module
# power: int = reader.get_power()
# print(f"Power: {power}")
 
# # 4. Set baud rate
# NEW_BAUD_RATE: int = 115200
# reader.set_baud_rate(NEW_BAUD_RATE)
# reader.close()
# reader = Reader(SERIAL_PORT, NEW_BAUD_RATE)
# response_save_config: Response = reader.save_config(violate=False)
# print(response_save_config)
 
reader.close()