learner

SelectorTrainingIncompleteError

Bases: ValueError

Raised when selector learning sees partially labeled traces.

HyDRALearner(regressor_factory, threshold, start_width=10, step_width=5, selector_learner=None, callback=None, trace_schema=None)

Bases: SupervisedLearner

Identify hybrid-system modes from trace inputs and derivatives.

regressor_factory must create fresh incremental supervised learners. The current learner supports single-output derivative training. When a selector learner is provided, HyDRA labels accurate trace segments and trains a selector to route future rows to learned modes.

Source code in src/flowcean/hydra/learner.py

def __init__(
    self,
    regressor_factory: Callable[[], SupervisedIncrementalLearner],
    threshold: float,
    start_width: int = 10,
    step_width: int = 5,
    selector_learner: HybridDecisionTreeLearner | None = None,
    callback: HyDRACallback | None = None,
    trace_schema: HyDRATraceSchema | None = None,
) -> None:
    super().__init__()
    if threshold < 0:
        message = "threshold must be non-negative."
        raise ValueError(message)
    if start_width <= 0:
        message = "start_width must be positive."
        raise ValueError(message)
    if step_width <= 0:
        message = "step_width must be positive."
        raise ValueError(message)
    self.regressor_factory = regressor_factory
    self.threshold = threshold
    self.start_width = start_width
    self.step_width = step_width
    self.selector_learner = selector_learner
    self.callback = callback or NoOpCallback()
    self.trace_schema = trace_schema

find_next_pending_segment(traces)

Find the first contiguous unlabeled segment across traces.

Source code in src/flowcean/hydra/learner.py

def find_next_pending_segment(
    traces: Sequence[HyDRATrace],
) -> TraceSegment | None:
    """Find the first contiguous unlabeled segment across traces."""
    for trace_index, trace in enumerate(traces):
        start_index: int | None = None
        for row_index, is_unlabeled in enumerate(trace.unlabeled_mask):
            if is_unlabeled and start_index is None:
                start_index = row_index
            elif not is_unlabeled and start_index is not None:
                return TraceSegment(
                    trace_index=trace_index,
                    start_index=start_index,
                    end_index=row_index - 1,
                )
        if start_index is not None:
            return TraceSegment(
                trace_index=trace_index,
                start_index=start_index,
                end_index=trace.height - 1,
            )
    return None