pySerial Захват долгого ответа
Привет, ребята, я работаю над сценарием, который будет получать данные от хоста с использованием стандарта передачи данных (разработано: Комитет по обработке данных Комитета по обработке линз The Vision Council), через последовательный порт и передавать данные в протокол ModBus для устройство для выполнения своих операций.
Поскольку у меня нет финансового доступа к хост-машине, я пытаюсь разработать дополнительный сценарий для эмуляции хоста. В настоящее время я нахожусь на этапе, когда мне нужно прочитать много информации из последовательного порта, и я получаю только часть данных. Я надеялся получить всю строку, отправленную функцией send_job() в моем скрипте эмулятора хоста.
Ребята, может, кто-нибудь из вас скажет мне, будет ли это хорошим подходом? единственное, что должен делать компьютер, - это получить 2 значения из ответа хоста и назначить их двум регистрам хранения Modbus.
ПРИМЕЧАНИЕ: функция инициализации жестко запрограммирована, потому что она всегда будет одинаковой, и фактические данные ответа не будут иметь значения, за исключением статуса. Кроме того, запрос задания жестко запрограммирован: я передаю только # задание, которое я получаю из регистра хранения Modbus, точная логика того, как хост разрешил это, не имеет значения, мне нужно только отправить отсканированный с устройства номер задания в этом формате.
основной скрипт:
def request_job_modbus(job):
data = F'[06][1c]req=33[0d][0a]job={job}[0d][0a][1e][1d]'.encode('ascii')
writer(data)
def get_job_from_serial():
response = serial_client.read_all()
resp = response.decode()
return resp
# TODO : SEND INIT SEQUENCE ONCE AND VERIFY IF REQUEST status=0
initiation_request()
init_response_status = get_init_status()
print('init method being active')
print(get_init_status())
while True:
# TODO: get job request data
job_serial = get_job_from_serial()
print(job_serial)
скрипт эмуляции хоста:
def send_job():
job_response = '''[06][1c]ans=33[0d]job=30925[0d]status=0;"ok"[0d]do=l[0d]add=;2.50[0d]ar=1[0d]
bcerin=;3.93[0d]bcerup=;-2.97[0d]crib=;64.00[0d]do=l[0d]ellh=;64.00[0d]engmask=;613l[0d]
erdrin=;0.00[0d]erdrup=;10.00[0d]ernrin=;2.00[0d]ernrup=;-8.00[0d]ersgin=;0.00[0d]
ersgup=;4.00[0d]gax=;0.00[0d]gbasex=;-5.30[0d]gcrosx=;-7.96[0d]kprva=;275[0d]kprvm=;0.55[0d]
ldpath=\\uscqx-tcpmain-at\lds\iot\do\800468.sdf[0d]lmatid=;151[0d]lmatname=;f50[0d]
lnam=;vsp_basic_fh15[0d]sgerin=;0.00[0d]sgerup=;0.00[0d]sval=;5.18[0d]text_11=;[0d]
text_12=;[0d]tind=;1.53[0d][1e][1d]'''.encode('ascii')
writer(job_response)
def get_init_request():
req = p.readline()
print(req)
request = req.decode()[4:11]
# print(request)
if request == 'req=ini':
print('request == req=ini??? <<<<<<< cumple condicion y enviala respuesta')
send_init_response()
send_job()
while True:
# print(get_init_request())
get_init_request()
что я получаю на экране: основной скрипт
init method being active
bce
erd
condition was met init status=0
outside loop
ers
condition was met init status=0
inside while loop
trigger reset <<<--------------------
5782
`:lmatid=;151[0d]lmatname=;f50[0d]
lnam=;vsp_basic_fh15[0d]sgerin=;0.00[0d]sgerup=;0.00[0d]sval=;5.18[0d]text_11=;[0d]
text_12=;[0d]tind=;1.53[0d][1e][1d]
outside loop
condition was met init status=0
outside loop
что я получаю на экране: скрипт эмуляции хоста
b'[1c]req=ini[0d][0a][1e][1d]'
request == req=ini??? <<<<<<< cumple condicion y enviala respuesta
b''
b'[06][1c]req=33[0d][0a]job=5782[0d][0a][1e][1d]'
b''
b''
b''
b''
b''
b''
1 ответ
Я подозреваю, что вы пытаетесь записать слишком много сразу в достаточно маленький аппаратный буфер. Особенно при работе с аппаратным обеспечением с низким энергопотреблением, предполагать, что вы можете поместить все сообщение в буфер, не всегда правильно. Даже на современных ПК иногда есть очень маленькие буферы для устаревшего оборудования, такого как последовательные порты. При переходе от разработки к реальному оборудованию вы можете обнаружить, что для определения того, когда отправлять или получать данные, необходимо использовать линии RTS и DTR. Это будет зависеть от того, кто разработал аппаратное обеспечение, к сожалению, поскольку они часто также игнорируются.
Я бы попробовал разбить вашу передачу данных на более мелкие биты в качестве теста, чтобы увидеть, проходит ли все сообщение целиком. Это быстрая и грязная первая попытка, которая может иметь ошибки, но она должна помочь вам выбрать правильный путь:
def get_job_from_serial():
response = b'' #buffer for response
while True:
try:
response += serial_client.read() #read any available data or wait for timeout
#this technically could only be reading 1 char at a time, but any
#remotely modern pc should easily keep up with 9600 baud
except serial.SerialTimeoutException: #timeout probably means end of data
#you could also presumably check the length of the buffer if it's always
#a fixed length to determine if the entire message has been sent yet.
break
return response
def writer(command):
written = 0 #how many bytes have we actually written
chunksize = 128 #the smaller you go, the less likely to overflow
# a buffer, but the slower you go.
while written < len(command):
#you presumably might have to wait for p.dtr() == True or similar
#though it's just as likely to not have been implemented.
written += p.write(command[written:written+chunksize])
p.flush() #probably don't actually need this
PS Мне пришлось перейти к исходному коду для p.read_all (по какой-то причине я не смог найти его в Интернете), и он не делает то, что, как я думаю, вы ожидаете. Точный код для этого:
def read_all(self):
"""\
Read all bytes currently available in the buffer of the OS.
"""
return self.read(self.in_waiting)
Нет концепции ожидания полного сообщения, это всего лишь сокращение для захвата всего, что доступно в данный момент.