import serial.tools.list_ports import logging import socket import lib.lomka as lomka import serial_asyncio class caser: #добавить по умолчанию с конфигом def __init__(self, device_hwid, config, logger=logging.getLogger(socket.gethostname()), baudrate=9600, bc_calculator=lomka.BCCCalculator, config_name="config.ini"): ''' Создает объект контроллера, к которому прикрепленна информация о существующем локере. connection - объект активного соединения с контроллером device_hwid - pid USB конвертера(необязательно USB) пример: "PID=1A86:7523" logger - logging.getLogger() object baudrate - скорость соединения bc_calculator - метод вычисления хеш-суммы (должен возвращать один байт в int) config_name - имя до файла конфигурации ''' try: config.read(config_name) self.device_hwid = config['DEFAULT']['device_hwid'] self.baudrate = config['DEFAULT']['baudrate'] self.controller_capacity = int(config['DEFAULT']['controller_capacity']) # узнать какое исключение вызывается конкретно, ибо нужно проверять каждое по отдельности except: self.device_hwid = device_hwid self.baudrate = baudrate self.controller_capacity: int self.logger = logger 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 #self.connection : serial.Serial self.async_connection : tuple self.instance_error_count : int = 0 # Определение базовых констант/команд self.default_consts = {True: 17, False: 0, "INFO": 51, "READ_HEADER": 128, "DEFAULT_MESSAGE_LENGTH": 5, "MANIPULATE_HEADER": 138, "RESPONSE_LENGHT": 5} def find_comport(self, some_open_ports : list): '''Находит и возвращает корректный comport (объект)''' for port in some_open_ports: self.logger.debug(f'{port.location}, {port.description}, {port.hwid}') if self.device_hwid in port.hwid: self.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 is None: self.comport = self.find_comport(serial.tools.list_ports.comports()) else: self.comport = port_device def set_controller_number(self, controller_number : int): '''Определяет номер контроллера. Значения хранятся в виде bytes''' self.controller_number = hex(controller_number) async def send_command(self, cell_number : int, command: int, bytes_count=5, header=138): try: # active_comport = serial.Serial(self.comport, baudrate=self.baudrate, timeout=2) reader, writer = await serial_asyncio.open_serial_connection( url=self.comport, baudrate=self.baudrate, timeout=2) self.logger.info(f"Connected to {self.comport}") # 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') writer.write(data_string) return await reader.read(bytes_count) except Exception as error: raise error async def read_controller(self, cell_number : int, command=51, header=128, bytes_count=5): try: # active_comport = serial.Serial(self.comport, baudrate=self.baudrate, timeout=2) reader, writer = await serial_asyncio.open_serial_connection( url=self.comport, baudrate=self.baudrate ) self.logger.info(f"Connected to {self.comport}") 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') writer.write(data_string) return await reader.read(bytes_count) except Exception as error: raise error async def open_cell(self, cell_number): '''открывает замок по номеру''' try: result = [] answer = await self.send_command(cell_number=cell_number, command=self.default_consts[True], bytes_count=self.default_consts["RESPONSE_LENGHT"], header=self.default_consts["MANIPULATE_HEADER"]) result.append({"CELL_NUMBER": cell_number, "CURRENT_STATUS": True if answer[-2] == 17 else False if answer[-2] == 0 else None, "FULL_ANSWER": answer}) return result except Exception as error: self.logger.error(f"{error}") self.instance_error_count += 1 return str(error) async def READ_ALL_CELLS(self): '''и возвращает состояние всех замков''' try: result = [] for i in range(1, self.controller_capacity + 1): answer = await self.send_command(cell_number=i, command=self.default_consts["INFO"], bytes_count=self.default_consts["RESPONSE_LENGHT"], header=self.default_consts["READ_HEADER"]) result.append({"CELL_NUMBER": i, "CURRENT_STATUS": True if answer[-2] == 17 else False if answer[-2] == 0 else None, "FULL_ANSWER": answer}) return result except Exception as error: self.logger.error(f"{error}") self.instance_error_count += 1 return str(error) async def set_controller_capacity_unsafe(self, capacity: int = 0): '''Устанавливает количество доступных замков''' if capacity >= 0: self.controller_capacity = capacity # else: # # target max capatity # self.controller_capacity = 128 # result = await self.READ_ALL_CELLS() # if isinstance(result, list): # self.controller_capacity = len(result) # self.logger.info(f"Controller capacity set to {self.controller_capacity}") # else: # self.logger.error("READ_ALL_CELLS() failed") # return False async def OPEN_ALL_CELLS_LEGACY(self, confirm : bool): '''Открывает все замки вручную и возвращает состояние всех замков Confirm - подтверждение операции''' if confirm: try: result = [] for i in range(1, self.controller_capacity + 1): answer = await self.send_command(cell_number=i, command=self.default_consts[True], bytes_count=self.default_consts["RESPONSE_LENGHT"], header=self.default_consts["MANIPULATE_HEADER"]) result.append({"CELL_NUMBER": i, "CURRENT_STATUS": True if answer[-2] == 17 else False if answer[-2] == 0 else None, "FULL_ANSWER": answer}) return result except Exception as error: self.logger.error(f"{error}") self.instance_error_count += 1 return str(error) else: return 1 async def set_controller_capacity(self, capacity=0) -> str: '''Устанавливает количество доступных замков''' if capacity != 0: self.controller_capacity = capacity return self.controller_capacity else: # тут должно быть автоматическое определение кол-ва замков на контроллере try: # написать count = await self.send_command(0, self.default_consts[True], self.default_consts["MANIPULATE_HEADER"], bytes_count=self.controller_capacity) except Exception as error: return str(error) async def watchdog(self): '''Слушает контролер в реальном времени и возвращает bytearray с событием''' try: reader, writer = await serial_asyncio.open_serial_connection( url=self.comport, baudrate=self.baudrate ) try: while True: data = await reader.read(self.default_consts["DEFAULT_MESSAGE_LENGTH"]) if data: self.logger.debug(f"Received: {data}") return data except Exception as error: self.logger.error(f"{error}") self.instance_error_count = self.instance_error_count + 1 writer.close() await writer.wait_closed() except Exception as error: self.logger.critical(f"{error}") self.instance_error_count = self.instance_error_count + 1