Merge ~kissiel/checkbox-support:zapper-proxy into checkbox-support:master

[Maciej Kisielewski]

Add zapper-proxy that lets us talk to zapper from checkbox[-support].

The AutoArgParse may change in the near future, as I still plan to develop some features over the weekend, but it works here beautifully.

[Paul Larson (pwlars)]

<insert usual grumbling about vendorized packages>, otherwise this looks really nice.
Two minor things below.

review: Needs Fixing

[Paul Larson (pwlars)]

Looks good and takes care of the few concerns I had. +1

review: Approve

diff --git a/checkbox_support/scripts/zapper_proxy.py b/checkbox_support/scripts/zapper_proxy.py
 new file mode 100644
index 0000000..afd998f
--- /dev/null
+++ b/checkbox_support/scripts/zapper_proxy.py
@@ -0,0 +1,141 @@
+# Copyright 2022 Canonical Ltd.
+# Written by:
+#   Maciej Kisielewski <[e-mail]>
+#
+# This is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# This file is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this file.  If not, see <http://www.gnu.org/licenses/>.
+"""
+This program acts as a proxy to Zapper Hardware.
+
+It uses internal Zapper Control RPyC API. Should this API change, additional
+ZapperControl classes should be added here.
+"""
+import os
+
+from abc import abstractmethod
+from importlib import import_module
+
+from checkbox_support.vendor.auto_argparse import AutoArgParser
+
+
+# Zapper Control is expected to change its RPC API.  Following adapter classes
+# serve the purpose of maintaining funcionality throughout the RPC API version
+# changes.
+
+class IZapperControl:
+    """Interface to the Zapper Control."""
+    @abstractmethod
+    def usb_get_state(self, address):
+        """
+        Get state of USBMUX addon.
+        Note that the address string is used as-is without any checks done on
+        this side. Any validation and use of that param will be done on the
+        remote end.
+
+        :param str address: address of USBMUX to get state from.
+        :return str: state of the USBMUX addon.
+        """
+
+    @abstractmethod
+    def usb_set_state(self, address, state):
+        """
+        Set state of USBMUX addon.
+        Note that both arguments are used as-is without any checks done on
+        this side. Any validation and use of those params will be done on the
+        remote end.
+
+        :param str address: address of USBMUX the state should be set on.
+        :param str state: state to set on the USBMUX addon
+        """
+
+
+class ZapperControlV1(IZapperControl):
+    """Control Zapper via RPyC using v1 of the API."""
+
+    def __init__(self, connection):
+        self._conn = connection
+
+    def usb_get_state(self, address):
+        ret = []
+        success = self._conn.root.zombiemux_get_state(address, ret)
+        if not success:
+            raise SystemExit(
+                "Failed to get state for address {}.".format(address))
+        print("State for address {} is {}".format(address, ret[0]))
+
+    def usb_set_state(self, address, state):
+        success = self._conn.root.zombiemux_get_state(address, state)
+        if not success:
+            raise SystemExit(
+                "Failed to set '{}' state for address {}.".format(
+                    state, address))
+        print("State '{}' set for the address {}.".format(state, address))
+
+
+def main():
+    """Entry point."""
+    # to not make checkbox-support dependant on RPyC let's use one available in
+    # the system, and if it's not available let's try loading one provided by
+    # Checkbox. Real world usecase would be to run this program from within
+    # Checkbox, so chances for not finding it are pretty slim.
+    try:
+        rpyc = import_module('rpyc')
+    except ImportError:
+        try:
+            rpyc = import_module('plainbox.vendor.rpyc')
+        except ImportError as exc:
+            raise SystemExit(
+                "RPyC not found. Neither from sys nor from Checkbox") from exc
+
+    # generate argparse from the interface of Zapper Control
+    parser = AutoArgParser(cls=IZapperControl)
+    parser.add_argument(
+        '--host', default=os.environ.get('ZAPPER_ADDRESS'),
+        help=("Address of Zapper to connect to. If not supplied, "
+              "ZAPPER_ADDRESS environment variable will be used.")
+    )
+    # turn Namespace into a normal dict
+    args = parser.parse_args()
+    # popping elements from the dict, so at the end the end the right method
+    # is called with only the item that are expected
+    host = args.host
+    if host is None:
+        raise SystemExit(
+            "You have to provide Zapper host, either via '--host' or via "
+            "ZAPPER_ADDRESS environment variable")
+    # connect and see which protol version should be used
+    conn = rpyc.connect(
+        host, 60000, config={"allow_all_attrs": True})
+    try:
+        version = conn.root.get_api_version()
+    except AttributeError:
+        # there was no "get_api_version" method on Zapper
+        # so this means the oldest version possible - 1
+        version = 1
+    # the following mapping could be replaced by something that generates
+    # a class name and tries looking it up in this module, but using dict
+    # feels simpler due to explicitness and can include classes defined in
+    # some other modules
+    control_cls = {
+        1: ZapperControlV1,
+    }.get(version, None)
+    if control_cls is None:
+        raise SystemExit((
+            "Zapper host returned unknown Zapper Control Version: {ver}\n"
+            "Implement ZapperControlV{ver} in checkbox_support!"
+        ).format(ver=version))
+    zapper_control = control_cls(conn)
+    parser.run(zapper_control)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/checkbox_support/vendor/auto_argparse.py b/checkbox_support/vendor/auto_argparse.py
 new file mode 100644
index 0000000..ca6074e
--- /dev/null
+++ b/checkbox_support/vendor/auto_argparse.py
@@ -0,0 +1,115 @@
+# Copyright (C) 2022 Canonical Ltd.
+#
+# Authors:
+#   Maciej Kisielewski <[e-mail]>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+An extended ArgParser that automatically creates subparsers.
+"""
+
+import argparse
+import inspect
+import re
+
+def _parse_docstring(doc):
+    # those are the defaults
+    param_descs = {}
+    return_desc = ''
+    info = ''
+    if not doc:
+        return info, param_descs, return_desc
+    lines = doc.strip().splitlines()
+    # let's extract the text preceeding param specification
+    # the first line that explains a param or a return value
+    # means the end of general info
+    info_lines = []
+    for lineno, line in enumerate(lines):
+        if line.startswith((':param', ':return')):
+            # we're at the param description section
+            # let's delete what we already read
+            lines = lines[lineno:]
+            break
+        info_lines.append(line)
+    info = '\n'.join(info_lines)
+    # the rest of lines should now contain information about params or the 
+    # return value
+    param_re = re.compile(
+        r":param(\s[A-Za-z_][A-Za-z0-9_]*)?\s([A-Za-z_][A-Za-z0-9_]*):(.*)")
+    return_re = re.compile(
+        r":returns(\s[A-Za-z_][A-Za-z0-9_]*)?:(.*)")
+    for line in lines:
+        param_match = param_re.match(line)
+        if param_match:
+            param_type, param_name, description = param_match.groups()
+            param_type = {
+                'str': str, 'int': int, 'float': float, 'bool': bool
+            }.get(param_type.strip() if param_type else None)
+            param_descs[param_name] = param_type, description.strip()
+            continue
+        return_match = return_re.match(line)
+        if return_match:
+            return_desc = return_match.groups()[1].strip()
+    return info, param_descs, return_desc
+
+class AutoArgParser(argparse.ArgumentParser):
+    def __init__(self, *args, cls=None, **kwargs):
+        if cls is None:
+            super().__init__(*args, **kwargs)
+            return
+        self._cls = cls
+        self._auto_args = []
+        self._args_already_parsed = False
+        cls_doc = (inspect.getdoc(cls) or '').strip()
+        super().__init__(*args, description=cls_doc, **kwargs)
+        subparsers = self.add_subparsers(dest='method')
+        methods = inspect.getmembers(cls, inspect.isroutine)
+        for name, method in methods:
+            if name.startswith('__'):
+                continue
+            doc = inspect.getdoc(method)
+            argspec = inspect.getfullargspec(method)
+            args = argspec.args
+            # if the function is a classmethod or instance method, let's pop
+            # the first arg (the bound object)
+            if args and args[0] in ['self', 'cls']:
+                args = args[1:]
+            info, params, return_desc = _parse_docstring(doc)
+            sub = subparsers.add_parser(method.__name__, help=info)
+            for arg in args:
+                self._auto_args.append(arg)
+                type, help = params.get(arg, (None, None))
+                sub.add_argument(arg, type=type, help=help)
+
+    def parse_args(self, *args, **kwargs):
+        args = super().parse_args(*args, **kwargs)
+        self._args_already_parsed = True
+        self._picked_method = args.method
+        # turn members of the args namespace into a normal dictionary so it's
+        # easier later on to use them as kwargs. But let's take only the one
+        # automatically generated.
+        self._method_args = {
+            a: v for a, v in vars(args).items() if a in self._auto_args}
+        return args
+
+
+    def run(self, obj=None):
+        if not self._args_already_parsed:
+            self.parse_args()
+        if obj is None:
+            obj = self._cls()
+        if not self._picked_method:
+            raise SystemExit(
+                "No sub-command chosen!\n\n{}".format(self.format_help()))
+            print("No subcommand chosen")
+        return getattr(obj, self._picked_method)(**self._method_args)
diff --git a/setup.py b/setup.py
index 0c398e6..2d5ea12 100755
--- a/setup.py
+++ b/setup.py
@@ -94,6 +94,8 @@ setup(
             ("checkbox-support-lsusb="
                 "checkbox_support.scripts.lsusb:main"),
             ("checkbox-support-parse=checkbox_support.parsers:main"),
+            ("checkbox-support-zapper-proxy="
+                "checkbox_support.scripts.zapper_proxy:main"),
         ],
     },
 )