import serial.tools.list_ports import time import configparser import logging import socket import lib.lomka as lomka #consts&confs config = configparser.ConfigParser() #bc_calculator = lomka.BCCCalculator() some_open_ports = serial.tools.list_ports.comports() #load config file config.read('config.ini') # device_hwid = config['DEFAULT']['device_hwid'] # baudrate = config['DEFAULT']['baudrate'] #log section logging.basicConfig( level=logging.INFO if config['DEFAULT']['debug'] == "false" else logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('lastest.log'), logging.StreamHandler()]) logger = logging.getLogger(socket.gethostname()) class caser: #добавить по умолчанию с конфигом def __init__(self, device_hwid, config, baudrate=9600, bc_calculator=lomka.BCCCalculator, config_name="config.ini"): ''' Создает объект контроллера, к которому прикрепленна информация о существующем локере. device_hwid - pid USB конвертера(необязательно USB) пример: "PID=1A86:7523" baudrate - скорость соединения bc_calculator - метод вычисления хеш-суммы (должен возвращать один байт в int) ''' try: config.read(config_name) self.device_hwid = config['DEFAULT']['device_hwid'] self.baudrate = config['DEFAULT']['baudrate'] except: self.device_hwid = device_hwid self.baudrate = baudrate self.bc_calculator = bc_calculator # xor_bcc method(returns int) self.comport: str # только адрес self.active_comport = serial.tools.list_ports.comports()[0] # объект порта (dummy) self.controller_number: bytes = 1 def find_comport(self, some_open_ports : list): '''Находит и возвращает корректный comport (объект)''' for port in some_open_ports: logger.debug(f'{port.location}, {port.description}, {port.hwid}') if self.device_hwid in port.hwid: logger.info(f"Found the correct port: {port.device}") return str(port.device) def set_port_device(self, port_device=None): '''Устанавливает com port''' if port_device == None: self.comport = self.find_comport(some_open_ports) else: self.comport = port_device def set_controller_number(self, controller_number : int): '''Определяет номер контроллера. Значения хранятся в виде bytes''' self.controller_number = hex(controller_number) def send_command(self, cell_number : int, command : int, header=138): active_comport = serial.Serial(self.comport, baudrate=self.baudrate) logger.info(f"Connected to {active_comport.name }") # 1 - header(8Ahex) (138dec), 2 - controller number(01), 3 - cell number(04), 4 - command(11), 5 - checksum(9E) # *4 - (11) - open, (00) - close data = [int(header).to_bytes(1, 'little'), int(self.controller_number, 16).to_bytes(1, 'little'), cell_number.to_bytes(1, 'little'), command.to_bytes(1, 'little')] data_string = b'' for i in data: data_string += i checksum = self.bc_calculator.xor_bcc(data_string) data_string += checksum.to_bytes(1, 'little') active_comport.write(data_string) return active_comport.read(5) def read_controller(self, cell_number : int, command=51, header=128): active_comport = serial.Serial(self.comport, baudrate=self.baudrate) logger.info(f"Connected to {active_comport.name }") data = [header.to_bytes(1, 'little'), int(self.controller_number, 16).to_bytes(1, 'little'), cell_number.to_bytes(1, 'little'), command.to_bytes(1, 'little')] data_string = b'' for i in data: data_string += i checksum = self.bc_calculator.xor_bcc(data_string) data_string += checksum.to_bytes(1, 'little') active_comport.write(data_string) return active_comport.read(5) zxc = caser(device_hwid="PID=1A86:7523", baudrate=9600, config=config) zxc.set_controller_number(1) zxc.set_port_device() # добавь *aargs print(zxc.read_controller(1)) print(zxc.send_command(1, 17)) print(zxc.read_controller(1))