opencaser/main.py

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_controller_capacity(self, capacity: int):
        '''Устанавливает количество доступных замков'''



    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["MANIPULATE_HEADER"],
                                bytes_count=self.controller_capacity)
            except Exception as error:
                return str(error)
        else:
            return 1

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))