Simplify custom operation handler's arguments

Signed-off-by: Rina Fujino <[email protected]>

Author: rina23q

operations/c8y_ModbusConfiguration

@@ -1,4 +1,4 @@
 [exec]
   topic = "c8y/devicecontrol/notifications"
   on_fragment = "c8y_ModbusConfiguration"
-  command = "python3 -m tedge_modbus.operations c8y_ModbusConfiguration ${.payload.c8y_ModbusConfiguration.transmitRate} ${.payload.c8y_ModbusConfiguration.pollingRate}"
+  command = "python3 -m tedge_modbus.operations c8y_ModbusConfiguration ${.payload.c8y_ModbusConfiguration}"

operations/c8y_ModbusDevice

@@ -1,4 +1,4 @@
 [exec]
   topic = "c8y/devicecontrol/notifications"
   on_fragment = "c8y_ModbusDevice"
-  command = "python3 -m tedge_modbus.operations c8y_ModbusDevice ${.payload.c8y_ModbusDevice.protocol} ${.payload.c8y_ModbusDevice.address} ${.payload.c8y_ModbusDevice.name} ${.payload.c8y_ModbusDevice.ipAddress} ${.payload.c8y_ModbusDevice.id} ${.payload.c8y_ModbusDevice.type}"
+  command = "python3 -m tedge_modbus.operations c8y_ModbusDevice ${.payload.c8y_ModbusDevice}"

operations/c8y_SerialConfiguration

@@ -1,4 +1,4 @@
 [exec]
   topic = "c8y/devicecontrol/notifications"
   on_fragment = "c8y_SerialConfiguration"
-  command = "python3 -m tedge_modbus.operations c8y_SerialConfiguration ${.payload.c8y_SerialConfiguration.baudRate} ${.payload.c8y_SerialConfiguration.stopBits} ${.payload.c8y_SerialConfiguration.parity} ${.payload.c8y_SerialConfiguration.dataBits}"
+  command = "python3 -m tedge_modbus.operations c8y_SerialConfiguration ${.payload.c8y_SerialConfiguration}"

tedge_modbus/operations/c8y_modbus_configuration.py

@@ -15,16 +15,17 @@
 
 def run(arguments, context: Context):
     """Run c8y_ModbusConfiguration operation handler"""
-    if len(arguments) != 2:
-        raise ValueError(f"Expected 2 arguments. Got {len(arguments)}")
+    if len(arguments) != 1:
+        raise ValueError(f"Expected 1 argument. Got {len(arguments)}")
     # Get device configuration
     modbus_config = context.base_config
     loglevel = modbus_config["modbus"]["loglevel"] or "INFO"
     logger.setLevel(getattr(logging, loglevel.upper(), logging.INFO))
     logger.info("New c8y_ModbusConfiguration operation")
     logger.debug("Current configuration: %s", modbus_config)
-    transmit_rate = int(arguments[0])
-    polling_rate = int(arguments[1])
+    data = json.loads(arguments[0])
+    transmit_rate = data["transmitRate"]
+    polling_rate = data["pollingRate"]
     logger.debug("transmitRate: %d, pollingRate: %d", transmit_rate, polling_rate)
 
     # Update configuration

tedge_modbus/operations/c8y_modbus_device.py

@@ -2,6 +2,7 @@
 """Cumulocity Modbus device operation handler"""
 import logging
 from dataclasses import dataclass
+import json
 import requests
 import toml
 
@@ -76,13 +77,14 @@ def get_device_from_mapping(target: ModebusDevice, mapping):
 
 def parse_arguments(arguments) -> ModebusDevice:
     """Parse operation arguments"""
+    data = json.loads(arguments[0])
     return ModebusDevice(
-        modbus_type=arguments[0],
-        modbus_address=arguments[1],
-        child_name=arguments[2],
-        modbus_server=arguments[3],
-        device_id=arguments[4],
-        mapping_path=arguments[5],
+        modbus_type=data["protocol"],
+        modbus_address=data["address"],
+        child_name=data["name"],
+        modbus_server=data["ipAddress"],
+        device_id=data["id"],
+        mapping_path=data["type"],
     )
 
 
@@ -92,8 +94,8 @@ def run(arguments, context: Context):
     logger.setLevel(getattr(logging, loglevel.upper(), logging.INFO))
     logger.info("New c8y_ModbusDevice operation")
     # Check and store arguments
-    if len(arguments) != 6:
-        raise ValueError("Expected 6 arguments. Got " + str(len(arguments)) + ".")
+    if len(arguments) != 1:
+        raise ValueError("Expected 1 argument. Got " + str(len(arguments)) + ".")
     config_path = context.config_dir / "devices.toml"
     target = parse_arguments(arguments)
 

tedge_modbus/operations/c8y_serial_configuration.py

@@ -13,20 +13,22 @@
 )
 
 
+# pylint: disable=duplicate-code
 def run(arguments, context: Context):
     """Run c8y_SerialConfiguration operation handler"""
-    if len(arguments) != 4:
-        raise ValueError(f"Expected 4 arguments. Got {len(arguments)}")
+    if len(arguments) != 1:
+        raise ValueError(f"Expected 1 argument. Got {len(arguments)}")
     # Get device configuration
     modbus_config = context.base_config
     loglevel = modbus_config["modbus"]["loglevel"] or "INFO"
     logger.setLevel(getattr(logging, loglevel.upper(), logging.INFO))
     logger.info("New c8y_SerialConfiguration operation")
     logger.debug("Current configuration: %s", modbus_config)
-    baud_rate = int(arguments[0])
-    stop_bits = int(arguments[1])
-    parity = arguments[2]
-    data_bits = int(arguments[3])
+    data = json.loads(arguments[0])
+    baud_rate = data["baudRate"]
+    stop_bits = data["stopBits"]
+    parity = data["parity"]
+    data_bits = data["dataBits"]
     logger.debug(
         "baudRate: %d, stopBits: %d, parity: %s, dataBits: %d",
         baud_rate,