Format all code using ruff

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)

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()

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
+        )

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

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,19 +18,20 @@ 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()
         self.model.set_session(now, 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

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()

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)
 
 

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))