Add real life simulator example.

Author: janiversen

examples/heatpump.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""Controlling a Daikin Altherma 3 heatpump, and presenting data in Home Assistant.
+
+This app is used to control a Daikin heatpump, by reading the temperatures
+of deposit for the underfloor heating.
+
+Based on the temperatures and the time of the day (electricity periods P1, P2 and P3) it
+is determined to start/stop the heat pump (Daikin Altherma3)
+
+The data is made available in modbus and observed by a Home Assistant server, and presented
+to the home users.
+
+Schematic of deposit:
+
+             +--------+
+    A) -->  -+        |  F)
+             |     E) +--+- --> C)
+             |        |  |
+    B) <--  -+        |  |
+             |        +--+- <-- D)
+             +--------+
+
+    A) Intake from heat pump (hot)
+    B) Return to heat pump (cold)
+    C) Output to underfloor circulation pump (hot)
+    D) return from underfloor pipes (cold)
+    E) Output from tank (hot)
+    F) Thermostatic mixing valve (output is fixed at max 30 degrees)
+
+The 5 point are measured.
+
+The algorithm is quite simple:
+
+    Difference between A) and heat pump setpoint shows the loss in the connection pipes.
+    Difference between D) and underfloor heating thermostats show if heat is required.
+
+Remark: the return from the underfloor heating is connected to a thermostatic valve,
+on the output to the circulation pump. This allows to e.g. heat the tank to 50 degrees,
+while still circulating 30 degrees. Using the thermostatic valve, dramatically reduces
+the number of times the heat pump is started (when started it runs longer) and thus saving
+electricity.
+
+The thermo meters are read via 1-wire protocol.
+
+usage::
+
+    heatpump.py [-h]
+                       [--log {critical,error,warning,info,debug}]
+                       [--port <PORT>]
+
+    -h, --help
+        show this help message and exit
+    -l, --log {critical,error,warning,info,debug}
+        set log level, default is info
+    -p, --port PORT
+        set port to listen on
+"""
+import argparse
+import asyncio
+
+from pymodbus import Log, pymodbus_apply_logging_config
+from pymodbus.server import ModbusTcpServer
+from pymodbus.simulator import DataType, SimData, SimDevice
+from pymodbus.constants import ExcCodes
+
+DEVICE_ID = 1
+
+def get_commandline(cmdline: list[str] | None = None) -> argparse.Namespace:
+    """Read and check command line arguments."""
+    parser = argparse.ArgumentParser(description="server_update")
+    parser.add_argument(
+        "-l",
+        "--log",
+        choices=["critical", "error", "warning", "info", "debug"],
+        help="set log level, default is info",
+        dest="log",
+        default="info",
+        type=str,
+    )
+    parser.add_argument(
+        "-p",
+        "--port",
+        help="set listen port, default is 5020",
+        dest="port",
+        default=5020,
+        type=str,
+    )
+    args = parser.parse_args(cmdline)
+    pymodbus_apply_logging_config(args.log.upper())
+    return args
+
+
+def setup_updating_server(cmdline=None):
+    """Run server setup."""
+    args = get_commandline(cmdline=cmdline)
+
+    device = SimDevice(DEVICE_ID, simdata=[
+        # bit 0, Heat pump requested active
+        # bit 1, Alive signal (toggles every minute)
+        # bit 2, Call signal (toggles with every call)
+        # bit 3, Heat pump set to be active
+        SimData(0, DataType.BITS, readonly=True),
+        # 4 thermo meters A,B,C,D
+        SimData(1, count=4, datatype=DataType.FLOAT32, readonly=True),
+        # Setpoint to activate heat pump (output to underfloor pipes)
+        # Setpoint to run heat pump at night (return to heat pump)
+        SimData(9, count=2, datatype=DataType.FLOAT32),
+    ])
+    server = ModbusTcpServer(
+        device,
+        address=("", args.port),
+        context=device,
+    )
+    return server
+
+
+async def updating_task(server):
+    """Update values in server.
+
+    This task runs continuously beside the server
+    It will increment some values each second.
+    """
+    func_code = 3
+    device_id = 0x01
+    address = 0x10
+    count = 6
+
+    # set values to zero
+    values = await server.async_getValues(device_id, func_code, address, count=count)
+    values = [0 for v in values]
+    await server.async_setValues(device_id, func_code, address, values)
+
+    txt = (
+        f"updating_task: started: initialised values: {values!s} at address {address!s}"
+    )
+    print(txt)
+    Log.debug(txt)
+
+    # incrementing loop
+    while True:
+        await asyncio.sleep(1)
+
+        values = await server.async_getValues(device_id, func_code, address, count=count)
+        values = [v + 1 for v in values]
+        await server.async_setValues(device_id, func_code, address, values)
+
+        txt = f"updating_task: incremented values: {values!s} at address {address!s}"
+        print(txt)
+        Log.debug(txt)
+
+async def catch_requests(
+        function_code: int,
+        start_address: int,
+        address: int,
+        count: int,
+        current_registers: list[int],
+        set_values: list[int] | list[bool] | None
+    ) -> :
+    """Run action."""
+
+
+async def run_updating_server(server):
+    """Start updating_task concurrently with the current task."""
+    task = asyncio.create_task(updating_task(server))
+    task.set_name("example updating task")
+    await server.serve_forever()  # start the server
+    task.cancel()
+
+
+async def main(cmdline=None):
+    """Combine setup and run."""
+    server = setup_updating_server(cmdline=cmdline)
+    await run_updating_server(server)
+
+
+if __name__ == "__main__":
+    asyncio.run(main(), debug=True)

examples/server_updating.py

@@ -4,6 +4,10 @@
 An example of an asynchronous server and
 a task that runs continuously alongside the server and updates values.
 
+A real world example controlling a heatpump can be found at
+
+    examples/heatpump.py
+
 usage::
 
     server_updating.py [-h]

pymodbus/__init__.py

@@ -6,16 +6,17 @@
 __all__ = [
     "ExceptionResponse",
     "FramerType",
+    "Log",
     "ModbusDeviceIdentification",
     "ModbusException",
     "__version__",
     "__version_full__",
-    "pymodbus_apply_logging_config"
+    "pymodbus_apply_logging_config",
 ]
 
 from .exceptions import ModbusException
 from .framer import FramerType
-from .logging import pymodbus_apply_logging_config
+from .logging import Log, pymodbus_apply_logging_config
 from .pdu import ExceptionResponse
 from .pdu.device import ModbusDeviceIdentification