Interface definition saving and loading prototype.

Author: jfjlaros

setup.cfg

@@ -20,6 +20,7 @@ classifiers =
 [options]
 packages = find:
 install_requires =
+    PyYAML==5.3.1
     pyserial==3.4
 
 [options.entry_points]

simple_rpc/cli.py

@@ -2,7 +2,7 @@
 from json import dumps, loads
 from json.decoder import JSONDecodeError
 from sys import stdout
-from typing import BinaryIO
+from typing import BinaryIO, TextIO
 
 from . import doc_split, usage, version
 from .extras import json_utf8_decode, json_utf8_encode
@@ -48,22 +48,28 @@ def _loads(string: str) -> str:
         return string
 
 
-def rpc_list(handle: BinaryIO, device: str, baudrate: int, wait: int) -> None:
+def rpc_list(
+        handle: BinaryIO, device: str, baudrate: int, wait: int, save: TextIO
+        ) -> None:
     """List the device methods.
 
     :arg handle: Output handle.
     :arg device: Device.
     :arg baudrate: Baud rate.
     :arg wait: Time in seconds before communication starts.
+    :arg save: Interface definition file.
     """
     with Interface(device, baudrate, wait) as interface:
-        for method in interface.methods.values():
-            handle.write(_describe_method(method) + '\n\n\n')
+        if not save:
+            for method in interface.methods.values():
+                handle.write(_describe_method(method) + '\n\n\n')
+        else:
+            interface.save(save)
 
 
 def rpc_call(
         handle: BinaryIO, device: str, baudrate: int, wait: int, name: str,
-        args: list) -> None:
+        args: list, load: TextIO) -> None:
     """Execute a method.
 
     :arg handle: Output handle.
@@ -72,10 +78,11 @@ def rpc_call(
     :arg wait: Time in seconds before communication starts.
     :arg name: Method name.
     :arg args: Method parameters.
+    :arg load: Interface definition file.
     """
     args_ = list(map(lambda x: json_utf8_encode(_loads(x)), args))
 
-    with Interface(device, baudrate, wait) as interface:
+    with Interface(device, baudrate, wait, True, load) as interface:
         result = interface.call_method(name, *args_)
 
         if result is not None:
@@ -110,16 +117,22 @@ def _arg_parser() -> object:
     subparser = subparsers.add_parser(
         'list', formatter_class=ArgumentDefaultsHelpFormatter,
         parents=[common_parser], description=doc_split(rpc_list))
+    subparser.add_argument(
+        '-s', dest='save', type=FileType('w'), default=None,
+        help='interface definition file')
     subparser.set_defaults(func=rpc_list)
 
     subparser = subparsers.add_parser(
         'call', formatter_class=ArgumentDefaultsHelpFormatter,
         parents=[common_parser], description=doc_split(rpc_call))
-    subparser.set_defaults(func=rpc_call)
     subparser.add_argument(
         'name', metavar='NAME', type=str, help='command name')
     subparser.add_argument(
         'args', metavar='ARG', type=str, nargs='*', help='command parameter')
+    subparser.add_argument(
+        '-l', dest='load', type=FileType('r'), default=None,
+        help='interface definition file')
+    subparser.set_defaults(func=rpc_call)
 
     return parser
 

simple_rpc/simple_rpc.py

@@ -1,17 +1,19 @@
 from functools import wraps
 from time import sleep
 from types import MethodType
+from typing import TextIO
 
 from serial import serial_for_url
 from serial.serialutil import SerialException
+from yaml import safe_dump, safe_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)
+_version = (3, 0, 0)  # TODO: Replace tuples with lists for serialisation.
 
 _list_req = 0xff
 
@@ -20,17 +22,20 @@ class _Interface(object):
     """Generic simpleRPC interface."""
     def __init__(
             self: object, device: str, baudrate: int=9600, wait: int=2,
-            autoconnect: bool=True) -> None:
+            autoconnect: bool=True, load: TextIO=None) -> None:
         """
         :arg device: Device name.
         :arg baudrate: Baud rate.
         :arg wait: Time in seconds before communication starts.
         :arg autoconnect: Automatically connect.
+        :arg load: Load interface definition from file.
         """
         self._wait = wait
 
         self._connection = serial_for_url(
             device, do_not_open=True, baudrate=baudrate)
+        self._load = load
+        # TODO: Put all of these in one object.
         self._version = (0, 0, 0)
         self._endianness = b'<'
         self._size_t = b'H'
@@ -118,7 +123,10 @@ def open(self: object) -> None:
         """Connect to device."""
         sleep(self._wait)
 
-        self.methods = self._get_methods()
+        if self._load:
+            self.load()
+        else:
+            self.methods = self._get_methods()
         for method in self.methods.values():
             setattr(
                 self, method['name'], MethodType(make_function(method), self))
@@ -160,6 +168,31 @@ def call_method(self: object, name: str, *args: list) -> any:
             return self._read(method['return']['fmt'])
         return None
 
+    def save(self: object, handle: TextIO) -> None:
+        """Save the interface definition to a file.
+
+        :arg handle: Open file handle.
+        """
+        safe_dump(
+            {
+                'version': self._version,
+                'endianness': self._endianness,
+                'size_t': self._size_t,
+                'methods': self.methods
+            },
+            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)
+        self._version = definition['version']
+        self._endianness = definition['endianness']
+        self._size_t = definition['size_t']
+        self.methods = definition['methods']
+
 
 class SerialInterface(_Interface):
     """Serial simpleRPC interface."""