fix lint errors and mypy warnings

asyncua/client/ua_client.py

@@ -456,7 +456,7 @@ async def register_server(self, registered_server):
         self.logger.debug(response)
         response.ResponseHeader.ServiceResult.check()
         # nothing to return for this service
-    
+
     async def unregister_server(self, registered_server):
         self.logger.debug("unregister_server")
         request = ua.RegisterServerRequest()
@@ -552,7 +552,7 @@ async def update_subscription(
         )
         return response.Parameters
 
-    modify_subscription = update_subscription # legacy support
+    modify_subscription = update_subscription  # legacy support
 
     async def delete_subscriptions(self, subscription_ids):
         self.logger.debug("delete_subscriptions %r", subscription_ids)

asyncua/common/copy_node_util.py

@@ -75,5 +75,5 @@ async def _read_and_copy_attrs(node_type, struct, addnode):
                 setattr(struct, name, results[idx].Value.Value)
         else:
             _logger.warning(f"Instantiate: while copying attributes from node type {str(node_type)},"
-                           f" attribute {str(name)}, statuscode is {str(results[idx].StatusCode)}")
+                            f" attribute {str(name)}, statuscode is {str(results[idx].StatusCode)}")
     addnode.NodeAttributes = struct

asyncua/common/event_objects.py

@@ -15,7 +15,6 @@
 __all__ = ["BaseEvent", "AuditEvent", "AuditSecurityEvent", "AuditChannelEvent", "AuditOpenSecureChannelEvent", "AuditSessionEvent", "AuditCreateSessionEvent", "AuditActivateSessionEvent", "AuditCancelEvent", "AuditCertificateEvent", "AuditCertificateDataMismatchEvent", "AuditCertificateExpiredEvent", "AuditCertificateInvalidEvent", "AuditCertificateUntrustedEvent", "AuditCertificateRevokedEvent", "AuditCertificateMismatchEvent", "AuditNodeManagementEvent", "AuditAddNodesEvent", "AuditDeleteNodesEvent", "AuditAddReferencesEvent", "AuditDeleteReferencesEvent", "AuditUpdateEvent", "AuditWriteUpdateEvent", "AuditHistoryUpdateEvent", "AuditUpdateMethodEvent", "SystemEvent", "DeviceFailureEvent", "BaseModelChangeEvent", "GeneralModelChangeEvent", "TransitionEvent", "AuditUpdateStateEvent", "ProgramTransitionEvent", "SemanticChangeEvent", "AuditUrlMismatchEvent", "Condition", "RefreshStartEvent", "RefreshEndEvent", "RefreshRequiredEvent", "AuditConditionEvent", "AuditConditionEnableEvent", "AuditConditionCommentEvent", "DialogCondition", "AcknowledgeableCondition", "AlarmCondition", "LimitAlarm", "AuditHistoryEventUpdateEvent", "AuditHistoryValueUpdateEvent", "AuditHistoryDeleteEvent", "AuditHistoryRawModifyDeleteEvent", "AuditHistoryAtTimeDeleteEvent", "AuditHistoryEventDeleteEvent", "EventQueueOverflowEvent", "ProgramTransitionAuditEvent", "AuditConditionRespondEvent", "AuditConditionAcknowledgeEvent", "AuditConditionConfirmEvent", "ExclusiveLimitAlarm", "ExclusiveLevelAlarm", "ExclusiveRateOfChangeAlarm", "ExclusiveDeviationAlarm", "NonExclusiveLimitAlarm", "NonExclusiveLevelAlarm", "NonExclusiveRateOfChangeAlarm", "NonExclusiveDeviationAlarm", "DiscreteAlarm", "OffNormalAlarm", "TripAlarm", "AuditConditionShelvingEvent", "ProgressEvent", "SystemStatusChangeEvent", "SystemOffNormalAlarm", "AuditProgramTransitionEvent", "TrustListUpdatedAuditEvent", "CertificateUpdatedAuditEvent", "CertificateExpirationAlarm", "AuditConditionResetEvent", "PubSubStatusEvent", "PubSubTransportLimitsExceedEvent", "PubSubCommunicationFailureEvent", "DiscrepancyAlarm", "AuditConditionSuppressionEvent", "AuditConditionSilenceEvent", "AuditConditionOutOfServiceEvent", "RoleMappingRuleChangedAuditEvent", "KeyCredentialAuditEvent", "KeyCredentialUpdatedAuditEvent", "KeyCredentialDeletedAuditEvent", "InstrumentDiagnosticAlarm", "SystemDiagnosticAlarm", "AuditHistoryAnnotationUpdateEvent", "TrustListOutOfDateAlarm", "AuditClientEvent", "AuditClientUpdateMethodResultEvent"]
 
 
-
 class BaseEvent(Event):
     """
     BaseEvent:
@@ -1197,4 +1196,4 @@ def __init__(self, sourcenode=None, message=None, severity=1):
     ua.ObjectIds.TrustListOutOfDateAlarmType: TrustListOutOfDateAlarm,
     ua.ObjectIds.AuditClientEventType: AuditClientEvent,
     ua.ObjectIds.AuditClientUpdateMethodResultEventType: AuditClientUpdateMethodResultEvent,
-    }
+}

asyncua/common/ua_utils.py

@@ -7,7 +7,7 @@
 from datetime import datetime
 from enum import Enum, IntEnum, IntFlag
 
-from dateutil import parser
+from dateutil import parser  # type: ignore[attr-defined]
 
 from asyncua import ua
 
@@ -315,6 +315,7 @@ def data_type_to_string(dtype):
         string = dtype.to_string()
     return string
 
+
 def copy_dataclass_attr(dc_source, dc_dest) -> None:
     """
     Copy the common attributes of dc_source to dc_dest

asyncua/common/xmlimporter.py

@@ -14,9 +14,11 @@
 
 _logger = logging.getLogger(__name__)
 
+
 def _parse_version(version_string: str) -> List[int]:
     return [int(v) for v in version_string.split('.')]
 
+
 class XmlImporter:
 
     def __init__(self, server, strict_mode=True):
@@ -162,7 +164,7 @@ async def _add_missing_reverse_references(self, new_nodes):
                                   ua.ObjectIds.FiniteTransitionVariableType, ua.ObjectIds.HasInterface}
         dangling_refs_to_missing_nodes = set(new_nodes)
 
-        RefSpecKey = Tuple[ua.NodeId, ua.NodeId, ua.NodeId] # (source_node_id, target_node_id, ref_type_id)
+        RefSpecKey = Tuple[ua.NodeId, ua.NodeId, ua.NodeId]  # (source_node_id, target_node_id, ref_type_id)
         node_reference_map: Dict[RefSpecKey, ua.ReferenceDescription] = {}
 
         for new_node_id in new_nodes:
@@ -189,8 +191,12 @@ async def _add_missing_reverse_references(self, new_nodes):
 
                 _logger.debug("Adding missing reference: %s <-> %s (%s)", target_node_id, source_node_id, ref.ReferenceTypeId)
 
-                new_ref = ua.AddReferencesItem(SourceNodeId=target_node_id, TargetNodeId=source_node_id,
-                    ReferenceTypeId=ref_type, IsForward=(not ref.IsForward))
+                new_ref = ua.AddReferencesItem(
+                    SourceNodeId=target_node_id,
+                    TargetNodeId=source_node_id,
+                    ReferenceTypeId=ref_type,
+                    IsForward=(not ref.IsForward)
+                )
                 reference_fixes.append(new_ref)
         await self._add_references(reference_fixes)
 

asyncua/common/xmlparser.py

@@ -459,7 +459,7 @@ def get_nodeset_namespaces(self) -> List[Tuple[str, ua.String, ua.DateTime]]:
                 date_time = model.attrib.get('PublicationDate')
                 if date_time is None:
                     date_time = ua.DateTime(1, 1, 1)
-                elif date_time is not None and date_time[-1]=="Z":
+                elif date_time is not None and date_time[-1] == "Z":
                     date_time = ua.DateTime.strptime(date_time, "%Y-%m-%dT%H:%M:%SZ")
                 else:
                     date_time = ua.DateTime.strptime(date_time, "%Y-%m-%dT%H:%M:%S%z")

asyncua/crypto/truststore.py

@@ -57,7 +57,7 @@ async def load(self):
 
     async def load_trust(self):
         """(re)load the trusted certificates"""
-        self._trust_store: crypto.X509Store = crypto.X509Store()
+        self._trust_store = crypto.X509Store()
         for location in self._trust_locations:
             await self._load_trust_location(location)
 
@@ -113,7 +113,7 @@ def is_revoked(self, certificate: x509.Certificate) -> bool:
         for revoked in self._revoked_list:
             if revoked.serial_number == certificate.serial_number:
                 subject_cn = certificate.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)[0].value
-                _logger.warning('Found revoked serial "%s" [CN=%s]', hex(certificate.serial_number),  subject_cn)
+                _logger.warning('Found revoked serial "%s" [CN=%s]', hex(certificate.serial_number), subject_cn)
                 is_revoked = True
                 break
         return is_revoked

asyncua/crypto/uacrypto.py

@@ -85,6 +85,7 @@ def der_from_x509(certificate):
         return b""
     return certificate.public_bytes(serialization.Encoding.DER)
 
+
 def pem_from_key(private_key: rsa.RSAPrivateKey) -> bytes:
     """dumps a private key in PEM format
 
@@ -96,6 +97,7 @@ def pem_from_key(private_key: rsa.RSAPrivateKey) -> bytes:
     """
     return private_key.private_bytes(encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption())
 
+
 def sign_sha1(private_key, data):
     return private_key.sign(
         data,

asyncua/server/address_space.py

@@ -24,7 +24,7 @@
         VariableAttributes,
         ObjectTypeAttributes,
         ObjectAttributes
-    ] # FIXME Check, if there are missing attribute types.
+    ]  # FIXME Check, if there are missing attribute types.
 
 from asyncua import ua
 
@@ -203,8 +203,8 @@ def _translate_browsepath_to_nodeid(self, path: ua.BrowsePath) -> ua.BrowsePathR
             return res
         for nodeid in target_nodeids:
             target = ua.BrowsePathTarget()
-            target.TargetId = nodeid # FIXME <<<< Type conflict
-            target.RemainingPathIndex = ua.Index(4294967295) # FIXME: magic number, why not Index.MAX?
+            target.TargetId = nodeid  # FIXME <<<< Type conflict
+            target.RemainingPathIndex = ua.Index(4294967295)  # FIXME: magic number, why not Index.MAX?
             res.Targets.append(target)
         # FIXME: might need to order these one way or another
         return res
@@ -274,7 +274,7 @@ def _add_node(self, item: ua.AddNodesItem, user: User, check: bool = True) -> ua
             # the namespace of the nodeid, this is an extention of the spec to allow
             # to requests the server to generate a new nodeid in a specified namespace
             # self.logger.debug("RequestedNewNodeId has null identifier, generating Identifier")
-            item.RequestedNewNodeId = self._aspace.generate_nodeid(item.RequestedNewNodeId.NamespaceIndex) # FIXME type conflict
+            item.RequestedNewNodeId = self._aspace.generate_nodeid(item.RequestedNewNodeId.NamespaceIndex)  # FIXME type conflict
         else:
             if item.RequestedNewNodeId in self._aspace:
                 self.logger.warning("AddNodesItem: Requested NodeId %s already exists", item.RequestedNewNodeId)
@@ -365,26 +365,26 @@ def _add_ref_from_parent(self, nodedata, item, parentdata):
         desc.BrowseName = item.BrowseName
         desc.DisplayName = item.NodeAttributes.DisplayName
         desc.TypeDefinition = item.TypeDefinition
-        desc.IsForward = True # FIXME in uaprotocol_auto.py
-        self._add_unique_reference(parentdata, desc) # FIXME return StatusCode is not evaluated
+        desc.IsForward = True  # FIXME in uaprotocol_auto.py
+        self._add_unique_reference(parentdata, desc)  # FIXME return StatusCode is not evaluated
 
     def _add_ref_to_parent(self, nodedata: NodeData, item: ua.AddNodesItem, parentdata: NodeData):
         addref = ua.AddReferencesItem()
         addref.ReferenceTypeId = item.ReferenceTypeId
         addref.SourceNodeId = nodedata.nodeid
         addref.TargetNodeId = item.ParentNodeId
-        addref.TargetNodeClass = parentdata.attributes[ua.AttributeIds.NodeClass].value.Value.Value # type: ignore[union-attr]
-        addref.IsForward = False # FIXME in uaprotocol_auto.py
-        self._add_reference_no_check(nodedata, addref) # FIXME return StatusCode is not evaluated
+        addref.TargetNodeClass = parentdata.attributes[ua.AttributeIds.NodeClass].value.Value.Value  # type: ignore[union-attr]
+        addref.IsForward = False  # FIXME in uaprotocol_auto.py
+        self._add_reference_no_check(nodedata, addref)  # FIXME return StatusCode is not evaluated
 
     def _add_type_definition(self, nodedata: NodeData, item: ua.AddNodesItem):
         addref = ua.AddReferencesItem()
         addref.SourceNodeId = nodedata.nodeid
-        addref.IsForward = True # FIXME in uaprotocol_auto.py
+        addref.IsForward = True  # FIXME in uaprotocol_auto.py
         addref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasTypeDefinition)
         addref.TargetNodeId = item.TypeDefinition
         addref.TargetNodeClass = ua.NodeClass.DataType
-        self._add_reference_no_check(nodedata, addref) # FIXME return StatusCode is not evaluated
+        self._add_reference_no_check(nodedata, addref)  # FIXME return StatusCode is not evaluated
 
     def delete_nodes(
         self,
@@ -427,7 +427,7 @@ def _delete_node_callbacks(self, nodedata: NodeData):
                         "Error calling delete node callback callback %s, %s, %s", nodedata, ua.AttributeIds.Value, ex
                     )
 
-    def add_references(self, refs: List[ua.AddReferencesItem], user: User = User(role=UserRole.Admin)): # FIXME return type
+    def add_references(self, refs: List[ua.AddReferencesItem], user: User = User(role=UserRole.Admin)):  # FIXME return type
         result = [self._add_reference(ref, user) for ref in refs]
         return result
 
@@ -612,7 +612,7 @@ class AddressSpace:
     https://reference.opcfoundation.org/Core/docs/Part3/
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.logger = logging.getLogger(__name__)
         self.force_server_timestamp: bool = True
         self._nodes: Dict[ua.NodeId, NodeData] = {}
@@ -625,7 +625,7 @@ def __getitem__(self, nodeid: ua.NodeId) -> NodeData:
         return self._nodes.__getitem__(nodeid)
 
     def get(self, nodeid: ua.NodeId) -> Union[NodeData, None]:
-        return self._nodes.get(nodeid, None) # Fixme This is another behaviour than __getitem__ where an KeyError exception is thrown, right?
+        return self._nodes.get(nodeid, None)  # Fixme This is another behaviour than __getitem__ where an KeyError exception is thrown, right?
 
     def __setitem__(self, nodeid: ua.NodeId, value: NodeData):
         return self._nodes.__setitem__(nodeid, value)
@@ -801,11 +801,11 @@ async def write_attribute_value(self, nodeid: ua.NodeId, attr: ua.AttributeIds,
 
     def _is_expected_variant_type(self, value: ua.DataValue, attval: AttributeValue, node: NodeData) -> bool:
         # FIXME Type hinting reveals that it is possible that Value (Optional) is None which would raise an exception
-        vtype = attval.value.Value.VariantType # type: ignore[union-attr] 
+        vtype = attval.value.Value.VariantType  # type: ignore[union-attr]
         if vtype == ua.VariantType.Null:
             # Node had a null value, many nodes are initialized with that value
             # we should check what the real type is
-            dtype = node.attributes[ua.AttributeIds.DataType].value.Value.Value # type: ignore[union-attr]
+            dtype = node.attributes[ua.AttributeIds.DataType].value.Value.Value  # type: ignore[union-attr]
             if dtype.NamespaceIndex == 0 and dtype.Identifier <= 25:
                 vtype = ua.VariantType(dtype.Identifier)
             else:
@@ -814,8 +814,12 @@ def _is_expected_variant_type(self, value: ua.DataValue, attval: AttributeValue,
                 return True
         if value.Value.VariantType == vtype:  # type: ignore[union-attr]
             return True
-        _logger.warning("Write refused: Variant: %s with type %s does not have expected type: %s",
-                value.Value, value.Value.VariantType, attval.value.Value.VariantType)  # type: ignore[union-attr] 
+        _logger.warning(
+            "Write refused: Variant: %s with type %s does not have expected type: %s",
+            value.Value,
+            value.Value.VariantType if value.Value else None,
+            attval.value.Value.VariantType if attval.value.Value else None,
+        )
         return False
 
     def add_datachange_callback(self, nodeid: ua.NodeId, attr: ua.AttributeIds, callback: Callable) -> Tuple[ua.StatusCode, int]:

asyncua/server/event_generator.py

@@ -5,6 +5,7 @@
 import sys
 
 from asyncua import ua
+from asyncua.server.internal_session import InternalSession
 from ..common import events, event_objects, Node
 
 
@@ -20,7 +21,7 @@ class EventGenerator:
         etype: The event type, either an objectId, a NodeId or a Node object
     """
 
-    def __init__(self, isession):
+    def __init__(self, isession: InternalSession):
         self.logger = logging.getLogger(__name__)
         self.isession = isession
         self.event: event_objects.BaseEvent = None
@@ -91,7 +92,7 @@ async def trigger(self, time_attr=None, message=None, subscription_id=None):
         self.event.LocalTime = ua.uaprotocol_auto.TimeZoneDataType()
         if sys.version_info.major > 2:
             localtime = time.localtime(self.event.Time.timestamp())
-            self.event.LocalTime.Offset = localtime.tm_gmtoff//60
+            self.event.LocalTime.Offset = localtime.tm_gmtoff // 60
         else:
             localtime = time.localtime(time.mktime(self.event.Time.timetuple()))
             self.event.LocalTime.Offset = -(time.altzone if localtime.tm_isdst else time.timezone)

asyncua/server/history_sql.py

@@ -22,7 +22,7 @@ class HistorySQLite(HistoryStorageInterface):
     note that PARSE_DECLTYPES is active so certain data types (such as datetime) will not be BLOBs
     """
 
-    def __init__(self, path="history.db", max_history_data_response_size=10000):
+    def __init__(self, path="history.db", max_history_data_response_size=10000) -> None:
         self.max_history_data_response_size = max_history_data_response_size
         self.logger = logging.getLogger(__name__)
         self._datachanges_period = {}

asyncua/server/internal_server.py

@@ -10,7 +10,6 @@
 from pathlib import Path
 import logging
 from urllib.parse import urlparse
-from typing import Coroutine
 
 from asyncua import ua
 from .user_managers import PermissiveUserManager, UserManager
@@ -151,7 +150,7 @@ async def load_standard_address_space(self, shelf_file=None):
             # path was supplied, but file doesn't exist - create one for next start up
             await asyncio.get_running_loop().run_in_executor(None, self.aspace.make_aspace_shelf, shelf_file)
 
-    async def _address_space_fixes(self) -> Coroutine:  # type: ignore
+    async def _address_space_fixes(self):  # type: ignore
         """
         Looks like the xml definition of address space has some error. This is a good place to fix them
         """

asyncua/server/internal_subscription.py

@@ -11,6 +11,7 @@
 from .monitored_item_service import MonitoredItemService
 from .address_space import AddressSpace
 
+
 class InternalSubscription:
     """
     Server internal subscription.