Combine the state types into Desk and Session

autodesk/api.py

@@ -1,7 +1,7 @@
 from aiohttp import web
 from autodesk.button import Button
 from autodesk.hardware.error import HardwareError
-from autodesk.states import deserialize_session, deserialize_desk
+from autodesk.states import Desk, Session, deserialize_session, deserialize_desk
 import aiohttp_jinja2
 import asyncio
 import autodesk.plots as plots
@@ -25,6 +25,7 @@ async def route_set_session(request: web.Request):
 
 async def route_get_session(request: web.Request):
     state = request.app["service"].get_session_state()
+    assert type(state) is Session
     return web.Response(text=state.test("inactive", "active"))
 
 
@@ -41,6 +42,7 @@ async def route_set_desk(request: web.Request):
 
 async def route_get_desk(request: web.Request):
     state = request.app["service"].get_desk_state()
+    assert type(state) is Desk
     return web.Response(text=state.test("down", "up"))
 
 

autodesk/application/autodeskservice.py

@@ -1,4 +1,5 @@
 from autodesk.hardware.error import HardwareError
+from autodesk.states import Desk, Session
 import logging
 
 
@@ -23,10 +24,10 @@ def init(self):
             self.logger.debug(error)
             self.timer.cancel()
 
-    def get_session_state(self):
+    def get_session_state(self) -> Session:
         return self.session_service.get()
 
-    def set_session(self, state):
+    def set_session(self, state: Session):
         try:
             self.session_service.set(state)
             self._update_timer()
@@ -39,10 +40,10 @@ def toggle_session(self):
         session_state = self.session_service.get()
         self.set_session(session_state.next())
 
-    def get_desk_state(self):
+    def get_desk_state(self) -> Desk:
         return self.desk_service.get()
 
-    def set_desk(self, state):
+    def set_desk(self, state: Desk):
         try:
             self.desk_service.set(state)
             self._update_timer()
@@ -58,8 +59,8 @@ def compute_hourly_count(self):
         return self.session_service.compute_hourly_count()
 
     def _update_timer(self):
-        session_state = self.session_service.get()
-        desk_state = self.desk_service.get()
+        session_state: Session = self.session_service.get()
+        desk_state: Desk = self.desk_service.get()
         now = self.time_service.now()
         if self.operation.allowed(session_state, now):
             active_time = self.session_service.get_active_time()

autodesk/sqlitedatastore.py

@@ -21,10 +21,8 @@ def convert_timestamp(val):
 sqlite3.register_converter("timestamp", convert_datetime)
 sqlite3.register_converter("unix_timestamp", convert_timestamp)
 
-sqlite3.register_adapter(states.Down, lambda _: 0)
-sqlite3.register_adapter(states.Up, lambda _: 1)
-sqlite3.register_adapter(states.Inactive, lambda _: 0)
-sqlite3.register_adapter(states.Active, lambda _: 1)
+sqlite3.register_adapter(states.Desk, lambda desk: desk.value())
+sqlite3.register_adapter(states.Session, lambda session: session.value())
 sqlite3.register_converter("desk", states.deserialize_desk)
 sqlite3.register_converter("session", states.deserialize_session)
 sqlite3.register_converter("desk_int", states.deserialize_desk_int)

autodesk/states.py

@@ -1,52 +1,42 @@
-class Up:
-    def next(self):
-        return Down()
-
-    def test(self, _, b):
-        return b
-
-    def __eq__(self, other):
-        return isinstance(other, Up)
+class Desk:
+    def __init__(self, state: bool):
+        self._state = state
 
+    def value(self):
+        return int(self._state)
 
-class Down:
     def next(self):
-        return Up()
+        return Desk(not self._state)
 
-    def test(self, a, _):
-        return a
+    def test[T](self, a: T, b: T) -> T:
+        return b if self._state else a
 
     def __eq__(self, other):
-        return isinstance(other, Down)
+        return isinstance(other, Desk) and self._state == other._state
 
 
-class Active:
-    def next(self):
-        return Inactive()
-
-    def test(self, _, b):
-        return b
-
-    def __eq__(self, other):
-        return isinstance(other, Active)
+class Session:
+    def __init__(self, state: bool):
+        self._state = state
 
+    def value(self):
+        return int(self._state)
 
-class Inactive:
     def next(self):
-        return Active()
+        return Session(not self._state)
 
-    def test(self, a, _):
-        return a
+    def test[T](self, a: T, b: T) -> T:
+        return b if self._state else a
 
     def __eq__(self, other):
-        return isinstance(other, Inactive)
+        return isinstance(other, Session) and self._state == other._state
 
 
-DOWN = Down()
-UP = Up()
+DOWN = Desk(False)
+UP = Desk(True)
 
-INACTIVE = Inactive()
-ACTIVE = Active()
+INACTIVE = Session(False)
+ACTIVE = Session(True)
 
 
 def deserialize_session(value: bytes | str):
@@ -58,7 +48,7 @@ def deserialize_session(value: bytes | str):
     raise ValueError('Incorrect session state "{0}".'.format(value))
 
 
-def deserialize_desk(value: bytes | str) -> Down | Up:
+def deserialize_desk(value: bytes | str):
     if value in (b"down", "down"):
         return DOWN
     if value in (b"up", "up"):
@@ -67,19 +57,19 @@ def deserialize_desk(value: bytes | str) -> Down | Up:
     raise ValueError('Incorrect desk state "{0}".'.format(value))
 
 
-def deserialize_session_int(value):
-    if value == b'0':
+def deserialize_session_int(value: bytes):
+    if value == b"0":
         return INACTIVE
-    if value == b'1':
+    if value == b"1":
         return ACTIVE
 
     raise ValueError('Incorrect session state "{0}".'.format(value))
 
 
-def deserialize_desk_int(value):
-    if value == b'0':
+def deserialize_desk_int(value: bytes):
+    if value == b"0":
         return DOWN
-    if value == b'1':
+    if value == b"1":
         return UP
 
     raise ValueError('Incorrect desk state "{0}".'.format(value))