84 lines
4.3 KiB
Python
84 lines
4.3 KiB
Python
# Демонстрационная модель шины и устройств для интерфейса без оборудования. Синтетические
|
|
# ответы служат проверке пользовательского сценария; они не являются измерениями физической
|
|
# линии.
|
|
|
|
"""@file bus_demo.py
|
|
@brief Генератор трафика полевой шины: разбор проверяется без железа.
|
|
|
|
Повторяет то, что мост выдаёт в RS485: готовые транспортные кадры со
|
|
всеми типами сообщений ProtoCAN. Модуль не импортирует Qt — планировщик
|
|
живёт во вкладке, а сам генератор проверяется host-тестами.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import random
|
|
from datetime import datetime
|
|
|
|
from . import can_transport as tr
|
|
from protocan import protocan as pc
|
|
|
|
#: Типы сообщений, которые генератор выдаёт по кругу случайно.
|
|
KINDS = ("pulse", "analog", "modbus", "gas", "status", "error", "discrete")
|
|
|
|
|
|
class BusDemoSource:
|
|
"""@brief Источник демонстрационных кадров шины.
|
|
|
|
@param rng Источник случайности; передаётся в тестах для повторяемости.
|
|
"""
|
|
|
|
def __init__(self, rng: random.Random | None = None) -> None:
|
|
self._rng = rng or random.Random()
|
|
self._seq = 0
|
|
|
|
def next_frame(self) -> bytes:
|
|
"""@brief Собирает очередной кадр так же, как его отдал бы мост.
|
|
|
|
@return Готовые байты транспортного кадра, включая SOF и CRC.
|
|
"""
|
|
self._seq = (self._seq + 1) & 0xFF
|
|
kind = self._rng.choice(KINDS)
|
|
dev_type, dev_id = 1, 2
|
|
if kind == "pulse":
|
|
cid = pc.ProtoCanId.build(1, 1, dev_type, dev_id, pc.MsgType.PULSE, 0)
|
|
data = bytes([self._seq])
|
|
elif kind == "analog":
|
|
sid = self._rng.randint(1, 40)
|
|
atype = self._rng.choice([pc.AnalogType.U, pc.AnalogType.I, pc.AnalogType.T])
|
|
cid = pc.ProtoCanId.build(0, 1, dev_type, dev_id, pc.MsgType.ANALOG,
|
|
pc.merge_analog(atype, sid))
|
|
tag = {pc.AnalogType.U: b"US", pc.AnalogType.I: b"IS",
|
|
pc.AnalogType.T: b"TS"}[atype]
|
|
data = tag + ("%04d" % sid).encode()
|
|
elif kind == "modbus":
|
|
adr, cnt = self._rng.randint(0, 0x200), self._rng.randint(1, 4)
|
|
cid = pc.ProtoCanId.build(0, 1, dev_type, dev_id, pc.MsgType.MODBUS_HOLDING,
|
|
pc.merge_modbus(adr, cnt))
|
|
data = b"".join(self._rng.randint(0, 0xFFFF).to_bytes(2, "little")
|
|
for _ in range(cnt))
|
|
elif kind == "gas":
|
|
adr = self._rng.randint(0, 0x400)
|
|
cid = pc.ProtoCanId.build(0, 1, dev_type, dev_id,
|
|
pc.MsgType.GENERAL_ADDRESS_SPACE, adr)
|
|
data = b"".join(self._rng.randint(0, 0xFFFF).to_bytes(2, "little")
|
|
for _ in range(self._rng.randint(1, 4)))
|
|
elif kind == "status":
|
|
now = datetime.now()
|
|
cid = pc.ProtoCanId.build(1, 1, dev_type, dev_id, pc.MsgType.BROADCAST,
|
|
pc.merge_broadcast(pc.BroadcastType.STATUS, 0))
|
|
data = bytes([now.hour, now.minute, now.second, now.year % 100,
|
|
now.month, now.day, now.isoweekday() % 7])
|
|
elif kind == "error":
|
|
cid = pc.ProtoCanId.build(0, 1, dev_type, dev_id, pc.MsgType.ERROR,
|
|
pc.merge_error(self._rng.randint(0, 5), 0xFF))
|
|
data = b""
|
|
else:
|
|
dtype = self._rng.choice(list(pc.DiscreteType)[:7])
|
|
cid = pc.ProtoCanId.build(0, 1, dev_type, dev_id, pc.MsgType.DISCRETE,
|
|
pc.merge_discrete(dtype, self._rng.randint(0, 20)))
|
|
data = bytes(self._rng.randint(0, 255)
|
|
for _ in range(self._rng.randint(1, 8)))
|
|
return tr.Frame(seq=self._seq, flags=tr.FLAG_IDE, can_id=cid,
|
|
data=data).encode()
|