SY-3616: Refactor Protocol Sims to use SimDAQ Infrastructure (#2001)

* Refactor Protocol Sims tro inheret from base simulator

* Update simulators to have a configurable rate

* Extract SimulatorCase, flatten SimulatorTaskCase, and use class attributes

Author: sy-nico

.github/PULL_REQUEST_TEMPLATE/rc.md

@@ -441,9 +441,6 @@ I can successfully:
 - **Read Task**
   - **Single Sampling**
     - [ ] Read timestamps from the OPC UA server.
-    - **Test the following array sizes:**
-      - [ ] 10
-      - [ ] 100
     - [ ] Read timestamps from the OPC UA server.
   - [ ] Obtain recommended Synnax channels based on the configured OPC UA node.
   - [ ] Connect to and read data from a physical device.
@@ -464,10 +461,6 @@ I can successfully:
   - [ ] Apply scaling to register values.
   - [ ] Enable and disable data saving.
   - [ ] Handle device disconnection gracefully.
-  - **Reliable data reading at the following sample rates:**
-    - [ ] 1 Hz
-    - [ ] 10 Hz
-    - [ ] 100 Hz
 - **Write Task**
   - [ ] Write to holding registers on a Modbus server.
   - [ ] Write to coils on a Modbus server.

client/py/examples/modbus/__init__.py

@@ -9,6 +9,6 @@
 
 """Modbus TCP example package."""
 
-from .server import run_server
+from .server import ModbusSim, run_server
 
-__all__ = ["run_server"]
+__all__ = ["run_server", "ModbusSim"]

client/py/examples/modbus/server.py

@@ -19,13 +19,16 @@
 )
 from pymodbus.server import StartAsyncTcpServer
 
+import synnax as sy
+from examples.simulators.device_sim import DeviceSim
+from synnax import modbus
 
-async def updating_writer(context):
+
+async def updating_writer(context, rate: sy.Rate = 50 * sy.Rate.HZ):
     """Update Modbus registers continuously with simulated sensor data."""
     slave_id = 0x00
     start_ref = time.time()
     i = 0
-    RATE = 50  # Hz
     SENSOR_COUNT = 5
 
     while True:
@@ -65,10 +68,38 @@ async def updating_writer(context):
         coil_values = [True, False, True, False, True]
         context[slave_id].setValues(1, 0, coil_values)
 
-        await asyncio.sleep(1 / RATE)
+        await asyncio.sleep(1 / rate)
+
+
+class ModbusSim(DeviceSim):
+    """Modbus TCP device simulator on port 5020."""
+
+    description = "Modbus TCP simulator on port 5020"
+    host = "127.0.0.1"
+    port = 5020
+    device_name = "Modbus TCP Test Server"
 
+    async def _run_server(self) -> None:
+        await run_server(self.host, self.port, self.rate)
 
-async def run_server() -> None:
+    @staticmethod
+    def create_device(rack_key: int) -> modbus.Device:
+        return modbus.Device(
+            host=ModbusSim.host,
+            port=ModbusSim.port,
+            name=ModbusSim.device_name,
+            location=f"{ModbusSim.host}:{ModbusSim.port}",
+            rack=rack_key,
+            swap_bytes=False,
+            swap_words=False,
+        )
+
+
+async def run_server(
+    host: str = ModbusSim.host,
+    port: int = ModbusSim.port,
+    rate: sy.Rate = 50 * sy.Rate.HZ,
+) -> None:
     """Run the Modbus TCP server."""
     # Initialize data store
     store = ModbusDeviceContext(
@@ -89,13 +120,9 @@ async def run_server() -> None:
     identity.ModelName = "Extended Simulator"
     identity.MajorMinorRevision = "1.0.0"
 
-    # Start the updating task
-    task = asyncio.create_task(updating_writer(context))
+    asyncio.create_task(updating_writer(context, rate))
 
-    # Start Modbus TCP server on localhost:5020
-    await StartAsyncTcpServer(
-        context=context, identity=identity, address=("127.0.0.1", 5020)
-    )
+    await StartAsyncTcpServer(context=context, identity=identity, address=(host, port))
 
 
 if __name__ == "__main__":

client/py/examples/opcua/__init__.py

@@ -9,6 +9,6 @@
 
 """OPC UA example package."""
 
-from .server import run_server
+from .server import OPCUASim, run_server
 
-__all__ = ["run_server"]
+__all__ = ["run_server", "OPCUASim"]

client/py/examples/opcua/server.py

@@ -14,12 +14,16 @@
 
 from asyncua import Server, ua
 
+import synnax as sy
+from examples.simulators.device_sim import DeviceSim
+from synnax import opcua
+
 # Configuration constants
 ARRAY_COUNT = 5
-ARRAY_SIZE = 5
+DEFAULT_ARRAY_SIZE = 5
 FLOAT_COUNT = 5
 BOOL_COUNT = 5
-RATE = 50  # Hz
+DEFAULT_RATE = 50  # Hz
 BOOL_OFFSET = 0.2  # seconds between each boolean transition
 
 # Error injection configuration
@@ -29,30 +33,30 @@
 
 
 # Initialization Functions
-async def create_array_variables(myobj, idx):
+async def create_array_variables(myobj, idx, array_size: int = DEFAULT_ARRAY_SIZE):
     """Create array variables with initial values."""
     arrays = []
     for i in range(ARRAY_COUNT):
-        initial_values = [float(j + i) for j in range(ARRAY_SIZE)]
+        initial_values = [float(j + i) for j in range(array_size)]
         arr = await myobj.add_variable(
             idx, f"my_array_{i}", initial_values, ua.VariantType.Float
         )
-        await arr.write_array_dimensions([ARRAY_SIZE])
+        await arr.write_array_dimensions([array_size])
         arrays.append(arr)
     return arrays
 
 
-async def create_time_array(myobj, idx):
+async def create_time_array(myobj, idx, array_size: int = DEFAULT_ARRAY_SIZE):
     """Create timestamp array variable."""
     now = datetime.datetime.now(datetime.timezone.utc)
     initial_times = [
-        now + datetime.timedelta(milliseconds=j) for j in range(ARRAY_SIZE)
+        now + datetime.timedelta(milliseconds=j) for j in range(array_size)
     ]
     mytimearray = await myobj.add_variable(
         idx, "my_time_array", initial_times, ua.VariantType.DateTime
     )
     await mytimearray.set_writable()
-    await mytimearray.write_array_dimensions([ARRAY_SIZE])
+    await mytimearray.write_array_dimensions([array_size])
     return mytimearray
 
 
@@ -116,7 +120,7 @@ def generate_sinewave_values(timestamps, start_ref):
 
 
 # Update Functions
-def inject_error(values):
+def inject_error(values, array_size: int = DEFAULT_ARRAY_SIZE):
     """
     Generate corrupted array data for error injection.
     """
@@ -129,7 +133,7 @@ def inject_error(values):
 
     # Array smaller than expected
     elif error_chance < 0.667:
-        size = random.randint(1, ARRAY_SIZE - 2)
+        size = random.randint(1, max(2, array_size - 2))
         return values[:size]
 
     # Array larger than expected
@@ -178,18 +182,24 @@ async def update_bools(bools, elapsed):
         await bool_var.set_value(square_wave, varianttype=ua.VariantType.Boolean)
 
 
-async def run_server() -> None:
+async def run_server(
+    endpoint: str = "",
+    rate: sy.Rate = DEFAULT_RATE * sy.Rate.HZ,
+    array_size: int = DEFAULT_ARRAY_SIZE,
+) -> None:
     # Initialize server
     server = Server()
     await server.init()
-    server.set_endpoint("opc.tcp://127.0.0.1:4841/freeopcua/server/")
+    if not endpoint:
+        endpoint = OPCUASim.endpoint
+    server.set_endpoint(endpoint)
     uri = "http://examples.freeopcua.github.io"
     idx = await server.register_namespace(uri)
 
     # Create OPC UA object and variables
     myobj = await server.nodes.objects.add_object(idx, "MyObject")
-    arrays = await create_array_variables(myobj, idx)
-    mytimearray = await create_time_array(myobj, idx)
+    arrays = await create_array_variables(myobj, idx, array_size)
+    mytimearray = await create_time_array(myobj, idx, array_size)
     floats = await create_float_variables(myobj, idx)
     bools = await create_bool_variables(myobj, idx)
     commands = await create_command_variables(myobj, idx)
@@ -224,7 +234,7 @@ async def run_server() -> None:
             elapsed = (start - start_ref).total_seconds()
 
             # Generate data
-            timestamps = generate_timestamps(start, RATE, ARRAY_SIZE)
+            timestamps = generate_timestamps(start, rate, array_size)
             sinewave_values = generate_sinewave_values(timestamps, start_ref)
 
             # Update all variables
@@ -237,7 +247,38 @@ async def run_server() -> None:
             duration = (
                 datetime.datetime.now(datetime.timezone.utc) - start
             ).total_seconds()
-            await asyncio.sleep((1 / RATE) - duration)
+            await asyncio.sleep(max(0, (1 / rate) - duration))
+
+
+class OPCUASim(DeviceSim):
+    """OPC UA device simulator on port 4841."""
+
+    description = "OPC UA simulator on port 4841"
+    host = "127.0.0.1"
+    port = 4841
+    device_name = "OPC UA Test Server"
+    endpoint = f"opc.tcp://{host}:{port}/freeopcua/server/"
+
+    def __init__(
+        self,
+        array_size: int = DEFAULT_ARRAY_SIZE,
+        rate: sy.Rate = 50 * sy.Rate.HZ,
+        verbose: bool = False,
+    ):
+        super().__init__(rate=rate, verbose=verbose)
+        self.array_size = array_size
+
+    async def _run_server(self) -> None:
+        await run_server(self.endpoint, self.rate, self.array_size)
+
+    @staticmethod
+    def create_device(rack_key: int) -> opcua.Device:
+        return opcua.Device(
+            endpoint=OPCUASim.endpoint,
+            name=OPCUASim.device_name,
+            location=OPCUASim.endpoint,
+            rack=rack_key,
+        )
 
 
 if __name__ == "__main__":

client/py/examples/simulators/README.md

@@ -1,15 +1,21 @@
 # Hardware Simulators
 
 This directory contains reusable hardware simulators for testing control sequences
-without real hardware. All simulators extend the `SimDAQ` base class which provides:
+without real hardware. There are two types of simulators:
+
+1. **SimDAQ** - Thread-based simulators that write directly to Synnax channels
+2. **DeviceSim** - Process-based simulators that expose network protocol endpoints
+
+Both extend the shared `Simulator` base class which provides `start()` / `stop()`
+lifecycle management and verbose logging.
+
+## SimDAQ Simulators
+
+SimDAQ simulators create Synnax channels and write data directly. They provide:
 
-- Thread-based lifecycle management (`start()` / `stop()`)
 - Automatic end command handling via a separate watcher thread
-- Verbose logging mode for standalone execution
 - Command-line interface for running standalone
 
-## Available Simulators
-
 ### PressSimDAQ (`press.py`)
 
 Simulates a pressurization system with:
@@ -46,9 +52,35 @@ Simulates a rocket engine tank pressurization control system with:
 uv run python -m examples.simulators.tpc --help
 ```
 
+## DeviceSim Simulators
+
+DeviceSim simulators expose network protocol endpoints (Modbus TCP, OPC UA) for testing
+driver integration. They run async servers in a subprocess and do NOT interact with
+Synnax directly - the C++ driver connects to the endpoint and writes data to Synnax.
+
+### ModbusSim (`examples/modbus/server.py`)
+
+Runs a Modbus TCP server on port 5020 with:
+
+- Holding registers (addresses 0-4): Sine wave data
+- Input registers (addresses 0-4): Sine wave data
+- Discrete inputs (addresses 0-3): Rotating binary patterns
+- Coils (addresses 0-4): Static digital outputs
+
+### OPCUASim (`examples/opcua/server.py`)
+
+Runs an OPC UA server on port 4841 with:
+
+- Float variables (`my_float_0` - `my_float_4`): Sine wave data
+- Boolean variables (`my_bool_0` - `my_bool_4`): Square wave patterns
+- Array variables (`my_array_0` - `my_array_4`): 5-element float arrays
+- Command variables (`command_0` - `command_2`): Writable floats
+
 ## Creating Custom Simulators
 
-To create a custom simulator, extend the `SimDAQ` base class:
+### Custom SimDAQ
+
+Extend the `SimDAQ` base class for simulators that write directly to Synnax:
 
 ```python
 from examples.simulators import SimDAQ
@@ -59,7 +91,6 @@ class MySimDAQ(SimDAQ):
     end_cmd_channel = "end_my_test_cmd"  # Optional: auto-stop on this channel
 
     def _create_channels(self) -> None:
-        # Create your Synnax channels here
         self.my_channel = self.client.channels.create(
             name="my_channel",
             data_type=sy.DataType.FLOAT32,
@@ -78,9 +109,37 @@ if __name__ == "__main__":
     MySimDAQ.main()
 ```
 
+### Custom DeviceSim
+
+Extend `DeviceSim` for simulators that expose a network protocol endpoint:
+
+```python
+from examples.simulators.device_sim import DeviceSim
+from synnax import modbus
+
+class MyDeviceSim(DeviceSim):
+    description = "My custom device simulator"
+    host = "127.0.0.1"
+    port = 5020
+    device_name = "My Test Device"
+
+    async def _run_server(self) -> None:
+        # Start your async server here
+        ...
+
+    @staticmethod
+    def create_device(rack_key: int) -> modbus.Device:
+        return modbus.Device(
+            host=MyDeviceSim.host,
+            port=MyDeviceSim.port,
+            name=MyDeviceSim.device_name,
+            rack=rack_key,
+        )
+```
+
 ## Usage in Tests
 
-Simulators can be used programmatically in tests:
+SimDAQ simulators can be used programmatically:
 
 ```python
 import synnax as sy
@@ -94,3 +153,16 @@ sim.start()
 
 sim.stop()
 ```
+
+DeviceSim simulators are used via the integration test framework:
+
+```python
+from examples.modbus import ModbusSim
+
+sim = ModbusSim()
+sim.start()   # Starts server subprocess on port 5020
+
+# Driver connects to the endpoint...
+
+sim.stop()    # Terminates server subprocess
+```