add type hints to struct_from_binary()

asyncua/client/client.py

@@ -681,7 +681,7 @@ class Client:
     def get_server_node(self):
         return self.get_node(ua.FourByteNodeId(ua.ObjectIds.Server))
 
-    def get_node(self, nodeid: Union[ua.NodeId, str]) -> Node:
+    def get_node(self, nodeid: Union[Node, ua.NodeId, str, bytes, int]) -> Node:
         """
         Get node using NodeId object or a string representing a NodeId.
         """

asyncua/common/node.py

@@ -46,7 +46,7 @@ class Node:
     OPC-UA protocol. Feel free to look at the code of this class and call
     directly UA services methods to optimize your code
     """
-    def __init__(self, session: AbstractSession, nodeid: Union[ua.NodeId, str, int]):
+    def __init__(self, session: AbstractSession, nodeid: Union["Node", ua.NodeId, str, bytes, int]):
         self.session = session
         self.nodeid: ua.NodeId
         if isinstance(nodeid, Node):

asyncua/sync.py

@@ -205,6 +205,7 @@ class _SubHandler:
     def status_change_notification(self, status: ua.StatusChangeNotification):
         self.sync_handler.status_change_notification(status)
 
+
 class Client:
     def __init__(self, url: str, timeout: int = 4, tloop=None):
         self.tloop = tloop
@@ -291,8 +292,9 @@ class Client:
     def get_namespace_index(self, url):
         pass
 
-    def get_node(self, nodeid):
-        return SyncNode(self.tloop, self.aio_obj.get_node(nodeid))
+    def get_node(self, nodeid: Union["SyncNode", ua.NodeId, str, bytes, int]):
+        aio_nodeid = nodeid.aio_obj if isinstance(nodeid, SyncNode) else nodeid
+        return SyncNode(self.tloop, self.aio_obj.get_node(aio_nodeid))
 
     def get_root_node(self):
         return SyncNode(self.tloop, self.aio_obj.get_root_node())
@@ -437,7 +439,6 @@ class Server:
         return Subscription(self.tloop, aio_sub)
 
 
-
 class EventGenerator:
     def __init__(self, tloop, aio_evgen):
         self.aio_obj = aio_evgen
@@ -459,7 +460,7 @@ def new_node(sync_node, nodeid):
 
 
 class SyncNode:
-    def __init__(self, tloop, aio_node):
+    def __init__(self, tloop: ThreadLoop, aio_node: node.Node):
         self.aio_obj = aio_node
         self.tloop = tloop
 

asyncua/ua/ua_binary.py

@@ -5,7 +5,7 @@ Binary protocol specific functions and constants
 import functools
 import struct
 import logging
-from typing import Any, Callable
+from typing import IO, Any, Callable, Optional, Sequence, Type, TypeVar, Union
 import typing
 import uuid
 from enum import Enum, IntFlag
@@ -17,6 +17,8 @@ from .uatypes import type_from_optional, type_is_list, type_is_union, type_from_
 
 logger = logging.getLogger('__name__')
 
+T = TypeVar('T')
+
 
 def test_bit(data, offset):
     mask = 1 << offset
@@ -373,7 +375,7 @@ def to_binary(uatype, val):
 
 
 @functools.lru_cache(maxsize=None)
-def create_list_serializer(uatype, recursive: bool = False) -> Callable[[Any], bytes]:
+def create_list_serializer(uatype, recursive: bool = False) -> Callable[[Optional[Sequence[Any]]], bytes]:
     """
     Given a type, return a function that takes a list of instances
     of that type and serializes it.
@@ -391,6 +393,7 @@ def create_list_serializer(uatype, recursive: bool = False) -> Callable[[Any], b
         return recursive_serialize
 
     type_serializer = create_type_serializer(uatype)
+
     def serialize(val):
         if val is None:
             return none_val
@@ -686,7 +689,7 @@ def _create_dataclass_deserializer(objtype):
     return decode
 
 
-def struct_from_binary(objtype, data):
+def struct_from_binary(objtype: Union[Type[T], str], data: IO) -> T:
     """
     unpack an ua struct. Arguments are an objtype as Python dataclass or string
     """

asyncua/ua/uatypes.py

@@ -915,7 +915,7 @@ class Variant:
 
     # FIXME: typing is wrong here
     Value: Any = None
-    VariantType: VariantType = None
+    VariantType: "VariantType" = None
     Dimensions: Optional[List[Int32]] = None
     is_array: Optional[bool] = None