Signal Processing¶

@abstractmethod
def discover(
    self,
    path_configs: str | Path,
    turbines: Sequence[str] | None = None,
) -> dict[str, Path]:
    """Return a turbine-to-config-path mapping.

    Parameters
    ----------
    path_configs
        Filesystem path to a directory of configuration files or a single
        configuration file.
    turbines
        Optional subset of turbine identifiers to retain from the
        discovered files.

    Returns
    -------
    dict[str, Path]
        Mapping from turbine identifier to configuration file path.
    """

def discover(
    self,
    path_configs: str | Path,
    turbines: Sequence[str] | None = None,
) -> dict[str, Path]:
    """Return available JSON config files keyed by turbine name.

    Parameters
    ----------
    path_configs
        Directory containing configuration files or a single JSON config
        path.
    turbines
        Optional subset of turbine stems to retain from the discovered
        files.

    Returns
    -------
    dict[str, Path]
        Mapping from turbine stem to configuration path.

    Raises
    ------
    ValueError
        If the path does not resolve to usable JSON files or the requested
        turbine subset is empty.
    """
    root = Path(path_configs)
    if root.is_dir():
        available = {
            path.stem: path for path in sorted(root.iterdir()) if path.is_file() and path.suffix == self.suffix
        }
    elif root.is_file() and root.suffix == self.suffix:
        available = {root.stem: root}
    else:
        raise ValueError(f"Could not discover configuration files from {root}.")

    if turbines is None:
        return available

    selected = {turbine: available[turbine] for turbine in turbines if turbine in available}
    missing = [turbine for turbine in turbines if turbine not in available]
    if missing:
        warnings.warn(
            "Some turbines from the provided list are not found in the "
            f"configurations directory. Using available turbines: {list(selected)}",
            stacklevel=2,
        )
    if not selected:
        raise ValueError("No valid turbines found in the provided list.")
    return selected

def matches(self, raw_key: str) -> bool:
    """Return ``True`` when the raw key belongs to a direct signal.

    Parameters
    ----------
    raw_key
        Raw configuration key to test.

    Returns
    -------
    bool
        Whether the key starts with one of the configured signal prefixes.
    """
    return raw_key.startswith(self.signal_prefixes)

def parse(self, raw_key: str) -> SignalEventKey | None:
    """Parse a raw key into a signal/property pair.

    Parameters
    ----------
    raw_key
        Raw configuration key containing a signal name and property name
        separated by :attr:`separator`.

    Returns
    -------
    SignalEventKey or None
        Parsed key, or *None* when the key does not match or lacks a
        separator.
    """
    if not self.matches(raw_key) or self.separator not in raw_key:
        return None

    signal_name, property_name = raw_key.split(self.separator, maxsplit=1)
    if not signal_name or not property_name:
        return None
    return SignalEventKey(signal_name=signal_name, property_name=property_name)

def process_events(self, events: Sequence[Mapping[str, Any]]) -> SignalProcessingResult:
    """Transform raw configuration events into typed signal records.

    Parameters
    ----------
    events
        Ordered raw configuration events loaded from one farm config.

    Returns
    -------
    SignalProcessingResult
        Typed signal and derived-signal records that can be converted to
        the archive-compatible uploader payload shape.

    Examples
    --------
    >>> from owi.metadatabase.shm.processing import (
    ...     ConfiguredSignalConfigProcessor,
    ...     DelimitedSignalKeyParser,
    ...     SignalProcessorSpec,
    ... )
    >>> spec = SignalProcessorSpec(
    ...     farm_name="Demo",
    ...     signal_key_parser=DelimitedSignalKeyParser(signal_prefixes=("WF_",)),
    ...     derived_signal_strategies={},
    ... )
    >>> processor = ConfiguredSignalConfigProcessor(path_configs='.', processor_spec=spec)
    >>> result = processor.process_events([{"WF_SIG/status": "ok"}])
    >>> result.to_legacy_data()[0]["WF_SIG"]["status"][0]["status"]
    'ok'
    """
    signals: dict[str, ProcessedSignalRecord] = {}
    derived_signals: dict[str, ProcessedDerivedSignalRecord] = {}
    current_time = self.processor_spec.default_initial_time

    for index, event in enumerate(events):
        current_time = self._resolve_event_time(
            event,
            index=index,
            current_time=current_time,
        )
        for raw_key, value in event.items():
            signal_key = self.processor_spec.signal_key_parser.parse(raw_key)
            if signal_key is not None:
                self._apply_signal_property(
                    signals=signals,
                    signal_key=signal_key,
                    value=value,
                    timestamp=current_time,
                )
                continue

            strategy = self.processor_spec.derived_signal_strategies.get(raw_key)
            if strategy is None:
                continue
            payload = _coerce_mapping(value, context=raw_key)
            self._apply_derived_updates(
                derived_signals=derived_signals,
                event_key=raw_key,
                updates=strategy.emit_updates(raw_key, payload),
                timestamp=current_time,
            )

    self._postprocess_signals(signals)
    return SignalProcessingResult(signals=signals, derived_signals=derived_signals)

def signal_preprocess_data(
    self,
    path_config: str | Path,
) -> tuple[LegacySignalMap, LegacySignalMap]:
    """Process one configuration file into archive-compatible mappings.

    Parameters
    ----------
    path_config
        JSON configuration file to load and process.

    Returns
    -------
    tuple[LegacySignalMap, LegacySignalMap]
        Main-signal and derived-signal mappings ready for uploader seams.
    """
    events = self._load_events(path_config)
    return self.process_events(events).to_legacy_data()

def signals_process_data(self) -> None:
    """Process all discovered configuration files under ``path_configs``.

    The processed results are stored on :attr:`signals_data` and
    :attr:`signals_derived_data`, keyed by turbine stem.
    """
    config_paths = self.processor_spec.config_discovery.discover(
        self.path_configs,
        turbines=self.turbines,
    )
    self.turbines = list(config_paths)
    for turbine, config_path in config_paths.items():
        signals_data, derived_data = self.signal_preprocess_data(config_path)
        self.signals_data[turbine] = signals_data
        self.signals_derived_data[turbine] = derived_data

def build_processor_spec(self) -> SignalProcessorSpec:
    """Return the explicit processor spec passed to the constructor.

    Returns
    -------
    SignalProcessorSpec
        The specification supplied at construction time.
    """
    return self._processor_spec

@classmethod
def from_yaml_spec(
    cls,
    *,
    path_configs: str | Path,
    processor_spec_path: str | Path,
    turbines: Sequence[str] | None = None,
) -> ConfiguredSignalConfigProcessor:
    """Construct a configured processor from a YAML-backed processor spec.

    Parameters
    ----------
    path_configs
        Directory or JSON file containing farm configuration events.
    processor_spec_path
        Path to a YAML processor specification file.
    turbines
        Optional subset of turbine stems to process during discovery.

    Returns
    -------
    ConfiguredSignalConfigProcessor
        Processor loaded with the given YAML spec.
    """
    return cls(
        path_configs=path_configs,
        processor_spec=load_signal_processor_spec(processor_spec_path),
        turbines=turbines,
    )

def build_processor_spec(self) -> SignalProcessorSpec:
    """Return the explicit processor spec passed to the constructor.

    Returns
    -------
    SignalProcessorSpec
        The specification supplied at construction time.
    """
    return self._processor_spec

def process_events(self, events: Sequence[Mapping[str, Any]]) -> SignalProcessingResult:
    """Transform raw configuration events into typed signal records.

    Parameters
    ----------
    events
        Ordered raw configuration events loaded from one farm config.

    Returns
    -------
    SignalProcessingResult
        Typed signal and derived-signal records that can be converted to
        the archive-compatible uploader payload shape.

    Examples
    --------
    >>> from owi.metadatabase.shm.processing import (
    ...     ConfiguredSignalConfigProcessor,
    ...     DelimitedSignalKeyParser,
    ...     SignalProcessorSpec,
    ... )
    >>> spec = SignalProcessorSpec(
    ...     farm_name="Demo",
    ...     signal_key_parser=DelimitedSignalKeyParser(signal_prefixes=("WF_",)),
    ...     derived_signal_strategies={},
    ... )
    >>> processor = ConfiguredSignalConfigProcessor(path_configs='.', processor_spec=spec)
    >>> result = processor.process_events([{"WF_SIG/status": "ok"}])
    >>> result.to_legacy_data()[0]["WF_SIG"]["status"][0]["status"]
    'ok'
    """
    signals: dict[str, ProcessedSignalRecord] = {}
    derived_signals: dict[str, ProcessedDerivedSignalRecord] = {}
    current_time = self.processor_spec.default_initial_time

    for index, event in enumerate(events):
        current_time = self._resolve_event_time(
            event,
            index=index,
            current_time=current_time,
        )
        for raw_key, value in event.items():
            signal_key = self.processor_spec.signal_key_parser.parse(raw_key)
            if signal_key is not None:
                self._apply_signal_property(
                    signals=signals,
                    signal_key=signal_key,
                    value=value,
                    timestamp=current_time,
                )
                continue

            strategy = self.processor_spec.derived_signal_strategies.get(raw_key)
            if strategy is None:
                continue
            payload = _coerce_mapping(value, context=raw_key)
            self._apply_derived_updates(
                derived_signals=derived_signals,
                event_key=raw_key,
                updates=strategy.emit_updates(raw_key, payload),
                timestamp=current_time,
            )

    self._postprocess_signals(signals)
    return SignalProcessingResult(signals=signals, derived_signals=derived_signals)

def signal_preprocess_data(
    self,
    path_config: str | Path,
) -> tuple[LegacySignalMap, LegacySignalMap]:
    """Process one configuration file into archive-compatible mappings.

    Parameters
    ----------
    path_config
        JSON configuration file to load and process.

    Returns
    -------
    tuple[LegacySignalMap, LegacySignalMap]
        Main-signal and derived-signal mappings ready for uploader seams.
    """
    events = self._load_events(path_config)
    return self.process_events(events).to_legacy_data()

def signals_process_data(self) -> None:
    """Process all discovered configuration files under ``path_configs``.

    The processed results are stored on :attr:`signals_data` and
    :attr:`signals_derived_data`, keyed by turbine stem.
    """
    config_paths = self.processor_spec.config_discovery.discover(
        self.path_configs,
        turbines=self.turbines,
    )
    self.turbines = list(config_paths)
    for turbine, config_path in config_paths.items():
        signals_data, derived_data = self.signal_preprocess_data(config_path)
        self.signals_data[turbine] = signals_data
        self.signals_derived_data[turbine] = derived_data

@classmethod
def from_yaml_spec(
    cls,
    *,
    path_configs: str | Path,
    processor_spec_path: str | Path,
    turbines: Sequence[str] | None = None,
) -> ConfiguredSignalConfigProcessor:
    """Construct a configured processor from a YAML-backed processor spec.

    Parameters
    ----------
    path_configs
        Directory or JSON file containing farm configuration events.
    processor_spec_path
        Path to a YAML processor specification file.
    turbines
        Optional subset of turbine stems to process during discovery.

    Returns
    -------
    ConfiguredSignalConfigProcessor
        Processor loaded with the given YAML spec.
    """
    return cls(
        path_configs=path_configs,
        processor_spec=load_signal_processor_spec(processor_spec_path),
        turbines=turbines,
    )

@abstractmethod
def build_processor_spec(self) -> SignalProcessorSpec:
    """Return the farm-specific processor specification.

    Returns
    -------
    SignalProcessorSpec
        Specification controlling signal key parsing, derived-signal
        strategies, and postprocessors.
    """

def process_events(self, events: Sequence[Mapping[str, Any]]) -> SignalProcessingResult:
    """Transform raw configuration events into typed signal records.

    Parameters
    ----------
    events
        Ordered raw configuration events loaded from one farm config.

    Returns
    -------
    SignalProcessingResult
        Typed signal and derived-signal records that can be converted to
        the archive-compatible uploader payload shape.

    Examples
    --------
    >>> from owi.metadatabase.shm.processing import (
    ...     ConfiguredSignalConfigProcessor,
    ...     DelimitedSignalKeyParser,
    ...     SignalProcessorSpec,
    ... )
    >>> spec = SignalProcessorSpec(
    ...     farm_name="Demo",
    ...     signal_key_parser=DelimitedSignalKeyParser(signal_prefixes=("WF_",)),
    ...     derived_signal_strategies={},
    ... )
    >>> processor = ConfiguredSignalConfigProcessor(path_configs='.', processor_spec=spec)
    >>> result = processor.process_events([{"WF_SIG/status": "ok"}])
    >>> result.to_legacy_data()[0]["WF_SIG"]["status"][0]["status"]
    'ok'
    """
    signals: dict[str, ProcessedSignalRecord] = {}
    derived_signals: dict[str, ProcessedDerivedSignalRecord] = {}
    current_time = self.processor_spec.default_initial_time

    for index, event in enumerate(events):
        current_time = self._resolve_event_time(
            event,
            index=index,
            current_time=current_time,
        )
        for raw_key, value in event.items():
            signal_key = self.processor_spec.signal_key_parser.parse(raw_key)
            if signal_key is not None:
                self._apply_signal_property(
                    signals=signals,
                    signal_key=signal_key,
                    value=value,
                    timestamp=current_time,
                )
                continue

            strategy = self.processor_spec.derived_signal_strategies.get(raw_key)
            if strategy is None:
                continue
            payload = _coerce_mapping(value, context=raw_key)
            self._apply_derived_updates(
                derived_signals=derived_signals,
                event_key=raw_key,
                updates=strategy.emit_updates(raw_key, payload),
                timestamp=current_time,
            )

    self._postprocess_signals(signals)
    return SignalProcessingResult(signals=signals, derived_signals=derived_signals)

def signal_preprocess_data(
    self,
    path_config: str | Path,
) -> tuple[LegacySignalMap, LegacySignalMap]:
    """Process one configuration file into archive-compatible mappings.

    Parameters
    ----------
    path_config
        JSON configuration file to load and process.

    Returns
    -------
    tuple[LegacySignalMap, LegacySignalMap]
        Main-signal and derived-signal mappings ready for uploader seams.
    """
    events = self._load_events(path_config)
    return self.process_events(events).to_legacy_data()

def signals_process_data(self) -> None:
    """Process all discovered configuration files under ``path_configs``.

    The processed results are stored on :attr:`signals_data` and
    :attr:`signals_derived_data`, keyed by turbine stem.
    """
    config_paths = self.processor_spec.config_discovery.discover(
        self.path_configs,
        turbines=self.turbines,
    )
    self.turbines = list(config_paths)
    for turbine, config_path in config_paths.items():
        signals_data, derived_data = self.signal_preprocess_data(config_path)
        self.signals_data[turbine] = signals_data
        self.signals_derived_data[turbine] = derived_data

def ensure_source_name(
    self,
    source_name: str,
    extra_fields: Mapping[str, Any] | None = None,
) -> None:
    """Initialize immutable source metadata for the derived signal.

    Parameters
    ----------
    source_name
        Event key that produced the derived signal.
    extra_fields
        Optional metadata merged into the legacy ``data`` mapping the first
        time the source name is set.
    """
    if not self.data_fields:
        self.data_fields = {"name": source_name}
        if extra_fields:
            self.data_fields.update(extra_fields)

def set_parent_signals(self, parent_signals: Sequence[str]) -> None:
    """Set parent signals when they are first known.

    Parameters
    ----------
    parent_signals
        Ordered parent signal identifiers for the derived signal.
    """
    if not self.parent_signals:
        self.parent_signals = tuple(parent_signals)

def add_calibration(self, timestamp: str, calibration_fields: Mapping[str, Any]) -> None:
    """Append a derived-signal calibration row.

    Parameters
    ----------
    timestamp
        Event timestamp associated with the calibration.
    calibration_fields
        Calibration fields emitted by the derived-signal strategy.
    """
    row = {"time": timestamp}
    row.update(calibration_fields)
    self.calibration_rows.append(row)

def to_legacy_dict(self) -> LegacyRecord:
    """Return the uploader-facing legacy mapping.

    Returns
    -------
    LegacyRecord
        Archive-compatible mapping consumed by uploader payload builders.
    """
    data: LegacyRecord = {}
    if self.data_fields:
        data["data"] = dict(self.data_fields)
    if self.calibration_rows:
        data["calibration"] = [dict(row) for row in self.calibration_rows]
    if self.parent_signals:
        data["parent_signals"] = list(self.parent_signals)
    return data

def set_scalar(self, property_name: str, value: Any) -> None:
    """Store a scalar property on the signal.

    Parameters
    ----------
    property_name
        Scalar field name from the raw configuration event.
    value
        Value to persist on the signal record.
    """
    self.scalar_fields[property_name] = value

def add_status(self, timestamp: str, status: Any) -> None:
    """Append a status row.

    Parameters
    ----------
    timestamp
        Event timestamp associated with the status.
    status
        Status value to store.
    """
    self.status_rows.append({"time": timestamp, "status": status})

def add_status_alias(self, timestamp: str, alias_name: str) -> None:
    """Append a status row that carries a legacy alias name.

    Parameters
    ----------
    timestamp
        Event timestamp associated with the alias.
    alias_name
        Legacy signal name that points at this record.
    """
    self.status_rows.append({"time": timestamp, "name": alias_name})

def add_offset(self, timestamp: str, offset: Any) -> None:
    """Append an offset row.

    Parameters
    ----------
    timestamp
        Event timestamp associated with the offset.
    offset
        Offset value to store.
    """
    self.offset_rows.append({"time": timestamp, "offset": offset})

def add_cwl(self, timestamp: str, cwl: Any) -> None:
    """Append a CWL row.

    Parameters
    ----------
    timestamp
        Event timestamp associated with the CWL value.
    cwl
        CWL value to store.
    """
    self.cwl_rows.append({"time": timestamp, "cwl": cwl})

def to_legacy_dict(self) -> LegacyRecord:
    """Return the uploader-facing legacy mapping.

    Returns
    -------
    LegacyRecord
        Archive-compatible mapping consumed by uploader payload builders.
    """
    data = dict(self.scalar_fields)
    if self.status_rows:
        data["status"] = [dict(row) for row in self.status_rows]
    if self.offset_rows:
        data["offset"] = [dict(row) for row in self.offset_rows]
    if self.cwl_rows:
        data["cwl"] = [dict(row) for row in self.cwl_rows]
    return data