Fixed type hinting bug.

Author: jfjlaros

simple_rpc/io.py

@@ -1,4 +1,4 @@
-from typing import BinaryIO
+from typing import Any, BinaryIO
 from struct import calcsize, pack, unpack
 
 
@@ -13,7 +13,7 @@ def _read_bytes_until(stream: BinaryIO, delimiter: bytes) -> bytes:
     return b''.join(until(lambda x: x == delimiter, stream.read, 1))
 
 
-def _read_basic(stream: BinaryIO, endianness: str, basic_type: str) -> any:
+def _read_basic(stream: BinaryIO, endianness: str, basic_type: str) -> Any:
     """Read a value of basic type from a stream.
 
     :arg stream: Stream object.
@@ -31,7 +31,7 @@ def _read_basic(stream: BinaryIO, endianness: str, basic_type: str) -> any:
 
 
 def _write_basic(
-        stream: BinaryIO, endianness: str, basic_type: str, value: any
+        stream: BinaryIO, endianness: str, basic_type: str, value: Any
         ) -> None:
     """Write a value of basic type to a stream.
 
@@ -66,8 +66,8 @@ def cast(c_type: str) -> object:
 
 
 def read(
-        stream: BinaryIO, endianness: str, size_t: str, obj_type: any
-        ) -> any:
+        stream: BinaryIO, endianness: str, size_t: str, obj_type: Any
+        ) -> Any:
     """Read an object from a stream.
 
     :arg stream: Stream object.
@@ -93,8 +93,8 @@ def read_byte_string(stream: BinaryIO) -> bytes:
 
 
 def write(
-        stream: BinaryIO, endianness: str, size_t: str, obj_type: any,
-        obj: any) -> None:
+        stream: BinaryIO, endianness: str, size_t: str, obj_type: Any,
+        obj: Any) -> None:
     """Write an object to a stream.
 
     :arg stream: Stream object.

simple_rpc/protocol.py

@@ -1,9 +1,9 @@
-from typing import BinaryIO
+from typing import Any, BinaryIO
 
 from .io import cast, read_byte_string
 
 
-def _parse_type(type_str: bytes) -> any:
+def _parse_type(type_str: bytes) -> Any:
     """Parse a type definition string.
 
     :arg type_str: Type definition string.
@@ -34,7 +34,7 @@ def _construct_type(tokens: tuple):
     return obj_type[0]
 
 
-def _type_name(obj_type: any) -> str:
+def _type_name(obj_type: Any) -> str:
     """Python type name of a C object type.
 
     :arg obj_type: C object type.

simple_rpc/simple_rpc.py

@@ -1,7 +1,7 @@
 from functools import wraps
 from time import sleep
 from types import MethodType
-from typing import TextIO
+from typing import Any, TextIO
 
 from serial import serial_for_url
 from serial.serialutil import SerialException
@@ -80,7 +80,7 @@ def _select(self: object, index: int) -> None:
         """
         self._write('B', index)
 
-    def _write(self: object, obj_type: any, obj: any) -> None:
+    def _write(self: object, obj_type: Any, obj: Any) -> None:
         """Provide parameters for a remote procedure call.
 
         :arg obj_type: Type of the parameter.
@@ -93,7 +93,7 @@ def _write(self: object, obj_type: any, obj: any) -> None:
     def _read_byte_string(self: object) -> bytes:
         return read_byte_string(self._connection)
 
-    def _read(self: object, obj_type: any) -> any:
+    def _read(self: object, obj_type: Any) -> Any:
         """Read a return value from a remote procedure call.
 
         :arg obj_type: Return type.
@@ -157,7 +157,7 @@ def close(self: object) -> None:
             delattr(self, method)
         self.device['methods'].clear()
 
-    def call_method(self: object, name: str, *args: list) -> any:
+    def call_method(self: object, name: str, *args: list) -> Any:
         """Execute a method.
 
         :arg name: Method name.
@@ -219,7 +219,7 @@ def _auto_open(f: callable) -> callable:
         """Decorator for automatic opening and closing of ethernet sockets."""
         @wraps(f)
         def _auto_open_wrapper(
-                self: object, *args: list, **kwargs: dict) -> any:
+                self: object, *args: list, **kwargs: dict) -> Any:
             self._open()
             result = f(self, *args, **kwargs)
             self._close()

tests/test_io.py

@@ -1,12 +1,13 @@
 from io import BytesIO
+from typing import Any
 
 from simple_rpc.io import (
     _read_basic, _read_bytes_until, _write_basic, cast, read, write)
 
 
 def _test_invariance_basic(
         f_read: callable, f_write: callable, endianness: str, basic_type: str,
-        data: bytes, value: any) -> None:
+        data: bytes, value: Any) -> None:
     stream = BytesIO(data)
     value_ = f_read(stream, endianness, basic_type)
     assert value_ == value
@@ -18,7 +19,7 @@ def _test_invariance_basic(
 
 def _test_invariance(
         f_read: callable, f_write: callable, endianness: str, size_t: str,
-        obj_def: any, data: bytes, obj: any) -> None:
+        obj_def: Any, data: bytes, obj: Any) -> None:
     stream = BytesIO(data)
     obj_ = f_read(stream, endianness, size_t, obj_def)
     assert obj_ == obj