104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
import argparse
|
|
import logging
|
|
import signal
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
from pyModbusTCP.server import DataBank, ModbusServer
|
|
|
|
|
|
LOGGER = logging.getLogger("nfs320-simulator")
|
|
REGISTER_COUNT = 600
|
|
|
|
STATUS_NORMAL = 5120
|
|
STATUS_INCIDENTE = 3088
|
|
STATUS_ACK = 3344
|
|
STATUS_REMOVIDO = 13312
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SimPoint:
|
|
address: int
|
|
label: str
|
|
values: tuple[int, ...]
|
|
|
|
|
|
SIM_POINTS = [
|
|
SimPoint(10, "Detector Recepcao", (STATUS_NORMAL, STATUS_NORMAL, STATUS_INCIDENTE, STATUS_ACK)),
|
|
SimPoint(11, "Detector CPD", (STATUS_NORMAL, STATUS_INCIDENTE, STATUS_INCIDENTE, STATUS_ACK)),
|
|
SimPoint(12, "Detector Almoxarifado", (STATUS_NORMAL, STATUS_NORMAL, STATUS_NORMAL, STATUS_REMOVIDO)),
|
|
SimPoint(165, "Modulo Bomba Incendio", (STATUS_NORMAL, STATUS_NORMAL, STATUS_ACK, STATUS_NORMAL)),
|
|
SimPoint(166, "Modulo Porta Corta-Fogo", (STATUS_NORMAL, STATUS_ACK, STATUS_NORMAL, STATUS_NORMAL)),
|
|
SimPoint(167, "Modulo Sirene Pavimento 1", (STATUS_NORMAL, STATUS_NORMAL, STATUS_NORMAL, STATUS_INCIDENTE)),
|
|
SimPoint(352, "Painel Repetidor", (STATUS_NORMAL, STATUS_NORMAL, STATUS_NORMAL, STATUS_NORMAL)),
|
|
]
|
|
|
|
|
|
def build_registers(step: int) -> list[int]:
|
|
registers = [0] * REGISTER_COUNT
|
|
for point in SIM_POINTS:
|
|
registers[point.address - 1] = point.values[step % len(point.values)]
|
|
return registers
|
|
|
|
|
|
def apply_step(data_bank: DataBank, step: int) -> None:
|
|
data_bank.set_holding_registers(1, build_registers(step))
|
|
status_summary = ", ".join(
|
|
f"{point.label}={point.values[step % len(point.values)]}" for point in SIM_POINTS
|
|
)
|
|
LOGGER.info("Ciclo %s aplicado: %s", step, status_summary)
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="Simulador Modbus/TCP de central Notifier NFS-320")
|
|
parser.add_argument("--host", default="127.0.0.1", help="Host de bind do simulador")
|
|
parser.add_argument("--port", default=1502, type=int, help="Porta TCP do simulador")
|
|
parser.add_argument("--interval", default=5.0, type=float, help="Intervalo entre mudancas de estado")
|
|
parser.add_argument("--log-level", default="INFO", help="Nivel de log")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
logging.basicConfig(
|
|
level=getattr(logging, args.log_level.upper(), logging.INFO),
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
)
|
|
|
|
stop_event = threading.Event()
|
|
data_bank = DataBank()
|
|
apply_step(data_bank, 0)
|
|
server = ModbusServer(host=args.host, port=args.port, no_block=True, data_bank=data_bank)
|
|
|
|
def rotate_states():
|
|
step = 1
|
|
while not stop_event.wait(args.interval):
|
|
apply_step(data_bank, step)
|
|
step += 1
|
|
|
|
def shutdown(*_args):
|
|
LOGGER.info("Encerrando simulador NFS-320")
|
|
stop_event.set()
|
|
server.stop()
|
|
|
|
signal.signal(signal.SIGINT, shutdown)
|
|
signal.signal(signal.SIGTERM, shutdown)
|
|
|
|
LOGGER.info("Subindo simulador NFS-320 em %s:%s", args.host, args.port)
|
|
LOGGER.info("Points simulados: %s", ", ".join(f"{point.address}:{point.label}" for point in SIM_POINTS))
|
|
server.start()
|
|
|
|
worker = threading.Thread(target=rotate_states, daemon=True, name="nfs320-state-rotator")
|
|
worker.start()
|
|
|
|
try:
|
|
while not stop_event.wait(1):
|
|
pass
|
|
finally:
|
|
shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|