Readable format in interface definition file, fixed tuple issues.

Author: jfjlaros

simple_rpc/cli.py

@@ -68,17 +68,17 @@ def rpc_list(
 
 
 def rpc_call(
-        handle: BinaryIO, device: str, baudrate: int, wait: int, name: str,
-        args: list, load: TextIO) -> None:
+        handle: BinaryIO, device: str, baudrate: int, wait: int, load: TextIO,
+        name: str, args: list) -> None:
     """Execute a method.
 
     :arg handle: Output handle.
     :arg device: Device.
     :arg baudrate: Baud rate.
     :arg wait: Time in seconds before communication starts.
+    :arg load: Interface definition file.
     :arg name: Method name.
     :arg args: Method parameters.
-    :arg load: Interface definition file.
     """
     args_ = list(map(lambda x: json_utf8_encode(_loads(x)), args))
 

simple_rpc/io.py

@@ -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: bytes, basic_type: bytes) -> 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.
@@ -22,16 +22,16 @@ def _read_basic(stream: BinaryIO, endianness: bytes, basic_type: bytes) -> any:
 
     :returns: Value of type {basic_type}.
     """
-    if basic_type == b's':
+    if basic_type == 's':
         return _read_bytes_until(stream, b'\0')
 
-    full_type = endianness + basic_type
+    full_type = (endianness + basic_type).encode('utf-8')
 
     return unpack(full_type, stream.read(calcsize(full_type)))[0]
 
 
 def _write_basic(
-        stream: BinaryIO, endianness: bytes, basic_type: bytes, value: any
+        stream: BinaryIO, endianness: str, basic_type: str, value: any
         ) -> None:
     """Write a value of basic type to a stream.
 
@@ -40,33 +40,33 @@ def _write_basic(
     :arg basic_type: Type of {value}.
     :arg value: Value to write.
     """
-    if basic_type == b's':
+    if basic_type == 's':
         stream.write(value + b'\0')
         return
 
-    full_type = endianness + basic_type
+    full_type = (endianness + basic_type).encode('utf-8')
 
     stream.write(pack(full_type, cast(basic_type)(value)))
 
 
-def cast(c_type: bytes) -> object:
+def cast(c_type: str) -> object:
     """Select the appropriate casting function given a C type.
 
     :arg c_type: C type.
 
     :returns: Casting function.
     """
-    if c_type == b'?':
+    if c_type == '?':
         return bool
-    if c_type in [b'c', b's']:
+    if c_type in ['c', 's']:
         return bytes
-    if c_type in [b'f', b'd']:
+    if c_type in ['f', 'd']:
         return float
     return int
 
 
 def read(
-        stream: BinaryIO, endianness: bytes, size_t: bytes, obj_type: any
+        stream: BinaryIO, endianness: str, size_t: str, obj_type: any
         ) -> any:
     """Read an object from a stream.
 
@@ -93,7 +93,7 @@ def read_byte_string(stream: BinaryIO) -> bytes:
 
 
 def write(
-        stream: BinaryIO, endianness: bytes, size_t: bytes, obj_type: any,
+        stream: BinaryIO, endianness: str, size_t: str, obj_type: any,
         obj: any) -> None:
     """Write an object to a stream.
 

simple_rpc/protocol.py

@@ -10,7 +10,7 @@ def _parse_type(type_str: bytes) -> any:
 
     :returns: Type object.
     """
-    def _construct_type(tokens):
+    def _construct_type(tokens: tuple):
         obj_type = []
 
         for token in tokens:
@@ -21,7 +21,7 @@ def _construct_type(tokens):
             elif token in (b')', b']'):
                 break
             else:
-                obj_type.append(token)
+                obj_type.append(token.decode())
 
         return obj_type
 
@@ -30,14 +30,14 @@ def _construct_type(tokens):
     if len(obj_type) > 1:
         raise ValueError('top level type can not be tuple')
     if not obj_type:
-        return b''
+        return ''
     return obj_type[0]
 
 
-def _type_name(obj_type: bytes) -> str:
-    """Python type name of a C type.
+def _type_name(obj_type: any) -> str:
+    """Python type name of a C object type.
 
-    :arg c_type: C type.
+    :arg obj_type: C object type.
 
     :returns: Python type name.
     """
@@ -84,7 +84,7 @@ def _strip_split(string: str, delimiter: str) -> list:
     return list(map(lambda x: x.strip(), string.split(delimiter)))
 
 
-def _add_doc(method: dict, doc: str) -> None:
+def _add_doc(method: dict, doc: bytes) -> None:
     """Add documentation to a method object.
 
     :arg method: Method object.

simple_rpc/simple_rpc.py

@@ -5,15 +5,15 @@
 
 from serial import serial_for_url
 from serial.serialutil import SerialException
-from yaml import safe_dump, safe_load
+from yaml import dump, load
 
 from .extras import make_function
 from .io import read, read_byte_string, until, write
 from .protocol import parse_line
 
 
-_protocol = b'simpleRPC'
-_version = (3, 0, 0)  # TODO: Replace tuples with lists for serialisation.
+_protocol = 'simpleRPC'
+_version = (3, 0, 0)
 
 _list_req = 0xff
 
@@ -37,8 +37,8 @@ def __init__(
         self._load = load
         # TODO: Put all of these in one object.
         self._version = (0, 0, 0)
-        self._endianness = b'<'
-        self._size_t = b'H'
+        self._endianness = '<'
+        self._size_t = 'H'
         self.methods = {}
 
         if autoconnect:
@@ -65,9 +65,9 @@ def _select(self: object, index: int) -> None:
 
         :arg index: Method index.
         """
-        self._write(b'B', index)
+        self._write('B', index)
 
-    def _write(self: object, obj_type: bytes, 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.
@@ -78,7 +78,7 @@ def _write(self: object, obj_type: bytes, obj: any) -> None:
     def _read_byte_string(self: object) -> bytes:
         return read_byte_string(self._connection)
 
-    def _read(self: object, obj_type: str) -> any:
+    def _read(self: object, obj_type: any) -> any:
         """Read a return value from a remote procedure call.
 
         :arg obj_type: Return type.
@@ -94,18 +94,18 @@ def _get_methods(self: object) -> dict:
         """
         self._select(_list_req)
 
-        if self._read_byte_string() != _protocol:
+        if self._read_byte_string().decode() != _protocol:
             raise ValueError('missing protocol header')
 
-        self._version = tuple(self._read(b'B') for _ in range(3))
+        self._version = tuple(self._read('B') for _ in range(3))
         if self._version[0] != _version[0] or self._version[1] > _version[1]:
             raise ValueError(
                 'version mismatch (device: {}, client: {})'.format(
                     '.'.join(map(str, self._version)),
                     '.'.join(map(str, _version))))
 
         self._endianness, self._size_t = (
-            bytes([c]) for c in self._read_byte_string())
+            chr(c) for c in self._read_byte_string())
 
         methods = {}
         for index, line in enumerate(
@@ -173,7 +173,7 @@ def save(self: object, handle: TextIO) -> None:
 
         :arg handle: Open file handle.
         """
-        safe_dump(
+        dump(
             {
                 'version': self._version,
                 'endianness': self._endianness,
@@ -183,11 +183,8 @@ def save(self: object, handle: TextIO) -> None:
             handle, width=76, default_flow_style=False)
 
     def load(self: object) -> None:
-        """Load the interface definition from a file.
-
-        :arg handle: Open file handle.
-        """
-        definition = safe_load(self._load)
+        """Load the interface definition from a file."""
+        definition = load(self._load)
         self._version = definition['version']
         self._endianness = definition['endianness']
         self._size_t = definition['size_t']

tests/test_cli.py

@@ -35,13 +35,13 @@ def test_describe_method() -> None:
 def test_rpc_list() -> None:
     handle = StringIO()
 
-    rpc_list(handle, _devices['serial'], 9600, 1)
+    rpc_list(handle, _devices['serial'], 9600, 1, None)
     assert 'ping data\n    Echo a value.\n' in handle.getvalue()
 
 
 @mark.test_device('serial')
 def test_rpc_call() -> None:
     handle = StringIO()
 
-    rpc_call(handle, _devices['serial'], 9600, 1, 'ping', ['10'])
+    rpc_call(handle, _devices['serial'], 9600, 1, None, 'ping', ['10'])
     assert handle.getvalue() == '10\n'

tests/test_device.py

@@ -26,15 +26,15 @@ def test_type_1(self: object) -> None:
         assert self._interface.methods['ping']['return']['typename'] == 'int'
 
     def test_fmt_1(self: object) -> None:
-        assert self._interface.methods['ping']['return']['fmt'] == b'B'
+        assert self._interface.methods['ping']['return']['fmt'] == 'B'
 
     def test_param_1(self: object) -> None:
         assert (
             self._interface.methods['ping']['parameters'][0]['typename'] ==
             'int')
 
     def test_param_2(self: object) -> None:
-        assert self._interface.methods['ping']['parameters'][0]['fmt'] == b'B'
+        assert self._interface.methods['ping']['parameters'][0]['fmt'] == 'B'
 
     def test_param_3(self: object) -> None:
         assert (

tests/test_io.py

@@ -5,8 +5,8 @@
 
 
 def _test_invariance_basic(
-        f_read: callable, f_write: callable, endianness: bytes,
-        basic_type: bytes, data: bytes, value: any) -> None:
+        f_read: callable, f_write: callable, endianness: str, basic_type: str,
+        data: bytes, value: any) -> None:
     stream = BytesIO(data)
     value_ = f_read(stream, endianness, basic_type)
     assert value_ == value
@@ -17,7 +17,7 @@ def _test_invariance_basic(
 
 
 def _test_invariance(
-        f_read: callable, f_write: callable, endianness: bytes, size_t: bytes,
+        f_read: callable, f_write: callable, endianness: str, size_t: str,
         obj_def: any, data: bytes, obj: any) -> None:
     stream = BytesIO(data)
     obj_ = f_read(stream, endianness, size_t, obj_def)
@@ -29,13 +29,13 @@ def _test_invariance(
 
 
 def test_cast() -> None:
-    assert cast(b'?') == bool
-    assert cast(b'c') == bytes
-    assert cast(b's') == bytes
-    assert cast(b'f') == float
-    assert cast(b'd') == float
-    assert cast(b'h') == int
-    assert cast(b'i') == int
+    assert cast('?') == bool
+    assert cast('c') == bytes
+    assert cast('s') == bytes
+    assert cast('f') == float
+    assert cast('d') == float
+    assert cast('h') == int
+    assert cast('i') == int
 
 
 def test_read_bytes_until() -> None:
@@ -46,65 +46,65 @@ def test_read_bytes_until() -> None:
 
 def test_basic_string() -> None:
     _test_invariance_basic(
-        _read_basic, _write_basic, b'<', b's', b'abcdef\0', b'abcdef')
+        _read_basic, _write_basic, '<', 's', b'abcdef\0', b'abcdef')
 
 
 def test_basic_int_le() -> None:
     _test_invariance_basic(
-        _read_basic, _write_basic, b'<', b'i', b'\2\0\0\0', 2)
+        _read_basic, _write_basic, '<', 'i', b'\2\0\0\0', 2)
 
 
 def test_basic_int_be() -> None:
     _test_invariance_basic(
-        _read_basic, _write_basic, b'>', b'i', b'\0\0\0\2', 2)
+        _read_basic, _write_basic, '>', 'i', b'\0\0\0\2', 2)
 
 
 def test_list_char() -> None:
     _test_invariance(
-        read, write, b'<', b'h', [b'c'], b'\3\0a\0c', [b'a', b'\0', b'c'])
+        read, write, '<', 'h', ['c'], b'\3\0a\0c', [b'a', b'\0', b'c'])
 
 
 def test_list_nibble() -> None:
     _test_invariance(
-        read, write, b'<', b'h', [b'h'], b'\3\0\1\0\2\0\3\0', [1, 2, 3])
+        read, write, '<', 'h', ['h'], b'\3\0\1\0\2\0\3\0', [1, 2, 3])
 
 
 def test_list_list() -> None:
     _test_invariance(
-        read, write, b'<', b'h', [[b'b']], b'\2\0\2\0\0\1\2\0\2\3',
+        read, write, '<', 'h', [['b']], b'\2\0\2\0\0\1\2\0\2\3',
         [[0, 1], [2, 3]])
 
 
 def test_object_char_int() -> None:
     _test_invariance(
-        read, write, b'<', b'h', (b'c', b'i'), b'a\3\0\0\0', (b'a', 3))
+        read, write, '<', 'h', ('c', 'i'), b'a\3\0\0\0', (b'a', 3))
 
 
 def test_object_nibble_string_char() -> None:
     _test_invariance(
-        read, write, b'<', b'h', (b'h', b's', b'c'), b'\2\0abcdef\0x',
+        read, write, '<', 'h', ('h', 's', 'c'), b'\2\0abcdef\0x',
         (2, b'abcdef', b'x'))
 
 
 def test_object_object() -> None:
     _test_invariance(
-        read, write, b'<', b'h', (((b'c', ), ), (b'c', ), ), b'ab',
+        read, write, '<', 'h', ((('c', ), ), ('c', ), ), b'ab',
         (((b'a', ), ), (b'b', )))
 
 
 def test_list_tuple() -> None:
     _test_invariance(
-        read, write, b'<', b'h', [b'c', b'c', b'c'], b'\2\0abcabc',
+        read, write, '<', 'h', ['c', 'c', 'c'], b'\2\0abcabc',
         [b'a', b'b', b'c', b'a', b'b', b'c'])
 
 
 def test_list_object() -> None:
     _test_invariance(
-        read, write, b'<', b'h', [(b'c', b'c', b'c')], b'\2\0abcabc',
+        read, write, '<', 'h', [('c', 'c', 'c')], b'\2\0abcabc',
         [(b'a', b'b', b'c'), (b'a', b'b', b'c')])
 
 
 def test_list_object_tuple() -> None:
     _test_invariance(
-        read, write, b'<', b'h', [(b'c', b'c'), b'c'], b'\2\0abcabc',
+        read, write, '<', 'h', [('c', 'c'), 'c'], b'\2\0abcabc',
         [(b'a', b'b'), b'c', (b'a', b'b'), b'c'])