Specify type hints with defaults in NeoHub.

Implement the three methods with optional datetimes.

Implement permit_join with optional timeout.
This commit is contained in:
Dave O'Connor 2021-01-28 19:12:20 +00:00
parent d21c54b4a6
commit 53a3d0b5a2
2 changed files with 34 additions and 25 deletions

View File

@ -261,44 +261,32 @@ class NeoHub:
result = await self._send(message, reply)
return result
async def set_date(self, date=None):
async def set_date(self, date: datetime.datetime = datetime.datetime.today()):
"""
Sets current date
By default, set to current date. Can be optionally passed datetime.datetime object
"""
if date is None:
date = datetime.datetime.today()
else:
if not isinstance(date, datetime.datetime):
raise NeoHubUsageError('date must be datetime.datetime object')
message = {"SET_DATE": [date.year, date.month, date.day]}
reply = {"result": "Date is set"}
result = await self._send(message, reply)
return result
async def set_time(self, time=None):
async def set_time(self, time: datetime.datetime = datetime.datetime.now()):
"""
Sets current time
By default, set to current time. Can be optionally passed datetime.datetime object
"""
if time is None:
time = datetime.datetime.now()
else:
raise NeoHubUsageError('time must be datetime.datetime object')
message = {"SET_TIME": [time.hour, time.minute]}
reply = {"result": "time set"}
result = await self._send(message, reply)
return result
async def set_datetime(self, date_time=None):
async def set_datetime(self, date_time: datetime.datetime = datetime.datetime.now()):
"""
Convenience method to set both date and time
"""
@ -319,7 +307,7 @@ class NeoHub:
result = await self._send(message, reply)
return result
async def set_dst(self, state: bool, region=None):
async def set_dst(self, state: bool, region: str = None):
"""
Enables/disables automatic DST handling.
@ -332,7 +320,7 @@ class NeoHub:
valid_timezones = ["UK", "EU", "NZ"]
if region not in valid_timezones:
return False
raise NeoHubUsageError(f'region must be in {valid_timezones}')
result = await self._send(message, reply)
return result
@ -376,7 +364,7 @@ class NeoHub:
return hub_data, devices
async def permit_join(self, name, timeout_s=120):
async def permit_join(self, name, timeout_s: int = 120):
"""
Permit new thermostat to join network

View File

@ -11,6 +11,7 @@ import argparse
import datetime
from functools import partial
from neohubapi.neohub import NeoHub
from neohubapi.neohub import NeoHubUsageError
from neohubapi.neostat import NeoStat
from neohubapi.enums import ScheduleFormat
@ -68,7 +69,7 @@ class NeoHubCLI(object):
# 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()
return await special_method()
hubmethod = getattr(self._hub, self._command)
sig = inspect.signature(hubmethod)
@ -161,17 +162,35 @@ class NeoHubCLI(object):
else:
raise NeoHubCLIInternalError(f'Unknown type {type(argtype)} {argtype} for {self._command}')
def _callable_set_date(self):
"""Build a callable for set_date."""
async def _optional_datetime(self):
# 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')
print(f'{self._command} 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])
real_arg = await self._parse_arg(self._args[0], datetime.datetime)
return partial(getattr(self._hub, self._command), *[real_arg])
else:
return partial(getattr(self._hub, 'set_date'), *[datetime.datetime.today()])
return getattr(self._hub, self._command)
# These take a single datetime (optional) argument.
_callable_set_date = _optional_datetime
_callable_set_time = _optional_datetime
_callable_set_datetime = _optional_datetime
async def _callable_permit_join(self):
if len(self._args) > 2 or len(self._args) == 0:
print(f'{self._command} takes either 1 or 2 arguments')
return None
real_name = self._args[0]
args = [real_name]
if len(self._args) == 2:
timeout_s = await self._parse_arg(self._args[1], int)
args.append(timeout_s)
return partial(getattr(self._hub, 'permit_join'), *args)
def output(self, raw_result, output_format='json'):
"""Produce output in a desired format."""
@ -263,6 +282,8 @@ async def main():
if m:
result = await m()
print(nhc.output(result, output_format=args.format))
except NeoHubUsageError as e:
print(f'Invalid API usage: {e}')
except Error as e:
print(e)