Introduces Occlusion masks table and related endpoints

client/pyroclient/client.py

@@ -25,6 +25,10 @@ class ClientRoute(str, Enum):
     # POSES
     POSES_CREATE = "poses/"
     POSES_BY_ID = "poses/{pose_id}"
+    POSES_MASKS_BY_ID = "poses/{pose_id}/occlusion_masks"
+    # OCCLUSION MASKS
+    OCCLUSION_MASKS_CREATE = "occlusion_masks"
+    OCCLUSION_MASKS_BY_ID = "occlusion_masks/{mask_id}"
     # DETECTIONS
     DETECTIONS_CREATE = "detections/"
     DETECTIONS_FETCH = "detections"
@@ -210,6 +214,83 @@ def delete_pose(self, pose_id: int) -> Response:
             timeout=self.timeout,
         )
 
+    def list_pose_masks(self, pose_id: int) -> Response:
+        """List occlusion masks for a pose (given its pose_id)
+
+        >>> api_client.list_pose_masks(pose_id=1)
+        """
+        return requests.get(
+            urljoin(self._route_prefix, ClientRoute.POSES_MASKS_BY_ID.format(pose_id=pose_id)),
+            headers=self.headers,
+            timeout=self.timeout,
+        )
+
+    # OCCLUSION MASKS
+
+    def create_occlusion_mask(
+        self,
+        pose_id: int,
+        mask: str,
+    ) -> Response:
+        """Create a create occlusion_mask for a pose
+
+        >>> api_client.create_occlusion_mask(pose_id=1, mask="(0.1,0.1,0.9,0.9)")
+        """
+        payload = {
+            "pose_id": pose_id,
+            "mask": mask,
+        }
+
+        return requests.post(
+            urljoin(self._route_prefix, ClientRoute.OCCLUSION_MASKS_CREATE),
+            headers=self.headers,
+            json=payload,
+            timeout=self.timeout,
+        )
+
+    def patch_occlusion_mask(
+        self,
+        mask_id: int,
+        mask: str,
+    ) -> Response:
+        """Update an occlusion mask
+
+        >>> api_client.patch_occlusion_mask(mask_id=1, mask="(0.1,0.2,0.9,0.3)")
+        """
+        payload = {"mask": mask}
+
+        return requests.patch(
+            urljoin(self._route_prefix, ClientRoute.OCCLUSION_MASKS_BY_ID.format(mask_id=mask_id)),
+            headers=self.headers,
+            json=payload,
+            timeout=self.timeout,
+        )
+
+    def get_occlusion_mask(
+        self,
+        mask_id: int,
+    ) -> Response:
+        """get mask from occlusion mask
+
+        >>> api_client.get_occlusion_mask(mask_id=1")
+        """
+        return requests.get(
+            urljoin(self._route_prefix, ClientRoute.OCCLUSION_MASKS_BY_ID.format(mask_id=mask_id)),
+            headers=self.headers,
+            timeout=self.timeout,
+        )
+
+    def delete_occlusion_mask(self, mask_id: int) -> Response:
+        """Delete an occlusion mask
+
+        >>> api_client.delete_occlusion_mask(mask_id=1)
+        """
+        return requests.delete(
+            urljoin(self._route_prefix, ClientRoute.OCCLUSION_MASKS_BY_ID.format(mask_id=mask_id)),
+            headers=self.headers,
+            timeout=self.timeout,
+        )
+
     # DETECTIONS
 
     def create_detection(

client/tests/test_client.py

@@ -56,6 +56,12 @@ def test_agent_workflow(test_cam_workflow, agent_token):
     assert len(response) == 1
     response = agent_client.label_sequence(response[0]["id"], "wildfire_smoke")
     assert response.status_code == 200, response.__dict__
+    response = agent_client.create_occlusion_mask(pose_id=1, mask="(0.1,0.1,0.9,0.9)")  # occlusion mask creation
+    assert response.status_code == 201, response.__dict__
+    print("reponseeee creation du mask")
+    print(response.json())
+    response = agent_client.delete_occlusion_mask(mask_id=response.json()["id"])  # occlusion mask deletion
+    assert response.status_code == 200, response.__dict__
 
 
 def test_user_workflow(test_cam_workflow, user_token):

src/app/api/api_v1/endpoints/occlusion_masks.py

@@ -0,0 +1,172 @@
+# Copyright (C) 2025-2026, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
+import re
+from typing import cast
+
+from fastapi import APIRouter, Body, Depends, HTTPException, Path, Security, status
+
+from app.api.dependencies import (
+    get_camera_crud,
+    get_jwt,
+    get_occlusion_mask_crud,
+    get_pose_crud,
+)
+from app.crud import CameraCRUD
+from app.crud.crud_occlusion_mask import OcclusionMaskCRUD
+from app.crud.crud_pose import PoseCRUD
+from app.models import Camera, OcclusionMask, Pose, UserRole
+from app.schemas.login import TokenPayload
+from app.schemas.occlusion_masks import (
+    OcclusionMaskCreate,
+    OcclusionMaskRead,
+    OcclusionMaskUpdate,
+)
+from app.services.telemetry import telemetry_client
+
+router = APIRouter()
+
+FLOAT_PATTERN = r"(0?\.[0-9]{1,3}|0|1)"
+MASK_PATTERN = rf"^\({FLOAT_PATTERN},{FLOAT_PATTERN},{FLOAT_PATTERN},{FLOAT_PATTERN}\)$"
+mask_regex = re.compile(MASK_PATTERN)
+
+
+def validate_mask(mask: str) -> None:
+    if not mask_regex.match(mask):
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=("Invalid mask format. Expected: (xmin, ymin, xmax, ymax) with float values in [0,1]."),
+        )
+
+
+@router.post(
+    "/",
+    status_code=status.HTTP_201_CREATED,
+    summary="Create an occlusion mask",
+)
+async def create_mask(
+    payload: OcclusionMaskCreate = Body(...),
+    poses: PoseCRUD = Depends(get_pose_crud),
+    cameras: CameraCRUD = Depends(get_camera_crud),
+    masks: OcclusionMaskCRUD = Depends(get_occlusion_mask_crud),
+    token_payload: TokenPayload = Security(
+        get_jwt,
+        scopes=[UserRole.ADMIN, UserRole.AGENT],
+    ),
+) -> OcclusionMaskRead:
+    # Validate mask format
+    validate_mask(payload.mask)
+
+    pose = cast(Pose, await poses.get(payload.pose_id, strict=True))
+    camera = cast(Camera, await cameras.get(pose.camera_id, strict=True))
+
+    if UserRole.ADMIN not in token_payload.scopes and token_payload.organization_id != camera.organization_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
+
+    telemetry_client.capture(
+        token_payload.sub,
+        event="occlusion_masks-create",
+        properties={"pose_id": payload.pose_id},
+    )
+
+    db_obj = await masks.create(payload)
+    return OcclusionMaskRead(**db_obj.model_dump())
+
+
+@router.get(
+    "/{mask_id}",
+    status_code=status.HTTP_200_OK,
+    summary="Get info about an occlusion mask",
+)
+async def get_mask(
+    mask_id: int = Path(..., gt=0),
+    masks: OcclusionMaskCRUD = Depends(get_occlusion_mask_crud),
+    poses: PoseCRUD = Depends(get_pose_crud),
+    cameras: CameraCRUD = Depends(get_camera_crud),
+    token_payload: TokenPayload = Security(
+        get_jwt,
+        scopes=[UserRole.ADMIN, UserRole.AGENT],
+    ),
+) -> OcclusionMaskRead:
+    mask = cast(OcclusionMask, await masks.get(mask_id, strict=True))
+    pose = cast(Pose, await poses.get(mask.pose_id, strict=True))
+    camera = cast(Camera, await cameras.get(pose.camera_id, strict=True))
+
+    if UserRole.ADMIN not in token_payload.scopes and token_payload.organization_id != camera.organization_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
+
+    telemetry_client.capture(
+        token_payload.sub,
+        event="occlusion_masks-get",
+        properties={"mask_id": mask_id},
+    )
+
+    return OcclusionMaskRead(**mask.model_dump())
+
+
+@router.patch(
+    "/{mask_id}",
+    status_code=status.HTTP_200_OK,
+    summary="Update an occlusion mask",
+)
+async def update_mask(
+    mask_id: int = Path(..., gt=0),
+    payload: OcclusionMaskUpdate = Body(...),
+    masks: OcclusionMaskCRUD = Depends(get_occlusion_mask_crud),
+    poses: PoseCRUD = Depends(get_pose_crud),
+    cameras: CameraCRUD = Depends(get_camera_crud),
+    token_payload: TokenPayload = Security(
+        get_jwt,
+        scopes=[UserRole.ADMIN, UserRole.AGENT],
+    ),
+) -> OcclusionMaskRead:
+    # Validate mask format
+    validate_mask(payload.mask)
+
+    mask = cast(OcclusionMask, await masks.get(mask_id, strict=True))
+    pose = cast(Pose, await poses.get(mask.pose_id, strict=True))
+    camera = cast(Camera, await cameras.get(pose.camera_id, strict=True))
+
+    if UserRole.ADMIN not in token_payload.scopes and token_payload.organization_id != camera.organization_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
+
+    telemetry_client.capture(
+        token_payload.sub,
+        event="occlusion_masks-update",
+        properties={"mask_id": mask_id},
+    )
+
+    db_obj = await masks.update(mask_id, payload)
+    return OcclusionMaskRead(**db_obj.model_dump())
+
+
+@router.delete(
+    "/{mask_id}",
+    status_code=status.HTTP_200_OK,
+    summary="Delete an occlusion mask",
+)
+async def delete_mask(
+    mask_id: int = Path(..., gt=0),
+    masks: OcclusionMaskCRUD = Depends(get_occlusion_mask_crud),
+    poses: PoseCRUD = Depends(get_pose_crud),
+    cameras: CameraCRUD = Depends(get_camera_crud),
+    token_payload: TokenPayload = Security(
+        get_jwt,
+        scopes=[UserRole.ADMIN, UserRole.AGENT],
+    ),
+) -> None:
+    mask = cast(OcclusionMask, await masks.get(mask_id, strict=True))
+    pose = cast(Pose, await poses.get(mask.pose_id, strict=True))
+    camera = cast(Camera, await cameras.get(pose.camera_id, strict=True))
+
+    if UserRole.ADMIN not in token_payload.scopes and token_payload.organization_id != camera.organization_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
+
+    telemetry_client.capture(
+        token_payload.sub,
+        event="occlusion_masks-delete",
+        properties={"mask_id": mask_id},
+    )
+
+    await masks.delete(mask_id)

src/app/api/api_v1/endpoints/poses.py

@@ -2,15 +2,17 @@
 
 # This program is licensed under the Apache License 2.0.
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
-from typing import cast
+from typing import List, cast
 
 from fastapi import APIRouter, Body, Depends, HTTPException, Path, Security, status
 
-from app.api.dependencies import get_camera_crud, get_jwt, get_pose_crud
+from app.api.dependencies import get_camera_crud, get_jwt, get_occlusion_mask_crud, get_pose_crud
 from app.crud import CameraCRUD
+from app.crud.crud_occlusion_mask import OcclusionMaskCRUD
 from app.crud.crud_pose import PoseCRUD
-from app.models import Camera, Pose, UserRole
+from app.models import Camera, Pose, Role, UserRole
 from app.schemas.login import TokenPayload
+from app.schemas.occlusion_masks import OcclusionMaskRead
 from app.schemas.poses import PoseCreate, PoseRead, PoseUpdate
 from app.services.telemetry import telemetry_client
 
@@ -83,3 +85,28 @@ async def delete_pose(
 ) -> None:
     telemetry_client.capture(token_payload.sub, event="poses-deletion", properties={"pose_id": pose_id})
     await poses.delete(pose_id)
+
+
+@router.get(
+    "/{pose_id}/occlusion_masks",
+    status_code=status.HTTP_200_OK,
+    summary="List occlusion masks for a pose",
+)
+async def list_pose_masks(
+    pose_id: int = Path(..., gt=0),
+    masks: OcclusionMaskCRUD = Depends(get_occlusion_mask_crud),
+    poses: PoseCRUD = Depends(get_pose_crud),
+    cameras: CameraCRUD = Depends(get_camera_crud),
+    token_payload: TokenPayload = Security(
+        get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER, Role.CAMERA]
+    ),
+) -> List[OcclusionMaskRead]:
+    telemetry_client.capture(token_payload.sub, event="occlusion_masks-list", properties={"pose_id": pose_id})
+    pose = cast(Pose, await poses.get(pose_id, strict=True))
+    camera = cast(Camera, await cameras.get(pose.camera_id, strict=True))
+
+    if token_payload.organization_id != camera.organization_id and UserRole.ADMIN not in token_payload.scopes:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
+
+    rows = await masks.get_by_pose(pose_id)
+    return [OcclusionMaskRead(**row.model_dump()) for row in rows]

src/app/api/api_v1/router.py

@@ -5,13 +5,24 @@
 
 from fastapi import APIRouter
 
-from app.api.api_v1.endpoints import cameras, detections, login, organizations, poses, sequences, users, webhooks
+from app.api.api_v1.endpoints import (
+    cameras,
+    detections,
+    login,
+    occlusion_masks,
+    organizations,
+    poses,
+    sequences,
+    users,
+    webhooks,
+)
 
 api_router = APIRouter(redirect_slashes=True)
 api_router.include_router(login.router, prefix="/login", tags=["login"])
 api_router.include_router(users.router, prefix="/users", tags=["users"])
 api_router.include_router(cameras.router, prefix="/cameras", tags=["cameras"])
 api_router.include_router(poses.router, prefix="/poses", tags=["poses"])
+api_router.include_router(occlusion_masks.router, prefix="/occlusion_masks", tags=["occlusion_masks"])
 api_router.include_router(detections.router, prefix="/detections", tags=["detections"])
 api_router.include_router(sequences.router, prefix="/sequences", tags=["sequences"])
 api_router.include_router(organizations.router, prefix="/organizations", tags=["organizations"])

src/app/api/dependencies.py

@@ -16,6 +16,7 @@
 
 from app.core.config import settings
 from app.crud import CameraCRUD, DetectionCRUD, OrganizationCRUD, SequenceCRUD, UserCRUD, WebhookCRUD
+from app.crud.crud_occlusion_mask import OcclusionMaskCRUD
 from app.crud.crud_pose import PoseCRUD
 from app.db import get_session
 from app.models import User, UserRole
@@ -49,6 +50,10 @@ def get_pose_crud(session: AsyncSession = Depends(get_session)) -> PoseCRUD:
     return PoseCRUD(session=session)
 
 
+def get_occlusion_mask_crud(session: AsyncSession = Depends(get_session)) -> OcclusionMaskCRUD:
+    return OcclusionMaskCRUD(session=session)
+
+
 def get_detection_crud(session: AsyncSession = Depends(get_session)) -> DetectionCRUD:
     return DetectionCRUD(session=session)