Merge branch 'Add-Multi-OpId-Search'

Author: mgarcia01752

src/pypnm/api/routes/advance/common/abstract/multi_capture_router.py

@@ -13,7 +13,11 @@
 
 from pypnm.api.routes.advance.common.abstract.service import AbstractService
 from pypnm.api.routes.advance.common.capture_service import AbstractCaptureService
+from pypnm.api.routes.advance.common.operation_kind import MultiCaptureOperation
 from pypnm.api.routes.advance.common.operation_manager import OperationManager
+from pypnm.api.routes.advance.common.schema.common_capture_schema import (
+    MultiCaptureOperationIdResponse,
+)
 from pypnm.api.routes.common.classes.file_capture.capture_group import CaptureGroup
 from pypnm.config.system_config_settings import SystemConfigSettings
 from pypnm.lib.types import GroupId, OperationId
@@ -132,3 +136,8 @@ def _repair_capture_group_from_service_samples(self, operation_id: OperationId,
                 exc,
             )
             return 0
+
+    def _build_operation_id_response(self, operation_name: MultiCaptureOperation) -> MultiCaptureOperationIdResponse:
+        """Return persisted operation records for a canonical multi-capture operation family."""
+        operations = OperationManager.list_operation_records_by_name(operation_name.value)
+        return MultiCaptureOperationIdResponse(status="success", message=None, operations=operations)

src/pypnm/api/routes/advance/common/capture_service.py

@@ -6,8 +6,13 @@
 import logging
 import time
 from abc import ABC, abstractmethod
+from enum import Enum
 from typing import Any, cast
 
+from pypnm.api.routes.advance.common.operation_kind import (
+    MultiCaptureOperation,
+    MultiCaptureOperationModel,
+)
 from pypnm.api.routes.advance.common.operation_manager import OperationManager
 from pypnm.api.routes.advance.common.operation_state import OperationState
 from pypnm.api.routes.common.classes.file_capture.capture_group import CaptureGroup
@@ -43,6 +48,9 @@ class AbstractCaptureService(ABC):
         logger (logging.Logger): Logger for operational messages.
     """
 
+    OPERATION_NAME: MultiCaptureOperation = MultiCaptureOperation.MULTI_RXMER
+    MEASURE_MODE: str = "standard"
+
     def __init__(
         self,
         duration: float,
@@ -104,10 +112,11 @@ async def start(self) -> tuple[GroupId, OperationId]:
             operation_metadata["mac_address"] = str(mac_str)
         if self._has_populated_system_description(self._system_description):
             operation_metadata["system_description"] = dict(self._system_description)
+        operation = self.get_operation_model()
 
         try:
             om = OperationManager(capture_group_id=group_id)
-            operation_id:OperationId = om.register(metadata=operation_metadata)
+            operation_id:OperationId = om.register(operation=operation, metadata=operation_metadata)
         except Exception as exc:
             self.logger.error(f"Failed to create operation manager, reason={exc}", exc_info=True)
             raise
@@ -283,6 +292,10 @@ def get_system_description(self) -> dict[str, str]:
         """Return the cached session-level sysDescr payload used for multi-capture metadata."""
         return dict(self._system_description)
 
+    def get_operation_model(self) -> MultiCaptureOperationModel:
+        """Return the persisted operation block associated with this capture service."""
+        return MultiCaptureOperationModel(name=self.OPERATION_NAME, measure_mode=self._normalize_measure_mode(self.MEASURE_MODE))
+
     def status(self, operation_id: OperationId) -> dict[str, Any]:
         """
         Get the current state and sample count for a capture operation.
@@ -433,6 +446,13 @@ def _process_captures(self, msg_rsp: MessageResponse) -> list[CaptureSample]:
 
         return samples
 
+    @staticmethod
+    def _normalize_measure_mode(measure_mode: Enum | str) -> str:
+        """Normalize enum or string measure-mode values for persistence."""
+        if isinstance(measure_mode, Enum):
+            return measure_mode.name.lower()
+        return str(measure_mode).strip().lower()
+
     @abstractmethod
     async def _capture_message_response(self) -> MessageResponse:
         """

src/pypnm/api/routes/advance/common/operation_kind.py

@@ -0,0 +1,22 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2026 Maurice Garcia
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from pypnm.lib.types import StringEnum
+
+
+class MultiCaptureOperation(StringEnum):
+    """Canonical operation families for persisted multi-capture records."""
+
+    MULTI_RXMER = "multi_rxmer"
+    MULTI_DS_CHANNEL_ESTIMATION = "multi_ds_channel_estimation"
+    MULTI_US_OFDMA_PRE_EQUALIZATION = "multi_us_ofdma_pre_equalization"
+
+
+class MultiCaptureOperationModel(BaseModel):
+    """Persisted operation identity for replaying a completed multi-capture."""
+
+    name: MultiCaptureOperation = Field(..., description="Canonical multi-capture operation family name.")
+    measure_mode: str = Field(..., description="Normalized measure mode name used when the operation was started.")

src/pypnm/api/routes/advance/common/operation_manager.py

@@ -9,6 +9,10 @@
 from pathlib import Path
 from typing import Any
 
+from pypnm.api.routes.advance.common.operation_kind import MultiCaptureOperationModel
+from pypnm.api.routes.advance.common.schema.common_capture_schema import (
+    MultiCapturePersistedRecordModel,
+)
 from pypnm.config.system_config_settings import SystemConfigSettings
 from pypnm.lib.constants import cast
 from pypnm.lib.db.json_file_lock import JsonFileLock
@@ -27,7 +31,11 @@ class OperationManager:
     {
         "<operation_id>": {
             "capture_group_id": "<group_id>",
-            "created": <unix_epoch_seconds>
+            "created": <unix_epoch_seconds>,
+            "operation": {
+                "name": "multi_rxmer",
+                "measure_mode": "ofdm_performance_1"
+            }
         },
         ...
     }
@@ -93,7 +101,11 @@ def _save(self, data: dict[str, Any]) -> None:
         except Exception as e:
             self.logger.error(f"Failed to save operation DB: {e}")
 
-    def register(self, metadata: dict[str, Any] | None = None) -> OperationId:
+    def register(
+        self,
+        operation: MultiCaptureOperationModel | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> OperationId:
         """
         Register this operation with its capture group ID in the DB.
 
@@ -121,6 +133,8 @@ def register(self, metadata: dict[str, Any] | None = None) -> OperationId:
                 "capture_group_id": self.capture_group_id,
                 "created": int(time.time())
             }
+            if operation is not None:
+                db[self.operation_id]["operation"] = operation.model_dump()
             if isinstance(metadata, dict) and metadata:
                 db[self.operation_id]["metadata"] = metadata
             self._save(db)
@@ -193,3 +207,49 @@ def get_operation_record(cls, operation_id: OperationId, db_path: Path | None =
         except Exception as e:
             logging.getLogger(cls.__name__).error(f"Error retrieving operation record for {operation_id}: {e}")
             return None
+
+    @classmethod
+    def list_operation_records_by_name(
+        cls,
+        operation_name: str,
+        db_path: Path | None = None,
+    ) -> dict[OperationId, MultiCapturePersistedRecordModel]:
+        """
+        Return persisted operation records filtered by operation.name.
+
+        Args:
+            operation_name: Canonical operation.name value to filter on.
+            db_path: Optional override for the operations DB file.
+
+        Returns:
+            Mapping of operation_id to typed persisted operation records.
+        """
+        if not db_path:
+            db_str = SystemConfigSettings.operation_db()
+            db_path = Path(db_str)
+        try:
+            with JsonFileLock(db_path), db_path.open("r", encoding="utf-8") as f:
+                db = json.load(f)
+        except Exception as e:
+            logging.getLogger(cls.__name__).error(f"Error listing operation records for {operation_name}: {e}")
+            return {}
+
+        records: dict[OperationId, MultiCapturePersistedRecordModel] = {}
+        for operation_id, record in db.items():
+            if not isinstance(record, dict):
+                continue
+            operation = record.get("operation")
+            if not isinstance(operation, dict):
+                continue
+            if operation.get("name") != operation_name:
+                continue
+            try:
+                records[cast(OperationId, operation_id)] = MultiCapturePersistedRecordModel(**record)
+            except Exception as e:
+                logging.getLogger(cls.__name__).warning(
+                    "Skipping invalid operation record %s for %s: %s",
+                    operation_id,
+                    operation_name,
+                    e,
+                )
+        return records

src/pypnm/api/routes/advance/common/schema/common_capture_schema.py

@@ -5,13 +5,14 @@
 
 from pydantic import BaseModel, Field
 
+from pypnm.api.routes.advance.common.operation_kind import MultiCaptureOperationModel
 from pypnm.api.routes.common.classes.common_endpoint_classes.common_req_resp import (
     CableModemPnmConfig,
 )
 from pypnm.api.routes.common.classes.common_endpoint_classes.schema.base_response import (
     BaseDeviceResponse,
 )
-from pypnm.lib.types import OperationId
+from pypnm.lib.types import GroupId, MacAddressStr, OperationId, TimestampSec
 
 
 class CaptureParameters(BaseModel):
@@ -36,3 +37,26 @@ class MultiCaptureParametersResponse(BaseDeviceResponse):
     time_remaining: int  = Field(..., description="Remaining time in seconds.")
     message:        str | None = Field(default="", description="Optional human-readable message or error detail.")
 
+
+class MultiCapturePersistedMetadataModel(BaseModel):
+    """Known persisted metadata attached to a multi-capture operation record."""
+
+    mac_address: MacAddressStr | None = Field(default=None, description="Cable modem MAC address captured when the operation was started.")
+    system_description: dict[str, str] = Field(default_factory=dict, description="Persisted system-description fields captured when the operation was started.")
+
+
+class MultiCapturePersistedRecordModel(BaseModel):
+    """Persisted multi-capture operation record stored by operation ID."""
+
+    capture_group_id: GroupId = Field(..., description="Capture-group identifier associated with the persisted operation.")
+    created: TimestampSec = Field(..., description="Unix epoch seconds when the operation record was created.")
+    operation: MultiCaptureOperationModel = Field(..., description="Canonical operation identity and measure-mode metadata.")
+    metadata: MultiCapturePersistedMetadataModel = Field(default_factory=MultiCapturePersistedMetadataModel, description="Additional persisted metadata such as MAC address and system description.")
+
+
+class MultiCaptureOperationIdResponse(BaseModel):
+    """Collection of persisted operation records keyed by operation ID."""
+
+    status: str | None = Field(default="success", description="Overall status for the operation-record listing request.")
+    message: str | None = Field(default=None, description="Additional information or error details for the operation-record listing request.")
+    operations: dict[OperationId, MultiCapturePersistedRecordModel] = Field(default_factory=dict, description="Persisted operation records keyed by operation ID for the requested operation family.")

src/pypnm/api/routes/advance/multi_ds_chan_est/router.py

@@ -20,7 +20,11 @@
 from pypnm.api.routes.advance.common.capture_data_aggregator import (
     CaptureDataAggregator,
 )
+from pypnm.api.routes.advance.common.operation_kind import MultiCaptureOperation
 from pypnm.api.routes.advance.common.operation_state import OperationState
+from pypnm.api.routes.advance.common.schema.common_capture_schema import (
+    MultiCaptureOperationIdResponse,
+)
 from pypnm.api.routes.advance.multi_ds_chan_est.schemas import (
     AnalysisDataModel,
     MultiChanEstAnalysisRequest,
@@ -123,10 +127,11 @@ async def start_multi_chan_estimation(request: MultiChanEstRequest) -> MultiChan
                                                     operation_id    =   operation_id)
 
 
-        @self.router.get("/status/{operation_id}",
+        @self.router.get("/status/{operationId}",
             response_model=MultiChanEstStatusResponse,
             summary="Get status of a multi-sample ChannelEstimation capture")
-        def get_status(operation_id: OperationId) -> MultiChanEstStatusResponse:
+        def get_status(operationId: OperationId) -> MultiChanEstStatusResponse:
+            operation_id = operationId
             service: MultiChannelEstimationService = cast(
                 MultiChannelEstimationService,
                 self._get_service_or_404(operation_id),
@@ -145,23 +150,31 @@ def get_status(operation_id: OperationId) -> MultiChanEstStatusResponse:
                     time_remaining  =   status["time_remaining"],
                     message         =   None))
 
-        @self.router.get("/results/{operation_id}",
+        @self.router.get("/operationId",
+            response_model=MultiCaptureOperationIdResponse,
+            summary="List persisted Multi-DS-Channel-Estimation operation IDs")
+        def get_operation_ids() -> MultiCaptureOperationIdResponse:
+            """Return persisted operation records for Multi-DS-Channel-Estimation keyed by operation ID."""
+            return self._build_operation_id_response(MultiCaptureOperation.MULTI_DS_CHANNEL_ESTIMATION)
+
+        @self.router.get("/results/{operationId}",
             summary="Download a ZIP archive of all ChannelEstimation capture files",
             responses={200: {"content": {"application/zip": {}},
                              "description": "ZIP archive of capture files"}})
-        def download_results_zip(operation_id: OperationId) -> StreamingResponse:
-
+        def download_results_zip(operationId: OperationId) -> StreamingResponse:
+            operation_id = operationId
             return self._build_results_zip_response(operation_id, "multiChannelEstimation")
 
 
-        @self.router.delete("/stop/{operation_id}",
+        @self.router.delete("/stop/{operationId}",
             response_model=MultiChanEstStatusResponse,
             summary="Stop a running multi-sample ChannelEstimation capture early")
-        def stop_capture(operation_id: OperationId) -> MultiChanEstStatusResponse:
+        def stop_capture(operationId: OperationId) -> MultiChanEstStatusResponse:
             """
 
 
             """
+            operation_id = operationId
             service: MultiChannelEstimationService = cast(
                 MultiChannelEstimationService,
                 self._get_service_or_404(operation_id),

src/pypnm/api/routes/advance/multi_ds_chan_est/service.py

@@ -5,6 +5,7 @@
 import logging
 
 from pypnm.api.routes.advance.common.capture_service import AbstractCaptureService
+from pypnm.api.routes.advance.common.operation_kind import MultiCaptureOperation
 from pypnm.api.routes.common.extended.common_measure_schema import (
     DownstreamOfdmParameters,
 )
@@ -35,6 +36,9 @@ class MultiChannelEstimationService(AbstractCaptureService):
       - duration: total measurement duration in seconds.
       - interval: interval between captures in seconds.
     """
+    OPERATION_NAME = MultiCaptureOperation.MULTI_DS_CHANNEL_ESTIMATION
+    MEASURE_MODE = "standard"
+
     def __init__(self, cm: CableModem,
                 tftp_servers: tuple[Inet, Inet] = PnmConfigManager.get_tftp_servers(),
                 tftp_path: str = PnmConfigManager.get_tftp_path(),