Draft : Started to work on ReadRangeService

bacpypes3/app.py

@@ -4,94 +4,75 @@
 
 from __future__ import annotations
 
-import asyncio
 import argparse
+import asyncio
 import dataclasses
-
 from functools import partial
+from typing import TYPE_CHECKING
+from typing import Any as _Any
+from typing import Callable, Dict, List, Optional, Set, Tuple, Union, cast
 
-from typing import (
-    TYPE_CHECKING,
-    cast,
-    Any as _Any,
-    Callable,
-    Dict,
-    List,
-    Optional,
-    Tuple,
-    Set,
-    Union,
-)
-
-from .debugging import bacpypes_debugging, DebugContents, ModuleLogger
-
-from .comm import ApplicationServiceElement, bind
-from .pdu import Address
-
+# for computing protocol services supported
 from .apdu import (
     APDU,
-    UnconfirmedRequestPDU,
-    ConfirmedRequestPDU,
-    SimpleAckPDU,
+    AbortPDU,
     ComplexAckPDU,
+    ConfirmedRequestPDU,
+    ConfirmedServiceChoice,
+    Error,
     ErrorPDU,
+    IAmRequest,
     RejectPDU,
-    AbortPDU,
-    Error,
-)
-
-from .errors import ExecutionError, UnrecognizedService, AbortException, RejectException
-
-# for computing protocol services supported
-from .apdu import (
+    SimpleAckPDU,
+    UnconfirmedRequestPDU,
     confirmed_request_types,
     unconfirmed_request_types,
-    IAmRequest,
 )
-from .primitivedata import ObjectType, ObjectIdentifier
+from .appservice import ApplicationServiceAccessPoint
 from .basetypes import (
+    BDTEntry,
+    HostNPort,
+    IPMode,
+    NetworkType,
+    ProtocolLevel,
     Segmentation,
     ServicesSupported,
-    ProtocolLevel,
-    NetworkType,
-    IPMode,
-    HostNPort,
-    BDTEntry,
 )
-from .object import Object, DeviceObject
-from .vendor import get_vendor_info
+from .comm import ApplicationServiceElement, bind
+from .debugging import DebugContents, ModuleLogger, bacpypes_debugging
+from .errors import AbortException, ExecutionError, RejectException, UnrecognizedService
+from .ipv4.link import BBMDLinkLayer as BBMDLinkLayer_ipv4
+from .ipv4.link import ForeignLinkLayer as ForeignLinkLayer_ipv4
+from .ipv4.link import NormalLinkLayer as NormalLinkLayer_ipv4
 
-from .appservice import ApplicationServiceAccessPoint
+# for serialized parameter initialization
+from .json import json_to_sequence
+
+# network port interpretation
+from .local.networkport import NetworkPortObject
+
+# objects with specialized interpreters
+from .local.schedule import ScheduleObject
 from .netservice import (
     NetworkServiceAccessPoint,
     NetworkServiceElement,
     RouterInfoCache,
 )
+from .object import DeviceObject, Object
+from .pdu import Address
+from .primitivedata import ObjectIdentifier, ObjectType
+from .service.cov import ChangeOfValueServices
 
 # basic services
-from .service.device import WhoIsIAmServices, WhoHasIHaveServices
+from .service.device import WhoHasIHaveServices, WhoIsIAmServices
 from .service.object import (
     ReadRangeServices,
-    ReadWritePropertyServices,
     ReadWritePropertyMultipleServices,
+    ReadWritePropertyServices,
 )
-from .service.cov import ChangeOfValueServices
-
-# network port interpretation
-from .local.networkport import NetworkPortObject
-from .ipv4.link import (
-    NormalLinkLayer as NormalLinkLayer_ipv4,
-    ForeignLinkLayer as ForeignLinkLayer_ipv4,
-    BBMDLinkLayer as BBMDLinkLayer_ipv4,
-)
+from .vendor import get_vendor_info
 from .vlan.link import VirtualLinkLayer
 
-# objects with specialized interpreters
-from .local.schedule import ScheduleObject
-
-# for serialized parameter initialization
-from .json import json_to_sequence
-
 if TYPE_CHECKING:
     # class is declared as generic in stubs but not at runtime
     APDUFuture = asyncio.Future[Optional[APDU]]
@@ -758,13 +739,17 @@ def get_services_supported(self):
         for service_choice, service_request_class in confirmed_request_types.items():
             service_helper = "do_" + service_request_class.__name__
             if hasattr(self, service_helper):
-                services_supported[service_choice] = 1
+                _key = ConfirmedServiceChoice(service_choice).attr
+                choice = getattr(services_supported, _key)
+                services_supported[choice] = 1
 
         # look through the unconfirmed services
         for service_choice, service_request_class in unconfirmed_request_types.items():
             service_helper = "do_" + service_request_class.__name__
             if hasattr(self, service_helper):
-                services_supported[service_choice] = 1
+                _key = ConfirmedServiceChoice(service_choice).attr
+                choice = getattr(services_supported, _key)
+                services_supported[choice] = 1
 
         # return the bit list
         return services_supported

bacpypes3/service/object.py

@@ -35,7 +35,7 @@
 )
 from ..constructeddata import Any, Array, List, SequenceOf
 from ..debugging import ModuleLogger, bacpypes_debugging
-from ..errors import ExecutionError, ObjectError, PropertyError
+from ..errors import ExecutionError, ObjectError, PropertyError, RejectException
 from ..object import DeviceObject
 from ..pdu import Address
 from ..primitivedata import Date, Null, ObjectIdentifier, Time, Unsigned
@@ -979,4 +979,100 @@ async def do_ReadRangeRequest(self, apdu: ReadRangeRequest) -> None:
         if _debug:
             ReadRangeServices._debug("do_ReadRangeRequest %r", apdu)
 
-        raise NotImplementedError
+        raise NotImplementedError()
+
+    """
+    TODO : Make that better
+        # extract the object identifier
+        objId = apdu.objectIdentifier
+
+        # get the object
+        obj = self.get_object_id(objId)
+        if not obj:
+            raise ExecutionError(errorClass="object", errorCode="unknownObject")
+
+        if _debug:
+            ReadRangeServices._debug("    - object: %r", obj)
+
+        # get the datatype
+        datatype = obj.get_property_type(apdu.propertyIdentifier)
+        if _debug:
+            ReadRangeServices._debug("    - datatype: %r", datatype)
+
+        # must be a list, or an array of lists
+        if issubclass(datatype, List):
+            pass
+        elif (
+            (apdu.propertyArrayIndex is not None)
+            and issubclass(datatype, Array)
+            and issubclass(datatype.subtype, List)
+        ):
+            pass
+        else:
+            raise ExecutionError(errorClass="property", errorCode="propertyIsNotAList")
+
+        # get the value
+        if _debug:
+            ReadRangeServices._debug(apdu.__dict__)
+        try:
+            value = await obj.read_property(
+                apdu.propertyIdentifier, apdu.propertyArrayIndex
+            )
+
+            if _debug:
+                ReadRangeServices._debug(
+                    f"    - value: {value.__repr__()} | of type {type(value)}"
+                )
+            if value is None:
+                raise PropertyError(errorCode="unknownProperty")
+            if isinstance(value, List):
+                ReadRangeServices._debug(
+                    "    - value is a list of: %r", datatype.subtype
+                )
+                # datatype = datatype.subtype
+
+            if apdu.range.byPosition:
+                range_by_position = apdu.range.byPosition
+                if _debug:
+                    ReadRangeServices._debug(
+                        "    - range_by_position: %r", range_by_position
+                    )
+
+            elif apdu.range.bySequenceNumber:
+                range_by_sequence_number = apdu.range.bySequenceNumber
+                if _debug:
+                    ReadRangeServices._debug(
+                        "    - range_by_sequence_number: %r", range_by_sequence_number
+                    )
+
+            elif apdu.range.byTime:
+                range_by_time = apdu.range.byTime
+                if _debug:
+                    ReadRangeServices._debug("    - range_by_time: %r", range_by_time)
+
+            else:
+                raise RejectException("missingRequiredParameter")
+        except AttributeError:
+            # exception if this is not a defined property
+            raise PropertyError(errorCode="unknownProperty")
+        # this is an ack
+        resp = ReadRangeACK(context=apdu)
+        resp.objectIdentifier = objId
+        resp.propertyIdentifier = apdu.propertyIdentifier
+        resp.propertyArrayIndex = apdu.propertyArrayIndex
+
+        resp.resultFlags = [1, 1, 0]
+        resp.itemCount = len(value)
+
+        # save the result in the item data
+        item_data = SequenceOf(Any)()
+        item_data.cast_in(value)
+        resp.itemData = item_data
+        if _debug:
+            ReadRangeServices._debug("    - itemData : %r", resp.itemData)
+            ReadRangeServices._debug("    - resp: %r", resp)
+        self.response(resp)
+
+        # return the result
+        await self.response(resp)
+"""