Function run_server

examples/Modbus PLC Simulator/plc_simulator.py:834–884

()

async def run_server():
    sim = HydraulicTestStand()

    if USE_SIM_API:
        # Registers are refreshed lazily: pymodbus calls this on every request,
        # so each poll sees the live simulation state.
        async def refresh_registers(_fc, _start, _address, _count, registers, _values):
            regs = sim.get_registers()
            registers[: len(regs)] = regs
            return None

        context = SimDevice(
            0,
            simdata=[
                SimData(0, count=NUM_REGISTERS, values=0, datatype=DataType.REGISTERS)
            ],
            action=refresh_registers,
        )
        publish = None
    else:
        store = SlaveContext(
            di=ModbusSequentialDataBlock(0, [0] * NUM_REGISTERS),
            co=ModbusSequentialDataBlock(0, [0] * NUM_REGISTERS),
            hr=ModbusSequentialDataBlock(0, [0] * NUM_REGISTERS),
            ir=ModbusSequentialDataBlock(0, [0] * NUM_REGISTERS),
        )
        context = ModbusServerContext(store, single=True)

        def publish(regs):
            store.setValues(3, 0, regs)

    asyncio.create_task(update_simulation(sim, publish))

    print()
    print("=" * 72)
    print("  HYDRAULIC TEST STAND SIMULATOR")
    print("=" * 72)
    print(
        f"  Server: {SERVER_HOST}:{SERVER_PORT}  |  Update: {UPDATE_INTERVAL*1000:.0f}ms ({1/UPDATE_INTERVAL:.0f}Hz)"
    )
    print()
    print("  Registers (FC03 Holding):")
    print("    0:E-Stop  1:Start  2:Temp(°F)  3:PSI  4:RPM  5:Valve(%)")
    print("    6:GPM(÷10)  7:Load(%)  8:Vibration(÷10 mm/s)")
    print()
    print("  Phases: STARTUP -> RUNNING -> PRESSURE_TEST -> (FAILURE -> SHUTDOWN)")
    print("  Flags:  E=E-Stop  A=Alarm(>2800PSI)  F=Failure")
    print("=" * 72)
    print()

    await StartAsyncTcpServer(context=context, address=(SERVER_HOST, SERVER_PORT))


if __name__ == "__main__":