Merge pull request #20 from max-pfeiffer/bugfix/cors_header

Bugfix/cors header

backend/app/adapters/waveshare.py

@@ -40,14 +40,18 @@ def __init__(
             ]
         self.relays: list[Relay] = relays
 
-    def get_relay(self, relay_position: int):
+    def get_relay(self, relay_position: int) -> Relay:
         """Return the relay object at the desired position."""
         if relay_position not in self.relay_position_mapping.keys():
             raise ValueError("Invalid relay position")
 
         relay: Relay = self.relays[self.relay_position_mapping[relay_position]]
         return relay
 
+    def get_relays(self) -> list[Relay]:
+        """Return all relay objects."""
+        return self.relays
+
     def on(self, relay_position: int):
         """Turn on relay at relay_position.
 

backend/app/api/v1/endpoints/relay.py

@@ -1,13 +1,28 @@
 """API endpoints for Relay objects."""
 from fastapi import APIRouter
 
-from app.api.v1.models import Relay
+from app.api.v1.models import Relay, RelayUpdate
 from app.config import relayBoardAdapter
-from app.services.relay import service_get_relay
+from app.services.relay import (
+    service_get_relay,
+    service_get_relays,
+    service_update_relay,
+)
 
 router = APIRouter()
 
 
+@router.get("/")
+def get_relays() -> list[Relay]:
+    """Get all relays.
+
+    :return:
+    """
+    data: list[dict] = service_get_relays(relayBoardAdapter)
+    relays: list[Relay] = [Relay(**item) for item in data]
+    return relays
+
+
 @router.get("/{position}")
 def get_relay(position: int) -> Relay:
     """Get relay by position.
@@ -20,13 +35,11 @@ def get_relay(position: int) -> Relay:
 
 
 @router.put("/{position}")
-def update_relay(relay: Relay):
+def update_relay(position: int, relay: RelayUpdate):
     """Update relay by position.
 
+    :param position:
     :param relay:
     :return:
     """
-    if relay.on:
-        relayBoardAdapter.on(relay.position)
-    else:
-        relayBoardAdapter.off(relay.position)
+    service_update_relay(relayBoardAdapter, position, relay.on)

backend/app/api/v1/endpoints/schedule.py

@@ -43,7 +43,7 @@ def update_schedule(request: Request, primary_key: int, schedule_data: ScheduleU
     """Update Schedule."""
     service_update_schedule(
         request.app.state.scheduler,
-        primary_key=primary_key,
+        primary_key,
         **schedule_data.model_dump(exclude_defaults=True),
     )
 

backend/app/api/v1/models.py

@@ -4,7 +4,6 @@
 
 from pydantic import BaseModel, Field, PositiveInt
 
-from app.adapters import RelayBoardType
 from app.scheduling import Repeat
 
 
@@ -16,8 +15,6 @@ class ScheduleResponse(BaseModel):
     duration: PositiveInt = Field(description="Duration in minutes")
     repeat: Repeat = Field(description="Specifies how the schedule is repeated")
     active: bool = Field(description="Whether the schedule is active")
-    relay_board_type: RelayBoardType = Field(description="Type of the relay board")
-    relay_position: PositiveInt = Field(description="Position of the relay")
 
 
 class ScheduleCreate(BaseModel):
@@ -27,7 +24,6 @@ class ScheduleCreate(BaseModel):
     duration: PositiveInt = Field(description="Duration in minutes")
     repeat: Repeat = Field(description="Specifies how the schedule is repeated")
     active: bool = Field(default=True, description="Whether the schedule is active")
-    relay_board_type: RelayBoardType = Field(description="Type of the relay board")
     relay_position: PositiveInt = Field(description="Position of the relay")
 
 
@@ -46,9 +42,6 @@ class ScheduleUpdate(BaseModel):
     active: Union[bool, None] = Field(
         default=None, description="Whether the schedule is active"
     )
-    relay_board_type: Union[RelayBoardType, None] = Field(
-        default=None, description="Type of the relay board"
-    )
     relay_position: Union[PositiveInt, None] = Field(
         default=None, description="Position of the relay"
     )
@@ -59,3 +52,9 @@ class Relay(BaseModel):
 
     position: PositiveInt = Field(description="Relay position on the board")
     on: bool = Field(description="Indicates if relay is switched on")
+
+
+class RelayUpdate(BaseModel):
+    """Update schema for Relay object."""
+
+    on: bool = Field(description="Indicates if relay is switched on")

backend/app/config.py

@@ -17,7 +17,7 @@ class ApplicationSettings(BaseSettings):
     database_name: str = "sqlite"
 
     @computed_field
-    def database_uri(self) -> str:
+    def  database_uri(self) -> str:
         """URI for database connection.
 
         :return:
@@ -57,9 +57,11 @@ def get_relay_board_adapter() -> WaveshareRpiRelayBoardAdapter:
 
         if pin_factory_type == "rpi_gpio":
             from gpiozero.pins.rpigpio import RPiGPIOFactory
+
             pin_factory = RPiGPIOFactory()
         elif pin_factory_type == "pigpio":
             from gpiozero.pins.pigpio import PiGPIOFactory
+
             pin_factory = PiGPIOFactory()
         elif pin_factory_type == "native":
             pin_factory = NativeFactory()

backend/app/database/models.py

@@ -5,7 +5,6 @@
 from pydantic import PositiveInt
 from sqlmodel import Field, SQLModel
 
-from app.adapters import RelayBoardType
 from app.scheduling import Repeat
 
 
@@ -19,10 +18,10 @@ class Schedule(BaseModel, table=True):
     """Schedule."""
 
     start_time: time
+    stop_time: time
     duration: PositiveInt
     repeat: Repeat
     active: bool
-    relay_board_type: RelayBoardType
     relay_position: PositiveInt
-    start_schedule_id: Optional[str]
-    stop_schedule_id: Optional[str]
+    start_job_id: Optional[str]
+    stop_job_id: Optional[str]

backend/app/main.py

@@ -47,4 +47,3 @@ def redirect_to_autodocs(request: Request) -> RedirectResponse:
 
 
 app.include_router(api_router, prefix="/v1")
-

backend/app/repositories.py

@@ -1,5 +1,6 @@
 """Repositories for data persistence."""
-from datetime import date, datetime, time, timedelta
+from datetime import time
+from typing import Tuple
 
 from apscheduler.job import Job
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
@@ -56,14 +57,13 @@ def create(self, **kwargs) -> Schedule:
         self.session.flush()
         return schedule
 
-    def update(self, **kwargs) -> Schedule:
+    def update(self, primary_key: int, **kwargs) -> Schedule:
         """Update a Schedule object.
 
         :param int primary_key:
         :param kwargs:
         :return:
         """
-        primary_key: int = kwargs.pop("primary_key")
         schedule: Schedule = self.get(primary_key)
 
         for key, value in kwargs.items():
@@ -93,19 +93,14 @@ def __init__(self, scheduler: AsyncIOScheduler) -> None:
         """
         self.scheduler: AsyncIOScheduler = scheduler
 
-    def _create_trigger_data(self, schedule_data: dict) -> tuple[dict, dict]:
+    def _create_trigger_data(
+        self, start_time: time, stop_time: time, repeat: Repeat
+    ) -> tuple[dict, dict]:
         """Create trigger date from Schedule object.
 
         :param dict schedule_data:
         :return dict:
         """
-        if not schedule_data["active"]:
-            raise Exception("Schedule is not active")
-
-        start_time: time = schedule_data["start_time"]
-        repeat: Repeat = schedule_data["repeat"]
-        duration: int = schedule_data["duration"]
-
         start_data: dict = {
             "hour": start_time.hour,
             "minute": start_time.minute,
@@ -114,10 +109,6 @@ def _create_trigger_data(self, schedule_data: dict) -> tuple[dict, dict]:
             "month": "*",
         }
 
-        stop_date_time: datetime = datetime.combine(
-            date.today(), start_time, tzinfo=start_time.tzinfo
-        ) + timedelta(minutes=duration)
-        stop_time: time = stop_date_time.timetz()
         stop_data: dict = {
             "hour": stop_time.hour,
             "minute": stop_time.minute,
@@ -161,64 +152,58 @@ def _create_trigger_data(self, schedule_data: dict) -> tuple[dict, dict]:
 
         return start_data, stop_data
 
-    def create(self, schedule_data: dict) -> dict:
+    def create(
+        self, start_time: time, stop_time: time, repeat: Repeat, relay_position: int
+    ) -> Tuple[str, str]:
         """Add a trigger to the scheduler.
 
         :param dict schedule_data:
         :return:
         """
-        start_data, stop_data = self._create_trigger_data(schedule_data)
+        start_data, stop_data = self._create_trigger_data(start_time, stop_time, repeat)
         start_job: Job = self.scheduler.add_job(
             task_switch_relay,
             CronTrigger(**start_data),
             args=[
-                schedule_data["relay_position"],
+                relay_position,
                 True,
             ],
         )
         stop_job: Job = self.scheduler.add_job(
             task_switch_relay,
             CronTrigger(**stop_data),
             args=[
-                schedule_data["relay_position"],
+                relay_position,
                 False,
             ],
         )
+        return start_job.id, stop_job.id
 
-        return_data: dict = schedule_data.copy()
-        return_data["start_schedule_id"] = start_job.id
-        return_data["stop_schedule_id"] = stop_job.id
-        return return_data
-
-    def create_multiple(self, schedule_data_list: list[dict]):
-        """Add multiple triggers to the scheduler.
-
-        :param list[dict] schedule_data_list:
-        :return:
-        """
-        return_data_list: list[dict] = []
-
-        for schedule_data in schedule_data_list:
-            return_data: dict = self.create(schedule_data)
-            return_data_list.append(return_data)
-
-        return return_data_list
-
-    def delete(self, schedule_data: dict):
+    def delete(self, primary_key: str):
         """Add a trigger to the scheduler.
 
         :param dict schedule_data:
         :return:
         """
-        self.scheduler.remove_job(schedule_data["start_schedule_id"])
-        self.scheduler.remove_job(schedule_data["stop_schedule_id"])
-
-    def update(self, schedule_data: dict) -> dict:
+        self.scheduler.remove_job(primary_key)
+
+    def update(
+        self,
+        start_time: time,
+        stop_time: time,
+        repeat: Repeat,
+        relay_position: int,
+        start_job_id: str,
+        stop_job_id: str,
+    ) -> Tuple[str, str]:
         """Add a trigger to the scheduler.
 
         :param dict schedule_data:
         :return:
         """
-        self.delete(schedule_data)
-        return_data: dict = self.create(schedule_data)
+        self.delete(start_job_id)
+        self.delete(stop_job_id)
+        return_data: Tuple[str, str] = self.create(
+            start_time, stop_time, repeat, relay_position
+        )
         return return_data

backend/app/scheduling.py

@@ -20,9 +20,7 @@ class Repeat(str, Enum):
     sunday = "sunday"
 
 
-def task_switch_relay(
-    relay_position: int, on: bool
-):
+def task_switch_relay(relay_position: int, on: bool):
     """Trigger function to switch relays.
 
     :param relay_position:

backend/app/services/relay.py

@@ -3,9 +3,7 @@
 from app.adapters.gpio_devices import Relay
 
 
-def service_get_relay(
-    adapter: RelayBoardAdapter, relay_position: int
-) -> dict:
+def service_get_relay(adapter: RelayBoardAdapter, relay_position: int) -> dict:
     """Get relay data.
 
     :param adapter:
@@ -18,3 +16,30 @@ def service_get_relay(
         "on": bool(relay.value),
     }
     return data
+
+
+def service_get_relays(adapter: RelayBoardAdapter) -> list[dict]:
+    """Get all relay data.
+
+    :param adapter:
+    :return:
+    """
+    relays: list[Relay] = adapter.get_relays()
+    data: list[dict] = []
+    for index, relay in enumerate(relays):
+        data.append({"position": index + 1, "on": bool(relay.value)})
+    return data
+
+
+def service_update_relay(adapter: RelayBoardAdapter, relay_position: int, on: bool):
+    """Update relay.
+
+    :param adapter:
+    :param relay_position:
+    :param on:
+    :return:
+    """
+    if on:
+        adapter.on(relay_position)
+    else:
+        adapter.off(relay_position)