Supports simple cmd payload without register info

Supports simple cmd payload without register info, made register address 6 and 7 writable and fixed writing float issue

Author: zhongys-c8y

config/devices.toml

@@ -19,6 +19,7 @@ multiplier=1
 divisor=1
 decimalshiftright=0
 input=false # true = Input Register false = Holding Register
+name="Test_Int16"
 #see https://thin-edge.github.io/thin-edge.io/html/architecture/thin-edge-json.html
 measurementmapping.templatestring="{\"Test\":{\"Int16\":%% }}" # tedge JSON format string, %% will be replaced with the calculated value
 #measurementmapping.combinemeasurements=true # Overrides device setting; Combines all measurements of a device to reduce the number of created measurements in the cloud
@@ -34,6 +35,7 @@ divisor=1
 decimalshiftright=0
 input=false
 datatype="float"
+name="Test_Float32"
 measurementmapping.templatestring="{\"Test\":{\"Float32\":%% }}"
 
 
@@ -42,4 +44,5 @@ number=2
 input=false
 alarmmapping.severity="MAJOR"
 alarmmapping.text="This alarm should be created once"
-alarmmapping.type="TestAlarm"
+alarmmapping.type="TestAlarm"
+name="TestAlarm"

images/simulator/modbus.json

@@ -34,7 +34,7 @@
         }
       },
       "invalid": [1],
-      "write": [3, 48],
+      "write": [3, 6, 7, 48],
       "bits": [
         48
       ],

tedge_modbus/operations/common.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import re
 
 import toml
 from pymodbus.client import ModbusSerialClient, ModbusTcpClient
@@ -32,24 +33,25 @@ def resolve_target_device(
     Returns (target_device, protocol).
     """
     if ip_address:
-        ip = ip_address or "127.0.0.1"
-        protocol = "TCP"
         target_device = {
             "protocol": "TCP",
-            "ip": ip,
+            "ip": ip_address or "127.0.0.1",
             "port": 502,
             "address": slave_id,
         }
-    else:
-        devices_cfg = toml.load(devices_path)
-        devices = devices_cfg.get("device", []) or []
-        target_device = next(
-            (d for d in devices if d.get("address") == slave_id), None
-        ) or next((d for d in devices if d.get("protocol") == "TCP"), None)
-        if target_device is None:
-            raise ValueError(f"No suitable device found in {devices_path}")
-        protocol = target_device.get("protocol")
-    return target_device, protocol  # type: ignore[return-value]
+        return target_device, "TCP"
+
+    devices_cfg = toml.load(devices_path)
+    devices = devices_cfg.get("device", []) or []
+    target_device = next(
+        (d for d in devices if d.get("address") == slave_id), None
+    ) or next((d for d in devices if d.get("protocol") == "TCP"), None)
+
+    if target_device is None:
+        raise ValueError(f"No suitable device found in {devices_path}")
+
+    protocol = target_device.get("protocol", "TCP")
+    return target_device, protocol
 
 
 def backfill_serial_defaults(
@@ -81,7 +83,7 @@ def build_modbus_client(target_device: dict, protocol: str):
             parity=target_device["parity"],
             bytesize=target_device["databits"],
         )
-    raise ValueError("Expected protocol to be RTU or TCP. Got " + str(protocol) + ".")
+    raise ValueError(f"Expected protocol to be RTU or TCP, got {protocol}")
 
 
 def close_client_quietly(client) -> None:
@@ -144,3 +146,138 @@ def compute_masked_value(
         raise ValueError(f"value must be within 0..{max_value} for noBits={num_bits}")
     mask = ((1 << num_bits) - 1) << start_bit
     return (current_value & ~mask) | ((write_value << start_bit) & mask)
+
+
+def extract_device_from_topic(topic: str) -> str:
+    """Extract device-id from topic.
+
+    Expected topic format: te/device/<device-id>///cmd/modbus_Set{Register|Coil}/<mapper-id>
+
+    Supports both SetRegister and SetCoil operations.
+
+    Returns device_id or empty string
+    """
+    # Match pattern like:
+    # - te/device/TestCase1///cmd/modbus_SetRegister/c8y-mapper-123
+    # - te/device/TestCase1///cmd/modbus_SetCoil/c8y-mapper-123
+    match = re.search(r"te/device/([^/]+)///cmd/modbus_Set(?:Register|Coil)/.+$", topic)
+    if match:
+        return match.group(1)
+    return ""
+
+
+def _match_from_metrics(  # pylint: disable=too-many-arguments
+    payload: dict,
+    device_name: str,
+    devices_path,
+    config_type: str,
+    id_key: str,
+    value_type,
+):
+    """Generic function to match register/coil from devices.toml.
+
+    Args:
+        payload: Payload containing metrics array
+        device_name: Name of the device to search
+        devices_path: Path to devices.toml
+        config_type: Type of config ("registers" or "coils")
+        id_key: ID field name ("name" for registers and coils)
+        value_type: Type to convert value to (float or int)
+
+    Returns:
+        Tuple of (config_dict, target_device, matched_id, value)
+    """
+    logger = logging.getLogger(__name__)
+
+    metric_name, value = _extract_metric_from_payload(payload, value_type, logger)
+    target_device = _load_target_device(devices_path, device_name)
+    if target_device is None:
+        return None, None, None, None
+
+    configs = target_device.get(config_type, []) or []
+    return _match_config_by_id(
+        configs, id_key, metric_name, target_device, value, logger
+    )
+
+
+def _extract_metric_from_payload(
+    payload: dict, value_type: type, logger
+) -> tuple[str, int | float]:
+    """Extract metric name and value from payload.
+
+    Args:
+        payload: Payload dictionary
+        value_type: Type to convert value to (int or float)
+        logger: Logger instance
+
+    Returns:
+        Tuple of (metric_name, converted_value)
+    """
+    metrics = payload.get("metrics", [])
+    if not metrics:
+        raise ValueError("No metrics found in payload")
+
+    if len(metrics) > 1:
+        logger.warning("Multiple metrics found, using first one")
+
+    metric = metrics[0]
+    return metric.get("name", ""), value_type(metric.get("value", 0))
+
+
+def _load_target_device(devices_path, device_name: str) -> dict | None:
+    """Load and find target device from devices.toml by name.
+
+    Args:
+        devices_path: Path to devices.toml file
+        device_name: Name of the device to find
+
+    Returns:
+        Device dictionary or None if not found
+    """
+    devices_cfg = toml.load(devices_path)
+    devices = devices_cfg.get("device", []) or []
+    return next((d for d in devices if d.get("name") == device_name), None)
+
+
+def _match_config_by_id(
+    configs, id_key, metric_name, target_device, value, logger
+):  # pylint: disable=too-many-arguments
+    """Match config by checking if metric name starts with or contains id."""
+    partial_match = None
+
+    for config in configs:
+        config_id = config.get(id_key)
+        if not config_id:
+            continue
+
+        # Exact prefix match (preferred)
+        if metric_name.startswith(config_id):
+            logger.info(
+                "Matched metric name '%s' with %s '%s'",
+                metric_name,
+                id_key,
+                config_id,
+            )
+            return config, target_device, config_id, value
+
+        # Partial match (fallback)
+        if config_id in metric_name and partial_match is None:
+            partial_match = (config, config_id)
+
+    # Return partial match if found
+    if partial_match:
+        config, config_id = partial_match
+        logger.info(
+            "Matched metric name '%s' with %s '%s' (partial match)",
+            metric_name,
+            id_key,
+            config_id,
+        )
+        return config, target_device, config_id, value
+
+    logger.warning(
+        "No matching %s found for metric name '%s'",
+        type(configs).__name__,
+        metric_name,
+    )
+    return None, None, None, None