From d21c54b4a6def66d2b0f0772b02a875ecae81f13 Mon Sep 17 00:00:00 2001 From: Dave O'Connor Date: Wed, 27 Jan 2021 12:17:24 +0000 Subject: [PATCH] Eliminate the boilerplate approach: - use method signatures to build a callable other fixes: Standardise capitalisation of NeoHub. Import inspect entirely. Still TODO: How to specify neostats on command line. --- scripts/neohub_cli.py | 154 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 132 insertions(+), 22 deletions(-) diff --git a/scripts/neohub_cli.py b/scripts/neohub_cli.py index 24d6505..50234de 100755 --- a/scripts/neohub_cli.py +++ b/scripts/neohub_cli.py @@ -6,36 +6,47 @@ import types import asyncio +import inspect import argparse -from inspect import ismethod, signature -from neohubapi import neohub +import datetime +from functools import partial +from neohubapi.neohub import NeoHub +from neohubapi.neostat import NeoStat +from neohubapi.enums import ScheduleFormat class Error(Exception): pass -class NeohubCLIUsageError(Error): +class NeoHubCLIUsageError(Error): pass -class NeohubCLIInternalError(Error): +class NeoHubCLIInternalError(Error): pass -class NeohubCLI(object): +class NeoHubCLIArgumentError(Error): + pass + + +class NeoHubCLI(object): """A runner for neohub_cli operations.""" def __init__(self, command, args, hub_ip=None, hub_port=4242): - self._hub = neohub.NeoHub(host=hub_ip, port=hub_port) + self._hub = NeoHub(host=hub_ip, port=hub_port) self._command = command self._args = args + # live data cached from the neohub. We assume this data will remain current + # throughout the execution of the script. + self._live_data = None if command == 'help': return if command not in self._hub_command_methods(): - raise NeohubCLIUsageError(f'Unknown command {command}') + raise NeoHubCLIUsageError(f'Unknown command {command}') def _hub_command_methods(self): """Return a list of NeoHub functions. @@ -44,27 +55,124 @@ class NeohubCLI(object): """ all_methods = [ - m for m in dir(self._hub) if ismethod(getattr(self._hub, m))] + m for m in dir(self._hub) if inspect.ismethod(getattr(self._hub, m))] return [m for m in all_methods if not m.startswith('_')] - def callable(self): + async def callable(self): """Return a bound callable for the method we want, or None.""" if self._command == 'help': print(self.get_help(self._args)) return None - # TODO(daveoc): Set special cases datetime, etc. + # Firstly, see if we have a separately-implemented exception below. + special_method = getattr(self, f'_callable_{self._command}', None) + if special_method: + return special_method() + hubmethod = getattr(self._hub, self._command) - sig = signature(hubmethod) + sig = inspect.signature(hubmethod) if len(sig.parameters) == 0: # No arguments return hubmethod - # TODO(daveoc): populate parameters and bind to methods that need them. - print(f'Cannot do {self._command} yet') + # See if we have the right numer of command line arguments. + if len(sig.parameters) != len(self._args): + print(f'Expecting {len(sig.parameters)} args for {self._command}, got {len(self._args)}') + print(self.get_help([self._command])) + return None + + # Next, see if all our arguments are annotated correctly. + # Several methods ask for [NeoStat] (i.e. a list of NeoStat objects). + # Cover for this case, without getting into expanding lists of other things. + # (Yes, this means we only support passing one neostat name per CLI command, for now). + arg_types = [] + for p in sig.parameters: + a = sig.parameters[p].annotation + if type(a) == type: + arg_types.append(a) + elif type(a) == list: + if a[0] == NeoStat: + arg_types.append(a[0]) + else: + raise NeoHubCLIInternalError(f'Command {self._command} argument not bindable: {a[0]}') + else: + raise NeoHubCLIInternalError(f'Unexpected para annotation in {self._command}: {a}') + + if inspect._empty not in arg_types: + # build a list comprising the 'real' objects represented. + real_args = [] + for i in range(len(self._args)): + try: + real_args.append(await self._parse_arg(self._args[i], arg_types[i])) + except NeoHubCLIArgumentError: + raise + except NeoHubCLIInternalError: + print('Internal Error:') + raise + return partial(getattr(self._hub, self._command), *real_args) + + # No special method and un-annotated params. + print(f'Cannot do {self._command} (yet)') return None + async def _parse_arg(self, arg, argtype): + """Return the desired type, given the string version of arg. + + This also does things liks parsing dates, etc. + """ + if argtype == int: + if not arg.isnumeric(): + raise NeoHubCLIArgumentError(f'argument to {self._command} must be numeric') + return int(arg) + elif argtype == str: + return str(arg) + elif argtype == bool: + if arg in ('1', 'True', 'true', 'on', 'y'): + away = True + elif arg in ('0', 'False', 'false', 'off', 'n'): + away = False + else: + raise NeoHubCLIArgumentError(f'\'{arg}\' not recognised as boolean') + return away + elif argtype == ScheduleFormat: + sf = getattr(ScheduleFormat, arg, None) + if not sf: + raise NeoHubCLIArgumentError( + f'argument must be in {[x.name for x in ScheduleFormat]}') + return sf + elif argtype == datetime.datetime: + try: + dt = datetime.datetime.strptime(arg, '%Y-%m-%d %H:%M:%S') + except ValueError: + raise NeoHubCLIArgumentError('dates must be in format "YYYY-MM-DD HH:MM:SS"') + return dt + elif argtype == NeoStat: + # We assume the cmdline argument is the thermostat name. + if not self._live_data: + # (hub_data, devices) + self._live_data = await self._hub.get_live_data() + found = [x for x in self._live_data[1]['thermostats'] if x.name == arg] + if found: + # methods always expect a list of these. + return [found[0]] + else: + raise NeoHubCLIArgumentError(f'No such thermostat: {arg}') + else: + raise NeoHubCLIInternalError(f'Unknown type {type(argtype)} {argtype} for {self._command}') + + def _callable_set_date(self): + """Build a callable for set_date.""" + # If we got an arg, assume it's a date, otherwise use today. + if len(self._args) > 1: + print('set_date takes zero or one argument') + return None + elif len(self._args) == 1: + real_arg = self._parse_arg(self._args[0], datetime.datetime) + return partial(getattr(self._hub, 'set_date'), *[real_arg]) + else: + return partial(getattr(self._hub, 'set_date'), *[datetime.datetime.today()]) + def output(self, raw_result, output_format='json'): """Produce output in a desired format.""" if output_format == 'raw': @@ -76,12 +184,15 @@ class NeohubCLI(object): if special_case: return special_case(raw_result, output_format) + if type(raw_result) == bool: + return 'Command Succeeded' if raw_result else 'Command Failed' + if type(raw_result) in (int, str): return f'{raw_result}' if type(raw_result) != types.SimpleNamespace: - raise NeohubCLIInternalError( - 'Unexpected type {type(raw_result)} in output()') + raise NeoHubCLIInternalError( + f'Unexpected type {type(raw_result)} in output()') return self._output_simplenamespace(raw_result, output_format) @@ -93,7 +204,7 @@ class NeohubCLI(object): if not a.startswith('_')]) return '\n'.join([f'{a}: {attrs[a]}' for a in attrs]) else: - raise NeohubCLIUsageError(f'Unknown output format {output_format}') + raise NeoHubCLIUsageError(f'Unknown output format {output_format}') def _output_get_live_data(self, raw_result, output_format): """Return special case output for get_live_data.""" @@ -119,7 +230,7 @@ class NeohubCLI(object): return f'Command {args[0]} not known' docstr = getattr(self._hub, args[0]).__doc__ or 'No help for {args[0]}' - sig = signature(getattr(self._hub, args[0])) + sig = inspect.signature(getattr(self._hub, args[0])) ret = f'{args[0]}:\n' @@ -134,27 +245,26 @@ class NeohubCLI(object): async def main(): argp = argparse.ArgumentParser(description='CLI to neohub devices') - argp.add_argument('--hub_ip', help='IP address of Neohub', default=None) + argp.add_argument('--hub_ip', help='IP address of NeoHub', default=None) argp.add_argument( - '--hub_port', help='Port number of Neohub to talk to', default=4242) + '--hub_port', help='Port number of NeoHub to talk to', default=4242) argp.add_argument('--format', help='Output format', default='list') argp.add_argument('command', help='Command to issue') argp.add_argument('arg', help='Arguments to command', nargs='*') args = argp.parse_args() try: - nhc = NeohubCLI( + nhc = NeoHubCLI( args.command, args.arg, hub_ip=args.hub_ip, hub_port=args.hub_port) - m = nhc.callable() + m = await nhc.callable() if m: result = await m() print(nhc.output(result, output_format=args.format)) except Error as e: print(e) - argp.print_help() if __name__ == '__main__':