diff --git a/AGENTS.md b/AGENTS.md index dd994543..68ee8c0d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,10 +14,7 @@ uv sync --dev uv run prek install # Run tests -uv run pytest -q -uv run pytest -s # verbose with stdout -uv run pytest --cov=. --cov-report=term-missing # with coverage -uv run pytest tests/test_foo.py # single test file +uv run pytest # Run linting/formatting (all pre-commit hooks) uv run prek run -a @@ -182,17 +179,6 @@ def set_configuration(self) -> None: Omitting either step means a `persist:` block in the YAML is silently ignored with no error. -## Code Quality - -Pre-commit hooks enforce: -- **mypy** strict mode -- **flake8** linting, **autopep8** formatting (max line 110) -- **bandit** security checks, **vulture** dead-code detection (70% threshold) -- **docformatter** docstring style - -Python 3.12 is required (see `.python-version`). - - # Git NEVER include "Co-Authored-By ..." in your commit or PR messages. diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/base.py b/src/detectmatelibrary/utils/persistency/event_data_structures/base.py index 08e4909b..31eb768a 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/base.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/base.py @@ -11,7 +11,7 @@ class EventDataStructure(ABC): template: str = "" @abstractmethod - def add_data(self, data_object: Any) -> None: ... + def add_data(self, data_object: Any, timestamp: float | None = None) -> None: ... @abstractmethod def get_data(self) -> Any: ... diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py index 53c32306..fac9c73c 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py @@ -24,7 +24,7 @@ class ChunkedEventDataFrame(EventDataStructure): chunks: list[pl.DataFrame] = field(default_factory=list) _rows: int = 0 - def add_data(self, data: pl.DataFrame) -> None: + def add_data(self, data: pl.DataFrame, timestamp: float | None = None) -> None: if data.height == 0: return self.chunks.append(data) diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py index 2a024883..5519393e 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py @@ -17,7 +17,7 @@ class EventDataFrame(EventDataStructure): """ data: pd.DataFrame = field(default_factory=pd.DataFrame) - def add_data(self, data: pd.DataFrame) -> None: + def add_data(self, data: pd.DataFrame, timestamp: float | None = None) -> None: if len(self.data) > 0: self.data = pd.concat([self.data, data], ignore_index=True) else: diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py index f43a22a7..05b1c6d7 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py @@ -27,9 +27,9 @@ def __init__( self.converter_function = converter_function self.multi_tracker = self.multi_tracker_type(single_tracker_type=self.single_tracker_type) - def add_data(self, data_object: Any) -> None: + def add_data(self, data_object: Any, timestamp: float | None = None) -> None: """Add data to the variable trackers.""" - self.multi_tracker.add_data(data_object) + self.multi_tracker.add_data(data_object, timestamp=timestamp) def get_data(self) -> Dict[str, SingleTracker]: """Retrieve the tracker's stored data.""" diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py index 08df8218..19aa2d28 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py @@ -13,12 +13,12 @@ def __init__(self, single_tracker_type: Callable[[], SingleTracker] = SingleTrac self.single_trackers: Dict[str, SingleTracker] = {} self.single_tracker_type: Callable[[], SingleTracker] = single_tracker_type - def add_data(self, data_object: Dict[str, Any]) -> None: + def add_data(self, data_object: Dict[str, Any], timestamp: float | None = None) -> None: """Add data to the appropriate feature trackers.""" for name, value in data_object.items(): if name not in self.single_trackers: self.single_trackers[name] = self.single_tracker_type() - self.single_trackers[name].add_value(value) + self.single_trackers[name].add_value(value, timestamp=timestamp) def get_trackers(self) -> Dict[str, SingleTracker]: """Get the current feature trackers.""" diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py index 01808d89..c5c12a43 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py @@ -15,7 +15,7 @@ class SingleTracker(ABC): """Tracks whether a single variable is converging to a constant value.""" @abstractmethod - def add_value(self, value: Any) -> None: + def add_value(self, value: Any, timestamp: float | None = None) -> None: """Add a new value to the tracker.""" @abstractmethod diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py index 25db8bc2..bf907758 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py @@ -18,10 +18,52 @@ def __init__(self, segment_thresholds: List[float], min_samples: int = 10): # for lists self.segment_means: List[float] = [] - def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: + def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = None) -> List[int]: + """Index boundaries of n_segments segments over total_len items. + + Equal-count by default. When timestamps are given (one per item, + chronological, non-zero span), boundaries are equal-DURATION + cuts of the observed time span, mapped back to indices. Falls + back to equal-count on missing/mismatched/non-finite timestamps + or zero span. + """ + try: + use_time = ( + timestamps is not None + and len(timestamps) == total_len + and total_len > 0 + and bool(np.all(np.isfinite(timestamps))) + and timestamps[-1] > timestamps[0] + ) + except TypeError: + # e.g. a None entry: not comparable/convertible -> equal-count + use_time = False + if use_time and timestamps is not None: # 2nd clause narrows for mypy + t_first, t_last = timestamps[0], timestamps[-1] + cuts = [ + t_first + k * (t_last - t_first) / self.n_segments + for k in range(self.n_segments + 1) + ] + boundaries = [int(np.searchsorted(timestamps, t, side="left")) for t in cuts] + boundaries[0] = 0 + boundaries[-1] = total_len + return boundaries + segment_size = total_len / self.n_segments + boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] + boundaries[-1] = total_len + return boundaries + + def is_stable( + self, + change_series: RLEList[bool] | List[bool], + timestamps: List[float] | None = None, + ) -> bool: """Determine if a list of segment means is stable. Works efficiently with RLEList without expanding to a full list. + When timestamps are given (one per occurrence, chronological), + segments are equal-duration cuts of the time span instead of + equal-count cuts of the series. """ # Handle both RLEList and regular list if isinstance(change_series, RLEList): @@ -29,10 +71,7 @@ def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: if total_len == 0: return True - # Calculate segment boundaries - segment_size = total_len / self.n_segments - segment_boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] - segment_boundaries[-1] = total_len + segment_boundaries = self._segment_boundaries(total_len, timestamps) # Compute segment means directly from RLE runs segment_sums = [0.0] * self.n_segments @@ -65,8 +104,15 @@ def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: for i in range(self.n_segments) ] else: - # Original implementation for regular lists - self.segment_means = self._compute_segment_means(change_series) + if timestamps is not None and len(timestamps) == len(change_series): + b = self._segment_boundaries(len(change_series), timestamps) + self.segment_means = [ + float(np.mean(change_series[b[i]:b[i + 1]])) if b[i + 1] > b[i] else np.nan + for i in range(self.n_segments) + ] + else: + # Original implementation for regular lists + self.segment_means = self._compute_segment_means(change_series) return all([not q >= thresh for q, thresh in zip(self.segment_means, self.segment_threshs)]) def _compute_segment_means(self, change_series: List[bool]) -> List[float]: @@ -80,8 +126,10 @@ def get_last_segment_means(self) -> List[float]: def get_segment_thresholds(self) -> List[float]: return self.segment_threshs - def __call__(self, change_series: RLEList[bool] | List[bool]) -> bool: - return self.is_stable(change_series) + def __call__( + self, change_series: RLEList[bool] | List[bool], timestamps: List[float] | None = None + ) -> bool: + return self.is_stable(change_series, timestamps=timestamps) def __repr__(self) -> str: return ( diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py index c32ff8ac..9ad5dbd5 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py @@ -14,14 +14,26 @@ class SingleStabilityTracker(SingleTracker): """Tracks stability of a single feature.""" - def __init__(self, min_samples: int = 3, add_value_fn: str = "default", - detector_config: "CoreDetectorConfig | None" = None) -> None: + def __init__( + self, + min_samples: int = 3, + expand_value: bool = False, + time_dependent: bool = False, + add_value_fn: str = "default", + detector_config: "CoreDetectorConfig | None" = None, + ) -> None: self.min_samples = min_samples + self.expand_value = expand_value + self.time_dependent = time_dependent self.change_series: RLEList[bool] = RLEList() self.unique_set: Set[Any] = set() self.stability_classifier: StabilityClassifier = StabilityClassifier( segment_thresholds=[1.1, 0.3, 0.1, 0.01], ) + self._accum = set.update if expand_value else set.add + # ponytail: O(N) timestamps; switch to fixed-width time buckets if + # this ever runs unbounded/streaming. + self.timestamps: List[float] = [] # Opaque slot for detectors to stash per-variable model state that # must survive save/load. Schema-free; the tracker does not interpret it. self.extra_state: Dict[str, Any] = {} @@ -35,14 +47,21 @@ def __init__(self, min_samples: int = 3, add_value_fn: str = "default", detector = detector() self.add_value = detector.add_value_fn.__get__(self, type(self)) # type: ignore[method-assign] - def add_value(self, value: Any) -> None: + def add_value(self, value: Any, timestamp: float | None = None) -> None: """Add a new value to the tracker.""" before = len(self.unique_set) - self.unique_set.add(value) + self._accum(self.unique_set, value) self.change_series.append(len(self.unique_set) > before) + if self.time_dependent and timestamp is not None: + self.timestamps.append(float(timestamp)) def classify(self) -> Classification: """Classify the variable.""" + timestamps = ( + self.timestamps + if self.time_dependent and len(self.timestamps) == len(self.change_series) + else None + ) if len(self.change_series) < self.min_samples: return Classification( type="INSUFFICIENT_DATA", @@ -58,7 +77,7 @@ def classify(self) -> Classification: type="RANDOM", reason=f"Unique set size equals number of samples ({len(self.change_series)})" ) - elif self.stability_classifier.is_stable(self.change_series): + elif self.stability_classifier.is_stable(self.change_series, timestamps=timestamps): return Classification( type="STABLE", reason=( @@ -79,6 +98,9 @@ def to_state(self) -> Dict[str, Any]: "type": self.__class__.__name__, "module": self.__class__.__module__, "min_samples": self.min_samples, + "expand_value": self.expand_value, + "time_dependent": self.time_dependent, + "timestamps": self.timestamps, "add_value_fn": self.add_value_fn, "detector_config": self.detector_config, "runs": self.change_series.runs(), @@ -92,6 +114,8 @@ def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": """Restore tracker from a state dict produced by to_state().""" tracker = cls( min_samples=state["min_samples"], + expand_value=state.get("expand_value", False), + time_dependent=state.get("time_dependent", False), add_value_fn=state["add_value_fn"], detector_config=state["detector_config"] ) @@ -104,6 +128,7 @@ def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": tracker.stability_classifier = StabilityClassifier( segment_thresholds=state["segment_thresholds"] ) + tracker.timestamps = [float(t) for t in state.get("timestamps", [])] tracker.extra_state = state.get("extra_state", {}) return tracker @@ -143,6 +168,8 @@ class EventStabilityTracker(EventTracker): def __init__( self, converter_function: Callable[[Any], Any] = lambda x: x, + expand_value: bool = False, + time_dependent: bool = False, add_value_fn: str = "default", detector_config: "CoreDetectorConfig | None" = None @@ -150,7 +177,12 @@ def __init__( self.multi_tracker: MultiStabilityTracker # for type hinting def make_tracker() -> SingleStabilityTracker: - return SingleStabilityTracker(add_value_fn=add_value_fn, detector_config=detector_config) + return SingleStabilityTracker( + expand_value=expand_value, + time_dependent=time_dependent, + add_value_fn=add_value_fn, + detector_config=detector_config, + ) # Mirror class identity onto the closure so dump()/load() can resolve # the underlying SingleStabilityTracker via its module + qualname. diff --git a/src/detectmatelibrary/utils/persistency/event_persistency.py b/src/detectmatelibrary/utils/persistency/event_persistency.py index ba9ae18c..8dd22d39 100644 --- a/src/detectmatelibrary/utils/persistency/event_persistency.py +++ b/src/detectmatelibrary/utils/persistency/event_persistency.py @@ -44,7 +44,8 @@ def ingest_event( event_id: int | str, event_template: str, variables: list[Any] = [], - named_variables: Dict[str, Any] = {} + named_variables: Dict[str, Any] = {}, + timestamp: float | None = None, ) -> None: """Ingest event data into the appropriate EventData store.""" with self._lock: @@ -60,7 +61,7 @@ def ingest_event( self.events_data[event_id] = data_structure data = data_structure.to_data(all_variables) - data_structure.add_data(data) + data_structure.add_data(data, timestamp=timestamp) # ponytail: fire callbacks outside the lock so a count-triggered save # doesn't hold the ingest lock across serialize + file I/O. for _cb in self._on_ingest_callbacks: diff --git a/tests/test_persistency/test_time_dependent_stability.py b/tests/test_persistency/test_time_dependent_stability.py new file mode 100644 index 00000000..c497fa80 --- /dev/null +++ b/tests/test_persistency/test_time_dependent_stability.py @@ -0,0 +1,175 @@ +"""Tests for the time_dependent option of the stability trackers.""" + +from detectmatelibrary.utils.persistency.rle_list import RLEList +from detectmatelibrary.utils.persistency import EventPersistency +from detectmatelibrary.utils.persistency.event_data_structures.trackers import ( + StabilityClassifier, + SingleStabilityTracker, + EventStabilityTracker, +) + +THRESHOLDS = [1.1, 0.3, 0.1, 0.01] # same defaults SingleStabilityTracker uses + + +def make_classifier() -> StabilityClassifier: + return StabilityClassifier(segment_thresholds=THRESHOLDS) + + +# Divergence fixture: 3 changes up front, then a quiet tail of 37. +# In *time*, the changes span most of the observed window and the quiet +# tail is a burst compressed into ~0.04s at the end. +DIVERGENT_SERIES = [True, True, True] + [False] * 37 +DIVERGENT_TIMES = [0.0, 40.0, 80.0] + [100.0 + 0.001 * i for i in range(37)] + + +class TestClassifierTimeBoundaries: + def test_count_mode_is_stable_on_divergent_fixture(self): + clf = make_classifier() + # count segments of 10: means [0.3, 0, 0, 0] -> all below thresholds + assert clf.is_stable(RLEList(DIVERGENT_SERIES)) is True + + def test_time_mode_is_unstable_on_divergent_fixture(self): + clf = make_classifier() + # time quarters put a lone change (mean 1.0) into segment 2 (thresh 0.3) + assert clf.is_stable(RLEList(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + + def test_uniform_timestamps_match_count_mode(self): + # N divisible by n_segments -> boundaries coincide exactly + series = [True, False, False, True, False, False, False, False] + ts = [float(i) for i in range(8)] + clf_count = make_classifier() + clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + clf_time.is_stable(RLEList(series), timestamps=ts) + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_plain_list_path_supports_timestamps(self): + clf = make_classifier() + assert clf.is_stable(list(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + + def test_zero_span_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + result = clf_time.is_stable(RLEList(series), timestamps=[5.0] * 8) + assert result == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_length_mismatch_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + assert clf_time.is_stable(RLEList(series), timestamps=[1.0, 2.0]) == expected + + def test_none_timestamp_entry_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + ts = [0.0, 1.0, None, 3.0, 4.0, 5.0, 6.0, 7.0] + assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_nan_timestamp_entry_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + ts = [0.0, 1.0, float("nan"), 3.0, 4.0, 5.0, 6.0, 7.0] + assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + +def feed_divergent(tracker: SingleStabilityTracker) -> None: + """3 new values spread over ~80s, then a repeated value bursting at + t~100.""" + values = ["a", "b", "c"] + ["c"] * 37 + for value, ts in zip(values, DIVERGENT_TIMES): + tracker.add_value(value, timestamp=ts) + + +class TestSingleStabilityTrackerTimeDependent: + def test_timestamps_stored_only_when_enabled(self): + on = SingleStabilityTracker(time_dependent=True) + on.add_value("a", timestamp=1.0) + on.add_value("b", timestamp=2.0) + assert on.timestamps == [1.0, 2.0] + + off = SingleStabilityTracker() # default time_dependent=False + off.add_value("a", timestamp=1.0) + assert off.timestamps == [] + + def test_classification_diverges_between_modes(self): + count_mode = SingleStabilityTracker() + feed_divergent(count_mode) + assert count_mode.classify().type == "STABLE" + + time_mode = SingleStabilityTracker(time_dependent=True) + feed_divergent(time_mode) + assert time_mode.classify().type == "UNSTABLE" + + def test_missing_timestamps_fall_back_to_count_mode(self): + # time_dependent on, but values arrive without timestamps + tracker = SingleStabilityTracker(time_dependent=True) + for value in ["a", "b", "c"] + ["c"] * 37: + tracker.add_value(value) + reference = SingleStabilityTracker() + for value in ["a", "b", "c"] + ["c"] * 37: + reference.add_value(value) + assert tracker.classify().type == reference.classify().type + + def test_round_trip_preserves_time_state(self): + tracker = SingleStabilityTracker(time_dependent=True) + feed_divergent(tracker) + restored = SingleStabilityTracker.from_state(tracker.to_state()) + assert restored.time_dependent is True + assert restored.timestamps == tracker.timestamps + assert restored.classify().type == "UNSTABLE" + + def test_legacy_state_without_time_keys_defaults_off(self): + tracker = SingleStabilityTracker() + tracker.add_value("hello") + state = tracker.to_state() + state.pop("time_dependent", None) # simulate pre-flag snapshot + state.pop("timestamps", None) + restored = SingleStabilityTracker.from_state(state) + assert restored.time_dependent is False + assert restored.timestamps == [] + + +class TestTimeDependentPlumbing: + def test_event_tracker_propagates_flag_and_timestamp(self): + event_tracker = EventStabilityTracker(time_dependent=True) + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + single = event_tracker.get_data()["var1"] + assert single.time_dependent is True + assert single.timestamps == [1.0, 2.0] + + def test_ingest_event_forwards_timestamp(self): + storage = EventPersistency( + EventStabilityTracker, + event_data_kwargs={"time_dependent": True}, + ) + storage.ingest_event(1, "tpl <*>", variables=["a"], timestamp=10.0) + storage.ingest_event(1, "tpl <*>", variables=["b"], timestamp=20.0) + single = storage.get_events_data()[1].get_data()["var_0"] + assert single.timestamps == [10.0, 20.0] + + def test_ingest_event_without_timestamp_still_works(self): + storage = EventPersistency(EventStabilityTracker) + storage.ingest_event(1, "tpl <*>", variables=["a"]) + single = storage.get_events_data()[1].get_data()["var_0"] + assert list(single.change_series) == [True] + assert single.timestamps == [] + + def test_event_tracker_dump_load_preserves_timestamps(self): + event_tracker = EventStabilityTracker(time_dependent=True) + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + restored = EventStabilityTracker.load(event_tracker.dump(), time_dependent=True) + single = restored.get_data()["var1"] + assert single.time_dependent is True + assert single.timestamps == [1.0, 2.0]