Ability to request RCC 5012 User Level

example_api_use.py

@@ -13,6 +13,7 @@ async def main():
     dataset = await XcomDataset.create(VOLTAGE.AC240) # or use VOLTAGE.AC120
     info_3023 = dataset.getByNr(3023, "xt")  # the "xt" part is optional but usefull for detecting mistakes
     info_6001 = dataset.getByNr(6001, "bsp")
+    param_5012 = dataset.getByNr(5012, "rcc")
     param_1107 = dataset.getByNr(1107, "xt")
     dataset = None  # Release memory of the dataset
 
@@ -30,11 +31,21 @@ async def main():
         value = await api.requestValue(info_6001, "BSP")    # bsp address range is only 601, or use "BSP"
         logger.info(f"BSP 6001: {value} {info_6001.unit} ({info_6001.name})")
 
-        # Update param 1107 on the first Xtender (Maximum current of AC source)
+        # Retrieve param #5012 from RCC (User Level)
+        value = await api.requestValue(param_5012, "RCC")    # rcc address range is only 501, or use "RCC"
+        logger.info(f"RCC 5012: {value} {param_5012.unit} ({param_5012.name})")
+
+        # Retrieve and Update param 1107 on the first Xtender (Maximum current of AC source)
+        value = await api.requestValue(param_1107, "XT1")
+        logger.info(f"XT1 1107: {value} {param_1107.unit} ({param_1107.name})")
+
         value = 4.0    # 4 Ampere
         if await api.updateValue(param_1107, value, "XT1"):
             logger.info(f"XT1 1107 updated to {value} {param_1107.unit} ({param_1107.name})")
 
+    except Exception as e:
+        logger.info(f"Unexpected exception: {e}")
+
     finally:
         await api.stop()
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "aioxcom"
-version = "1.5.1"
+version = "1.6.0"
 authors = [
   { name="Anko Hanse", email="[email protected]" },
 ]

src/aioxcom/xcom_api.py

@@ -8,6 +8,7 @@
 
 from .xcom_const import (
     FORMAT,
+    LEVEL,
     OBJ_TYPE,
     SCOM_OBJ_TYPE,
     SCOM_SERVICE,
@@ -234,6 +235,10 @@ async def updateValue(self, parameter: XcomDatapoint, value, dstAddr = 100, retr
             _LOGGER.warn(f"Ignoring attempt to update readonly infos value {parameter}")
             return None
 
+        if parameter.level == LEVEL.VO:
+            _LOGGER.warn(f"Ignoring attempt to update view-only params value {parameter}")
+            return None
+
         if type(dstAddr) is str:
             dstAddr = XcomDeviceFamilies.getAddrByCode(dstAddr)
 

src/aioxcom/xcom_const.py

@@ -46,7 +46,7 @@ class LEVEL(IntEnum):
     EXPERT = 0x0020
     INST   = 0x0030 # Installer
     QSP    = 0x0040 # Qualified Service Person
-    VO     = 0xFFFF # View Only? Used for param 5012, marked as 'not supported'
+    VO     = 0xFFFF # View Only. Used for param 5012
 
     @staticmethod
     def from_str(s: str):

src/aioxcom/xcom_datapoints.py

@@ -88,7 +88,7 @@ def obj_type(self):
         if self.level in [LEVEL.INFO]:
             return OBJ_TYPE.INFO
 
-        if self.level in [LEVEL.BASIC, LEVEL.EXPERT, LEVEL.INST, LEVEL.QSP]:
+        if self.level in [LEVEL.VO, LEVEL.BASIC, LEVEL.EXPERT, LEVEL.INST, LEVEL.QSP]:
             return OBJ_TYPE.PARAMETER
 
         _LOGGER.debug(f"Unknown obj_type for datapoint {self.nr} with level {self.level} and format {self.format}")

src/aioxcom/xcom_datapoints_240v.json

@@ -587,7 +587,7 @@
     {   "fam": "rcc",   "lvl": "Basic",   "pnr": 5036,   "nr": 5039,   "name": "  Choice of the third language",   "unit": "",   "def": 4,   "min": 1,   "max": 128,   "fmt": "LONG ENUM",   "inc": "Only 1 bit",   "opt": {"1":"English", "2":"French", "4":"German", "8":"Spanish", "16":"Dutch", "32":"Latinoelinka", "64":"Italian", "128":"Slovak"} },
     {   "fam": "rcc",   "lvl": "Basic",   "pnr": 5036,   "nr": 5040,   "name": "  Choice of the fourth language",   "unit": "",   "def": 8,   "min": 1,   "max": 128,   "fmt": "LONG ENUM",   "inc": "Only 1 bit",   "opt": {"1":"English", "2":"French", "4":"German", "8":"Spanish", "16":"Dutch", "32":"Latinoelinka", "64":"Italian", "128":"Slovak"} },
     {   "fam": "rcc",   "lvl": "Basic",   "pnr": 50,     "nr": 5002,   "name": "Date",   "unit": "Seconds",   "def": 0,   "min": 0,   "max": 0,   "fmt": "INT32",   "inc": 1 },
-    {   "fam": "rcc",   "lvl": "V.O.",    "pnr": 50,     "nr": 5012,   "name": "User level",   "unit": "",   "def": 16,   "min": 0,   "max": 111,   "fmt": "Not supported",   "inc": "" },
+    {   "fam": "rcc",   "lvl": "V.O.",    "pnr": 50,     "nr": 5012,   "name": "User level",   "unit": "",   "def": 16,   "min": 0,   "max": 111,   "fmt": "INT32",   "inc": "" },
     {   "fam": "rcc",   "lvl": "Expert",  "pnr": 50,     "nr": 5019,   "name": "Force remote control to user BASIC level",   "unit": "",   "def": "S",   "min": "S",   "max": "S",   "fmt": "INT32",   "inc": "Signal" },
     {   "fam": "rcc",   "lvl": "Expert",  "pnr": 50,     "nr": 5057,   "name": "DATALOGGER",   "unit": "",   "def": "",   "min": "",   "max": "",   "fmt": "ONLY LEVEL",   "inc": "Menu" },
     {   "fam": "rcc",   "lvl": "Expert",  "pnr": 5057,   "nr": 5101,   "name": "  Datalogger enabled",   "unit": "",   "def": 1,   "min": 1,   "max": 4,   "fmt": "LONG ENUM",   "inc": "Only 1 bit",   "opt": {"1":"Automatic", "2":"Yes", "4":"No"} },

tests/test_api.py

@@ -99,7 +99,8 @@ async def test_connect(name, start_server, start_client, wait_server, exp_server
         ("request param ok",     1107, 100, None, 100, SCOM_SERVICE.READ, SCOM_OBJ_TYPE.PARAMETER, 1107, SCOM_QSP_ID.UNSAVED_VALUE, 0x02, XcomData.pack(1234.0, FORMAT.FLOAT), 1234.0, None),
         ("update param ok",      1107, 100, 4.0,  100, SCOM_SERVICE.WRITE, SCOM_OBJ_TYPE.PARAMETER, 1107, SCOM_QSP_ID.UNSAVED_VALUE, 0x02, b'', True, None),
         ("update param err",     1107, 100, 4.0,  100, SCOM_SERVICE.WRITE, SCOM_OBJ_TYPE.PARAMETER, 1107, SCOM_QSP_ID.UNSAVED_VALUE, 0x03, SCOM_ERROR_CODES.WRITE_PROPERTY_FAILED, None, XcomApiResponseIsError),
-        ("update param timeout", 1107, 100, 4.0,  100, SCOM_SERVICE.WRITE, SCOM_OBJ_TYPE.PARAMETER, 1107, SCOM_QSP_ID.UNSAVED_VALUE, 0x00, b'', None, XcomApiTimeoutException),
+        ("update param timeout", 1107, 100, 4.0,  100, SCOM_SERVICE.WRITE, SCOM_OBJ_TYPE.PARAMETER, 1107, SCOM_QSP_ID.UNSAVED_VALUE, 0x00, b'', True, XcomApiTimeoutException),
+        ("request param vo",     5012, 501, None, 501, SCOM_SERVICE.READ, SCOM_OBJ_TYPE.PARAMETER, 5012, SCOM_QSP_ID.UNSAVED_VALUE, 0x02, XcomData.pack(32, FORMAT.INT32), 32, None),
     ]
 )
 async def test_request(name, test_nr, test_dest, test_value_update, exp_dst_addr, exp_svc_id, exp_obj_type, exp_obj_id, exp_prop_id, rsp_flags, rsp_data, exp_value, exp_except, request):

tests/test_datapoints.py

@@ -43,6 +43,12 @@ async def test_nr():
     assert param.format == FORMAT.FLOAT
     assert param.obj_type == OBJ_TYPE.INFO
 
+    param = dataset.getByNr(5012, "rcc")
+    assert param.family_id == "rcc"
+    assert param.nr == 5012
+    assert param.format == FORMAT.INT32
+    assert param.obj_type == OBJ_TYPE.PARAMETER
+
     with pytest.raises(XcomDatapointUnknownException):
         param = dataset.getByNr(9999)