Sync

lib/OCer.py

@@ -0,0 +1,167 @@
+import serial.tools.list_ports
+import time
+import configparser
+import logging
+import socket
+import lib.lomka as lomka
+
+class caser:
+    #добавить по умолчанию с конфигом
+    def __init__(self, device_hwid, config, 
+                 logger=logging.getLogger(socket.gethostname()), baudrate=9600, 
+                 bc_calculator=lomka.BCCCalculator, config_name="config.ini"):
+        '''
+        Создает объект контроллера, к которому прикрепленна информация о существующем
+        локере.
+
+        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 = config['DEFAULT']['controller_capacity']
+
+        # узнать какое исключение вызывается конкретно, ибо нужно проверять каждое по отдельности
+        except:
+            self.device_hwid = device_hwid
+            self.baudrate = baudrate
+            self.controller_capacity: int
+
+        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.default_consts = {True: 17,
+                          False: 0,
+                          "READ_HEADER": 128,
+                          "MANIPULATE_HEADER": 138}
+
+    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, 
+                     bytes_count=5, 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(bytes_count)
+
+    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)
+    
+    def OPEN_ALL_LOCKS(self, confirm : bool):
+        '''и возвращает состояние всех замков'''
+        if confirm:
+            try:
+                return self.send_command(0, self.default_consts[True], 
+                                self.default_consts["READ_HEADER"],
+                                bytes_count=self.controller_capacity)
+            except Exception as error:
+                return str(error)
+        else:
+            return 1
+   
+    def set_controller_capacity(self, capacity=0) -> str:
+        '''Устанавливает количество доступных замков'''
+        if capacity != 0:
+            self.controller_capacity = capacity
+            return self.controller_capacity
+        else:
+            # тут должно быть автоматическое определение кол-ва замков на контроллере
+            # try:
+            #     return 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)
+            pass
+
+    def watchdog(self):
+        pass
+
+if __name__ == "__main__":
+    #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())
+
+    zxc = caser(device_hwid="PID=1A86:7523", baudrate=9600,
+                config=config, logger=logger)
+    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))

lib/lomka.py

@@ -0,0 +1,42 @@
+class BCCCalculator:
+    """
+    Класс для работы с BCC
+    """
+    
+    @staticmethod
+    def xor_bcc(data):
+        """XOR BCC"""
+        if isinstance(data, str):
+            data = data.encode('utf-8')
+        elif isinstance(data, list):
+            data = bytes(data)
+        
+        bcc = 0
+        for byte in data:
+            bcc ^= byte
+        return bcc
+    
+    @staticmethod
+    def sum_bcc(data):
+        """Суммирование BCC"""
+        if isinstance(data, str):
+            data = data.encode('utf-8')
+        elif isinstance(data, list):
+            data = bytes(data)
+        
+        total = sum(data)
+        return total & 0xFF
+    
+    @staticmethod
+    def verify_bcc(data_with_bcc):
+        """
+        Проверка BCC в данных
+        """
+        if len(data_with_bcc) < 2:
+            return False
+        
+        data = data_with_bcc[:-1]
+        expected_bcc = data_with_bcc[-1]
+        calculated_bcc = BCCCalculator.xor_bcc(data)
+        
+        return expected_bcc == calculated_bcc