Supports simple cmd payload without register info

Author: zhongys-c8y

config/devices.toml

@@ -19,6 +19,7 @@ multiplier=1
 divisor=1
 decimalshiftright=0
 input=false # true = Input Register false = Holding Register
+registerid="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"
+registerid="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"
+coilid="TestAlarm"

tedge_modbus/operations/common.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import re
 
 import toml
 from pymodbus.client import ModbusSerialClient, ModbusTcpClient
@@ -144,3 +145,114 @@ 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 ("registerid" or "coilid")
+        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, value_type, logger):
+    """Extract metric name and value from payload."""
+    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):
+    """Load and find target device from devices.toml."""
+    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 id."""
+    partial_match = None
+
+    for config in configs:
+        config_id = config.get(id_key)
+        if config_id:
+            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
+
+            if config_id in metric_name and partial_match is None:
+                partial_match = (config, config_id)
+
+    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'",
+        configs.__class__.__name__,
+        metric_name,
+    )
+    return None, None, None, None

tedge_modbus/operations/set_coil.py

@@ -9,6 +9,8 @@
     prepare_client,
     apply_loglevel,
     close_client_quietly,
+    extract_device_from_topic,
+    _match_from_metrics,
 )
 
 logger = logging.getLogger(__name__)
@@ -17,41 +19,49 @@
 )
 
 
-def run(arguments: str | list[str]) -> None:
+def run(arguments: str | list[str], topic: str | None = None) -> None:
     """Run set coil operation handler
-    Expected arguments (JSON):
-    {
-        "input": false,
-        "address": < Fieldbusaddress >,
-        "coil": < coilnumber >,
-        "value": < 0 | 1 >
-    }
-    Parse JSON payload"""
-    payload = parse_json_arguments(arguments)
 
-    # Create context with default config directory
-    context = Context()
+    Supports two payload formats:
+
+    1. Legacy format (full coil details):
+        {
+            "input": false,
+            "address": < Fieldbusaddress >,
+            "coil": < coilnumber >,
+            "value": < 0 | 1 >
+        }
 
-    # Load configs and set log level
+    2. New format (coilid-based):
+        {
+            "timestamp": "2025-09-23T00:00:00Z",
+            "uuid": "device-id",
+            "metrics": [{
+                "name": "<coilid>_xxxxxxxx",
+                "timestamp": "2025-09-23T01:00:00Z",
+                "value": 0 or 1
+            }]
+        }
+        Requires topic: te/device/<device-id>///cmd/modbus_SetCoil/<mapper-id>
+    """
+    payload = parse_json_arguments(arguments)
+    context = Context()
     modbus_config = context.base_config
     apply_loglevel(logger, modbus_config)
-    logger.info("New set coil operation. args=%s", arguments)
-
-    try:
-        slave_id = int(payload["address"])  # Fieldbus address
-        coil_number = int(payload["coil"])  # Coil address
-        value_int = int(payload["value"])  # 0 or 1
-    except KeyError as err:
-        raise ValueError(f"Missing required field: {err}") from err
-    except (TypeError, ValueError) as err:
-        raise ValueError(f"Invalid numeric field: {err}") from err
+    logger.info("New set coil operation")
 
-    if value_int not in (0, 1):
-        raise ValueError("value must be 0 or 1 for a coil write")
+    if "metrics" in payload and topic:
+        slave_id, coil_number, value_int, ip_address = _process_new_format_coil(
+            payload, topic, context
+        )
+    else:
+        slave_id, coil_number, value_int, ip_address = _process_legacy_format_coil(
+            payload
+        )
 
     # Prepare client (resolve target, backfill defaults, build client)
     client = prepare_client(
-        payload["ipAddress"],
+        ip_address,
         slave_id,
         context.config_dir / "devices.toml",
         modbus_config,
@@ -74,3 +84,72 @@ def run(arguments: str | list[str]) -> None:
         raise
     finally:
         close_client_quietly(client)
+
+
+def _process_new_format_coil(payload: dict, topic: str, context):
+    """Process new format coil payload."""
+    logger.info("Processing new format payload with metrics array")
+    device_name = extract_device_from_topic(topic)
+    if not device_name:
+        raise ValueError(f"Could not extract device name from topic: {topic}")
+
+    coil_config, target_device, coil_id, write_value = _match_coil_from_metrics(
+        payload, device_name, context.config_dir / "devices.toml"
+    )
+
+    if not coil_config:
+        raise ValueError("Could not match any coil for metrics in payload")
+    if not target_device:
+        raise ValueError(f"Could not find device '{device_name}' in devices.toml")
+
+    logger.info("Matched CoilID: %s, Value: %s", coil_id, write_value)
+    coil_num = coil_config.get("number")
+    if coil_num is None:
+        raise ValueError(
+            f"Coil configuration missing 'number' field for coilid '{coil_id}'"
+        )
+
+    value_int = int(write_value)
+    if value_int not in (0, 1):
+        raise ValueError("Coil value must be 0 or 1")
+
+    return (
+        target_device.get("address"),
+        coil_num,
+        value_int,
+        target_device.get("ip", ""),
+    )
+
+
+def _process_legacy_format_coil(payload: dict):
+    """Process legacy format coil payload."""
+    logger.info("Processing legacy format payload")
+    try:
+        slave_id = int(payload["address"])
+        coil_number = int(payload["coil"])
+        value_int = int(payload["value"])
+    except KeyError as err:
+        raise ValueError(f"Missing required field: {err}") from err
+    except (TypeError, ValueError) as err:
+        raise ValueError(f"Invalid numeric field: {err}") from err
+
+    if value_int not in (0, 1):
+        raise ValueError("value must be 0 or 1 for a coil write")
+
+    return slave_id, coil_number, value_int, payload.get("ipAddress", "")
+
+
+def _match_coil_from_metrics(payload: dict, device_name: str, devices_path):
+    """Match coil from devices.toml by checking if metric name starts with coilid.
+
+    Args:
+        payload: Payload containing metrics array
+        device_name: Name of the device to search
+        devices_path: Path to devices.toml
+
+    Returns:
+        Tuple of (coil_config, target_device, coilid, value)
+    """
+    return _match_from_metrics(
+        payload, device_name, devices_path, "coils", "coilid", int
+    )