136 lines
5.9 KiB
Python
136 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Имитатор сервера Modbus Serial прибора Щ3.
|
|
Использовать для отладки. Только сохраняет данные регистров.
|
|
"""
|
|
|
|
import logger
|
|
import os
|
|
import argparse
|
|
import json
|
|
from socket import *
|
|
from pymodbus.server.sync import StartSerialServer
|
|
from pymodbus.datastore import ModbusSlaveContext
|
|
from pymodbus.datastore import ModbusServerContext
|
|
from pymodbus.transaction import ModbusRtuFramer
|
|
from threading import Thread
|
|
|
|
DEF_SERIAL = 'COM2' if os.name == 'nt' else '/dev/tty3'
|
|
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>': 'ошибка: получена неверная команда'
|
|
}
|
|
|
|
MAIN_ADDR = ('localhost', 44103)
|
|
DST_ADDR = ('localhost', 44203)
|
|
|
|
""" Адрес получателя """
|
|
|
|
FX_CODES = {2: 'di', 4: 'ir', 3: 'hr', 6: 'hr', 16: 'hr', 22: 'hr', 23: 'hr', 1: 'co', 5: 'co', 15: 'co'}
|
|
""" Кодировка регистров в pymodbus """
|
|
|
|
|
|
def on_rcv(*args):
|
|
store = args[0]
|
|
sock = store.sock_udp
|
|
dst_addr = store.dst_addr
|
|
|
|
json_decoder = json.JSONDecoder()
|
|
json_encoder = json.JSONEncoder()
|
|
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'] = []
|
|
|
|
if message['code'] == 'write':
|
|
data = message['data']
|
|
store.setValues(3, data[0], data[1:])
|
|
tx_data = json_encoder.encode({'code': 'response', 'data': ''})
|
|
sock.sendto(bytes(tx_data, 'utf-8'), dst_addr)
|
|
elif message['code'] == 'write_input':
|
|
data = message['data']
|
|
store.setValues(4, data[0], data[1:])
|
|
tx_data = json_encoder.encode({'code': 'response', 'data': ''})
|
|
sock.sendto(bytes(tx_data, 'utf-8'), dst_addr)
|
|
elif message['code'] == 'read_holding':
|
|
data = message['data']
|
|
read_data = store.getValues(3, data[0], data[1])
|
|
read_data.insert(0, data[0])
|
|
tx_data = json_encoder.encode({'code': 'response', 'data': read_data})
|
|
sock.sendto(bytes(tx_data, 'utf-8'), dst_addr)
|
|
elif message['code'] == 'read_input':
|
|
data = message['data']
|
|
read_data = store.getValues(4, data[0], data[1])
|
|
read_data.insert(0, data[0])
|
|
tx_data = json_encoder.encode({'code': 'response', 'data': read_data})
|
|
sock.sendto(bytes(tx_data, 'utf-8'), dst_addr)
|
|
|
|
|
|
class CallbackModbusSlaveContext(ModbusSlaveContext):
|
|
""" Блок данных, который сохраняет новое значение
|
|
в памяти и передаёт операцию в очередь сообщений
|
|
для дальнейшей процедуры обработки. """
|
|
|
|
def __init__(self):
|
|
""" """
|
|
self.dst_addr = DST_ADDR
|
|
self.json_enc = json.JSONEncoder()
|
|
self.sock_udp = socket(AF_INET, SOCK_DGRAM)
|
|
self.sock_udp.bind(MAIN_ADDR)
|
|
super(CallbackModbusSlaveContext, self).__init__()
|
|
|
|
def setValues(self, fx, address, values):
|
|
""" Устанавливает требуемые значения хранилища данных.
|
|
:param fx:
|
|
:param address: Начальный адрес.
|
|
:param values: Новые значения, которые необходимо установить.
|
|
"""
|
|
super(CallbackModbusSlaveContext, self).setValues(fx, address, values)
|
|
fx = FX_CODES[fx] if fx in FX_CODES else '**error**'
|
|
log.debug('setValues() fx:%s address:%s values:%s' % (fx, address, values))
|
|
|
|
def getValues(self, fx, address, count=1):
|
|
""" Устанавливает требуемые значения хранилища данных.
|
|
:param fx:
|
|
:param address: Начальный адрес
|
|
:param count:
|
|
"""
|
|
values = super(CallbackModbusSlaveContext, self).getValues(fx, address, count)
|
|
fx = FX_CODES[fx] if fx in FX_CODES else '**error**'
|
|
log.debug('getValues() fx:%s address:%s values:%s' % (fx, address, values))
|
|
return values
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
|
|
parser.add_argument('--serial-baud', default=115200, type=int, help='Скорость последовательного порта, бод')
|
|
parser.add_argument('--serial-port', default=DEF_SERIAL, type=str, help='Последовательный порт для подключения клиента')
|
|
parser.add_argument('--timeout', default=0.005, type=str, help='Таймаут последовательного порта')
|
|
opts, opts_unk = parser.parse_known_args()
|
|
print_opts_error(opts_unk)
|
|
store = CallbackModbusSlaveContext()
|
|
context = ModbusServerContext(slaves=store, single=True)
|
|
log.info("серевер-заглушка Modbus serial работает: %s:%d" % (opts.serial_port, opts.serial_baud))
|
|
t = Thread(target=on_rcv, args=[store], daemon=True)
|
|
t.start()
|
|
StartSerialServer(context=context, framer=ModbusRtuFramer, port=opts.serial_port, timeout=opts.timeout, baudrate=opts.serial_baud)
|
|
|
|
|
|
def print_opts_error(opts: list) -> None:
|
|
for opt in opts:
|
|
log.debug('внимание: неизвестный аргумент: \"%s\"\n' % opt)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|