Merge branch 'master' into fix/ruff-gh-action

src/sdc11073/consumer/consumerimpl.py

@@ -12,7 +12,7 @@
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urlparse
 
-from lxml import etree as etree_
+from lxml import etree
 
 import sdc11073.certloader
 from sdc11073 import commlog, loghelper, network, xml_utils
@@ -100,7 +100,7 @@ def _read_wsdl(self, soap_client: SoapClientProtocol, wsdl_url: str):
                 encoding = 'UTF-8'
             self.wsdl_string = self.wsdl_bytes.decode(encoding)
             logging.getLogger(commlog.WSDL).debug(self.wsdl_string)
-        except etree_.XMLSyntaxError as ex:
+        except etree.XMLSyntaxError as ex:
             self._logger.error(  # noqa: PLE1205
                 'could not read wsdl from {}: error={}, data=\n{}', actual_path, ex, self.wsdl_bytes)
 

src/sdc11073/consumer/serviceclients/contextservice.py

@@ -14,7 +14,7 @@
 if TYPE_CHECKING:
     from concurrent.futures import Future
 
-    from lxml.etree import QName
+    from lxml import etree
 
     from sdc11073.consumer.manipulator import RequestManipulatorProtocol
     from sdc11073.mdib.statecontainers import AbstractMultiStateProtocol
@@ -92,7 +92,7 @@ def get_context_states(self, handles: list[str] | None = None,
         return GetRequestResult(received_message_data, report)
 
     def get_context_state_by_identification(self, identifications: list[InstanceIdentifier],
-                                            context_type: QName | None = None,
+                                            context_type: etree.QName | None = None,
                                             request_manipulator: RequestManipulatorProtocol | None = None) \
             -> GetRequestResult:
         """Send a GetContextStatesByIdentification request.

src/sdc11073/consumer/serviceclients/serviceclientbase.py

@@ -10,7 +10,7 @@
 if TYPE_CHECKING:
     from concurrent.futures import Future
 
-    from lxml.etree import QName
+    from lxml import etree
 
     from sdc11073.consumer.consumerimpl import SdcConsumer
     from sdc11073.consumer.manipulator import RequestManipulatorProtocol
@@ -55,7 +55,7 @@ def action(self) -> str | None:
         return self._received_message.action
 
     @property
-    def msg_name(self) -> QName:
+    def msg_name(self) -> etree.QName:
         """Return the QName of message body."""
         return self._received_message.q_name.localname
 
@@ -76,7 +76,7 @@ class HostedServiceClient:
     def __init__(self, sdc_consumer: SdcConsumer,
                  soap_client: SoapClientProtocol,
                  dpws_hosted: HostedServiceType,
-                 port_type: QName):
+                 port_type: etree.QName):
         """Construct a HostedServiceClient.
 
         :param sdc_consumer:

src/sdc11073/consumer/subscription.py

@@ -8,7 +8,7 @@
 from typing import TYPE_CHECKING, Callable, Protocol
 from urllib.parse import urlparse
 
-from lxml import etree as etree_
+from lxml import etree
 
 from sdc11073 import loghelper
 from sdc11073 import observableproperties as properties
@@ -64,7 +64,7 @@ class ConsumerSubscription:
 
     notification_msg = properties.ObservableProperty()
     notification_data = properties.ObservableProperty()
-    IDENT_TAG = etree_.QName('http.local.com', 'MyClIdentifier')
+    IDENT_TAG = etree.QName('http.local.com', 'MyClIdentifier')
     is_subscribed = properties.ObservableProperty(False)
 
     def __init__(self, msg_factory: MessageFactory,  # noqa: PLR0913
@@ -198,7 +198,7 @@ def renew(self, expires: int = 3600) -> float:
                     return self.granted_expires
                 self.is_subscribed = False
                 self._logger.warning('renew failed: {}',  # noqa: PLE1205
-                                     etree_.tostring(message_data.p_msg.body_node, pretty_print=True))
+                                     etree.tostring(message_data.p_msg.body_node, pretty_print=True))
             return 0.0
 
     def unsubscribe(self):
@@ -518,9 +518,9 @@ def mk_subscription(self, dpws_hosted: HostedServiceType, filter_type: FilterTyp
                                             dpws_hosted, filter_type,
                                             self._notification_url,
                                             self._end_to_url, self.log_prefix)
-        subscription.notify_to_identifier = etree_.Element(ConsumerSubscription.IDENT_TAG)
+        subscription.notify_to_identifier = etree.Element(ConsumerSubscription.IDENT_TAG)
         subscription.notify_to_identifier.text = uuid.uuid4().urn
-        subscription.end_to_identifier = etree_.Element(ConsumerSubscription.IDENT_TAG)
+        subscription.end_to_identifier = etree.Element(ConsumerSubscription.IDENT_TAG)
         subscription.end_to_identifier.text = uuid.uuid4().urn
 
         filter_ = filter_type.text

src/sdc11073/definitions_base.py

@@ -6,7 +6,7 @@
 if TYPE_CHECKING:
     from types import ModuleType
 
-    from lxml.etree import QName
+    from lxml import etree
 
     from .mdib.descriptorcontainers import AbstractDescriptorProtocol
     from .mdib.statecontainers import AbstractStateProtocol
@@ -30,10 +30,10 @@ class AbstractDataModel(ABC):
     """Abstract base class for DataModelProtocol implementation."""
 
     @abstractmethod
-    def get_descriptor_container_class(self, type_qname: QName) -> type[AbstractDescriptorProtocol]:
+    def get_descriptor_container_class(self, type_qname: etree.QName) -> type[AbstractDescriptorProtocol]:
         """Get the class that represents a BICEPS descriptor entity with given QName."""
 
-    def mk_descriptor_container(self, type_qname: QName, handle: str, parent_descriptor: Any) -> Any:
+    def mk_descriptor_container(self, type_qname: etree.QName, handle: str, parent_descriptor: Any) -> Any:
         """Create an instance that represents a BICEPS entity with given QName."""
         cls = self.get_descriptor_container_class(type_qname)
         if parent_descriptor is not None:
@@ -44,7 +44,7 @@ def mk_descriptor_container(self, type_qname: QName, handle: str, parent_descrip
         return ret
 
     @abstractmethod
-    def get_state_container_class(self, type_qname: QName) -> type[AbstractStateProtocol]:
+    def get_state_container_class(self, type_qname: etree.QName) -> type[AbstractStateProtocol]:
         """Get the class that represents a BICEPS state entity with given QName."""
 
     def get_state_class_for_descriptor(
@@ -98,14 +98,14 @@ class BaseDefinitions(metaclass=ProtocolsRegistry):
     """
 
     # set the following values in derived classes:
-    MedicalDeviceType: QName = None  # a QName, needed for types_match method
+    MedicalDeviceType: etree.QName = None  # a QName, needed for types_match method
     ActionsNamespace: str = None  # needed for wsdl generation
     PortTypeNamespace: str = None  # needed for wsdl generation
-    MedicalDeviceTypesFilter: tuple[QName] | None = None  # QNames that are used / expected in "types" of wsdiscovery
+    MedicalDeviceTypesFilter: tuple[etree.QName] | None = None  # QNames that are used / expected in "types" of wsdiscovery
     Actions = None
     data_model: AbstractDataModel = None
 
     @classmethod
-    def types_match(cls, types: list[QName]) -> bool:
+    def types_match(cls, types: list[etree.QName]) -> bool:
         """Check if this definition can be used for the provided types."""
         return cls.MedicalDeviceType in types

src/sdc11073/definitions_sdc.py

@@ -11,7 +11,7 @@
 if TYPE_CHECKING:
     from types import ModuleType
 
-    from lxml.etree import QName
+    from lxml import etree
 
     from .namespaces import NamespaceHelper
 
@@ -23,11 +23,11 @@ def __init__(self):
         super().__init__()
         self._ns_hlp = ns_hlp
 
-    def get_descriptor_container_class(self, type_qname: QName) -> type:
+    def get_descriptor_container_class(self, type_qname: etree.QName) -> type:
         """Get the class that represents a BICEPS descriptor entity with given QName."""
         return get_descriptor_container_class(type_qname)
 
-    def get_state_container_class(self, type_qname: QName) -> type:
+    def get_state_container_class(self, type_qname: etree.QName) -> type:
         """Get the class that represents a BICEPS state entity with given QName."""
         return get_state_container_class(type_qname)
 

src/sdc11073/dispatch/dispatchkey.py

@@ -11,7 +11,7 @@
 from sdc11073.pysoap.soapenvelope import Fault, faultcodeEnum
 
 if TYPE_CHECKING:
-    from lxml.etree import QName
+    from lxml import etree
 
 
 @dataclass(frozen=True)
@@ -22,7 +22,7 @@ class DispatchKey:
     """
 
     action: str
-    message_tag: QName | None
+    message_tag: etree.QName | None
 
     def __repr__(self) -> str:
         return f'{self.__class__.__name__} action={self.action} msg={self.message_tag}'

src/sdc11073/mdib/containerbase.py

@@ -4,7 +4,7 @@
 import inspect
 from typing import Any
 import math
-from lxml.etree import Element, SubElement, QName
+from lxml import etree
 
 from sdc11073 import observableproperties as properties
 from sdc11073.namespaces import QN_TYPE, NamespaceHelper
@@ -14,7 +14,7 @@
 class ContainerBase:
     """Common base class for descriptors and states."""
 
-    NODETYPE: QName = None  # overwrite in derived classes! This is the BICEPS Type.
+    NODETYPE: etree.QName = None  # overwrite in derived classes! This is the BICEPS Type.
     node = properties.ObservableProperty()
     is_state_container = False
     is_descriptor_container = False
@@ -35,7 +35,7 @@ def get_actual_value(self, attr_name: str) -> Any:
         return getattr(self.__class__, attr_name).get_actual_value(self)
 
     def mk_node(self,
-                tag: QName,
+                tag: etree.QName,
                 ns_helper: NamespaceHelper,
                 parent_node: xml_utils.LxmlElement | None = None,
                 set_xsi_type: bool = False) -> xml_utils.LxmlElement:
@@ -51,9 +51,9 @@ def mk_node(self,
                                        ns_helper.MSG,
                                        ns_helper.XSI)
         if parent_node is not None:
-            node = SubElement(parent_node, tag, nsmap=ns_map)
+            node = etree.SubElement(parent_node, tag, nsmap=ns_map)
         else:
-            node = Element(tag, nsmap=ns_map)
+            node = etree.Element(tag, nsmap=ns_map)
 
         self.update_node(node, ns_helper, set_xsi_type)
         return node

src/sdc11073/mdib/descriptorcontainers.py

@@ -17,10 +17,9 @@
     from collections.abc import Iterable
     from decimal import Decimal
 
-    from lxml import etree as etree_
+    from lxml import etree
 
     from sdc11073 import xml_utils
-    from sdc11073.namespaces import NamespaceHelper
     from sdc11073.xml_types.isoduration import DurationType
     from sdc11073.xml_types.xml_structure import ExtensionLocalValue
 
@@ -34,8 +33,8 @@ class ChildDescriptorMapping:
     pm.Metric for all classes derived from AbstractMetricDescriptor.
     """
 
-    child_qname: etree_.QName
-    node_types: tuple[etree_.QName, ...] = None
+    child_qname: etree.QName
+    node_types: tuple[etree.QName, ...] = None
 
     def __repr__(self) -> str:
         if self.node_types is None:
@@ -59,8 +58,8 @@ def sorted_child_data(obj: Any, member_name: str):
 class AbstractDescriptorProtocol(Protocol):
     """The common Interface of all descriptors."""
 
-    NODETYPE: etree_.QName
-    STATE_QNAME: etree_.QName
+    NODETYPE: etree.QName
+    STATE_QNAME: etree.QName
     is_descriptor_container: bool
     is_system_context_descriptor: bool
     is_realtime_sample_array_metric_descriptor: bool
@@ -125,7 +124,7 @@ 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: ClassVar[dict[etree_.QName, type[pm_types.PropertyBasedPMType]]] = {
+    extension_class_lookup: ClassVar[dict[etree.QName, type[pm_types.PropertyBasedPMType]]] = {
         msg.Retrievability: pm_types.Retrievability
     }
 
@@ -205,7 +204,7 @@ def diff(self, other: AbstractDescriptorContainer, ignore_property_names: list[s
                     ret.append(f'parent_handle={my_value}, other={other_value}')
         return None if len(ret) == 0 else ret
 
-    def tag_name_for_child_descriptor(self, node_type: etree_.QName) -> (etree_.QName, bool):
+    def tag_name_for_child_descriptor(self, node_type: etree.QName) -> (etree.QName, bool):
         """Determine the tag name of a child descriptor.
 
         This isneeded when the xml tree of the descriptor is created.
@@ -796,6 +795,6 @@ class EnsembleContextDescriptorContainer(AbstractContextDescriptorContainer):
 _name_class_lookup.update(_name_class_xtra_lookup)
 
 
-def get_container_class(qname: etree_.QName) -> type[AbstractDescriptorContainer]:
+def get_container_class(qname: etree.QName) -> type[AbstractDescriptorContainer]:
     """:param qname: a QName instance"""
     return _name_class_lookup.get(qname)