Enabled CR1000X device to calculate XML data type

Author: lewis-chambers

src/iotdevicesimulator/devices.py

@@ -324,6 +324,60 @@ def __init__(
         if table_name is not None:
             self.table_name = str(table_name)
 
+    @staticmethod
+    def _get_xsd_type(value: str | int | float | bool | object) -> str:
+        """Converts a value to the XML data type expected.
+        Args:
+            value: The item to convert.
+
+        Returns:
+            str: A string of the XML datatype.
+        """
+
+        if isinstance(value, datetime):
+            return "xsd:dateTime"
+
+        if isinstance(value, bool):
+            return "xsd:boolean"
+
+        if isinstance(value, str):
+            try:
+                datetime.fromisoformat(value)
+                return "xsd:dateTime"
+            except ValueError:
+                pass
+
+            return "xsd:string"
+
+        if isinstance(value, int):
+            if (value > 0 and value <= 32767) or (value < 0 and value >= -32768):
+                return "xsd:short"
+            if (
+                value == 0
+                or (value < 0 and value >= -2147483648)
+                or (value > 0 and value <= 2147483647)
+            ):
+                return "xsd:int"
+            if (value > 0 and value <= 9223372036854775807) or (
+                value < 0 and value >= -9223372036854775808
+            ):
+                return "xsd:long"
+            else:
+                return "xsd:integer"
+
+        if isinstance(value, float):
+            if abs(value) > 0 and (
+                abs(value) < 1.1754943508222875e-38
+                or abs(value) > 3.4028234663852886e38
+            ):
+                return "xsd:double"
+            else:
+                return "xsd:float"
+
+        raise TypeError(
+            f"Couldnt find XML datatype for value `{value}` and type: `{type(value)}`."
+        )
+
     @staticmethod
     def _build_field(
         name: str,
@@ -393,12 +447,16 @@ def _format_payload(self, payload: dict) -> dict:
             if not date_found:
                 time = datetime.now().isoformat()
 
-            f_payload["head"]["fields"] = [self._build_field(k) for k in keys]
+            f_payload["head"]["fields"] = [
+                self._build_field(k, data_type=self._get_xsd_type(v))
+                for k, v in zip(keys, vals)
+            ]
             f_payload["data"] = {"time": time, "vals": vals}
 
         elif hasattr(payload, "__iter__"):
             f_payload["head"]["fields"] = [
-                self._build_field(f"_{i}") for i in range(len(payload))
+                self._build_field(f"_{i}", data_type=self._get_xsd_type(payload[i]))
+                for i in range(len(payload))
             ]
             f_payload["data"] = {"time": datetime.now().isoformat(), "vals": payload}
 

src/tests/test_devices.py

@@ -493,7 +493,7 @@ def test_table_name_set(self, arg):
         self.assertEqual(inst.table_name, str(arg))
 
     def test_list_payload_formatting(self):
-        payload = [1,"data", "true", True]
+        payload = [1,"data", 0.0, True]
 
         device = CR1000XDevice("my_device", self.db, self.conn)
 
@@ -514,28 +514,28 @@ def test_list_payload_formatting(self):
                 "fields": [
                     {
                         "name": "_0",
-                        "type": "",
+                        "type": "xsd:short",
                         "units": "",
                         "process": "",
                         "settable": False
                     },
                     {
                         "name": "_1",
-                        "type": "",
+                        "type": "xsd:string",
                         "units": "",
                         "process": "",
                         "settable": False
                     },
                     {
                         "name": "_2",
-                        "type": "",
+                        "type": "xsd:float",
                         "units": "",
                         "process": "",
                         "settable": False
                     },
                     {
                         "name": "_3",
-                        "type": "",
+                        "type": "xsd:boolean",
                         "units": "",
                         "process": "",
                         "settable": False
@@ -557,7 +557,7 @@ def test_list_payload_formatting(self):
         datetime.fromisoformat(formatted["data"]["time"])
 
     def test_dict_payload_formatting(self):
-        payload = {"temp": 17.16, "door_open": False, "BattV": 74, "BattLevel": 99}
+        payload = {"temp": 17.16, "door_open": False, "BattV": int(1e20), "BattLevel": 1e-50}
 
         device = CR1000XDevice("my_dict_device", self.db, self.conn)
 
@@ -578,28 +578,28 @@ def test_dict_payload_formatting(self):
                 "fields": [
                     {
                         "name": "temp",
-                        "type": "",
+                        "type": "xsd:float",
                         "units": "",
                         "process": "",
                         "settable": False
                     },
                     {
                         "name": "door_open",
-                        "type": "",
+                        "type": "xsd:boolean",
                         "units": "",
                         "process": "",
                         "settable": False
                     },
                     {
                         "name": "BattV",
-                        "type": "",
+                        "type": "xsd:integer",
                         "units": "",
                         "process": "",
                         "settable": False
                     },
                     {
                         "name": "BattLevel",
-                        "type": "",
+                        "type": "xsd:double",
                         "units": "",
                         "process": "",
                         "settable": False
@@ -631,5 +631,44 @@ def test_dict_payload_formatting(self):
 
         formatted = device._format_payload(payload)
         datetime.fromisoformat(formatted["data"]["time"])
+
+    @parameterized.expand([
+        [0, "xsd:int"],
+        [32767, "xsd:short"],
+        [-32768, "xsd:short"],
+        [-2147483648, "xsd:int"],
+        [2147483647, "xsd:int"],
+        [-9223372036854775808, "xsd:long"],
+        [9223372036854775807, "xsd:long"],
+        [-9923372036854775808, "xsd:integer"],
+        [9923372036854775807, "xsd:integer"],
+        [-3.4028234663852886e+38, "xsd:float"],
+        [3.4028234663852886e+38, "xsd:float"],
+        [-1.1754943508222875e-38, "xsd:float"],
+        [1.1754943508222875e-38, "xsd:float"],
+        [0.0, "xsd:float"],
+        [1e39, "xsd:double"],
+        [-1e39, "xsd:double"],
+        [1e-39, "xsd:double"],
+        [-1e-39, "xsd:double"],
+        [True, "xsd:boolean"],
+        [False, "xsd:boolean"],
+        [datetime.now().isoformat(), "xsd:dateTime"],
+        [datetime.now(), "xsd:dateTime"],
+        ["value", "xsd:string"]
+
+    ])
+    def test_data_type_matching(self, value, expected):
+        """Tests that the `xsd` data type can be extracted."""
+
+        result = CR1000XDevice._get_xsd_type(value)
+
+        self.assertEqual(result, expected)
+
+    def test_error_if_xsd_type_not_valid(self):
+
+        with self.assertRaises(TypeError):
+            CR1000XDevice._get_xsd_type(None)
+            
 if __name__ == "__main__":
     unittest.main()