108 lines
4.9 KiB
Python
108 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""
|
|
Драйвер клиента для интерфейса Modbus работающего через последовательный порт.
|
|
(приложение | UDP | JSON) <- (этот драйвер клиента) -> (сервер | Serial | Modbus RTU)
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import argparse
|
|
import time
|
|
from socket import *
|
|
|
|
import logger
|
|
from pymodbus.client.sync import ModbusSerialClient
|
|
|
|
DEF_SERIAL = 'COM1' if os.name == 'nt' else '/dev/tty0'
|
|
log_file_dir = os.path.join(os.path.expanduser('~'))
|
|
log = logger.get_logger(__file__, log_file_dir)
|
|
EXCEPT_MSGS = {
|
|
'write': 'ошибка: запись в регистры не выполнена',
|
|
'read_holding': 'ошибка: чтение регистров хранения не выполнено',
|
|
'read_input': 'ошибка: чтение регистров входов не выполнено',
|
|
'<unknown>': 'ошибка: получена неверная команда'
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
|
|
parser.add_argument('--modbus-unit', default=1, type=int, help='Номер устройства Modbus')
|
|
parser.add_argument('--serial-baud', default=115200, type=int, help='Скорость последовательного порта, бод')
|
|
parser.add_argument('--serial-port', default=DEF_SERIAL, type=str, help='Последовательный порт для подключения к Modbus серверу')
|
|
parser.add_argument('--udp-addr', default='localhost', type=str, help='Адрес для отправки сообщений приложению')
|
|
parser.add_argument('--udp-port-rx', default=44100, type=int, help='Порт для приёма сообщений от приложения')
|
|
parser.add_argument('--udp-port-tx', default=48000, type=int, help='Порт для отправки сообщений приложению')
|
|
opts, opts_unk = parser.parse_known_args()
|
|
print_opts_error(opts_unk)
|
|
run_sync_client(opts.udp_addr, opts.udp_port_rx, opts.udp_port_tx, opts.serial_port, opts.serial_baud, opts.modbus_unit)
|
|
|
|
|
|
def run_sync_client(udp_addr, udp_port_rx, udp_port_tx, serial_port, serial_baud, unit_num):
|
|
sock = socket(AF_INET, SOCK_DGRAM)
|
|
sock.bind((udp_addr, udp_port_rx))
|
|
log.info('привязан к адресу %s:%d' % (udp_addr, udp_port_rx))
|
|
|
|
json_decoder = json.JSONDecoder()
|
|
json_encoder = json.JSONEncoder()
|
|
|
|
client = ModbusSerialClient(method='rtu', port=serial_port, baudrate=int(serial_baud))
|
|
while not client.connect(): # Ждать подключения к серверу
|
|
time.sleep(3)
|
|
|
|
log.info('подключен к серверу через \"%s\" на скорости %d бод' % (serial_port, serial_baud))
|
|
|
|
while True:
|
|
udp_data = sock.recv(2048) # Ждать команду от приложения
|
|
udp_data = udp_data.decode('utf-8')
|
|
message = json_decoder.decode(udp_data) # type: dict
|
|
|
|
|
|
if 'code' not in message:
|
|
message['code'] = '<unknown>'
|
|
|
|
if 'data' not in message:
|
|
message['data'] = []
|
|
|
|
rc = False
|
|
if message['code'] == 'write':
|
|
data = message['data']
|
|
rq = client.write_registers(data[0], data[1:], unit=unit_num)
|
|
rc = rq.isError()
|
|
elif message['code'] == 'read_holding':
|
|
data = message['data']
|
|
print('client_serial_modbus.py: data', data)
|
|
rq = client.read_holding_registers(data[0], data[1:], unit=unit_num)
|
|
rc = rq.isError()
|
|
if not rc:
|
|
tx_data = json_encoder.encode({'code': 'response', 'data': rq.registers})
|
|
sock.sendto(bytes(tx_data), (udp_addr, udp_port_tx))
|
|
elif message['code'] == 'read_input':
|
|
data = message['data']
|
|
rq = client.read_input_registers(data[0], data[1:], unit=unit_num)
|
|
rc = rq.isError()
|
|
if not rc:
|
|
tx_data = json_encoder.encode({'code': 'response', 'data': rq.registers})
|
|
sock.sendto(bytes(tx_data), (udp_addr, udp_port_tx))
|
|
elif message['code'] == '<unknown>':
|
|
rc = False
|
|
|
|
if rc:
|
|
code = message['code']
|
|
data = message['data']
|
|
rc_msg = EXCEPT_MSGS[code]
|
|
tx_data = json_encoder.encode({'code': 'exception', 'data': rc_msg})
|
|
sock.sendto(bytes(tx_data), (udp_addr, udp_port_tx))
|
|
log.error(
|
|
'ошибка при выполнении команды: \"%s\". данные: \"%s\", порт: \"%s\", скорость: %s'
|
|
% (code, data, serial_port, serial_baud))
|
|
|
|
|
|
def print_opts_error(opts: list) -> None:
|
|
for opt in opts:
|
|
log.warning('внимание: неизвестный аргумент: \"%s\"' % opt)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|