api: implement operating locks

api/src/dooris_api/app.py

@@ -240,9 +240,37 @@ async def list_locks(
 
 
 @app.patch(
-    "/api/locks/{name}",
+    "/api/locks/{lock_id}",
     tags=["locks"],
-    responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}},
+    responses={
+        status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail},
+        status.HTTP_404_NOT_FOUND: {"model": models.HttpProblemDetail},
+    },
 )
-async def operate_lock(name: str):
+async def operate_lock(req: Request, lock_id: str, requested_op: models.LockOperation, ccujack: deps.CCUJackClient, _current_user: deps.CurrentUser) -> None:
-    pass
+    # TODO: Validate that the user is authorized
+    # find appropriate lock from ccujack
+    for i_lock, lock_channels in ccujack.locks:
+        if i_lock.identifier == lock_id:
+            for i_channel, channel_params in lock_channels:
+                if i_channel.type == "DOOR_LOCK_STATE_TRANSMITTER":
+                    for i_param in channel_params:
+                        if i_param.id == "LOCK_TARGET_LEVEL":
+                            addr = f"{i_lock.address}/{i_channel.index}/{i_param.id}"
+
+                            # match readable request parameter to ccujack value
+                            match requested_op.desired_state:
+                                case "closed":
+                                    ccujack_value = 0
+                                case "open":
+                                    ccujack_value = 1
+
+                            # write to ccujack
+                            await ccujack.set_param_value(addr, ccujack_value)
+                            return
+                    
+                    
+    else:
+        raise exceptions.HttpProblemException(
+            models.HttpProblemDetail.new_lock_not_found(lock_id, req.url)
+        )

api/src/dooris_api/ccujack.py

@@ -83,13 +83,14 @@ class CCUJackClient:
             ]
                 
 
-    async def query_param_value(self, address):
+    async def query_param_value(self, address: str):
         logger.debug("Querying parameter value from '%s'", address)
         async with self.http.get(f"/device/{address}/~pv") as resp:
             return CCUValue.model_validate(await resp.json())
 
-    # async def toggle_lock(self, lock_id, new_state):
+    async def set_param_value(self, address: str, value: Any):
-    #     pass
+        logger.debug("Writing parameter value '%s' to '%s'", value, address)
+        await self.http.put(f"/device/{address}/~pv", json={"v": value})
 
     async def _inspect_ccu_device(self, device_ref: CCURef) -> Tuple[CCUDeviceInfo, List[Tuple[CCUChannelInfo, List[CCUParamInfo]]]]:
         logger.debug("Inspecting device '%s' (%s)", device_ref.href, device_ref.title)

api/src/dooris_api/models.py

@@ -13,7 +13,7 @@ class HttpProblemType(Enum):
     """
 
     UNAUTHORIZED = "type:noc@hamburg.ccc.de,2026:UNAUTHORIZED"
-    DOOR_NOT_FOUND = "type:noc@hamburg.ccc.de,2026:DOOR_NOT_FOUND"
+    LOCK_NOT_FOUND = "type:noc@hamburg.ccc.de,2026:LOCK_NOT_FOUND"
 
 
 class HttpProblemDetail(BaseModel):
@@ -38,12 +38,12 @@ class HttpProblemDetail(BaseModel):
         )
 
     @classmethod
-    def new_door_not_found(cls, requested_door: str, request_uri: str | URL) -> Self:
+    def new_lock_not_found(cls, requested_lock: str, request_uri: str | URL) -> Self:
         return cls(
-            type=HttpProblemType.DOOR_NOT_FOUND,
+            type=HttpProblemType.LOCK_NOT_FOUND,
             status=status.HTTP_404_NOT_FOUND,
-            title="Door not found",
+            title="Lock not found",
-            detail=f"You tried to interact with door {requested_door!r} that is not known to dooris",
+            detail=f"You tried to interact with lock {requested_lock!r} that is not known to dooris",
             instance=str(request_uri),
         )
 
@@ -73,3 +73,7 @@ class Lock(BaseModel):
     name: str
     id: str
     status: LockStatus
+
+
+class LockOperation(BaseModel):
+    desired_state: Literal["open", "closed"]