diff --git a/autodesk/api.py b/autodesk/api.py index 54bf780..f8f29b8 100644 --- a/autodesk/api.py +++ b/autodesk/api.py @@ -13,50 +13,49 @@ async def route_set_session(request): body = await request.text() try: state = deserialize_session(body) - request.app['service'].set_session(state) + request.app["service"].set_session(state) return web.Response() except ValueError as error: return web.Response(text=str(error), status=400) async def route_get_session(request): - state = request.app['service'].get_session_state() - return web.Response(text=state.test('inactive', 'active')) + state = request.app["service"].get_session_state() + return web.Response(text=state.test("inactive", "active")) async def route_set_desk(request): body = await request.text() try: state = deserialize_desk(body) - okay = request.app['service'].set_desk(state) + okay = request.app["service"].set_desk(state) return web.Response(status=200 if okay else 403) except ValueError as error: return web.Response(text=str(error), status=400) async def route_get_desk(request): - state = request.app['service'].get_desk_state() - return web.Response(text=state.test('down', 'up')) + state = request.app["service"].get_desk_state() + return web.Response(text=state.test("down", "up")) -@aiohttp_jinja2.template('index.html') +@aiohttp_jinja2.template("index.html") async def route_index(request): - service = request.app['service'] - session_state = service.get_session_state().test('inactive', 'active') - desk_state = service.get_desk_state().test('down', 'up') + service = request.app["service"] + session_state = service.get_session_state().test("inactive", "active") + desk_state = service.get_desk_state().test("down", "up") active_time = service.get_active_time() - counts_figure = plots.plot_weekday_hourly_count( - service.compute_hourly_count()) + counts_figure = plots.plot_weekday_hourly_count(service.compute_hourly_count()) return { - 'session': session_state, - 'desk': desk_state, - 'active_time': active_time, - 'statistics': plots.figure_to_base64(counts_figure) + "session": session_state, + "desk": desk_state, + "active_time": active_time, + "statistics": plots.figure_to_base64(counts_figure), } async def poll_button(button, polling_delay, hardware_error_delay): - logger = logging.getLogger('poll_button') + logger = logging.getLogger("poll_button") while True: try: button.poll() @@ -68,37 +67,36 @@ async def poll_button(button, polling_delay, hardware_error_delay): async def init(app): loop = asyncio.get_running_loop() - service = app['factory'].create(loop) + service = app["factory"].create(loop) service.init() - button = Button(app['button_pin'], service) - app['poll_button_task'] = loop.create_task(poll_button( - button, app['button_polling_delay'], app['hardware_error_delay'])) - del app['button_pin'] - del app['factory'] - app['service'] = service + button = Button(app["button_pin"], service) + app["poll_button_task"] = loop.create_task( + poll_button(button, app["button_polling_delay"], app["hardware_error_delay"]) + ) + del app["button_pin"] + del app["factory"] + app["service"] = service async def cleanup(app): - app['poll_button_task'].cancel() + app["poll_button_task"].cancel() -def setup_app( - button_pin, factory, button_polling_delay=0.1, hardware_error_delay=5): +def setup_app(button_pin, factory, button_polling_delay=0.1, hardware_error_delay=5): app = web.Application() - app['button_polling_delay'] = button_polling_delay - app['hardware_error_delay'] = hardware_error_delay - app['button_pin'] = button_pin - app['factory'] = factory + app["button_polling_delay"] = button_polling_delay + app["hardware_error_delay"] = hardware_error_delay + app["button_pin"] = button_pin + app["factory"] = factory - aiohttp_jinja2.setup( - app, loader=jinja2.PackageLoader('autodesk', 'templates')) + aiohttp_jinja2.setup(app, loader=jinja2.PackageLoader("autodesk", "templates")) - app.router.add_get('/', route_index) - app.router.add_get('/api/session', route_get_session) - app.router.add_put('/api/session', route_set_session) - app.router.add_get('/api/desk', route_get_desk) - app.router.add_put('/api/desk', route_set_desk) + app.router.add_get("/", route_index) + app.router.add_get("/api/session", route_get_session) + app.router.add_put("/api/session", route_set_session) + app.router.add_get("/api/desk", route_get_desk) + app.router.add_put("/api/desk", route_set_desk) app.on_startup.append(init) app.on_cleanup.append(cleanup) diff --git a/autodesk/application/autodeskservice.py b/autodesk/application/autodeskservice.py index d24a2f9..0acb558 100644 --- a/autodesk/application/autodeskservice.py +++ b/autodesk/application/autodeskservice.py @@ -3,9 +3,10 @@ class AutoDeskService: - def __init__(self, operation, scheduler, timer, time_service, - session_service, desk_service): - self.logger = logging.getLogger('autodeskservice') + def __init__( + self, operation, scheduler, timer, time_service, session_service, desk_service + ): + self.logger = logging.getLogger("autodeskservice") self.operation = operation self.timer = timer self.scheduler = scheduler @@ -18,7 +19,7 @@ def init(self): self.session_service.init() self._update_timer() except HardwareError as error: - self.logger.warning('hardware failure, timer not started') + self.logger.warning("hardware failure, timer not started") self.logger.debug(error) self.timer.cancel() @@ -30,7 +31,7 @@ def set_session(self, state): self.session_service.set(state) self._update_timer() except HardwareError as error: - self.logger.warning('hardware failure, timer cancelled') + self.logger.warning("hardware failure, timer cancelled") self.logger.debug(error) self.timer.cancel() @@ -46,7 +47,7 @@ def set_desk(self, state): self.desk_service.set(state) self._update_timer() except HardwareError as error: - self.logger.warning('hardware failure, timer cancelled') + self.logger.warning("hardware failure, timer cancelled") self.logger.debug(error) self.timer.cancel() @@ -64,6 +65,7 @@ def _update_timer(self): active_time = self.session_service.get_active_time() self.timer.schedule( self.scheduler.compute_delay(active_time, desk_state), - lambda: self.set_desk(desk_state.next())) + lambda: self.set_desk(desk_state.next()), + ) else: self.timer.cancel() diff --git a/autodesk/application/autodeskservicefactory.py b/autodesk/application/autodeskservicefactory.py index 1a6b05b..ff4ad61 100644 --- a/autodesk/application/autodeskservicefactory.py +++ b/autodesk/application/autodeskservicefactory.py @@ -12,8 +12,9 @@ class AutoDeskServiceFactory: - def __init__(self, database_path, pin_factory, limits, delay, motor_pins, - light_pins): + def __init__( + self, database_path, pin_factory, limits, delay, motor_pins, light_pins + ): self.database_path = database_path self.pin_factory = pin_factory self.limits = limits @@ -30,14 +31,14 @@ def create(self, loop): self.delay, self.pin_factory.create_output(self.motor_pins[0]), self.pin_factory.create_output(self.motor_pins[1]), - self.pin_factory.create_output(self.light_pins[0])) + self.pin_factory.create_output(self.light_pins[0]), + ) light_controller = LightController( - self.pin_factory.create_output(self.light_pins[1])) + self.pin_factory.create_output(self.light_pins[1]) + ) timer_service = TimeService() - session_service = SessionService( - model, light_controller, timer_service) - desk_service = DeskService( - operation, model, desk_controller, timer_service) + session_service = SessionService(model, light_controller, timer_service) + desk_service = DeskService(operation, model, desk_controller, timer_service) return AutoDeskService( - operation, scheduler, timer, timer_service, session_service, - desk_service) + operation, scheduler, timer, timer_service, session_service, desk_service + ) diff --git a/autodesk/application/deskservice.py b/autodesk/application/deskservice.py index 86ef9de..8b18fc1 100644 --- a/autodesk/application/deskservice.py +++ b/autodesk/application/deskservice.py @@ -4,7 +4,7 @@ class DeskService: def __init__(self, operation, model, desk_controller, time_service): - self.logger = logging.getLogger('deskservice') + self.logger = logging.getLogger("deskservice") self.operation = operation self.model = model self.desk_controller = desk_controller @@ -17,14 +17,13 @@ def set(self, desk): now = self.time_service.now() session = self.model.get_session_state() if not self.operation.allowed(session, now): - self.logger.warning('desk operation not allowed at this time') + self.logger.warning("desk operation not allowed at this time") return try: self.desk_controller.move(desk) self.model.set_desk(self.time_service.now(), desk) except HardwareError as error: - self.logger.error( - 'hardware failure, desk state not updated in model') + self.logger.error("hardware failure, desk state not updated in model") self.logger.debug(error) raise diff --git a/autodesk/application/sessionservice.py b/autodesk/application/sessionservice.py index 47767da..b210878 100644 --- a/autodesk/application/sessionservice.py +++ b/autodesk/application/sessionservice.py @@ -4,7 +4,7 @@ class SessionService: def __init__(self, model, light_controller, time_service): - self.logger = logging.getLogger('sessionservice') + self.logger = logging.getLogger("sessionservice") self.model = model self.light_controller = light_controller self.time_service = time_service @@ -18,11 +18,13 @@ def get(self): def get_active_time(self): return self.model.get_active_time( - self.time_service.min, self.time_service.now()) + self.time_service.min, self.time_service.now() + ) def compute_hourly_count(self): return self.model.compute_hourly_count( - self.time_service.min, self.time_service.now()) + self.time_service.min, self.time_service.now() + ) def set(self, state): now = self.time_service.now() @@ -30,7 +32,6 @@ def set(self, state): try: self.light_controller.set(state) except HardwareError as error: - self.logger.warning( - 'hardware failure, could not set session light') + self.logger.warning("hardware failure, could not set session light") self.logger.debug(error) raise diff --git a/autodesk/hardware/__init__.py b/autodesk/hardware/__init__.py index aab3f90..3bf589e 100644 --- a/autodesk/hardware/__init__.py +++ b/autodesk/hardware/__init__.py @@ -3,29 +3,32 @@ def create_pin_factory(kind): - logger = logging.getLogger('hardware') + logger = logging.getLogger("hardware") - if kind == 'raspberrypi': - logger.info('using rpi hardware') + if kind == "raspberrypi": + logger.info("using rpi hardware") return LoggingPinFactory(create_raspberry_pi()) - if kind == 'ft232h': - logger.info('using ft232h hardware') + if kind == "ft232h": + logger.info("using ft232h hardware") return LoggingPinFactory(create_ft232h()) - logger.info('using noop hardware') + logger.info("using noop hardware") return LoggingPinFactory(create_noop()) def create_raspberry_pi(): from autodesk.hardware.raspberrypi import RaspberryPiPinFactory + return RaspberryPiPinFactory() def create_ft232h(): from autodesk.hardware.ft232h import Ft232hPinFactory + return Ft232hPinFactory() def create_noop(): from autodesk.hardware.noop import NoopPinFactory + return NoopPinFactory() diff --git a/autodesk/hardware/ft232h.py b/autodesk/hardware/ft232h.py index a182af9..1316308 100644 --- a/autodesk/hardware/ft232h.py +++ b/autodesk/hardware/ft232h.py @@ -22,7 +22,7 @@ def __init__(self): def _setup(self): try: self.controller.configure( - 'ftdi://ftdi:ft232h/1', + "ftdi://ftdi:ft232h/1", frequency=100, direction=0xFFFF, initial=0x0000, @@ -32,10 +32,8 @@ def _setup(self): used_pins = self.output_pins | self.input_pins invalid_pins = ((valid_pins | used_pins) & ~valid_pins) & used_pins if invalid_pins != 0: - formatted = \ - [i for i in range(0, 16) if invalid_pins & 1 << i != 0] - raise HardwareError( - "Cannot use pin(s) {} as GPIO.".format(formatted)) + formatted = [i for i in range(0, 16) if invalid_pins & 1 << i != 0] + raise HardwareError("Cannot use pin(s) {} as GPIO.".format(formatted)) # A low bit (equal to 0) indicates an input pin. # A high bit (equal to 1) indicates an output pin. new_direction = self.output_pins & ~self.input_pins @@ -93,12 +91,10 @@ def _reconnect_and_try_again(self, action): raise HardwareError(error1) def read(self, pin): - return self._reconnect_and_try_again( - lambda: self._read_no_error_handling(pin)) + return self._reconnect_and_try_again(lambda: self._read_no_error_handling(pin)) def write(self, pin, value): - self._reconnect_and_try_again( - lambda: self._write_no_error_handling(pin, value)) + self._reconnect_and_try_again(lambda: self._write_no_error_handling(pin, value)) class Ft232hOutputPin: @@ -108,8 +104,7 @@ def __init__(self, wrapper, pin): def write(self, value): if value != 0 and value != 1: - raise ValueError( - 'Pin value must be 0 or 1 but got {0}'.format(value)) + raise ValueError("Pin value must be 0 or 1 but got {0}".format(value)) self.wrapper.write(self.pin, value) diff --git a/autodesk/hardware/logging.py b/autodesk/hardware/logging.py index 501bfd3..ab9b14d 100644 --- a/autodesk/hardware/logging.py +++ b/autodesk/hardware/logging.py @@ -3,32 +3,32 @@ class LoggingPin: def __init__(self, inner): - self.logger = logging.getLogger('hardware') + self.logger = logging.getLogger("hardware") self.inner = inner def read(self): value = self.inner.read() - self.logger.debug('read %d %d', self.inner.pin, value) + self.logger.debug("read %d %d", self.inner.pin, value) return value def write(self, value): - self.logger.info('write %d %d', self.inner.pin, value) + self.logger.info("write %d %d", self.inner.pin, value) self.inner.write(value) class LoggingPinFactory: def __init__(self, inner): - self.logger = logging.getLogger('hardware') + self.logger = logging.getLogger("hardware") self.inner = inner def close(self): - self.logger.info('close') + self.logger.info("close") self.inner.close() def create_input(self, pin): - self.logger.info('create %d', pin) + self.logger.info("create %d", pin) return LoggingPin(self.inner.create_input(pin)) def create_output(self, pin): - self.logger.info('create %d', pin) + self.logger.info("create %d", pin) return LoggingPin(self.inner.create_output(pin)) diff --git a/autodesk/hardware/raspberrypi.py b/autodesk/hardware/raspberrypi.py index de441b6..4b23b09 100644 --- a/autodesk/hardware/raspberrypi.py +++ b/autodesk/hardware/raspberrypi.py @@ -19,8 +19,7 @@ def __init__(self, pin): def write(self, value): if value != 0 and value != 1: - raise ValueError( - 'Pin value must be 0 or 1 but got {0}'.format(value)) + raise ValueError("Pin value must be 0 or 1 but got {0}".format(value)) gpio_value = GPIO.LOW if value == 0 else GPIO.HIGH GPIO.output(self.pin, gpio_value) diff --git a/autodesk/model.py b/autodesk/model.py index a9e6c7e..08a07bd 100644 --- a/autodesk/model.py +++ b/autodesk/model.py @@ -34,7 +34,7 @@ def cut(start, end, spans): yield ( start if span.start < start else span.start, end if span.end > end else span.end, - span.state + span.state, ) @@ -56,18 +56,18 @@ def get_desk_spans(self, initial, final): default_state=DOWN, initial=initial, final=final, - events=self.datastore.get_desk_events()) - return pd.DataFrame.from_records( - spans, columns=['start', 'end', 'state']) + events=self.datastore.get_desk_events(), + ) + return pd.DataFrame.from_records(spans, columns=["start", "end", "state"]) def get_session_spans(self, initial, final): spans = collect( default_state=INACTIVE, initial=initial, final=final, - events=self.datastore.get_session_events()) - return pd.DataFrame.from_records( - spans, columns=['start', 'end', 'state']) + events=self.datastore.get_session_events(), + ) + return pd.DataFrame.from_records(spans, columns=["start", "end", "state"]) def get_session_state(self): events = self.datastore.get_session_events() @@ -84,10 +84,10 @@ def get_active_time(self, initial, final): return Timedelta(0) desk_spans = self.get_desk_spans(initial, final) - current_spans = pd.DataFrame.from_records(cut( - desk_spans.iloc[-1].start, - desk_spans.iloc[-1].end, - session_spans), columns=['start', 'end', 'state']) + current_spans = pd.DataFrame.from_records( + cut(desk_spans.iloc[-1].start, desk_spans.iloc[-1].end, session_spans), + columns=["start", "end", "state"], + ) active_spans = current_spans[current_spans.state == ACTIVE] return (active_spans.end - active_spans.start).sum() @@ -97,20 +97,25 @@ def compute_hourly_count(self, initial, final): rows = np.zeros((7 * 24)) for span in spans[spans.state == ACTIVE].itertuples(): - for (day, hour) in enumerate_hours(span.start, span.end): + for day, hour in enumerate_hours(span.start, span.end): rows[day * 24 + hour] += 1 - weekdays = pd.Series([ - 'Monday', - 'Tuesday', - 'Wednesday', - 'Thursday', - 'Friday', - 'Saturday', - 'Sunday' - ], dtype='string') - return pd.DataFrame({ - 'weekday': np.repeat(weekdays, 24), - 'hour': np.tile(np.arange(24), 7), - 'counts': rows - }) + weekdays = pd.Series( + [ + "Monday", + "Tuesday", + "Wednesday", + "Thursday", + "Friday", + "Saturday", + "Sunday", + ], + dtype="string", + ) + return pd.DataFrame( + { + "weekday": np.repeat(weekdays, 24), + "hour": np.tile(np.arange(24), 7), + "counts": rows, + } + ) diff --git a/autodesk/operation.py b/autodesk/operation.py index f10512b..5686b8a 100644 --- a/autodesk/operation.py +++ b/autodesk/operation.py @@ -5,7 +5,7 @@ class Operation: def __init__(self): - self.logger = logging.getLogger('operation') + self.logger = logging.getLogger("operation") self.allowance_start = time(7, 0, 0) self.allowance_end = time(20, 0, 0) @@ -17,7 +17,9 @@ def _check_time(self, at): friday = 4 time_at = time(at.hour, at.minute, at.second) weekday = at.weekday() - return \ - time_at >= self.allowance_start and \ - time_at <= self.allowance_end and \ - weekday >= monday and weekday <= friday + return ( + time_at >= self.allowance_start + and time_at <= self.allowance_end + and weekday >= monday + and weekday <= friday + ) diff --git a/autodesk/plots.py b/autodesk/plots.py index 383698c..4ba8beb 100644 --- a/autodesk/plots.py +++ b/autodesk/plots.py @@ -5,29 +5,27 @@ def figure_to_base64(figure): tmpfile = BytesIO() - metadata = {'Software': 'AutoDesk https://github.com/daoo/autodesk'} - figure.savefig(tmpfile, format='png', metadata=metadata) - return base64.b64encode(tmpfile.getvalue()).decode('utf-8') + metadata = {"Software": "AutoDesk https://github.com/daoo/autodesk"} + figure.savefig(tmpfile, format="png", metadata=metadata) + return base64.b64encode(tmpfile.getvalue()).decode("utf-8") def plot_weekday_hourly_count(full_week, dpi=80): - working_days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday'] + working_days = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"] filtered = full_week[ (full_week.weekday.isin(working_days)) & (full_week.hour >= 6) & (full_week.hour <= 20) ] - counts = filtered['counts'] + counts = filtered["counts"] relative = (counts - counts.min()) / (counts.max() - counts.min()) figure = plt.figure(figsize=(800 / dpi, 400 / dpi), dpi=dpi) - figure.suptitle('Weekday Hourly Relative Presence') + figure.suptitle("Weekday Hourly Relative Presence") ax = figure.add_subplot() - ax.set_xlabel('Hour') + ax.set_xlabel("Hour") ax.scatter( - x=filtered.hour, - y=filtered.weekday.index, - s=relative * 500, - color='grey') + x=filtered.hour, y=filtered.weekday.index, s=relative * 500, color="grey" + ) ax.set_yticks(range(len(working_days)), working_days) return figure diff --git a/autodesk/program.py b/autodesk/program.py index 9c56d08..b7220e3 100644 --- a/autodesk/program.py +++ b/autodesk/program.py @@ -9,40 +9,39 @@ import yaml logging.basicConfig( - level=logging.INFO, - format='%(asctime)s %(name)s %(levelname)s %(message)s' + level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s" ) -logger = logging.getLogger('program') +logger = logging.getLogger("program") -config_path = os.getenv('AUTODESK_CONFIG', 'config/testing.yml') -database = os.getenv('AUTODESK_DATABASE', ':memory:') -address = os.getenv('AUTODESK_ADDRESS', '127.0.0.1') -port = int(os.getenv('AUTODESK_PORT', '8080')) +config_path = os.getenv("AUTODESK_CONFIG", "config/testing.yml") +database = os.getenv("AUTODESK_DATABASE", ":memory:") +address = os.getenv("AUTODESK_ADDRESS", "127.0.0.1") +port = int(os.getenv("AUTODESK_PORT", "8080")) logger.info('Reading config "%s"', config_path) config = None -with open(config_path, 'r') as file: +with open(config_path, "r") as file: config = yaml.load(file, Loader=yaml.SafeLoader) -with closing(create_pin_factory(config['hardware'])) as pin_factory: - button_pin = pin_factory.create_input(config['button_pin']) +with closing(create_pin_factory(config["hardware"])) as pin_factory: + button_pin = pin_factory.create_input(config["button_pin"]) factory = AutoDeskServiceFactory( database, pin_factory, ( - Timedelta(seconds=config['limits']['down']), - Timedelta(seconds=config['limits']['up']), + Timedelta(seconds=config["limits"]["down"]), + Timedelta(seconds=config["limits"]["up"]), ), - config['delay'], + config["delay"], ( - config['motor_pins']['down'], - config['motor_pins']['up'], + config["motor_pins"]["down"], + config["motor_pins"]["up"], ), ( - config['light_pins']['desk'], - config['light_pins']['session'], + config["light_pins"]["desk"], + config["light_pins"]["session"], ), ) app = setup_app(button_pin, factory) diff --git a/autodesk/sqlitedatastore.py b/autodesk/sqlitedatastore.py index 243edff..d619461 100644 --- a/autodesk/sqlitedatastore.py +++ b/autodesk/sqlitedatastore.py @@ -4,28 +4,29 @@ import sqlite3 -sqlite3.register_adapter(states.Down, lambda _: 'down') -sqlite3.register_adapter(states.Up, lambda _: 'up') -sqlite3.register_adapter(states.Inactive, lambda _: 'inactive') -sqlite3.register_adapter(states.Active, lambda _: 'active') -sqlite3.register_converter('desk', states.deserialize_desk) -sqlite3.register_converter('session', states.deserialize_session) +sqlite3.register_adapter(states.Down, lambda _: "down") +sqlite3.register_adapter(states.Up, lambda _: "up") +sqlite3.register_adapter(states.Inactive, lambda _: "inactive") +sqlite3.register_adapter(states.Active, lambda _: "active") +sqlite3.register_converter("desk", states.deserialize_desk) +sqlite3.register_converter("session", states.deserialize_session) class SqliteDataStore: def __init__(self, path): - self.logger = logging.getLogger('sqlite3') - self.logger.info('Opening database %s', path) - self.connection = sqlite3.connect( - path, detect_types=sqlite3.PARSE_DECLTYPES) + self.logger = logging.getLogger("sqlite3") + self.logger.info("Opening database %s", path) + self.connection = sqlite3.connect(path, detect_types=sqlite3.PARSE_DECLTYPES) self.connection.execute( - 'CREATE TABLE IF NOT EXISTS session(' - 'date TIMESTAMP NOT NULL,' - 'state SESSION NOT NULL)') + "CREATE TABLE IF NOT EXISTS session(" + "date TIMESTAMP NOT NULL," + "state SESSION NOT NULL)" + ) self.connection.execute( - 'CREATE TABLE IF NOT EXISTS desk(' - 'date TIMESTAMP NOT NULL,' - 'state DESK NOT NULL)') + "CREATE TABLE IF NOT EXISTS desk(" + "date TIMESTAMP NOT NULL," + "state DESK NOT NULL)" + ) def close(self): self.connection.close() @@ -34,25 +35,19 @@ def _get(self, query): return pd.read_sql_query(query, self.connection) def get_desk_events(self): - return self._get('SELECT * FROM desk ORDER BY date ASC') + return self._get("SELECT * FROM desk ORDER BY date ASC") def get_session_events(self): - return self._get('SELECT * FROM session ORDER BY date ASC') + return self._get("SELECT * FROM session ORDER BY date ASC") def set_desk(self, date, state): - self.logger.debug( - 'set desk %s %s', - date, - state.test('down', 'up')) + self.logger.debug("set desk %s %s", date, state.test("down", "up")) values = (date.to_pydatetime(), state) - self.connection.execute('INSERT INTO desk values(?, ?)', values) + self.connection.execute("INSERT INTO desk values(?, ?)", values) self.connection.commit() def set_session(self, date, state): - self.logger.debug( - 'set session %s %s', - date, - state.test('inactive', 'active')) + self.logger.debug("set session %s %s", date, state.test("inactive", "active")) values = (date.to_pydatetime(), state) - self.connection.execute('INSERT INTO session values(?, ?)', values) + self.connection.execute("INSERT INTO session values(?, ?)", values) self.connection.commit() diff --git a/autodesk/states.py b/autodesk/states.py index 8f91eb9..98fa35b 100644 --- a/autodesk/states.py +++ b/autodesk/states.py @@ -50,18 +50,18 @@ def __eq__(self, other): def deserialize_session(value): - if value in (b'inactive', 'inactive'): + if value in (b"inactive", "inactive"): return INACTIVE - if value in (b'active', 'active'): + if value in (b"active", "active"): return ACTIVE raise ValueError('Incorrect session state "{0}".'.format(value)) def deserialize_desk(value): - if value in (b'down', 'down'): + if value in (b"down", "down"): return DOWN - if value in (b'up', 'up'): + if value in (b"up", "up"): return UP raise ValueError('Incorrect desk state "{0}".'.format(value)) diff --git a/autodesk/timer.py b/autodesk/timer.py index 55336d2..de8c1be 100644 --- a/autodesk/timer.py +++ b/autodesk/timer.py @@ -3,18 +3,18 @@ class Timer: def __init__(self, loop): - self.logger = logging.getLogger('timer') + self.logger = logging.getLogger("timer") self.loop = loop self.timer = None def schedule(self, delay, callback): - self.logger.info('scheduling in %s', delay) + self.logger.info("scheduling in %s", delay) if self.timer: self.timer.cancel() self.timer = self.loop.call_later(delay.total_seconds(), callback) def cancel(self): - self.logger.info('cancling') + self.logger.info("cancling") if self.timer: self.timer.cancel() self.timer = None diff --git a/bin/ft232h-button-test.py b/bin/ft232h-button-test.py index 4c0a1e0..be6a590 100755 --- a/bin/ft232h-button-test.py +++ b/bin/ft232h-button-test.py @@ -6,10 +6,7 @@ delay = 0.1 controller = GpioMpsseController() try: - controller.configure( - "ftdi://ftdi:ft232h/1", - frequency=1e6, - direction=0x0000) + controller.configure("ftdi://ftdi:ft232h/1", frequency=1e6, direction=0x0000) gpio = controller.get_gpio() while True: print(gpio.read()[0] >> pin & 1) diff --git a/bin/ft232h-control.py b/bin/ft232h-control.py index 910b8f3..28fdec2 100755 --- a/bin/ft232h-control.py +++ b/bin/ft232h-control.py @@ -5,10 +5,10 @@ def toggle(gpio, delay): gpio.write(0x0000 & gpio.direction) - print('off', bin(gpio.read()[0])) + print("off", bin(gpio.read()[0])) time.sleep(delay) gpio.write(0xFFFF & gpio.direction) - print('on ', bin(gpio.read()[0])) + print("on ", bin(gpio.read()[0])) time.sleep(delay) @@ -16,10 +16,8 @@ def toggle(gpio, delay): controller = GpioMpsseController() try: controller.configure( - "ftdi://ftdi:ft232h/1", - frequency=100, - direction=0xFFFF, - initial=0x0000) + "ftdi://ftdi:ft232h/1", frequency=100, direction=0xFFFF, initial=0x0000 + ) gpio = controller.get_gpio() while True: toggle(gpio, delay) diff --git a/bin/ft232h-list.py b/bin/ft232h-list.py index c7c4c1a..a61eb2b 100755 --- a/bin/ft232h-list.py +++ b/bin/ft232h-list.py @@ -1,7 +1,7 @@ #!/usr/bin/env python from pyftdi.ftdi import Ftdi -ft232h_pid = (Ftdi.PRODUCT_IDS[Ftdi.FTDI_VENDOR]['ft232h']) +ft232h_pid = Ftdi.PRODUCT_IDS[Ftdi.FTDI_VENDOR]["ft232h"] id_tuple = (Ftdi.FTDI_VENDOR, ft232h_pid) devices = Ftdi.find_all([id_tuple]) @@ -9,4 +9,4 @@ for device in devices: print(device) else: - print('No FT232H devices found.') + print("No FT232H devices found.") diff --git a/bin/ft232h-reconnect-test.py b/bin/ft232h-reconnect-test.py index 277cd2f..ff0cb5e 100755 --- a/bin/ft232h-reconnect-test.py +++ b/bin/ft232h-reconnect-test.py @@ -3,30 +3,26 @@ import traceback from pyftdi.usbtools import UsbTools -print('Creating, configuring, writing and closing a GpioMpsseController...') +print("Creating, configuring, writing and closing a GpioMpsseController...") controller = GpioMpsseController() controller.configure( - 'ftdi://ftdi:ft232h/1', - frequency=100, - direction=0xFFFF, - initial=0x0000) + "ftdi://ftdi:ft232h/1", frequency=100, direction=0xFFFF, initial=0x0000 +) gpio = controller.get_gpio() gpio.write(0xFFFF & gpio.direction) value = gpio.read()[0] -print(' Success (read {0})!'.format(hex(value))) +print(" Success (read {0})!".format(hex(value))) -input('Now, disconnect and reconnect hardware and press enter.') +input("Now, disconnect and reconnect hardware and press enter.") try: UsbTools.release_device(controller._ftdi._usb_dev) controller.close() UsbTools.flush_cache() controller.configure( - 'ftdi://ftdi:ft232h/1', - frequency=100, - direction=0xFFFF, - initial=0x0000) + "ftdi://ftdi:ft232h/1", frequency=100, direction=0xFFFF, initial=0x0000 + ) gpio = controller.get_gpio() gpio.write(0xFFFF & gpio.direction) - print(' Success!') + print(" Success!") except Exception: print(traceback.format_exc()) diff --git a/bin/logger.py b/bin/logger.py index 42b08a0..80552e1 100755 --- a/bin/logger.py +++ b/bin/logger.py @@ -8,33 +8,34 @@ def notify(url, active): - state = b'active' if active else b'inactive' + state = b"active" if active else b"inactive" requests.put(url, data=state) def properties_handler(hostname, interface, changed, invalidated): print(datetime.now(), interface, changed, invalidated) - if interface == 'org.freedesktop.login1.Session': - if 'Active' in changed: - notify(hostname, changed['Active']) + if interface == "org.freedesktop.login1.Session": + if "Active" in changed: + notify(hostname, changed["Active"]) def program(hostname): bus = SystemBus() - login = bus.get('org.freedesktop.login1') - for (sid, uid, uname, seat, path) in login.ListSessions(): - print('Connecting to {}'.format(path)) - sessionbus = bus.get('org.freedesktop.login1', path) + login = bus.get("org.freedesktop.login1") + for sid, uid, uname, seat, path in login.ListSessions(): + print("Connecting to {}".format(path)) + sessionbus = bus.get("org.freedesktop.login1", path) sessionbus.PropertiesChanged.connect( lambda interface, changed, invalidated: properties_handler( - hostname, interface, changed, invalidated)) + hostname, interface, changed, invalidated + ) + ) GLib.MainLoop().run() def main(): if len(sys.argv) != 2: - sys.stderr.write( - 'Usage: {} URL\n'.format(sys.argv[0])) + sys.stderr.write("Usage: {} URL\n".format(sys.argv[0])) sys.exit(1) try: diff --git a/bin/migrator.py b/bin/migrator.py index 6c1d66c..22f9d32 100644 --- a/bin/migrator.py +++ b/bin/migrator.py @@ -3,29 +3,25 @@ import sys old = sqlite3.connect(sys.argv[1]) -desks = pd.read_sql_query('SELECT * FROM desk ORDER BY date ASC', old) -sessions = pd.read_sql_query('SELECT * FROM session ORDER BY date ASC', old) +desks = pd.read_sql_query("SELECT * FROM desk ORDER BY date ASC", old) +sessions = pd.read_sql_query("SELECT * FROM session ORDER BY date ASC", old) old.close() new = sqlite3.connect(sys.argv[2]) new.execute( - 'CREATE TABLE session(' - 'date TIMESTAMP NOT NULL,' - 'state SESSION NOT NULL)') -new.execute( - 'CREATE TABLE desk(' - 'date TIMESTAMP NOT NULL,' - 'state DESK NOT NULL)') + "CREATE TABLE session(date TIMESTAMP NOT NULL, state SESSION NOT NULL)" +) +new.execute("CREATE TABLE desk(date TIMESTAMP NOT NULL, state DESK NOT NULL)") for desk in desks.itertuples(): - state = 'up' if desk.state == 1 else 'down' - print('INSERT INTO desk values({0}, {1})'.format(desk.date, state)) - new.execute('INSERT INTO desk values(?, ?)', (desk.date, state)) + state = "up" if desk.state == 1 else "down" + print("INSERT INTO desk values({0}, {1})".format(desk.date, state)) + new.execute("INSERT INTO desk values(?, ?)", (desk.date, state)) for session in sessions.itertuples(): - state = 'active' if session.active == 1 else 'inactive' - print('INSERT INTO session values({0}, {1})'.format(session.date, state)) - new.execute('INSERT INTO session values(?, ?)', (session.date, state)) + state = "active" if session.active == 1 else "inactive" + print("INSERT INTO session values({0}, {1})".format(session.date, state)) + new.execute("INSERT INTO session values(?, ?)", (session.date, state)) new.commit() diff --git a/bin/plot.py b/bin/plot.py index 28b4f1f..690e1eb 100644 --- a/bin/plot.py +++ b/bin/plot.py @@ -6,8 +6,6 @@ model = Model(SqliteDataStore(sys.argv[1])) figure = plot_weekday_hourly_count( - model.compute_hourly_count( - Timestamp.min, - Timestamp.now() - )) -figure.savefig(sys.argv[2], format='png') + model.compute_hourly_count(Timestamp.min, Timestamp.now()) +) +figure.savefig(sys.argv[2], format="png") diff --git a/setup.py b/setup.py index e989477..504706d 100644 --- a/setup.py +++ b/setup.py @@ -1,31 +1,27 @@ import setuptools -with open('README.md', 'r') as fh: +with open("README.md", "r") as fh: long_description = fh.read() setuptools.setup( - name='autodesk', - description='Automatic standing desk controller.', + name="autodesk", + description="Automatic standing desk controller.", long_description=long_description, - long_description_content_type='text/markdown', - version='1.0', - author='Daniel Oom', - author_email='oom.daniel@gmail.com', - url='https://github.com/daoo/autodesk', + long_description_content_type="text/markdown", + version="1.0", + author="Daniel Oom", + author_email="oom.daniel@gmail.com", + url="https://github.com/daoo/autodesk", packages=setuptools.find_packages(), - package_data={ - 'autodesk': [ - 'templates/*.*' - ] - }, + package_data={"autodesk": ["templates/*.*"]}, include_package_data=True, install_requires=[ - 'aiohttp>=3.9.1, <3.10', - 'aiohttp-jinja2>=1.6, <1.7', - 'matplotlib>=3.8.2, <3.9', - 'numpy>=1.26.2, <1.27', - 'pandas>=2.1.3, <2.2', - 'pyyaml>=6.0.1, <6.1', - 'pyftdi>=0.55, <0.56', + "aiohttp>=3.9.1, <3.10", + "aiohttp-jinja2>=1.6, <1.7", + "matplotlib>=3.8.2, <3.9", + "numpy>=1.26.2, <1.27", + "pandas>=2.1.3, <2.2", + "pyyaml>=6.0.1, <6.1", + "pyftdi>=0.55, <0.56", ], ) diff --git a/tests/autodeskservice.py b/tests/autodeskservice.py index a4c2015..5b1855a 100644 --- a/tests/autodeskservice.py +++ b/tests/autodeskservice.py @@ -7,40 +7,43 @@ TIME_ALLOWED = Timestamp(2018, 4, 23, 13, 0) TIME_DENIED = Timestamp(2018, 4, 23, 21, 0) -DESK_DENIED = [ - (ACTIVE, TIME_DENIED), - (INACTIVE, TIME_ALLOWED), - (INACTIVE, TIME_DENIED) -] +DESK_DENIED = [(ACTIVE, TIME_DENIED), (INACTIVE, TIME_ALLOWED), (INACTIVE, TIME_DENIED)] def create_service( - mocker, - now, - session_state, - active_time, - desk_state, - limits=(Timedelta(0), Timedelta(0))): - timer_fake = mocker.patch( - 'autodesk.timer.Timer', autospec=True) + mocker, + now, + session_state, + active_time, + desk_state, + limits=(Timedelta(0), Timedelta(0)), +): + timer_fake = mocker.patch("autodesk.timer.Timer", autospec=True) time_service_fake = mocker.patch( - 'autodesk.application.timeservice.TimeService', autospec=True) + "autodesk.application.timeservice.TimeService", autospec=True + ) time_service_fake.now.return_value = now session_service_fake = mocker.patch( - 'autodesk.application.sessionservice.SessionService', autospec=True) + "autodesk.application.sessionservice.SessionService", autospec=True + ) session_service_fake.get.return_value = session_state session_service_fake.get_active_time.return_value = active_time + def set_session_get_return(state): session_service_fake.get.return_value = state + session_service_fake.set.side_effect = set_session_get_return desk_service_fake = mocker.patch( - 'autodesk.application.deskservice.DeskService', autospec=True) + "autodesk.application.deskservice.DeskService", autospec=True + ) desk_service_fake.get.return_value = desk_state + def set_desk_get_return(state): desk_service_fake.get.return_value = state + desk_service_fake.set.side_effect = set_desk_get_return service = AutoDeskService( @@ -49,5 +52,6 @@ def set_desk_get_return(state): timer_fake, time_service_fake, session_service_fake, - desk_service_fake) + desk_service_fake, + ) return (timer_fake, session_service_fake, desk_service_fake, service) diff --git a/tests/component/test_api.py b/tests/component/test_api.py index a59f2b9..398f8b6 100644 --- a/tests/component/test_api.py +++ b/tests/component/test_api.py @@ -22,11 +22,9 @@ (Timestamp(2019, 4, 16, 18, 0), INACTIVE), (Timestamp(2019, 4, 23, 14, 0), ACTIVE), (Timestamp(2019, 4, 23, 18, 0), INACTIVE), - # Wednesdays (Timestamp(2019, 4, 24, 13, 0), ACTIVE), (Timestamp(2019, 4, 24, 14, 0), INACTIVE), - # Thursdays (Timestamp(2019, 4, 25, 8, 0), ACTIVE), (Timestamp(2019, 4, 25, 9, 0), INACTIVE), @@ -43,19 +41,16 @@ @pytest.fixture def client(mocker, event_loop, aiohttp_client): time_service = mocker.patch( - 'autodesk.application.timeservice.TimeService', autospec=True) + "autodesk.application.timeservice.TimeService", autospec=True + ) time_service.min = Timestamp.min time_service.now.return_value = Timestamp(2019, 4, 25, 12, 0) - model = Model(StubDataStore( - session_events=SESSION_EVENTS, - desk_events=DESK_EVENTS)) + model = Model(StubDataStore(session_events=SESSION_EVENTS, desk_events=DESK_EVENTS)) button_pin = NoopPin(4) - timer = mocker.patch( - 'autodesk.timer.Timer', autospec=True) - desk_controller = DeskController( - 0, NoopPin(0), NoopPin(1), NoopPin(3)) + timer = mocker.patch("autodesk.timer.Timer", autospec=True) + desk_controller = DeskController(0, NoopPin(0), NoopPin(1), NoopPin(3)) light_controller = LightController(NoopPin(2)) operation = Operation() limits = (Timedelta(minutes=30), Timedelta(minutes=30)) @@ -63,16 +58,18 @@ def client(mocker, event_loop, aiohttp_client): session_service = SessionService(model, light_controller, time_service) desk_service = DeskService(operation, model, desk_controller, time_service) service = AutoDeskService( - operation, scheduler, timer, time_service, session_service, - desk_service) + operation, scheduler, timer, time_service, session_service, desk_service + ) factory = mocker.patch( - 'autodesk.application.autodeskservicefactory.AutoDeskServiceFactory', - autospec=True) + "autodesk.application.autodeskservicefactory.AutoDeskServiceFactory", + autospec=True, + ) factory.create.return_value = service return event_loop.run_until_complete( - aiohttp_client(api.setup_app(button_pin, factory))) + aiohttp_client(api.setup_app(button_pin, factory)) + ) @pytest.fixture @@ -82,20 +79,19 @@ def testdir(request): @pytest.fixture def expected_figure(testdir): - with open(os.path.join(testdir, 'expected_figure.png'), 'rb') as file: + with open(os.path.join(testdir, "expected_figure.png"), "rb") as file: return file.read() @pytest.mark.asyncio async def test_index(client, expected_figure): - response = await client.get('/') + response = await client.get("/") text = await response.text() - expected_state_string = \ - 'Currently active with ' + \ - 'desk up for 0 days 00:30:00.' - expected_figure_string = \ - base64.b64encode(expected_figure).decode('utf-8') + expected_state_string = ( + "Currently active with " + "desk up for 0 days 00:30:00." + ) + expected_figure_string = base64.b64encode(expected_figure).decode("utf-8") assert 200 == response.status assert expected_state_string in text @@ -104,11 +100,11 @@ async def test_index(client, expected_figure): @pytest.mark.asyncio async def test_set_desk_invalid(client): - response = await client.put('/api/desk', data=b'invalid state string') + response = await client.put("/api/desk", data=b"invalid state string") assert 400 == response.status @pytest.mark.asyncio async def test_set_session_invalid(client): - response = await client.put('/api/session', data=b'invalid state string') + response = await client.put("/api/session", data=b"invalid state string") assert 400 == response.status diff --git a/tests/integration/test_program.py b/tests/integration/test_program.py index a558dde..b06d1f9 100644 --- a/tests/integration/test_program.py +++ b/tests/integration/test_program.py @@ -6,21 +6,21 @@ @pytest.fixture def process(): - cmd = ['python3', '-u', '-m', 'autodesk.program'] + cmd = ["python3", "-u", "-m", "autodesk.program"] env = os.environ.copy() - env['AUTODESK_ADDRESS'] = '127.0.0.1' - env['AUTODESK_CONFIG'] = 'config/testing.yml' - env['AUTODESK_DATABASE'] = ':memory:' - env['AUTODESK_PORT'] = '8081' - process = subprocess.Popen(cmd, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, encoding='utf-8', - env=env) + env["AUTODESK_ADDRESS"] = "127.0.0.1" + env["AUTODESK_CONFIG"] = "config/testing.yml" + env["AUTODESK_DATABASE"] = ":memory:" + env["AUTODESK_PORT"] = "8081" + process = subprocess.Popen( + cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8", env=env + ) while True: line = process.stdout.readline() if not line: raise RuntimeError(process.stderr.read()) - elif 'Running on ' in line: + elif "Running on " in line: break yield process @@ -32,4 +32,4 @@ def process(): def test_get_index(process): - assert requests.get('http://localhost:8081').ok + assert requests.get("http://localhost:8081").ok diff --git a/tests/stubdatastore.py b/tests/stubdatastore.py index 1b7c93d..3a6a1dc 100644 --- a/tests/stubdatastore.py +++ b/tests/stubdatastore.py @@ -3,10 +3,8 @@ class StubDataStore: def __init__(self, session_events, desk_events): - self.session_events = pd.DataFrame( - session_events, columns=['date', 'state']) - self.desk_events = pd.DataFrame( - desk_events, columns=['date', 'state']) + self.session_events = pd.DataFrame(session_events, columns=["date", "state"]) + self.desk_events = pd.DataFrame(desk_events, columns=["date", "state"]) def close(self): pass diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index ea6d209..c178d96 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -8,26 +8,32 @@ @pytest.fixture def service_mock(mocker): return mocker.patch( - 'autodesk.application.autodeskservice.AutoDeskService', autospec=True) + "autodesk.application.autodeskservice.AutoDeskService", autospec=True + ) @pytest.fixture def button_pin_stub(mocker): - return mocker.patch('autodesk.hardware.noop.NoopPin') + return mocker.patch("autodesk.hardware.noop.NoopPin") @pytest.fixture def client(event_loop, mocker, button_pin_stub, service_mock, aiohttp_client): factory = mocker.patch( - 'autodesk.application.autodeskservicefactory.AutoDeskServiceFactory', - autospec=True) + "autodesk.application.autodeskservicefactory.AutoDeskServiceFactory", + autospec=True, + ) factory.create.return_value = service_mock return event_loop.run_until_complete( - aiohttp_client(api.setup_app( - button_pin_stub, - factory, - button_polling_delay=0.1, - hardware_error_delay=0.1))) + aiohttp_client( + api.setup_app( + button_pin_stub, + factory, + button_polling_delay=0.1, + hardware_error_delay=0.1, + ) + ) + ) @pytest.mark.asyncio @@ -38,27 +44,27 @@ async def test_setup(client, service_mock): @pytest.mark.asyncio async def test_get_desk_down(client, service_mock): service_mock.get_desk_state.return_value = DOWN - response = await client.get('/api/desk') - assert 'down' == await response.text() + response = await client.get("/api/desk") + assert "down" == await response.text() @pytest.mark.asyncio async def test_get_desk_up(client, service_mock): service_mock.get_desk_state.return_value = UP - response = await client.get('/api/desk') - assert 'up' == await response.text() + response = await client.get("/api/desk") + assert "up" == await response.text() @pytest.mark.asyncio async def test_set_desk_down(client, service_mock): - response = await client.put('/api/desk', data=b'down') + response = await client.put("/api/desk", data=b"down") assert 200 == response.status service_mock.set_desk.assert_called_with(DOWN) @pytest.mark.asyncio async def test_set_desk_up(client, service_mock): - response = await client.put('/api/desk', data=b'up') + response = await client.put("/api/desk", data=b"up") assert 200 == response.status service_mock.set_desk.assert_called_with(UP) @@ -66,34 +72,34 @@ async def test_set_desk_up(client, service_mock): @pytest.mark.asyncio async def test_set_desk_not_allowed(client, service_mock): service_mock.set_desk.return_value = False - response = await client.put('/api/desk', data=b'up') + response = await client.put("/api/desk", data=b"up") assert 403 == response.status @pytest.mark.asyncio async def test_get_session_inactive(client, service_mock): service_mock.get_session_state.return_value = INACTIVE - response = await client.get('/api/session') - assert 'inactive' == await response.text() + response = await client.get("/api/session") + assert "inactive" == await response.text() @pytest.mark.asyncio async def test_get_session_active(client, service_mock): service_mock.get_session_state.return_value = ACTIVE - response = await client.get('/api/session') - assert 'active' == await response.text() + response = await client.get("/api/session") + assert "active" == await response.text() @pytest.mark.asyncio async def test_set_session_inactive(client, service_mock): - response = await client.put('/api/session', data=b'inactive') + response = await client.put("/api/session", data=b"inactive") service_mock.set_session.assert_called_with(INACTIVE) assert 200 == response.status @pytest.mark.asyncio async def test_set_session_active(client, service_mock): - response = await client.put('/api/session', data=b'active') + response = await client.put("/api/session", data=b"active") service_mock.set_session.assert_called_with(ACTIVE) assert 200 == response.status @@ -106,10 +112,8 @@ async def test_button_press(client, button_pin_stub, service_mock): @pytest.mark.asyncio -async def test_button_press_after_hardware_error( - client, button_pin_stub, service_mock): - button_pin_stub.read.side_effect = HardwareError( - 'Stubbed error for unit testing.') +async def test_button_press_after_hardware_error(client, button_pin_stub, service_mock): + button_pin_stub.read.side_effect = HardwareError("Stubbed error for unit testing.") await asyncio.sleep(0.1) button_pin_stub.read.side_effect = None button_pin_stub.read.return_value = 1 diff --git a/tests/unit/test_autodeskservice_desk.py b/tests/unit/test_autodeskservice_desk.py index 0898c02..b520459 100644 --- a/tests/unit/test_autodeskservice_desk.py +++ b/tests/unit/test_autodeskservice_desk.py @@ -6,10 +6,8 @@ @pytest.mark.parametrize("session,now", DESK_DENIED) -def test_set_desk_down_denied_timer_not_scheduled( - mocker, session, now): - (timer_mock, _, _, service) = create_service( - mocker, now, session, Timedelta(0), UP) +def test_set_desk_down_denied_timer_not_scheduled(mocker, session, now): + (timer_mock, _, _, service) = create_service(mocker, now, session, Timedelta(0), UP) service.set_desk(DOWN) @@ -18,8 +16,13 @@ def test_set_desk_down_denied_timer_not_scheduled( def test_set_desk_down_allowed_timer_scheduled_right_time(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(10), UP, - limits=(Timedelta(20), Timedelta(30))) + mocker, + TIME_ALLOWED, + ACTIVE, + Timedelta(10), + UP, + limits=(Timedelta(20), Timedelta(30)), + ) service.set_desk(DOWN) @@ -28,8 +31,13 @@ def test_set_desk_down_allowed_timer_scheduled_right_time(mocker): def test_set_desk_up_allowed_timer_scheduled_right_time(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(10), UP, - limits=(Timedelta(20), Timedelta(30))) + mocker, + TIME_ALLOWED, + ACTIVE, + Timedelta(10), + UP, + limits=(Timedelta(20), Timedelta(30)), + ) service.set_desk(UP) @@ -39,7 +47,8 @@ def test_set_desk_up_allowed_timer_scheduled_right_time(mocker): @pytest.mark.parametrize("desk", [DOWN, UP]) def test_set_desk_hardware_error_timer_cancelled(mocker, desk): (timer_mock, _, desk_service_stub, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), desk.next()) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), desk.next() + ) desk_service_stub.set.side_effect = HardwareError(RuntimeError()) service.set_desk(desk) diff --git a/tests/unit/test_autodeskservice_init.py b/tests/unit/test_autodeskservice_init.py index a329af9..52ff484 100644 --- a/tests/unit/test_autodeskservice_init.py +++ b/tests/unit/test_autodeskservice_init.py @@ -6,7 +6,8 @@ def test_init_active_denied_timer_not_scheduled(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_DENIED, ACTIVE, Timedelta(0), DOWN) + mocker, TIME_DENIED, ACTIVE, Timedelta(0), DOWN + ) service.init() @@ -15,7 +16,8 @@ def test_init_active_denied_timer_not_scheduled(mocker): def test_init_inactive_operation_allowed_timer_not_scheduled(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_ALLOWED, INACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, INACTIVE, Timedelta(0), DOWN + ) service.init() @@ -24,8 +26,13 @@ def test_init_inactive_operation_allowed_timer_not_scheduled(mocker): def test_init_active_operation_allowed_timer_scheduled(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN, - limits=(Timedelta(10), Timedelta(20))) + mocker, + TIME_ALLOWED, + ACTIVE, + Timedelta(0), + DOWN, + limits=(Timedelta(10), Timedelta(20)), + ) service.init() @@ -34,7 +41,8 @@ def test_init_active_operation_allowed_timer_scheduled(mocker): def test_init_hardware_error_timer_cancelled(mocker): (timer_mock, session_service_stub, _, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN + ) session_service_stub.init.side_effect = HardwareError(RuntimeError()) service.init() diff --git a/tests/unit/test_button.py b/tests/unit/test_button.py index a7c3992..2b2d660 100644 --- a/tests/unit/test_button.py +++ b/tests/unit/test_button.py @@ -4,14 +4,14 @@ @pytest.fixture def pin_stub(mocker): - return mocker.patch('autodesk.hardware.noop.NoopPin', autospec=True) + return mocker.patch("autodesk.hardware.noop.NoopPin", autospec=True) @pytest.fixture def autodeskservice_mock(mocker): return mocker.patch( - 'autodesk.application.autodeskservice.AutoDeskService', - autospec=True) + "autodesk.application.autodeskservice.AutoDeskService", autospec=True + ) @pytest.fixture @@ -19,8 +19,7 @@ def button(pin_stub, autodeskservice_mock): return Button(pin_stub, autodeskservice_mock) -def test_poll_read0_toggle_session_not_called( - pin_stub, autodeskservice_mock, button): +def test_poll_read0_toggle_session_not_called(pin_stub, autodeskservice_mock, button): pin_stub.read.return_value = 0 button.poll() @@ -28,8 +27,7 @@ def test_poll_read0_toggle_session_not_called( autodeskservice_mock.toggle_session.assert_not_called() -def test_poll_read1_toggle_session_called( - pin_stub, autodeskservice_mock, button): +def test_poll_read1_toggle_session_called(pin_stub, autodeskservice_mock, button): pin_stub.read.return_value = 1 button.poll() @@ -38,7 +36,8 @@ def test_poll_read1_toggle_session_called( def test_poll_read1_twice_toggle_session_called_only_once( - pin_stub, autodeskservice_mock, button): + pin_stub, autodeskservice_mock, button +): pin_stub.read.return_value = 1 button.poll() diff --git a/tests/unit/test_deskcontroller.py b/tests/unit/test_deskcontroller.py index e4bf18b..5234e5e 100644 --- a/tests/unit/test_deskcontroller.py +++ b/tests/unit/test_deskcontroller.py @@ -5,23 +5,22 @@ @pytest.fixture def sleep_fake(mocker): - return mocker.patch('time.sleep', autospec=True) + return mocker.patch("time.sleep", autospec=True) def create_pin_fake(mocker): - return mocker.patch('autodesk.hardware.noop.NoopPin') + return mocker.patch("autodesk.hardware.noop.NoopPin") def create_controller(mocker, delay): pin_down_fake = create_pin_fake(mocker) pin_up_fake = create_pin_fake(mocker) pin_light_fake = create_pin_fake(mocker) - controller = DeskController( - delay, pin_down_fake, pin_up_fake, pin_light_fake) + controller = DeskController(delay, pin_down_fake, pin_up_fake, pin_light_fake) return (pin_down_fake, pin_up_fake, pin_light_fake, controller) -@pytest.mark.parametrize('direction', [DOWN, UP]) +@pytest.mark.parametrize("direction", [DOWN, UP]) def test_move_sleeps_for_specified_delay(mocker, sleep_fake, direction): sleep_mock = sleep_fake (_, _, _, controller) = create_controller(mocker, 1) @@ -36,8 +35,7 @@ def test_move_down_correct_pin_pulled_high_and_low(mocker, sleep_fake): controller.move(DOWN) - pin_down_mock.write.assert_has_calls( - [mocker.call(1), mocker.call(0)]) + pin_down_mock.write.assert_has_calls([mocker.call(1), mocker.call(0)]) def test_move_up_correct_pin_pulled_high_and_low(mocker, sleep_fake): @@ -45,15 +43,13 @@ def test_move_up_correct_pin_pulled_high_and_low(mocker, sleep_fake): controller.move(UP) - pin_up_mock.write.assert_has_calls( - [mocker.call(1), mocker.call(0)]) + pin_up_mock.write.assert_has_calls([mocker.call(1), mocker.call(0)]) -@pytest.mark.parametrize('direction', [DOWN, UP]) +@pytest.mark.parametrize("direction", [DOWN, UP]) def test_move_light_pin_pulled_high_and_low(mocker, sleep_fake, direction): (_, _, pin_light_mock, controller) = create_controller(mocker, 1) controller.move(direction) - pin_light_mock.write.assert_has_calls( - [mocker.call(1), mocker.call(0)]) + pin_light_mock.write.assert_has_calls([mocker.call(1), mocker.call(0)]) diff --git a/tests/unit/test_deskservice.py b/tests/unit/test_deskservice.py index 7299bdd..031647c 100644 --- a/tests/unit/test_deskservice.py +++ b/tests/unit/test_deskservice.py @@ -8,46 +8,34 @@ TIME_ALLOWED = Timestamp(2018, 4, 23, 13, 0) TIME_DENIED = Timestamp(2018, 4, 23, 21, 0) -DESK_DENIED = [ - (ACTIVE, TIME_DENIED), - (INACTIVE, TIME_ALLOWED), - (INACTIVE, TIME_DENIED) -] - - -def create_service( - mocker, - now, - session_state, - active_time, - desk_state): - model_fake = mocker.patch( - 'autodesk.model.Model', autospec=True) +DESK_DENIED = [(ACTIVE, TIME_DENIED), (INACTIVE, TIME_ALLOWED), (INACTIVE, TIME_DENIED)] + + +def create_service(mocker, now, session_state, active_time, desk_state): + model_fake = mocker.patch("autodesk.model.Model", autospec=True) model_fake.get_active_time.return_value = active_time model_fake.get_session_state.return_value = session_state model_fake.get_desk_state.return_value = desk_state time_service_fake = mocker.patch( - 'autodesk.application.timeservice.TimeService', autospec=True) + "autodesk.application.timeservice.TimeService", autospec=True + ) time_service_fake.now.return_value = now desk_controller_fake = mocker.patch( - 'autodesk.deskcontroller.DeskController', autospec=True) + "autodesk.deskcontroller.DeskController", autospec=True + ) service = DeskService( - Operation(), - model_fake, - desk_controller_fake, - time_service_fake) - return ( - model_fake, - desk_controller_fake, - service) + Operation(), model_fake, desk_controller_fake, time_service_fake + ) + return (model_fake, desk_controller_fake, service) @pytest.mark.parametrize("session,now", DESK_DENIED) def test_set_denied_desk_controller_not_called(mocker, session, now): (_, desk_controller_mock, service) = create_service( - mocker, now, session, Timedelta(0), DOWN) + mocker, now, session, Timedelta(0), DOWN + ) service.set(UP) @@ -56,18 +44,18 @@ def test_set_denied_desk_controller_not_called(mocker, session, now): @pytest.mark.parametrize("session,now", DESK_DENIED) def test_set_denied_desk_model_not_updated(mocker, session, now): - (model_mock, _, service) = create_service( - mocker, now, session, Timedelta(0), DOWN) + (model_mock, _, service) = create_service(mocker, now, session, Timedelta(0), DOWN) service.set(UP) model_mock.set_desk.assert_not_called() -@pytest.mark.parametrize('direction', [DOWN, UP]) +@pytest.mark.parametrize("direction", [DOWN, UP]) def test_set_allowed_model_updated(mocker, direction): (model_mock, _, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN + ) service.set(direction) @@ -76,7 +64,8 @@ def test_set_allowed_model_updated(mocker, direction): def test_set_allowed_desk_controller_called(mocker): (_, desk_controller_mock, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN + ) service.set(DOWN) @@ -85,7 +74,8 @@ def test_set_allowed_desk_controller_called(mocker): def test_set_hardware_error_is_passed_up(mocker): (model_mock, desk_controller_stub, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN + ) desk_controller_stub.move.side_effect = HardwareError(RuntimeError()) with pytest.raises(HardwareError): @@ -94,7 +84,8 @@ def test_set_hardware_error_is_passed_up(mocker): def test_set_hardware_error_model_not_updated(mocker): (model_mock, desk_controller_stub, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN + ) desk_controller_stub.move.side_effect = HardwareError(RuntimeError()) try: diff --git a/tests/unit/test_hardware.py b/tests/unit/test_hardware.py index 8d95e13..ac9b75b 100644 --- a/tests/unit/test_hardware.py +++ b/tests/unit/test_hardware.py @@ -3,29 +3,29 @@ def patch_module(mocker, module): fake = mocker.MagicMock() - mocker.patch.dict('sys.modules', {module: fake}) + mocker.patch.dict("sys.modules", {module: fake}) return fake def test_create_pin_factory_raspberry_pi(mocker): - raspberrypi_mock = patch_module(mocker, 'autodesk.hardware.raspberrypi') + raspberrypi_mock = patch_module(mocker, "autodesk.hardware.raspberrypi") - create_pin_factory('raspberrypi') + create_pin_factory("raspberrypi") raspberrypi_mock.RaspberryPiPinFactory.assert_called_once() def test_create_pin_factory_ft232h(mocker): - ft232h_mock = patch_module(mocker, 'autodesk.hardware.ft232h') + ft232h_mock = patch_module(mocker, "autodesk.hardware.ft232h") - create_pin_factory('ft232h') + create_pin_factory("ft232h") ft232h_mock.Ft232hPinFactory.assert_called_once() def test_create_pin_factory_noop(mocker): - noop_mock = patch_module(mocker, 'autodesk.hardware.noop') + noop_mock = patch_module(mocker, "autodesk.hardware.noop") - create_pin_factory('noop') + create_pin_factory("noop") noop_mock.NoopPinFactory.assert_called_once() diff --git a/tests/unit/test_hardware_ft232h.py b/tests/unit/test_hardware_ft232h.py index 7e37d54..f400608 100644 --- a/tests/unit/test_hardware_ft232h.py +++ b/tests/unit/test_hardware_ft232h.py @@ -8,9 +8,7 @@ @pytest.fixture def controller_fake(mocker): - return mocker \ - .patch('autodesk.hardware.ft232h.GpioMpsseController') \ - .return_value + return mocker.patch("autodesk.hardware.ft232h.GpioMpsseController").return_value @pytest.fixture @@ -19,6 +17,7 @@ def gpio_fake(controller_fake): def assign_direction(_, direction): fake.direction = direction + fake.set_direction.side_effect = assign_direction return fake @@ -57,7 +56,8 @@ def test_factory_create_input_followed_by_read(gpio_fake, factory): expected_pin_mask = 0b0010 expected_direction = 0b0000 gpio_fake.set_direction.assert_called_once_with( - expected_pin_mask, expected_direction) + expected_pin_mask, expected_direction + ) def test_factory_create_output_followed_by_write(gpio_fake, factory): @@ -70,7 +70,8 @@ def test_factory_create_output_followed_by_write(gpio_fake, factory): expected_pin_mask = 0b0010 expected_direction = 0b0010 gpio_fake.set_direction.assert_called_once_with( - expected_pin_mask, expected_direction) + expected_pin_mask, expected_direction + ) def test_factory_create_input_invalid_pin(gpio_fake, factory): @@ -146,7 +147,8 @@ def test_pin_write_invalid_value(factory): def test_pin_read_failure_recovery(mocker, gpio_fake, factory): def fail_once(): gpio_fake.read.side_effect = None - raise FtdiError('Stubbed error for unit testing.') + raise FtdiError("Stubbed error for unit testing.") + gpio_fake.read.side_effect = fail_once pin_number = 1 gpio_fake.all_pins = 0b0110 @@ -155,16 +157,17 @@ def fail_once(): actual = pin.read() - gpio_fake.read.assert_has_calls([ - mocker.call(), # failed attempt - mocker.call(), - ]) + gpio_fake.read.assert_has_calls( + [ + mocker.call(), # failed attempt + mocker.call(), + ] + ) assert actual == 0 def test_pin_read_two_failures_raises(mocker, gpio_fake, factory): - gpio_fake.read.side_effect = FtdiError( - 'Stubbed error for unit testing.') + gpio_fake.read.side_effect = FtdiError("Stubbed error for unit testing.") pin_number = 1 gpio_fake.all_pins = 0b0110 gpio_fake.read.return_value = (0b0100,) @@ -173,16 +176,19 @@ def test_pin_read_two_failures_raises(mocker, gpio_fake, factory): with pytest.raises(HardwareError): pin.read() - gpio_fake.read.assert_has_calls([ - mocker.call(), # failed attempt - mocker.call(), # failed attempt - ]) + gpio_fake.read.assert_has_calls( + [ + mocker.call(), # failed attempt + mocker.call(), # failed attempt + ] + ) def test_pin_write_failure_recovery(mocker, gpio_fake, factory): def fail_once(value): gpio_fake.write.side_effect = None - raise FtdiError('Stubbed error for unit testing.') + raise FtdiError("Stubbed error for unit testing.") + gpio_fake.write.side_effect = fail_once pin_number = 1 gpio_fake.all_pins = 0b0110 @@ -191,15 +197,16 @@ def fail_once(value): pin.write(1) - gpio_fake.write.assert_has_calls([ - mocker.call(0b0010), # failed attempt - mocker.call(0b0010), - ]) + gpio_fake.write.assert_has_calls( + [ + mocker.call(0b0010), # failed attempt + mocker.call(0b0010), + ] + ) def test_pin_write_two_failures_raises(mocker, gpio_fake, factory): - gpio_fake.write.side_effect = FtdiError( - 'Stubbed error for unit testing.') + gpio_fake.write.side_effect = FtdiError("Stubbed error for unit testing.") pin_number = 1 gpio_fake.all_pins = 0b0110 gpio_fake.read.return_value = (0b0100,) @@ -208,10 +215,12 @@ def test_pin_write_two_failures_raises(mocker, gpio_fake, factory): with pytest.raises(HardwareError): pin.write(1) - gpio_fake.write.assert_has_calls([ - mocker.call(0b0010), # failed attempt - mocker.call(0b0010), # failed attempt - ]) + gpio_fake.write.assert_has_calls( + [ + mocker.call(0b0010), # failed attempt + mocker.call(0b0010), # failed attempt + ] + ) def test_two_reads_connects_once(controller_fake, gpio_fake, factory): @@ -240,7 +249,8 @@ def test_two_writes_connects_once(controller_fake, gpio_fake, factory): def test_input_connect_usb_tools_error(controller_fake, gpio_fake, factory): controller_fake.configure.side_effect = UsbToolsError( - 'Stubbed error for unit testing.') + "Stubbed error for unit testing." + ) pin_number = 1 gpio_fake.all_pins = 0b0110 gpio_fake.read.return_value = (0b0100,) @@ -251,8 +261,7 @@ def test_input_connect_usb_tools_error(controller_fake, gpio_fake, factory): def test_input_connect_usb_error(controller_fake, gpio_fake, factory): - controller_fake.configure.side_effect = USBError( - 'Stubbed error for unit testing.') + controller_fake.configure.side_effect = USBError("Stubbed error for unit testing.") pin_number = 1 gpio_fake.all_pins = 0b0110 gpio_fake.read.return_value = (0b0100,) @@ -264,7 +273,8 @@ def test_input_connect_usb_error(controller_fake, gpio_fake, factory): def test_output_connect_usb_tools_error(controller_fake, gpio_fake, factory): controller_fake.configure.side_effect = UsbToolsError( - 'Stubbed error for unit testing.') + "Stubbed error for unit testing." + ) pin_number = 1 gpio_fake.all_pins = 0b0110 gpio_fake.read.return_value = (0b0100,) @@ -275,8 +285,7 @@ def test_output_connect_usb_tools_error(controller_fake, gpio_fake, factory): def test_output_connect_usb_error(controller_fake, gpio_fake, factory): - controller_fake.configure.side_effect = USBError( - 'Stubbed error for unit testing.') + controller_fake.configure.side_effect = USBError("Stubbed error for unit testing.") pin_number = 1 gpio_fake.all_pins = 0b0110 gpio_fake.read.return_value = (0b0100,) diff --git a/tests/unit/test_hardware_logging.py b/tests/unit/test_hardware_logging.py index 816bc04..441d380 100644 --- a/tests/unit/test_hardware_logging.py +++ b/tests/unit/test_hardware_logging.py @@ -4,13 +4,12 @@ @pytest.fixture def mock_pin(mocker): - return mocker.patch('autodesk.hardware.noop.NoopPin') + return mocker.patch("autodesk.hardware.noop.NoopPin") @pytest.fixture def mock_factory(mocker, mock_pin): - factory = mocker.patch( - 'autodesk.hardware.noop.NoopPinFactory', autospec=True) + factory = mocker.patch("autodesk.hardware.noop.NoopPinFactory", autospec=True) factory.create_output.return_value = mock_pin factory.create_input.return_value = mock_pin return factory @@ -47,7 +46,7 @@ def test_factory_create_output(mock_factory, factory): mock_factory.create_output.assert_called_once_with(pin_number) -@pytest.mark.parametrize('value', [0, 1]) +@pytest.mark.parametrize("value", [0, 1]) def test_pin_read_return_same_as_mock(mock_pin, factory, value): pin_number = 0 mock_pin.read.return_value = value diff --git a/tests/unit/test_hardware_raspberrypi.py b/tests/unit/test_hardware_raspberrypi.py index 02fe762..bb1c1a3 100644 --- a/tests/unit/test_hardware_raspberrypi.py +++ b/tests/unit/test_hardware_raspberrypi.py @@ -13,11 +13,15 @@ def gpio_mock(rpi_stub): @pytest.fixture def factory(mocker, rpi_stub, gpio_mock): - mocker.patch.dict('sys.modules', **{ - 'RPi': rpi_stub, - 'RPi.GPIO': gpio_mock, - }) + mocker.patch.dict( + "sys.modules", + **{ + "RPi": rpi_stub, + "RPi.GPIO": gpio_mock, + }, + ) from autodesk.hardware.raspberrypi import RaspberryPiPinFactory + factory = RaspberryPiPinFactory() yield factory factory.close() @@ -25,10 +29,13 @@ def factory(mocker, rpi_stub, gpio_mock): def test_factory_constructor(mocker, rpi_stub, gpio_mock): - mocker.patch.dict('sys.modules', **{ - 'RPi': rpi_stub, - 'RPi.GPIO': gpio_mock, - }) + mocker.patch.dict( + "sys.modules", + **{ + "RPi": rpi_stub, + "RPi.GPIO": gpio_mock, + }, + ) from autodesk.hardware.raspberrypi import RaspberryPiPinFactory RaspberryPiPinFactory() @@ -37,11 +44,15 @@ def test_factory_constructor(mocker, rpi_stub, gpio_mock): def test_factory_close(mocker, rpi_stub, gpio_mock): - mocker.patch.dict('sys.modules', **{ - 'RPi': rpi_stub, - 'RPi.GPIO': gpio_mock, - }) + mocker.patch.dict( + "sys.modules", + **{ + "RPi": rpi_stub, + "RPi.GPIO": gpio_mock, + }, + ) from autodesk.hardware.raspberrypi import RaspberryPiPinFactory + factory = RaspberryPiPinFactory() factory.close() diff --git a/tests/unit/test_lightcontroller.py b/tests/unit/test_lightcontroller.py index bc1cfe0..66ca3c3 100644 --- a/tests/unit/test_lightcontroller.py +++ b/tests/unit/test_lightcontroller.py @@ -5,7 +5,7 @@ @pytest.fixture def pin_mock(mocker): - return mocker.patch('autodesk.hardware.noop.NoopPin', autospec=True) + return mocker.patch("autodesk.hardware.noop.NoopPin", autospec=True) def test_set_inactive(pin_mock): diff --git a/tests/unit/test_model.py b/tests/unit/test_model.py index 60bb4a7..777bb37 100644 --- a/tests/unit/test_model.py +++ b/tests/unit/test_model.py @@ -9,12 +9,12 @@ def make_spans(records): - return pd.DataFrame(records, columns=['start', 'end', 'state']) + return pd.DataFrame(records, columns=["start", "end", "state"]) @pytest.fixture() def inmemory_model(): - model = Model(SqliteDataStore(':memory:')) + model = Model(SqliteDataStore(":memory:")) yield model model.close() @@ -45,10 +45,7 @@ def test_get_desk_spans_one_up_span(): t1 = Timestamp(2018, 1, 1) t2 = Timestamp(2018, 1, 2) t3 = Timestamp(2018, 1, 3) - model = Model(StubDataStore( - session_events=[], - desk_events=[(t2, UP)] - )) + model = Model(StubDataStore(session_events=[], desk_events=[(t2, UP)])) result = model.get_desk_spans(t1, t3) @@ -60,10 +57,7 @@ def test_get_session_spans_one_active_span(): t1 = Timestamp(2018, 1, 1) t2 = Timestamp(2018, 1, 2) t3 = Timestamp(2018, 1, 3) - model = Model(StubDataStore( - session_events=[(t2, ACTIVE)], - desk_events=[] - )) + model = Model(StubDataStore(session_events=[(t2, ACTIVE)], desk_events=[])) result = model.get_session_spans(t1, t3) @@ -88,30 +82,21 @@ def test_get_active_time_empty(): def test_get_active_time_active_zero(): t = Timestamp(2018, 1, 1) - model = Model(StubDataStore( - session_events=[(t, ACTIVE)], - desk_events=[] - )) + model = Model(StubDataStore(session_events=[(t, ACTIVE)], desk_events=[])) assert model.get_active_time(Timestamp.min, t) == Timedelta(0) def test_get_active_time_active_for_10_minutes(): t1 = Timestamp(2018, 1, 1, 0, 0, 0) t2 = Timestamp(2018, 1, 1, 0, 10, 0) - model = Model(StubDataStore( - session_events=[(t1, ACTIVE)], - desk_events=[] - )) + model = Model(StubDataStore(session_events=[(t1, ACTIVE)], desk_events=[])) assert model.get_active_time(Timestamp.min, t2) == Timedelta(minutes=10) def test_get_active_time_just_after_desk_change(): t1 = Timestamp(2018, 1, 1, 0, 0, 0) t2 = Timestamp(2018, 1, 1, 0, 10, 0) - model = Model(StubDataStore( - session_events=[(t1, ACTIVE)], - desk_events=[(t2, UP)] - )) + model = Model(StubDataStore(session_events=[(t1, ACTIVE)], desk_events=[(t2, UP)])) assert model.get_active_time(Timestamp.min, t2) == Timedelta(0) @@ -119,34 +104,25 @@ def test_get_active_time_active_20_minutes_with_changed_desk_state(): t1 = Timestamp(2018, 1, 1, 0, 0, 0) t2 = Timestamp(2018, 1, 1, 0, 10, 0) t3 = Timestamp(2018, 1, 1, 0, 20, 0) - model = Model(StubDataStore( - session_events=[(t1, ACTIVE)], - desk_events=[(t2, UP)] - )) + model = Model(StubDataStore(session_events=[(t1, ACTIVE)], desk_events=[(t2, UP)])) assert model.get_active_time(Timestamp.min, t3) == Timedelta(minutes=10) def test_compute_hourly_count_active_30_minutes(): t1 = Timestamp(2017, 4, 12, 10, 0, 0) t2 = Timestamp(2017, 4, 12, 10, 30, 0) - model = Model(StubDataStore( - session_events=[(t1, ACTIVE), (t2, INACTIVE)], - desk_events=[] - )) + model = Model( + StubDataStore(session_events=[(t1, ACTIVE), (t2, INACTIVE)], desk_events=[]) + ) result = model.compute_hourly_count(t1, t2) - specific_hour = result[ - (result.weekday == 'Wednesday') & (result.hour == 10) - ] + specific_hour = result[(result.weekday == "Wednesday") & (result.hour == 10)] assert specific_hour.counts.iloc[0] == 1 def test_compute_hourly_count_active_0_minutes(): t1 = Timestamp(2017, 4, 12, 10, 0, 0) t2 = Timestamp(2017, 4, 12, 10, 30, 0) - model = Model(StubDataStore( - session_events=[(t1, INACTIVE)], - desk_events=[] - )) + model = Model(StubDataStore(session_events=[(t1, INACTIVE)], desk_events=[])) result = model.compute_hourly_count(t1, t2) assert result.counts.sum() == 0 diff --git a/tests/unit/test_operation.py b/tests/unit/test_operation.py index 03b6c21..36723ae 100644 --- a/tests/unit/test_operation.py +++ b/tests/unit/test_operation.py @@ -6,8 +6,7 @@ def combine(dates, times): - return [Timestamp.combine(day, stroke) - for day in dates for stroke in times] + return [Timestamp.combine(day, stroke) for day in dates for stroke in times] ondates = [ @@ -40,10 +39,11 @@ def combine(dates, times): ] ondatetimes = combine(ondates, ontimes) -offdatetimes = \ - combine(offdates, offtimes) + \ - combine(ondates, offtimes) + \ - combine(offdates, ontimes) +offdatetimes = ( + combine(offdates, offtimes) + + combine(ondates, offtimes) + + combine(offdates, ontimes) +) @pytest.mark.parametrize("at", ondatetimes) diff --git a/tests/unit/test_sessionservice.py b/tests/unit/test_sessionservice.py index 41f9c87..c997dd9 100644 --- a/tests/unit/test_sessionservice.py +++ b/tests/unit/test_sessionservice.py @@ -6,20 +6,18 @@ def create_service(mocker, session_state, now=Timestamp.min): - model_fake = mocker.patch( - 'autodesk.model.Model', autospec=True) + model_fake = mocker.patch("autodesk.model.Model", autospec=True) model_fake.get_session_state.return_value = session_state time_service_fake = mocker.patch( - 'autodesk.application.timeservice.TimeService', autospec=True) + "autodesk.application.timeservice.TimeService", autospec=True + ) time_service_fake.now.return_value = now light_controller_fake = mocker.patch( - 'autodesk.lightcontroller.LightController', autospec=True) - service = SessionService( - model_fake, - light_controller_fake, - time_service_fake) + "autodesk.lightcontroller.LightController", autospec=True + ) + service = SessionService(model_fake, light_controller_fake, time_service_fake) return (model_fake, light_controller_fake, service) @@ -83,8 +81,7 @@ def test_set_hardware_error_is_passed_up(mocker): def test_set_hardware_error_model_still_called(mocker): now = Timestamp(2019, 8, 1, 13, 0) - (model_mock, light_controller_stub, service) = create_service( - mocker, INACTIVE, now) + (model_mock, light_controller_stub, service) = create_service(mocker, INACTIVE, now) light_controller_stub.set.side_effect = HardwareError(RuntimeError()) try: diff --git a/tests/unit/test_timer.py b/tests/unit/test_timer.py index cc87da3..8727805 100644 --- a/tests/unit/test_timer.py +++ b/tests/unit/test_timer.py @@ -5,7 +5,7 @@ @pytest.fixture def loop(mocker): - return mocker.patch('asyncio.AbstractEventLoop', autospec=True) + return mocker.patch("asyncio.AbstractEventLoop", autospec=True) @pytest.fixture diff --git a/tests/unit/tests_autodeskservice_session.py b/tests/unit/tests_autodeskservice_session.py index 43d11db..9b2bcbd 100644 --- a/tests/unit/tests_autodeskservice_session.py +++ b/tests/unit/tests_autodeskservice_session.py @@ -6,8 +6,13 @@ def test_set_session_active_desk_down_timer_scheduled_right_time(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_ALLOWED, INACTIVE, Timedelta(10), DOWN, - limits=(Timedelta(20), Timedelta(30))) + mocker, + TIME_ALLOWED, + INACTIVE, + Timedelta(10), + DOWN, + limits=(Timedelta(20), Timedelta(30)), + ) service.set_session(ACTIVE) @@ -16,8 +21,13 @@ def test_set_session_active_desk_down_timer_scheduled_right_time(mocker): def test_set_session_active_desk_up_timer_scheduled_right_time(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_ALLOWED, INACTIVE, Timedelta(10), UP, - limits=(Timedelta(20), Timedelta(30))) + mocker, + TIME_ALLOWED, + INACTIVE, + Timedelta(10), + UP, + limits=(Timedelta(20), Timedelta(30)), + ) service.set_session(ACTIVE) @@ -26,7 +36,8 @@ def test_set_session_active_desk_up_timer_scheduled_right_time(mocker): def test_set_session_inactive_timer_cancelled(mocker): (timer_mock, _, _, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), UP) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), UP + ) service.set_session(INACTIVE) @@ -35,7 +46,8 @@ def test_set_session_inactive_timer_cancelled(mocker): def test_set_session_timer_lambda_called_desk_service_called(mocker): (timer_stub, _, desk_service_mock, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), DOWN + ) service.set_session(ACTIVE) timer_stub.schedule.call_args[0][1]() @@ -45,7 +57,8 @@ def test_set_session_timer_lambda_called_desk_service_called(mocker): def test_set_session_timer_lambda_called_model_updated(mocker): (timer_stub, _, desk_service_mock, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), UP) + mocker, TIME_ALLOWED, ACTIVE, Timedelta(0), UP + ) service.set_session(ACTIVE) timer_stub.schedule.call_args[0][1]() @@ -55,7 +68,8 @@ def test_set_session_timer_lambda_called_model_updated(mocker): def test_set_session_hardware_error_timer_cancelled(mocker): (timer_mock, session_service_stub, _, service) = create_service( - mocker, TIME_ALLOWED, INACTIVE, Timedelta(0), DOWN) + mocker, TIME_ALLOWED, INACTIVE, Timedelta(0), DOWN + ) session_service_stub.set.side_effect = HardwareError(RuntimeError()) service.set_session(ACTIVE) @@ -65,8 +79,13 @@ def test_set_session_hardware_error_timer_cancelled(mocker): def test_toggle_session_from_inactive(mocker): (_, session_service_mock, _, service) = create_service( - mocker, TIME_ALLOWED, INACTIVE, Timedelta(10), DOWN, - limits=(Timedelta(20), Timedelta(30))) + mocker, + TIME_ALLOWED, + INACTIVE, + Timedelta(10), + DOWN, + limits=(Timedelta(20), Timedelta(30)), + ) service.toggle_session() @@ -75,8 +94,13 @@ def test_toggle_session_from_inactive(mocker): def test_toggle_session_from_active(mocker): (_, session_service_mock, _, service) = create_service( - mocker, TIME_ALLOWED, ACTIVE, Timedelta(10), DOWN, - limits=(Timedelta(20), Timedelta(30))) + mocker, + TIME_ALLOWED, + ACTIVE, + Timedelta(10), + DOWN, + limits=(Timedelta(20), Timedelta(30)), + ) service.toggle_session()