Now save works as all CharacterString values read from bacpypes3 are …

…now converted to str internally.

BAC0/__init__.py

@@ -36,7 +36,9 @@
 
     from .scripts.Lite import Lite as lite  # to maintain compatibility with old code
     from .scripts.Lite import Lite as connect  # to maintain compatibility with old code
-    from .scripts.Lite import Lite as start  # this would be the new preferred way to start a BAC0 app
+    from .scripts.Lite import (
+        Lite as start,
+    )  # this would be the new preferred way to start a BAC0 app
 
 except ImportError as error:
     print("=" * 80)

BAC0/core/devices/Device.py

@@ -534,8 +534,10 @@ async def _buildPointList(self):
             self.segmentation_supported = False
             await self.new_state(DeviceDisconnected)
 
-        self.properties.name = await self.properties.network.read(
-            f"{self.properties.address} device {self.properties.device_id} objectName"
+        self.properties.name = str(
+            await self.properties.network.read(
+                f"{self.properties.address} device {self.properties.device_id} objectName"
+            )
         )
         self.properties.vendor_id = await self.properties.network.read(
             "{} device {} vendorIdentifier".format(
@@ -815,7 +817,7 @@ async def update_description(self, value):
             value=value,
             prop_id="description",
         )
-        self.properties.description = await self.read_property("description")
+        self.properties.description = str(await self.read_property("description"))
 
     async def ping(self):
         try:

BAC0/core/devices/Points.py

@@ -126,11 +126,11 @@ def __init__(
         self.properties.history_size = history_size
 
         self.properties.device = device
-        self.properties.name = pointName
+        self.properties.name = str(pointName)
         self.properties.type = pointType
         self.properties.address = pointAddress
 
-        self.properties.description = description
+        self.properties.description = str(description)
         self.properties.units_state = units_state
         self.properties.simulated = (False, 0)
         self.properties.overridden = (False, 0)
@@ -667,13 +667,13 @@ async def subscribe_cov(self, confirmed: bool = False, lifetime: int = 900):
     async def cancel_cov(self):
         self.log(f"Canceling COV subscription for {self.properties.name}", level="info")
         process_identifer = self.cov_task.process_identifier
-        
+
         if process_identifer not in Point._running_cov_tasks:
             await self.response("COV subscription not found")
             return
         cov_subscription = Point._running_cov_tasks.pop(process_identifer)
         cov_subscription.stop()
-        #await cov_subscription.task
+        # await cov_subscription.task
 
     def update_description(self, value):
         asyncio.create_task(self._update_description(value=value))
@@ -690,7 +690,7 @@ async def _update_description(self, value):
             value=value,
             prop_id="description",
         )
-        self.properties.description = self.read_property("description")
+        self.properties.description = str(self.read_property("description"))
 
     def tag(self, tag_id, tag_value, lst=None):
         """
@@ -886,6 +886,7 @@ def __init__(
             units_state=units_state,
             history_size=history_size,
         )
+        self.properties.units_state = tuple(str(x) for x in units_state)
 
     def _trend(self, res):
         res = "1: active" if res == BinaryPV.active else "0: inactive"
@@ -1017,6 +1018,7 @@ def __init__(
             units_state=units_state,
             history_size=history_size,
         )
+        self.properties.units_state = [str(x) for x in units_state]
 
     def _trend(self, res):
         res = f"{res}: {self.get_state(res)}"
@@ -1492,7 +1494,6 @@ def __init__(
         self.process_identifier = Point._last_cov_identifier + 1
         Point._last_cov_identifier = self.process_identifier
 
-
         self.point = point
         self.lifetime = lifetime
         self.confirmed = confirmed
@@ -1515,7 +1516,9 @@ async def run(self):
                 while not self.cov_fini.is_set():
                     incoming: asyncio.Future = asyncio.ensure_future(scm.get_value())
                     done, pending = await asyncio.wait(
-                        [incoming,],
+                        [
+                            incoming,
+                        ],
                         return_when=asyncio.FIRST_COMPLETED,
                     )
                     for task in pending:
@@ -1528,9 +1531,7 @@ async def run(self):
                             level="info",
                         )
                         if property_identifier == PropertyIdentifier.presentValue:
-                            val = extract_value_from_primitive_data(
-                                property_value
-                            )
+                            val = extract_value_from_primitive_data(property_value)
                             self.point._trend(val)
                         elif property_identifier == PropertyIdentifier.statusFlags:
                             self.point.properties.status_flags = property_value
@@ -1544,7 +1545,8 @@ async def run(self):
 
     def stop(self):
         self.point.log(
-            f"Stopping COV subscription class for {self.point.properties.name}", level="debug"
+            f"Stopping COV subscription class for {self.point.properties.name}",
+            level="debug",
         )
         self.cov_fini.set()
         self.point.cov_registered = False

BAC0/core/devices/Trends.py

@@ -121,6 +121,7 @@ async def update_properties(self) -> None:
                     oid=str(self.properties.oid),
                 )
             )
+            self.properties.description = str(self.properties.description)
         except Exception as error:
             raise Exception(f"Problem reading trendLog informations: {error}")
 

BAC0/core/devices/legacy/Device.py

@@ -801,7 +801,7 @@ def update_description(self, value):
             value=value,
             prop_id="description",
         )
-        self.properties.description = self.read_property("description")
+        self.properties.description = str(self.read_property("description"))
 
     def ping(self):
         try:

BAC0/core/devices/local/factory.py

@@ -303,7 +303,7 @@ def _create(definition, **kwargs):
     return ObjectFactory.from_dict(_definition)
 
 
-def make_state_text(list_of_string:list[str]):
+def make_state_text(list_of_string: list[str]):
     _arr = ArrayOf(CharacterString)
     _lst = [CharacterString(each) for each in list_of_string]
     return _arr(_lst)
@@ -409,9 +409,9 @@ def multistate(**kwargs):
             # "numberOfStates": Unsigned(2),
         },
         "description": "No description",
-        #"presentValue": Unsigned(1),
-        #"is_commandable": False,
-        #"relinquishDefault": Unsigned(1),
+        # "presentValue": Unsigned(1),
+        # "is_commandable": False,
+        # "relinquishDefault": Unsigned(1),
     }
     return _create(definition, **kwargs)
 

BAC0/core/devices/mixins/read_mixin.py

@@ -313,7 +313,7 @@ def _find_propid_index(key):
                 elif obj_type == "multi":
                     presentValue = int(presentValue)
             try:
-                point_description = point_infos[_find_propid_index("description")]
+                point_description = str(point_infos[_find_propid_index("description")])
             except KeyError:
                 point_description = ""
             try:

BAC0/core/functions/Alias.py

@@ -57,7 +57,9 @@ def iam(self, address=None):
 
         _app.i_am(address=address)
 
-    async def whois_router_to_network(self, network=None, *, destination=None, timeout=3):
+    async def whois_router_to_network(
+        self, network=None, *, destination=None, timeout=3
+    ):
         """
         Send a Who-Is-Router-To-Network request. This request is used to discover routers
         on the network that can route messages to a specific network.
@@ -74,13 +76,17 @@ async def whois_router_to_network(self, network=None, *, destination=None, timeo
         _this_application: BAC0Application = self.this_application
         _app: Application = _this_application.app
         try:
-            network_numbers = await asyncio.wait_for(_app.nse.who_is_router_to_network(), timeout)
+            network_numbers = await asyncio.wait_for(
+                _app.nse.who_is_router_to_network(), timeout
+            )
             return network_numbers
         except asyncio.TimeoutError:
             # Handle the timeout error
-            self.log("Request timed out for whois_router_to_network, no response", level='warning')
+            self.log(
+                "Request timed out for whois_router_to_network, no response",
+                level="warning",
+            )
             return []
-
 
     async def init_routing_table(self, address):
         """
@@ -107,11 +113,16 @@ async def what_is_network_number(self, destination=None, timeout=3):
         _this_application: BAC0Application = self.this_application
         _app: Application = _this_application.app
         try:
-            network_number = await asyncio.wait_for(_app.nse.what_is_network_number(), timeout)
+            network_number = await asyncio.wait_for(
+                _app.nse.what_is_network_number(), timeout
+            )
             return network_number
         except asyncio.TimeoutError:
             # Handle the timeout error
-            self.log("Request timed out for what_is_network_number, no response", level='warning')
+            self.log(
+                "Request timed out for what_is_network_number, no response",
+                level="warning",
+            )
             return None
 
     async def whohas(

BAC0/core/functions/GetIPAddr.py

@@ -19,6 +19,7 @@
 import re
 
 import struct
+
 # from bacpypes.pdu import Address as legacy_Address
 from bacpypes3.pdu import Address
 
@@ -136,7 +137,7 @@ def _old_findSubnetMask(self, ip: str) -> str:
                 "we'll consider 255.255.255.0 (/24).\nYou can install netifaces using 'pip install netifaces'."
             )
             return "255.255.255.0"
-        
+
     def _findSubnetMask(self, ip: str) -> str:
         """
         Retrieve the broadcast IP address connected to internet... used as
@@ -149,12 +150,12 @@ def _findSubnetMask(self, ip: str) -> str:
         if platform.system() == "Windows":
             try:
                 # Run the ipconfig command
-                result = subprocess.run(['ipconfig'], capture_output=True, text=True)
+                result = subprocess.run(["ipconfig"], capture_output=True, text=True)
                 output = result.stdout
 
                 # Regular expression to match IP and subnet mask
-                ip_pattern = re.compile(r'IPv4 Address[. ]*: ([\d.]+)')
-                mask_pattern = re.compile(r'Subnet Mask[. ]*: ([\d.]+)')
+                ip_pattern = re.compile(r"IPv4 Address[. ]*: ([\d.]+)")
+                mask_pattern = re.compile(r"Subnet Mask[. ]*: ([\d.]+)")
 
                 # Parse the output
                 lines = output.splitlines()
@@ -175,24 +176,27 @@ def _findSubnetMask(self, ip: str) -> str:
                 return None
         elif platform.system() == "Linux":
             import fcntl
+
             def get_interface_info(ifname):
                 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                 return fcntl.ioctl(
                     s.fileno(),
-                    0x891b,  # SIOCGIFNETMASK
-                    struct.pack('256s', ifname[:15].encode('utf-8'))
+                    0x891B,  # SIOCGIFNETMASK
+                    struct.pack("256s", ifname[:15].encode("utf-8")),
                 )
 
             try:
                 interfaces = socket.if_nameindex()
                 for ifindex, ifname in interfaces:
                     try:
                         netmask = socket.inet_ntoa(get_interface_info(ifname)[20:24])
-                        ip_address = socket.inet_ntoa(fcntl.ioctl(
-                            socket.socket(socket.AF_INET, socket.SOCK_DGRAM),
-                            0x8915,  # SIOCGIFADDR
-                            struct.pack('256s', ifname[:15].encode('utf-8'))
-                        )[20:24])
+                        ip_address = socket.inet_ntoa(
+                            fcntl.ioctl(
+                                socket.socket(socket.AF_INET, socket.SOCK_DGRAM),
+                                0x8915,  # SIOCGIFADDR
+                                struct.pack("256s", ifname[:15].encode("utf-8")),
+                            )[20:24]
+                        )
                         if ip is None or ip_address == ip:
                             return netmask
                     except IOError:
@@ -219,4 +223,3 @@ def validate_ip_address(ip: Address) -> bool:
     finally:
         s.close()
     return result
-

BAC0/core/functions/Reinitialize.py

@@ -23,6 +23,7 @@
 from ..io.IOExceptions import ApplicationNotStarted, NoResponseFromController
 from ..utils.notes import note_and_log
 from ...core.app.asyncApp import BAC0Application
+
 # --- standard Python modules ---
 
 
@@ -46,14 +47,13 @@ def reinitialize(self, address=None, password=None, state="coldstart"):
 
         # build a request
         request = ReinitializeDeviceRequest()
-        request.reinitializedStateOfDevice = (
-            getattr(ReinitializeDeviceRequestReinitializedStateOfDevice, state)
+        request.reinitializedStateOfDevice = getattr(
+            ReinitializeDeviceRequestReinitializedStateOfDevice, state
         )
         request.pduDestination = Address(address)
         request.password = CharacterString(password)
 
         self.log(f"{'- request:':>12} {request}", level="debug")
         _app.request(request)
-
 
         self.log(f"Reinitialize request sent to device : {address}", level="info")

BAC0/core/io/IOExceptions.py

@@ -172,5 +172,6 @@ class WrongParameter(Exception):
 class NumerousPingFailures(Exception):
     pass
 
+
 class NotReadyError(Exception):
-    pass
+    pass

BAC0/core/io/Write.py

@@ -50,7 +50,8 @@ def write()
 _debug = 0
 _LOG = ModuleLogger(globals())
 WRITE_REGEX = r"(?P<address>\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b|(\b\d+:\d+\b)) (?P<objId>(@obj_)?[-\w:]*[: ]*\d*) (?P<propId>(@prop_)?\w*(-\w*)?)[ ]?(?P<value>-*\w*)?[ ]?(?P<indx>-|\d*)?[ ]?(?P<priority>(1[0-6]|[0-9]))?"
-write_pattern = re.compile(WRITE_REGEX)        
+write_pattern = re.compile(WRITE_REGEX)
+
 
 @note_and_log
 class WriteProperty:

BAC0/core/proprietary_objects/__init__.py

@@ -1,11 +1,24 @@
-from bacpypes3.primitivedata import Atomic, Enumerated, Real, Unsigned, Boolean, CharacterString, TagList, TagClass, TagNumber, Tag
+from bacpypes3.primitivedata import (
+    Atomic,
+    Enumerated,
+    Real,
+    Unsigned,
+    Boolean,
+    CharacterString,
+    TagList,
+    TagClass,
+    TagNumber,
+    Tag,
+)
 from bacpypes3.errors import DecodingError, InvalidTag, PropertyError
 from typing import Any as _Any, Callable, Optional
 
+
 class OptionalUnsigned(Unsigned):
     """
     This is a special case where a vendor will send NULL values for Unsigned
     """
+
     @classmethod
     def decode(cls, tag_list: TagList) -> Optional[Unsigned]:
         """Decode an unsigned element from a tag list."""
@@ -20,7 +33,9 @@ def decode(cls, tag_list: TagList) -> Optional[Unsigned]:
                 if tag.tag_number == TagNumber.null:
                     return None
                 else:
-                    raise InvalidTag(f"unsigned application tag expected, got {tag.tag_number}")
+                    raise InvalidTag(
+                        f"unsigned application tag expected, got {tag.tag_number}"
+                    )
         elif tag.tag_class == TagClass.context:
             if cls._context is None:
                 raise InvalidTag("unsigned application tag expected")
@@ -37,4 +52,4 @@ def decode(cls, tag_list: TagList) -> Optional[Unsigned]:
             value = (value << 8) + c
 
         # return an instance of this thing
-        return cls(value)
+        return cls(value)