pw_rpc.console_tools: Context class; function helpers
- Rework the cumbersome multi_client_terminal_variables function into a
pw_rpc.console_tools.Context class. The class can be extended if
needed.
- Provide functions.help_as_repr that wraps a function so that its
__repr__ displays help information. This is helpful for defining
commands in an interactive console.
- Display all variables, not just RPCs, in the CommandHelper.
Requires: pigweed-internal:11382
Change-Id: I9a4dfc25473f6cf878f7855602a4f2334313de29
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/41120
Reviewed-by: Keir Mierle <keir@google.com>
Commit-Queue: Wyatt Hepler <hepler@google.com>
diff --git a/pw_rpc/py/BUILD.gn b/pw_rpc/py/BUILD.gn
index 07fd6b4..71faaee 100644
--- a/pw_rpc/py/BUILD.gn
+++ b/pw_rpc/py/BUILD.gn
@@ -32,6 +32,7 @@
"pw_rpc/codegen_raw.py",
"pw_rpc/console_tools/__init__.py",
"pw_rpc/console_tools/console.py",
+ "pw_rpc/console_tools/functions.py",
"pw_rpc/console_tools/watchdog.py",
"pw_rpc/descriptors.py",
"pw_rpc/ids.py",
@@ -43,7 +44,8 @@
tests = [
"tests/callback_client_test.py",
"tests/client_test.py",
- "tests/console_tools_test.py",
+ "tests/console_tools/console_tools_test.py",
+ "tests/console_tools/functions_test.py",
"tests/descriptors_test.py",
"tests/ids_test.py",
"tests/packets_test.py",
diff --git a/pw_rpc/py/docs.rst b/pw_rpc/py/docs.rst
index 26c31c2..5aeea46 100644
--- a/pw_rpc/py/docs.rst
+++ b/pw_rpc/py/docs.rst
@@ -20,4 +20,4 @@
pw_rpc.console_tools
====================
.. automodule:: pw_rpc.console_tools
- :members: multi_client_terminal_variables, ClientInfo, Watchdog
+ :members: Context, ClientInfo, Watchdog, help_as_repr
diff --git a/pw_rpc/py/pw_rpc/console_tools/__init__.py b/pw_rpc/py/pw_rpc/console_tools/__init__.py
index 7b469bd..3f157f1 100644
--- a/pw_rpc/py/pw_rpc/console_tools/__init__.py
+++ b/pw_rpc/py/pw_rpc/console_tools/__init__.py
@@ -13,6 +13,6 @@
# the License.
"""Utilities for building tools that interact with pw_rpc."""
-from pw_rpc.console_tools.console import (CommandHelper, ClientInfo,
- multi_client_terminal_variables)
+from pw_rpc.console_tools.console import Context, CommandHelper, ClientInfo
+from pw_rpc.console_tools.functions import help_as_repr
from pw_rpc.console_tools.watchdog import Watchdog
diff --git a/pw_rpc/py/pw_rpc/console_tools/console.py b/pw_rpc/py/pw_rpc/console_tools/console.py
index 6b21624..29797f5 100644
--- a/pw_rpc/py/pw_rpc/console_tools/console.py
+++ b/pw_rpc/py/pw_rpc/console_tools/console.py
@@ -18,13 +18,14 @@
import inspect
import textwrap
import types
-from typing import Any, Collection, Dict, Iterable, NamedTuple
+from typing import Any, Collection, Dict, Iterable, Mapping, NamedTuple
import pw_status
from pw_protobuf_compiler import python_protos
import pw_rpc
from pw_rpc.descriptors import Method
+from pw_rpc.console_tools import functions
_INDENT = ' '
@@ -34,21 +35,31 @@
@classmethod
def from_methods(cls,
methods: Iterable[Method],
+ variables: Mapping[str, object],
header: str,
footer: str = '') -> 'CommandHelper':
- return cls({m.full_name: m for m in methods}, header, footer)
+ return cls({m.full_name: m
+ for m in methods}, variables, header, footer)
- def __init__(self, methods: Dict[str, Any], header: str, footer: str = ''):
+ def __init__(self,
+ methods: Mapping[str, object],
+ variables: Mapping[str, object],
+ header: str,
+ footer: str = ''):
self._methods = methods
+ self._variables = variables
self.header = header
self.footer = footer
- def help(self, item: Any = None) -> str:
+ def help(self, item: object = None) -> str:
"""Returns a help string with a command or all commands listed."""
if item is None:
+ all_vars = '\n'.join(sorted(self._variables_without_methods()))
all_rpcs = '\n'.join(self._methods)
return (f'{self.header}\n\n'
+ f'All variables:\n\n{textwrap.indent(all_vars, _INDENT)}'
+ '\n\n'
f'All commands:\n\n{textwrap.indent(all_rpcs, _INDENT)}'
f'\n\n{self.footer}'.strip())
@@ -67,7 +78,16 @@
return inspect.getdoc(item) or f'No documentation for {item!r}.'
- def __call__(self, item: Any = None) -> None:
+ def _variables_without_methods(self) -> Mapping[str, object]:
+ packages = frozenset(
+ n.split('.', 1)[0] for n in self._methods if '.' in n)
+
+ return {
+ name: var
+ for name, var in self._variables.items() if name not in packages
+ }
+
+ def __call__(self, item: object = None) -> None:
"""Prints the help string."""
print(self.help(item))
@@ -82,82 +102,104 @@
name: str
# An object to use in the console for the client. May be a pw_rpc.Client.
- client: Any
+ client: object
# The pw_rpc.Client; may be the same object as client.
rpc_client: pw_rpc.Client
-def multi_client_terminal_variables(clients: Collection[ClientInfo],
- default_client: Any,
- protos: python_protos.Library,
- help_header: str = '') -> Dict[str, Any]:
- """Creates a dict with variables for use in an interactive RPC console.
+class Context:
+ """The Context class is used to set up an interactive RPC console.
- Protos and RPC services are accessible by their proto package and name. The
- target for these can be set with the set_target function.
+ The Context manages a set of variables that make it easy to access RPCs and
+ protobufs in a REPL.
- For example, this function could be used to set up a console with IPython:
+ As an example, this class can be used to set up a console with IPython:
.. code-block:: python
- variables = console_tools.multi_client_terminal_variables(
- clients, default_client, protos, WELCOME_MESSAGE)
+ context = console_tools.Context(
+ clients, default_client, protos, help_header=WELCOME_MESSAGE)
IPython.terminal.embed.InteractiveShellEmbed().mainloop(
- module=types.SimpleNamespace(**variables))
-
- Args:
- clients: ClientInfo objects that represent the clients this console uses
- to communicate with other devices
- default_client: default ClientInfo object; must be one of the clients
- protos: protobufs to use for RPCs for all clients
- help_header: Message to display for the help command
-
- Returns:
- dict that maps names to variables for use in an RPC console
+ module=types.SimpleNamespace(**context.variables()))
"""
- assert clients, 'At least one client must be provided!'
+ def __init__(self,
+ client_info: Collection[ClientInfo],
+ default_client: Any,
+ protos: python_protos.Library,
+ *,
+ help_header: str = '') -> None:
+ """Creates an RPC console context.
- variables: Dict[str, Any] = dict(
- # Make pw_status.Status available in the console.
- Status=pw_status.Status,
- # Keep the original built-in help function as 'python_help'
- python_help=help,
- # Implement the custom help function using the CommandHelper.
- help=CommandHelper.from_methods(
- chain.from_iterable(c.rpc_client.methods() for c in clients),
- help_header,
- 'Type a command and hit Enter to see detailed help information.'),
- )
+ Protos and RPC services are accessible by their proto package and name.
+ The target for these can be set with the set_target function.
- # Make the RPC clients and protos available in the console.
- variables.update((c.name, c.client) for c in clients)
+ Args:
+ client_info: ClientInfo objects that represent the clients this
+ console uses to communicate with other devices
+ default_client: default client object; must be one of the clients
+ protos: protobufs to use for RPCs for all clients
+ help_header: Message to display for the help command
+ """
+ assert client_info, 'At least one client must be provided!'
- # Make the proto package hierarchy directly available in the console.
- for package in protos.packages:
- variables[package._package] = package # pylint: disable=protected-access
+ self.client_info = client_info
+ self.current_client = default_client
+ self.protos = protos
- # Monkey patch the message types to use an improved repr function.
- for message_type in protos.messages():
- message_type.__repr__ = python_protos.proto_repr
+ # Store objects with references to RPC services, sorted by package.
+ self._services: Dict[str, types.SimpleNamespace] = defaultdict(
+ types.SimpleNamespace)
- # Store objects with references to RPC services, sorted by package.
- services_by_package: Dict[str, types.SimpleNamespace] = defaultdict(
- types.SimpleNamespace)
+ self._variables: Dict[str, object] = dict(
+ Status=pw_status.Status,
+ set_target=functions.help_as_repr(self.set_target),
+ # The original built-in help function is available as 'python_help'.
+ python_help=help,
+ )
- def set_target(selected_client: Any, channel_id: int = None) -> None:
+ # Make the RPC clients and protos available in the console.
+ self._variables.update((c.name, c.client) for c in self.client_info)
+
+ # Make the proto package hierarchy directly available in the console.
+ for package in self.protos.packages:
+ self._variables[package._package] = package # pylint: disable=protected-access
+
+ # Monkey patch the message types to use an improved repr function.
+ for message_type in self.protos.messages():
+ message_type.__repr__ = python_protos.proto_repr
+
+ # Set up the 'help' command.
+ all_methods = chain.from_iterable(c.rpc_client.methods()
+ for c in self.client_info)
+ self._helper = CommandHelper.from_methods(
+ all_methods, self._variables, help_header,
+ 'Type a command and hit Enter to see detailed help information.')
+
+ self._variables['help'] = self._helper
+
+ # Call set_target to set up for the default target.
+ self.set_target(self.current_client)
+
+ def variables(self) -> Dict[str, Any]:
+ """Returns a mapping of names to variables for use in an RPC console."""
+ return self._variables
+
+ def set_target(self,
+ selected_client: object,
+ channel_id: int = None) -> None:
"""Sets the default target for commands."""
# Make sure the variable is one of the client variables.
name = ''
rpc_client: Any = None
- for name, client, rpc_client in clients:
+ for name, client, rpc_client in self.client_info:
if selected_client is client:
print('CURRENT RPC TARGET:', name)
break
else:
raise ValueError('Supported targets :' +
- ', '.join(c.name for c in clients))
+ ', '.join(c.name for c in self.client_info))
# Update the RPC services to use the newly selected target.
for service_client in rpc_client.channel(channel_id).rpcs:
@@ -167,16 +209,11 @@
method.response_type.__repr__ = python_protos.proto_repr
service = service_client._service # pylint: disable=protected-access
- setattr(services_by_package[service.package], service.name,
+ setattr(self._services[service.package], service.name,
service_client)
# Add the RPC methods to their proto packages.
- for package_name, rpcs in services_by_package.items():
- protos.packages[package_name]._add_item(rpcs) # pylint: disable=protected-access
+ for package_name, rpcs in self._services.items():
+ self.protos.packages[package_name]._add_item(rpcs) # pylint: disable=protected-access
- variables['set_target'] = set_target
-
- # Call set_target to set up for the default target.
- set_target(default_client)
-
- return variables
+ self.current_client = selected_client
diff --git a/pw_rpc/py/pw_rpc/console_tools/functions.py b/pw_rpc/py/pw_rpc/console_tools/functions.py
new file mode 100644
index 0000000..dea6251
--- /dev/null
+++ b/pw_rpc/py/pw_rpc/console_tools/functions.py
@@ -0,0 +1,90 @@
+# Copyright 2021 The Pigweed Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+"""Code for improving interactive use of Python functions."""
+
+import inspect
+import textwrap
+from typing import Callable
+
+
+def _annotation_name(annotation: object) -> str:
+ if isinstance(annotation, str):
+ return annotation
+
+ return getattr(annotation, '__name__', repr(annotation))
+
+
+def format_parameter(param: inspect.Parameter) -> str:
+ """Formats a parameter for printing in a function signature."""
+ if param.kind == param.VAR_POSITIONAL:
+ name = '*' + param.name
+ elif param.kind == param.VAR_KEYWORD:
+ name = '**' + param.name
+ else:
+ name = param.name
+
+ if param.default is param.empty:
+ default = ''
+ else:
+ default = f' = {param.default}'
+
+ if param.annotation is param.empty:
+ annotation = ''
+ else:
+ annotation = f': {_annotation_name(param.annotation)}'
+
+ return f'{name}{annotation}{default}'
+
+
+def format_signature(name: str, signature: inspect.Signature) -> str:
+ """Formats a function signature as if it were source code.
+
+ Does not yet handle / and * markers.
+ """
+ params = ', '.join(
+ format_parameter(arg) for arg in signature.parameters.values())
+ if signature.return_annotation is signature.empty:
+ return_annotation = ''
+ else:
+ return_annotation = ' -> ' + _annotation_name(
+ signature.return_annotation)
+
+ return f'{name}({params}){return_annotation}'
+
+
+def format_function_help(function: Callable) -> str:
+ """Formats a help string with a declaration and docstring."""
+ signature = format_signature(
+ function.__name__, inspect.signature(function, follow_wrapped=False))
+
+ docs = inspect.getdoc(function) or '(no docstring)'
+ return f'{signature}:\n\n{textwrap.indent(docs, " ")}'
+
+
+def help_as_repr(function: Callable) -> Callable:
+ """Wraps a function so that its repr() and docstring provide detailed help.
+
+ This is useful for creating commands in an interactive console. In a
+ console, typing a function's name and hitting Enter shows rich documentation
+ with the full function signature, type annotations, and docstring when the
+ function is wrapped with help_as_repr.
+ """
+ def display_help(_):
+ return format_function_help(function)
+
+ return type(
+ function.__name__, (),
+ dict(__call__=staticmethod(function),
+ __doc__=format_function_help(function),
+ __repr__=display_help))()
diff --git a/pw_rpc/py/tests/console_tools_test.py b/pw_rpc/py/tests/console_tools/console_tools_test.py
old mode 100644
new mode 100755
similarity index 68%
rename from pw_rpc/py/tests/console_tools_test.py
rename to pw_rpc/py/tests/console_tools/console_tools_test.py
index e34bbc3..3dbb5ce
--- a/pw_rpc/py/tests/console_tools_test.py
+++ b/pw_rpc/py/tests/console_tools/console_tools_test.py
@@ -22,8 +22,7 @@
from pw_protobuf_compiler import python_protos
import pw_rpc
from pw_rpc import callback_client
-from pw_rpc.console_tools import (CommandHelper, ClientInfo, Watchdog,
- multi_client_terminal_variables)
+from pw_rpc.console_tools import CommandHelper, Context, ClientInfo, Watchdog
class TestWatchdog(unittest.TestCase):
@@ -84,8 +83,9 @@
class TestCommandHelper(unittest.TestCase):
def setUp(self) -> None:
self._commands = {'command_a': 'A', 'command_B': 'B'}
- self._helper = CommandHelper(self._commands, 'The header',
- 'The footer')
+ self._variables = {'hello': 1, 'world': 2}
+ self._helper = CommandHelper(self._commands, self._variables,
+ 'The header', 'The footer')
def test_help_contents(self) -> None:
help_contents = self._helper.help()
@@ -93,6 +93,9 @@
self.assertTrue(help_contents.startswith('The header'))
self.assertIn('The footer', help_contents)
+ for var_name in self._variables:
+ self.assertIn(var_name, help_contents)
+
for cmd_name in self._commands:
self.assertIn(cmd_name, help_contents)
@@ -120,8 +123,8 @@
"""
-class TestMultiClientTerminal(unittest.TestCase):
- """Tests console_tools.console.multi_client_terminal_variables."""
+class TestConsoleContext(unittest.TestCase):
+ """Tests console_tools.console.Context."""
def setUp(self) -> None:
self._protos = python_protos.Library.from_strings(_PROTO)
@@ -133,10 +136,10 @@
], self._protos.modules()))
def test_sets_expected_variables(self) -> None:
- variables = multi_client_terminal_variables(
- clients=[self._info],
- default_client=self._info.client,
- protos=self._protos)
+ variables = Context([self._info],
+ default_client=self._info.client,
+ protos=self._protos).variables()
+
self.assertIn('set_target', variables)
self.assertIsInstance(variables['help'], CommandHelper)
@@ -154,63 +157,76 @@
[client_2_channel],
self._protos.modules()))
- variables = multi_client_terminal_variables(
- clients=[self._info, info_2],
- default_client=self._info.client,
- protos=self._protos)
+ context = Context([self._info, info_2],
+ default_client=self._info.client,
+ protos=self._protos)
# Make sure the RPC service switches from one client to the other.
- self.assertIs(variables['the'].pkg.Service.Unary.channel,
+ self.assertIs(context.variables()['the'].pkg.Service.Unary.channel,
client_1_channel)
- variables['set_target'](info_2.client)
+ context.set_target(info_2.client)
- self.assertIs(variables['the'].pkg.Service.Unary.channel,
+ self.assertIs(context.variables()['the'].pkg.Service.Unary.channel,
client_2_channel)
def test_default_client_must_be_in_clients(self) -> None:
with self.assertRaises(ValueError):
- multi_client_terminal_variables(clients=[self._info],
- default_client='something else',
- protos=self._protos)
+ Context([self._info],
+ default_client='something else',
+ protos=self._protos)
def test_set_target_invalid_channel(self) -> None:
- variables = multi_client_terminal_variables(
- clients=[self._info],
- default_client=self._info.client,
- protos=self._protos)
+ context = Context([self._info],
+ default_client=self._info.client,
+ protos=self._protos)
with self.assertRaises(KeyError):
- variables['set_target'](self._info.client, 100)
+ context.set_target(self._info.client, 100)
def test_set_target_non_default_channel(self) -> None:
channel_1 = self._info.rpc_client.channel(1).channel
channel_2 = self._info.rpc_client.channel(2).channel
- variables = multi_client_terminal_variables(
- clients=[self._info],
- default_client=self._info.client,
- protos=self._protos)
+ context = Context([self._info],
+ default_client=self._info.client,
+ protos=self._protos)
+ variables = context.variables()
self.assertIs(variables['the'].pkg.Service.Unary.channel, channel_1)
- variables['set_target'](self._info.client, 2)
+ context.set_target(self._info.client, 2)
self.assertIs(variables['the'].pkg.Service.Unary.channel, channel_2)
with self.assertRaises(KeyError):
- variables['set_target'](self._info.client, 100)
+ context.set_target(self._info.client, 100)
def test_set_target_requires_client_object(self) -> None:
- variables = multi_client_terminal_variables(
- clients=[self._info],
- default_client=self._info.client,
- protos=self._protos)
+ context = Context([self._info],
+ default_client=self._info.client,
+ protos=self._protos)
with self.assertRaises(ValueError):
- variables['set_target'](self._info.rpc_client)
+ context.set_target(self._info.rpc_client)
+ context.set_target(self._info.client)
+
+ def test_derived_context(self) -> None:
+ called_derived_set_target = False
+
+ class DerivedContext(Context):
+ def set_target(self,
+ unused_selected_client,
+ unused_channel_id: int = None) -> None:
+ nonlocal called_derived_set_target
+ called_derived_set_target = True
+
+ variables = DerivedContext(client_info=[self._info],
+ default_client=self._info.client,
+ protos=self._protos).variables()
variables['set_target'](self._info.client)
+ self.assertTrue(called_derived_set_target)
if __name__ == '__main__':
diff --git a/pw_rpc/py/tests/console_tools/functions_test.py b/pw_rpc/py/tests/console_tools/functions_test.py
new file mode 100644
index 0000000..85f352a
--- /dev/null
+++ b/pw_rpc/py/tests/console_tools/functions_test.py
@@ -0,0 +1,67 @@
+# Copyright 2021 The Pigweed Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+"""Tests the pw_rpc.console_tools.functions module."""
+
+import unittest
+
+from pw_rpc.console_tools import functions
+
+
+def func(one, two: int, *a: bool, three=3, four: 'int' = 4, **kw) -> None: # pylint: disable=unused-argument
+ """This is the docstring.
+
+ More stuff.
+ """
+
+
+_EXPECTED_HELP = """\
+func(one, two: int, *a: bool, three = 3, four: int = 4, **kw) -> None:
+
+ This is the docstring.
+
+ More stuff."""
+
+
+class TestFunctions(unittest.TestCase):
+ def test_format_no_args_function_help(self) -> None:
+ def simple_function():
+ pass
+
+ self.assertEqual(functions.format_function_help(simple_function),
+ 'simple_function():\n\n (no docstring)')
+
+ def test_format_complex_function_help(self) -> None:
+ self.assertEqual(functions.format_function_help(func), _EXPECTED_HELP)
+
+ def test_help_as_repr_with_docstring_help(self) -> None:
+ wrapped = functions.help_as_repr(func)
+ self.assertEqual(repr(wrapped), _EXPECTED_HELP)
+
+ def test_help_as_repr_decorator(self) -> None:
+ @functions.help_as_repr
+ def no_docs():
+ pass
+
+ self.assertEqual(repr(no_docs), 'no_docs():\n\n (no docstring)')
+
+ def test_help_as_repr_call_no_args(self) -> None:
+ self.assertEqual(functions.help_as_repr(lambda: 9876)(), 9876)
+
+ def test_help_as_repr_call_with_arg(self) -> None:
+ value = object()
+ self.assertIs(functions.help_as_repr(lambda arg: arg)(value), value)
+
+
+if __name__ == '__main__':
+ unittest.main()