118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
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, header : int, cell_number : int, command : int):
|
||
try:
|
||
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 = [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')]
|
||
|
||
#проблемы тут
|
||
checksum = self.bc_calculator.xor_bcc(data)
|
||
exit()
|
||
packet = data + [checksum]
|
||
active_comport.write(b''.join(packet))
|
||
# time.sleep(0.2)
|
||
|
||
# добавь таймаут на чтение ответа от устройства
|
||
line = active_comport.read(10)
|
||
|
||
if line:
|
||
print("Received:", line)
|
||
|
||
except ValueError as ve:
|
||
logger.error(f"Invalid port or baudrate: {str(ve)}")
|
||
|
||
except serial.SerialException as se:
|
||
logger.error(f"Serial port error: {str(se)}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Unexpected error: {str(e)}")
|
||
|
||
zxc = caser(device_hwid="PID=1A86:7523", baudrate=9600,
|
||
config=config)
|
||
zxc.set_controller_number(1)
|
||
zxc.set_port_device()
|
||
# у тебя беды с хексами и типами данных
|
||
zxc.send_command(138, 1, 17)
|
||
exit() |