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.
This commit is contained in:
Dave O'Connor 2021-01-27 12:17:24 +00:00 committed by Andrius Štikonas
parent e10b1de9ae
commit d21c54b4a6
1 changed files with 132 additions and 22 deletions

View File

@ -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__':