Merge branch 'master' into v2.0.0a6

src/sdc11073/definitions_base.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 if TYPE_CHECKING:
     from types import ModuleType
@@ -16,11 +16,12 @@
 class ProtocolsRegistry(type):
     """base class that has the only purpose to register classes that use this as metaclass."""
 
-    protocols = []
+    protocols: ClassVar[list[type[BaseDefinitions]]] = []
 
     def __new__(cls, name: str, *arg, **kwarg):
-        new_cls = super().__new__(cls, name, *arg, **kwarg)
+        new_cls: ProtocolsRegistry = super().__new__(cls, name, *arg, **kwarg)
         if name != 'BaseDefinitions':  # ignore the base class itself
+            new_cls: type[BaseDefinitions]
             cls.protocols.append(new_cls)
         return new_cls
 
@@ -100,7 +101,7 @@ class BaseDefinitions(metaclass=ProtocolsRegistry):
     MedicalDeviceType: QName = None  # a QName, needed for types_match method
     ActionsNamespace: str = None  # needed for wsdl generation
     PortTypeNamespace: str = None  # needed for wsdl generation
-    MedicalDeviceTypesFilter: list[QName] = None  # list of QNames that are used / expected in "types" of wsdiscovery
+    MedicalDeviceTypesFilter: tuple[QName] | None = None  # QNames that are used / expected in "types" of wsdiscovery
     Actions = None
     data_model: AbstractDataModel = None
 

src/sdc11073/definitions_sdc.py

@@ -64,6 +64,6 @@ class SDC_v1_Definitions(BaseDefinitions):  # pylint: disable=invalid-name
     MedicalDeviceType = ns_hlp.MDPWS.tag('MedicalDevice')
     ActionsNamespace = ns_hlp.SDC.namespace
     PortTypeNamespace = ns_hlp.SDC.namespace
-    MedicalDeviceTypesFilter = [DpwsDeviceType, MedicalDeviceType]
+    MedicalDeviceTypesFilter = (DpwsDeviceType, MedicalDeviceType)
     Actions = actions.Actions
     data_model = V1Model()

src/sdc11073/httpserver/compression.py

@@ -1,8 +1,9 @@
-"""Compression module for http. """
+"""Compression module for http."""
+import contextlib
 import zlib
 from abc import ABC, abstractmethod
 from collections import OrderedDict
-from typing import Type
+from typing import ClassVar
 
 try:
     import lz4.frame
@@ -30,13 +31,14 @@ def decompress_payload(payload):
 
 class CompressionHandler:
     """Compression handler.
-    Should be used by servers and clients that are supposed to handle compression
+    Should be used by servers and clients that are supposed to handle compression.
     """
-    available_encodings = []  # initial default
-    handlers = {}
+
+    available_encodings: ClassVar[list[str]] = []  # initial default
+    handlers: ClassVar[dict[str, type[AbstractDataCompressor]]] = {}
 
     @classmethod
-    def register_handler(cls, handler: Type[AbstractDataCompressor]):
+    def register_handler(cls, handler: type[AbstractDataCompressor]):
         for alg in handler.algorithms:
             if alg.lower() in cls.available_encodings:
                 raise ValueError(f'Algorithm {alg} already registered, class = {cls.__name__} ')
@@ -56,7 +58,7 @@ def compress_payload(cls, algorithm: str, payload: bytes):
         return cls.get_handler(algorithm).compress_payload(payload)
 
     @classmethod
-    def decompress_payload(cls, algorithm: str, payload:bytes):
+    def decompress_payload(cls, algorithm: str, payload: bytes):
         """Decompresses payload based on required algorithm.
         Raises CompressionException if algorithm is not supported.
 
@@ -68,12 +70,9 @@ def decompress_payload(cls, algorithm: str, payload:bytes):
 
     @classmethod
     def get_handler(cls, algorithm: str):
-        """
-
-        :param algorithm: one of strings provided by registered compression handlers
+        """:param algorithm: one of strings provided by registered compression handlers
         :return: AbstractDataCompressor implementation
         """
-
         handler = cls.handlers.get(algorithm.lower())
         if not handler:
             txt = f"{algorithm} compression is not supported. Only {cls.available_encodings} are supported."
@@ -82,14 +81,13 @@ def get_handler(cls, algorithm: str):
 
     @staticmethod
     def parse_header(header):
-        """
-        Examples of headers are:  Examples of its use are:
+        """Examples of headers are:  Examples of its use are:
 
-       Accept-Encoding: compress, gzip
-       Accept-Encoding:
-       Accept-Encoding: *
-       Accept-Encoding: compress;q=0.5, gzip;q=1.0
-       Accept-Encoding: gzip;q=1.0, identity; q=0.5, *;q=0
+        Accept-Encoding: compress, gzip
+        Accept-Encoding:
+        Accept-Encoding: *
+        Accept-Encoding: compress;q=0.5, gzip;q=1.0
+        Accept-Encoding: gzip;q=1.0, identity; q=0.5, *;q=0
 
         returns sorted list of compression algorithms by priority
         """
@@ -100,26 +98,24 @@ def parse_header(header):
             for alg in (x.split(";") for x in header.split(",")):
                 alg_name = alg[0].strip()
                 parsed_headers[alg_name] = 1  # default
-                try:
+                with contextlib.suppress(ValueError, IndexError):
                     parsed_headers[alg_name] = float(alg[1].split("=")[1])
-                except (ValueError, IndexError):
-                    pass
+
         return [pair[0] for pair in sorted(parsed_headers.items(), key=lambda kv: kv[1], reverse=True)]
 
 
 class GzipCompressionHandler(AbstractDataCompressor):
     algorithms = ('gzip',)
 
     @staticmethod
-    def compress_payload(payload:bytes):
+    def compress_payload(payload: bytes):
         if not isinstance(payload, bytes):
             raise TypeError(f'a bytes-like object is required, not "{payload.__class__.__name__}", payload={payload}')
         gzip_compress = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, 16 + zlib.MAX_WBITS)
-        data = gzip_compress.compress(payload) + gzip_compress.flush()
-        return data
+        return gzip_compress.compress(payload) + gzip_compress.flush()
 
     @staticmethod
-    def decompress_payload(payload:bytes):
+    def decompress_payload(payload: bytes):
         return zlib.decompress(payload, 16 + zlib.MAX_WBITS)
 
 
@@ -130,11 +126,11 @@ class Lz4CompressionHandler(AbstractDataCompressor):
     algorithms = ('x-lz4', 'lz4')
 
     @staticmethod
-    def compress_payload(payload:bytes):
+    def compress_payload(payload: bytes):
         return lz4.frame.compress(payload)
 
     @staticmethod
-    def decompress_payload(payload:bytes):
+    def decompress_payload(payload: bytes):
         return lz4.frame.decompress(payload)
 
 

src/sdc11073/mdib/consumermdibxtra.py

@@ -7,7 +7,7 @@
 from enum import IntEnum
 from statistics import mean, stdev
 from threading import Lock
-from typing import TYPE_CHECKING, Callable
+from typing import TYPE_CHECKING, Callable, ClassVar
 
 from sdc11073 import observableproperties as properties
 from sdc11073.exceptions import ApiUsageError
@@ -43,7 +43,7 @@ class DeterminationTimeWarner:
 
     ST_IN_RANGE = 0
     ST_OUT_OF_RANGE = 1
-    result_lookup = {
+    result_lookup: ClassVar[dict[tuple[int, int], tuple[_WarningState, bool]]] = {
         (ST_IN_RANGE, ST_IN_RANGE): (_WarningState.A_NO_LOG, False),
         (ST_IN_RANGE, ST_OUT_OF_RANGE): (_WarningState.A_OUT_OF_RANGE, False),
         (ST_OUT_OF_RANGE, ST_OUT_OF_RANGE): (_WarningState.A_STILL_OUT_OF_RANGE, True),

src/sdc11073/mdib/descriptorcontainers.py

@@ -3,7 +3,7 @@
 import inspect
 import sys
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Protocol
+from typing import TYPE_CHECKING, Any, Protocol, ClassVar
 
 from sdc11073 import observableproperties as properties
 from sdc11073.xml_types import ext_qnames as ext
@@ -124,7 +124,9 @@ class AbstractDescriptorContainer(ContainerBase):
     _props = ('Handle', 'DescriptorVersion', 'SafetyClassification', 'Extension', 'Type')
     _child_elements_order = (ext.Extension, pm_qnames.Type)  # child elements in BICEPS order
     STATE_QNAME = None
-    extension_class_lookup = {msg.Retrievability: pm_types.Retrievability}
+    extension_class_lookup: ClassVar[dict[etree_.QName, type[pm_types.PropertyBasedPMType]]] = {
+        msg.Retrievability: pm_types.Retrievability
+    }
 
     def __init__(self, handle: str | None, parent_handle: str | None):
         """Parent Handle can only be None for a Mds Descriptor. every other descriptor has a parent."""

src/sdc11073/mdib/statecontainers.py

@@ -154,7 +154,7 @@ class AllowedValuesType(pm_types.PropertyBasedPMType):  # pylint: disable=invali
     """Represents a list of values, in xml it is a list of pm.Value elements with one value as text."""
 
     Value: list[str] = x_struct.SubElementStringListProperty(pm.Value)
-    _props = ['Value']
+    _props = ('Value',)
 
     def is_empty(self) -> bool:
         """Return True if Value is empty."""

src/sdc11073/provider/porttypes/porttypebase.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections import namedtuple
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, Optional, ClassVar
 
 from lxml import etree as etree_
 
@@ -37,7 +37,7 @@ class DPWSPortTypeBase:
     port_type_name: Optional[etree_.QName] = None
     WSDLOperationBindings = ()  # overwrite in derived classes
     WSDLMessageDescriptions = ()  # overwrite in derived classes
-    additional_namespaces: List[PrefixNamespace] = []  # for special namespaces
+    additional_namespaces: ClassVar[list[PrefixNamespace]] = []  # for special namespaces
 
     def __init__(self, sdc_device, log_prefix: Optional[str] = None):
         """

src/sdc11073/provider/providerimpl.py

@@ -329,7 +329,7 @@ def publish(self):
         scopes = self._components.scopes_factory(self._mdib)
         x_addrs = self.get_xaddrs()
         self._wsdiscovery.publish_service(self.epr_urn,
-                                          self._mdib.sdc_definitions.MedicalDeviceTypesFilter,
+                                          list(self._mdib.sdc_definitions.MedicalDeviceTypesFilter),
                                           scopes,
                                           x_addrs)
 

src/sdc11073/pysoap/soapenvelope.py

@@ -109,30 +109,30 @@ class reasontext(ElementWithText):  # noqa: N801
     """Text with language attribute."""
 
     lang: str = struct.StringAttributeProperty(ns_hlp.XML.tag('lang'), default_py_value='en-US')
-    _props = ['lang']
+    _props = ('lang',)
 
 
 class faultreason(XMLTypeBase):  # noqa: N801
     """List of reasontext."""
 
     Text: list[reasontext] = struct.SubElementListProperty(ns_hlp.S12.tag('Text'), value_class=reasontext)
-    _props = ['Text']
+    _props = ('Text',)
 
 
 class subcode(XMLTypeBase):  # noqa: N801
     """Value is a QName."""
 
     Value: etree_.QName = struct.NodeTextQNameProperty(ns_hlp.S12.tag('Value'))
     # optional Subcode Element intentionally omitted, it is of type subcode => recursion, bad idea!
-    _props = ['Value']
+    _props = ('Value',)
 
 
 class faultcode(XMLTypeBase):  # noqa: N801
     """Code wit subcode."""
 
     Value = struct.NodeEnumQNameProperty(ns_hlp.S12.tag('Value'), faultcodeEnum)
     Subcode = struct.SubElementProperty(ns_hlp.S12.tag('Subcode'), value_class=subcode, is_optional=True)
-    _props = ['Value', 'Subcode']
+    _props = ('Value', 'Subcode')
 
 
 class Fault(MessageType):
@@ -147,8 +147,8 @@ class Fault(MessageType):
     Role = struct.AnyUriTextElement(ns_hlp.S12.tag('Role'), is_optional=True)
     # Schema says Detail is an "any" type. Here it is modelled as a string that becomes the text of the Detail node
     Detail = struct.NodeStringProperty(ns_hlp.S12.tag('Detail'), is_optional=True)
-    _props = ['Code', 'Reason', 'Node', 'Role', 'Detail']
-    additional_namespaces = [ns_hlp.XML, ns_hlp.WSE]
+    _props = ('Code', 'Reason', 'Node', 'Role', 'Detail')
+    additional_namespaces = (ns_hlp.XML, ns_hlp.WSE)
 
     def add_reason_text(self, text: str, lang: str = 'en-US'):
         """Add reason text to list."""

src/sdc11073/wsdiscovery/wsdimpl.py

@@ -102,10 +102,10 @@ def _is_scope_in_list(uri: str, match_by: str, srv_sc: wsd_types.ScopesType) ->
 
 
 def matches_filter(service: Service,
-                   types: list[QName] | None,
+                   types: Iterable[QName] | None,
                    scopes: wsd_types.ScopesType | None) -> bool:
     """Check if service matches the types and scopes."""
-    if types is not None and len(types) > 0:
+    if types is not None:
         for ttype in types:
             if not _is_type_in_list(ttype, service.types):
                 return False
@@ -117,7 +117,7 @@ def matches_filter(service: Service,
 
 
 def filter_services(services: Iterable[Service],
-                    types: list[QName] | None,
+                    types: Iterable[QName] | None,
                     scopes: wsd_types.ScopesType | None) -> list[Service]:
     """Filter services that match types and scopes."""
     return [service for service in services if matches_filter(service, types, scopes)]
@@ -183,7 +183,7 @@ def stop(self):
             self._server_started = False
 
     def search_services(self,
-                        types: list[QName] | None = None,
+                        types: Iterable[QName] | None = None,
                         scopes: wsd_types.ScopesType | None = None,
                         timeout: int | float | None = 5,
                         repeat_probe_interval: int | None = 3) -> list[Service]:
@@ -198,6 +198,7 @@ def search_services(self,
         if not self._server_started:
             raise RuntimeError("Server not started")
 
+        types = list(types) if types is not None else None
         start = time.monotonic()
         end = start + timeout
         now = time.monotonic()
@@ -545,7 +546,8 @@ def _send_probe_match(self, services: list[Service], relates_to: str, addr: str)
                                                                                 ns_map=nsh.partial_map(nsh.WSD)))
             self._networking_thread.add_unicast_message(created_message, addr[0], addr[1])
 
-    def _send_probe(self, types: list[QName] | None = None, scopes: wsd_types.ScopesType | None = None):
+    def _send_probe(self, types: Iterable[QName] | None = None, scopes: wsd_types.ScopesType | None = None):
+        types = list(types) if types is not None else None  # enforce iteration
         self._logger.debug('sending probe types=%r scopes=%r', types_info(types), scopes)
         payload = wsd_types.ProbeType()
         payload.Types = types

src/sdc11073/xml_types/addressing_types.py

@@ -22,21 +22,21 @@ class EndpointReferenceType(XMLTypeBase):
     Address = struct.NodeStringProperty(nsh.WSA.tag('Address'))
     ReferenceParameters = struct.AnyEtreeNodeListProperty(nsh.WSA.tag('ReferenceParameters'), is_optional=True)
     PortType = struct.NodeTextQNameProperty(nsh.WSA.tag('PortType'), is_optional=True)
-    _props = ['Address', 'ReferenceParameters', 'PortType']
+    _props = ('Address', 'ReferenceParameters', 'PortType')
 
 
 class Relationship(ElementWithText):
     """Relationship type of ws-addressing."""
 
     RelationshipType = struct.QNameAttributeProperty('RelationshipType')
-    _props = ['RelationshipType']
+    _props = ('RelationshipType',)
 
 
 class MustUnderStandTextElement(ElementWithText):
     """XML Element with text and mustUnderstand attribute."""
 
     _must_understand = struct.BooleanAttributeProperty(nsh.S12.tag('mustUnderstand'), default_py_value=True)
-    _props = ['_must_understand']
+    _props = ('_must_understand',)
 
     def __init__(self, text: str | None = None):
         super().__init__()
@@ -59,7 +59,7 @@ class HeaderInformationBlock(XMLTypeBase):
     From = struct.SubElementProperty(nsh.WSA.tag('From'),
                                      value_class=EndpointReferenceType,
                                      is_optional=True)
-    _props = ['MessageID', 'RelatesTo', 'To', 'Action', 'From']
+    _props = ('MessageID', 'RelatesTo', 'To', 'Action', 'From')
 
     def __init__(self, action: str | None = None,
                  message_id: str | None = None,