coco_pipe.descriptors.extractors.base ===================================== .. py:module:: coco_pipe.descriptors.extractors.base .. autoapi-nested-parse:: Base interfaces for descriptor extraction backends. This module defines the internal contracts shared by built-in descriptor extractors. The module exposes: - `BaseDescriptorExtractor` for families that consume validated raw signal batches - `BasePSDDescriptorExtractor` for families that consume shared PSD batches - `_DescriptorBlock` as the private family output payload - `make_failure_record` as the shared normalized failure-record helper The surrounding descriptors stack uses these interfaces to provide: - explicit runtime dispatch from `DescriptorPipeline` - deterministic sensor-level descriptor naming - family-wise metadata and failure collection - safe merging of family outputs into one stable result dictionary .. rubric:: Notes `BaseDescriptorExtractor` is an internal extension point for descriptor families. Unlike dim-reduction reducers, descriptor extractors are stateless at runtime and do not expose `fit`, persistence, or model objects. Author: Hamza Abdelhedi (hamza.abdelhedi@umontreal.ca) Classes ------- .. autoapisummary:: coco_pipe.descriptors.extractors.base.BaseDescriptorExtractor coco_pipe.descriptors.extractors.base.BasePSDDescriptorExtractor Functions --------- .. autoapisummary:: coco_pipe.descriptors.extractors.base.make_failure_record Module Contents --------------- .. py:function:: make_failure_record(family: str, obs_index: int, obs_id: Any = None, channel_index: int | None = None, channel_name: str | None = None, exception_type: str | None = None, message: str | None = None) -> dict[str, Any] Create one normalized extractor failure record. .. py:class:: BaseDescriptorExtractor(config: Any) Bases: :py:obj:`abc.ABC` Abstract base class for descriptor extraction families. Subclasses receive already validated NumPy inputs and must return one `_DescriptorBlock` aligned on the observation axis. The base class keeps the extractor API narrow and provides a shared helper for sensor-level finalization and deterministic descriptor naming. :param config: Typed family configuration parsed by `DescriptorConfig`. :type config: Any .. attribute:: config Stored family-specific configuration object. :type: Any .. attribute:: family_name Stable family identifier used in failure records and merged metadata. :type: str .. rubric:: Notes Extractors are stateless at runtime. They do not learn parameters across calls; all runtime state is provided explicitly through `extract()`. Concrete extractors are expected to: 1. compute family-specific values with shape ``(n_obs, n_channels)`` for each metric 2. pass those values through :meth:`_finalize_descriptor` 3. return one `_DescriptorBlock` with aligned names, metadata, and failures .. rubric:: Examples A minimal concrete extractor typically looks like: >>> class MeanOverTimeExtractor(BaseDescriptorExtractor): ... family_name = "toy" ... ... def extract( ... self, ... X, ... sfreq, ... channel_names, ... ids, ... runtime, ... ): ... values = X.mean(axis=-1) ... X_out, names = self._finalize_descriptor( ... values, ... family_prefix="toy", ... metric_name="mean", ... channel_names=channel_names, ... ) ... return _DescriptorBlock( ... family=self.family_name, ... X=X_out, ... descriptor_names=names, ... ) .. py:attribute:: family_name :value: 'base' .. py:attribute:: config .. py:property:: capabilities :type: dict[str, Any] Return static extractor capability metadata. :returns: Static metadata describing optional dependencies and general execution properties for the extractor. :rtype: dict[str, Any] .. rubric:: Notes The descriptors pipeline currently uses this mapping only as lightweight backend metadata. It is intentionally much smaller than the reducer capability surface in `dim_reduction`. .. py:method:: extract(X: numpy.ndarray, sfreq: float | None, channel_names: list[str] | None, ids: numpy.ndarray | None, runtime: coco_pipe.descriptors.configs.DescriptorRuntimeConfig, obs_offset: int = 0) -> _DescriptorBlock :abstractmethod: Extract descriptors from a validated input array. :param X: Input array with shape ``(n_obs, n_channels, n_times)``. :type X: np.ndarray :param sfreq: Sampling frequency in Hertz. :type sfreq: float, optional :param channel_names: Explicit channel labels aligned with axis 1 of ``X``. :type channel_names: list of str, optional :param ids: Observation identifiers aligned with axis 0 of ``X``. :type ids: np.ndarray, optional :param runtime: Runtime execution controls shared across extractors. :type runtime: DescriptorRuntimeConfig :param obs_offset: Global observation offset applied to any collected failure records. :type obs_offset: int, default=0 :returns: Family-specific descriptor matrix plus metadata and failures. :rtype: _DescriptorBlock :raises ImportError: If an optional backend required by the extractor is unavailable. :raises ValueError: If the extractor encounters an invalid runtime condition and the configured error policy requires raising. .. rubric:: Notes The recommended pattern is to keep family-specific computation local to the extractor and delegate sensor-level naming behavior to :meth:`_finalize_descriptor`. .. py:method:: _finalize_descriptor(values: numpy.ndarray, family_prefix: str, metric_name: str, channel_names: list[str] | None) -> tuple[numpy.ndarray, list[str]] Build deterministic sensor-level descriptor names. :param values: Family metric values with shape ``(n_obs, n_channels)`` or ``(n_obs,)``. :type values: np.ndarray :param family_prefix: Stable family prefix, for example ``"band"`` or ``"param"``. :type family_prefix: str :param metric_name: Family-local metric identifier used in the descriptor name. :type metric_name: str :param channel_names: Channel labels used when building channel-resolved descriptor names. :type channel_names: list of str, optional :returns: ``(X_metric, names)`` where ``X_metric`` is the finalized metric matrix and ``names`` is the aligned list of descriptor names. :rtype: tuple .. rubric:: Notes This helper assumes ``values`` already represents descriptor values, not raw signals. It therefore only handles the stable sensor-level naming convention used by the public extract result. .. rubric:: Examples Given ``channel_names=["Fz", "Cz", "Pz"]`` and ``metric_name="abs_alpha"``: - yields ``["band_abs_alpha_ch-Fz", "band_abs_alpha_ch-Cz", "band_abs_alpha_ch-Pz"]`` .. py:class:: BasePSDDescriptorExtractor(config: Any) Bases: :py:obj:`BaseDescriptorExtractor` Abstract base class for descriptor families that consume PSD batches. PSD-consuming families still participate in the shared descriptor contract, but they expose one additional explicit entry point: - `extract_psd(...)` consumes precomputed `psds, freqs` - `psd_request()` tells the planner which PSD range and method is needed This keeps the generic raw-signal interface narrow while still giving the planner one formal PSD-consumer contract shared by spectral and parametric families. .. rubric:: Notes PSD consumers may still expose `extract()` to satisfy the generic family interface, but the shared planner uses `psd_request()` and `extract_psd()` exclusively once PSD intermediates have been materialized. .. py:method:: psd_request() -> dict[str, Any] :abstractmethod: Describe the PSD requirements for the shared planner. :returns: Minimal request payload containing the PSD method and the required frequency range for this family. :rtype: dict[str, Any] .. py:method:: parametric_fit_requirements() -> dict[str, Any] Describe whether this PSD consumer needs a shared parametric fit. :returns: Shared-fit requirements with the keys: - `needed` - `metrics` - `periodic_psds` - `config` :rtype: dict[str, Any] .. py:method:: extract_psd(psds: numpy.ndarray, freqs: numpy.ndarray, channel_names: list[str] | None, ids: numpy.ndarray | None, runtime: coco_pipe.descriptors.configs.DescriptorRuntimeConfig, obs_offset: int = 0, fit_batch: Any | None = None) -> _DescriptorBlock :abstractmethod: Extract descriptors from explicit PSD intermediates. :param psds: PSD batch with shape ``(n_obs, n_channels, n_freqs)``. :type psds: np.ndarray :param freqs: Frequency grid aligned with the last axis of ``psds``. :type freqs: np.ndarray :param channel_names: Explicit channel labels aligned with the channel axis. :type channel_names: list of str, optional :param ids: Observation identifiers aligned with the observation axis. :type ids: np.ndarray, optional :param runtime: Runtime execution controls shared across extractors. :type runtime: DescriptorRuntimeConfig :param obs_offset: Global observation offset applied to collected failure records. :type obs_offset: int, default=0 :param fit_batch: Additional shared fit payload required by some PSD consumers. :type fit_batch: Any, optional :returns: Family-specific descriptor block aligned with the input PSD batch. :rtype: _DescriptorBlock