FPM-1360 - manual flags

docs/operations.md

@@ -77,6 +77,18 @@ All QC methods are registered in `operations/quality_control/qc_methods.py`.
          )
 ```
 
+### Manual flagging
+
+Most QC methods take their thresholds from metadata. The `manual_removal` check is different - the data points it
+flags come from `__assets__/manual_flagging/<network>/manual_flags.csv`, a hand-maintained list of site, variable
+and date range for points that have been checked by a person and found to be bad. Each network has its own file,
+and COSMOS is the only one with a list so far.
+
+The check needs the network and the short site code (e.g. `ALIC1`) to look a data point up. `QCPipeline` puts both
+into every method config before the checks run, taken from the container, so the metadata configuration needs no
+arguments. The file is read by the `ManualRemoval` class itself, where a variable of `ALL` means every variable at
+that site and an empty end date means the period runs to the end of the data.
+
 ## Infilling
 
 Infilling replaces missing or removed data values to produce a complete time series. Infilling may reference

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "dri-timeseries-processor"
-version = "0.7.34"
+version = "0.7.35"
 description = "Timeseries processor service."
 readme = "README.md"
 license = { file = "LICENSE" }

src/dritimeseriesprocessor/operations/quality_control/qc_methods.py

@@ -1,10 +1,15 @@
 from abc import ABC, abstractmethod
+from collections import defaultdict
 from datetime import datetime
+from functools import lru_cache
+from pathlib import Path
 
 import polars as pl
 import time_stream as ts
 from time_stream.operation import Operation
+from time_stream.utils import get_date_filter
 
+from dritimeseriesprocessor import PACKAGE_ROOT
 from dritimeseriesprocessor.models.domain_models.processing_config import DataProcessingMethodConfig
 from dritimeseriesprocessor.utils.enums import ConfigurationType
 
@@ -179,3 +184,75 @@ class FluxQcFlag(QcMethod):
     def run(self, tf: ts.TimeFrame, config: DataProcessingMethodConfig) -> pl.Series:
         col = tf.metadata["column_name"]
         return tf.df[col] == 2
+
+
+@QcMethod.register
+class ManualRemoval(QcMethod):
+    """Flag data points that appear in the manual flagging file.
+
+    The file is a hand-maintained list of data points that have been checked by a person and found to be bad, for
+    example during sensor maintenance or a known fault. Each row gives a site, one or more variables, and the date
+    range that is affected.
+
+    This is a temporary method until a proper manual flagging process has been implemented within the FDRI system.
+    """
+
+    name = "manual_removal"
+
+    def run(self, tf: ts.TimeFrame, config: DataProcessingMethodConfig) -> pl.Series:
+        # The network and site code are put into the params by QCPipeline, taken from the container.
+        periods = self.flag_periods(config.params["network"], config.params["site_id"], tf.metadata["column_name"])
+
+        flagged = pl.repeat(False, pl.len())
+        for period in periods:
+            flagged = flagged | get_date_filter(tf.time_name, period)
+
+        return tf.df.select(flagged).to_series()
+
+    @staticmethod
+    @lru_cache
+    def load_manual_flags(file_path: Path) -> dict[tuple[str, str], list[tuple[datetime, datetime | None]]]:
+        """Read the manual flagging file and group the flagged periods by site and variable.
+
+        Args:
+            file_path: The manual flagging file to read.
+
+        Returns:
+            Flagged periods keyed by (site ID, variable name).
+        """
+        flags = pl.read_csv(file_path).with_columns(
+            pl.col("SITE_ID").str.strip_chars().str.to_uppercase(),
+            pl.col("START_DATETIME").str.to_datetime("%Y-%m-%d %H:%M:%S"),
+            pl.col("END_DATETIME").str.to_datetime("%Y-%m-%d %H:%M:%S"),
+            pl.col("VARIABLES_AFFECTED").str.split(";").alias("VARIABLE"),
+        )
+        flags = flags.explode("VARIABLE", empty_as_null=False).with_columns(
+            pl.col("VARIABLE").str.strip_chars().str.to_uppercase()
+        )
+
+        # A trailing ";" leaves an empty variable name behind, which flags nothing.
+        flags = flags.filter(pl.col("VARIABLE") != "")
+
+        periods = defaultdict(list)
+        for row in flags.iter_rows(named=True):
+            periods[(row["SITE_ID"], row["VARIABLE"])].append((row["START_DATETIME"], row["END_DATETIME"]))
+
+        return dict(periods)
+
+    def flag_periods(self, network: str, site_id: str, column_name: str) -> list[tuple[datetime, datetime | None]]:
+        """Get the periods that have been manually flagged for a site and variable.
+
+        Includes any periods recorded against "ALL" for the site, which apply to every variable there.
+
+        Args:
+            network: Name of the network the site belongs to, e.g. "cosmos". Each network has its own file.
+            site_id: Short site code, e.g. "ALIC1".
+            column_name: Name of the data column, e.g. "TDT2_TSOIL".
+
+        Returns:
+            The flagged periods, or an empty list if nothing has been flagged for this site and variable.
+        """
+        manual_flags_file = PACKAGE_ROOT / "__assets__" / "manual_flagging" / network / "manual_flags.csv"
+        periods = self.load_manual_flags(manual_flags_file)
+        site_id = site_id.upper()
+        return periods.get((site_id, column_name.upper()), []) + periods.get((site_id, "ALL"), [])

src/dritimeseriesprocessor/operations/quality_control/qc_pipeline.py

@@ -33,6 +33,12 @@ def run(
         remove_flagged: bool = True,
     ) -> ts.TimeFrame:
         """Run QC checks and, if this is the last QC block in the plan, remove data that failed."""
+        # Checks that need to know which site they are running on (e.g. manual_removal) read these. Other checks
+        # ignore them.
+        for cfg in config.method_configs:
+            cfg.params["network"] = container.network
+            cfg.params["site_id"] = container.source_site_identifier
+
         tf = super().run(container, dataset_repository, config)
         if remove_flagged and container.has_flags():
             logger.info("Removing data that has failed QC checks")

tests/dritimeseriesprocessor/operations/quality_control/test_qc_methods.py

@@ -1,6 +1,8 @@
 from datetime import datetime
+from pathlib import Path
 
 import polars as pl
+import pytest
 from polars.testing import assert_series_equal
 
 from dritimeseriesprocessor.models.domain_models.processing_config import DataProcessingMethodConfig
@@ -9,6 +11,7 @@
     ErrorCode,
     FluxQcFlag,
     HeatFluxPlateRemoval,
+    ManualRemoval,
     Nr01Temp,
     PluvioDiagnostic,
     Range,
@@ -19,6 +22,8 @@
 )
 from utils.data_creation import create_timeframe
 
+MANUAL_FLAGS_HEADER = '"SITE_ID","VARIABLES_AFFECTED","START_DATETIME","END_DATETIME"'
+
 
 def create_method_config(
     start_date: datetime | None = None,
@@ -329,3 +334,161 @@ def test_no_poor_quality_rows(self) -> None:
 
         expected = pl.Series([False, False, False, False])
         assert_series_equal(result, expected, check_names=False)
+
+
+class TestManualRemoval:
+    def use_periods(self, monkeypatch: pytest.MonkeyPatch, *periods: tuple) -> None:
+        """Use the given flagged periods rather than the file shipped with the package.
+
+        Args:
+            monkeypatch: Fixture used to replace the lookup.
+            periods: (start, end) pairs to return for any site and variable.
+        """
+        monkeypatch.setattr(ManualRemoval, "flag_periods", lambda self, network, site_id, column_name: list(periods))
+
+    def use_loaded_flags(self, monkeypatch: pytest.MonkeyPatch, loaded_flags: dict) -> None:
+        """Use the given loaded flags rather than reading the file shipped with the package.
+
+        Args:
+            monkeypatch: Fixture used to replace the file read.
+            loaded_flags: Flagged periods keyed by (site ID, variable name).
+        """
+        monkeypatch.setattr(ManualRemoval, "load_manual_flags", staticmethod(lambda file_path: loaded_flags))
+
+    def write_manual_flags(self, tmp_path: Path, *rows: str) -> Path:
+        """Write a manual flagging file made up of the given rows.
+
+        Args:
+            tmp_path: Directory to write the file into.
+            rows: Rows to write below the header.
+
+        Returns:
+            Path of the file that was written.
+        """
+        file_path = tmp_path / "manual_flags.csv"
+        file_path.write_text("\n".join([MANUAL_FLAGS_HEADER, *rows]) + "\n")
+        return file_path
+
+    def test_manual_removal_simple(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that values inside a flagged period are flagged."""
+        self.use_periods(monkeypatch, (datetime(2025, 1, 1, 2), datetime(2025, 1, 1, 4)))
+        tf = create_timeframe([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0])
+        config = create_method_config(site_id="ALIC1", network="cosmos")
+
+        result = ManualRemoval().run(tf, config)
+
+        expected = pl.Series([False, False, True, True, True, False, False])
+        assert_series_equal(result, expected, check_names=False)
+
+    def test_manual_removal_nothing_flagged(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that nothing is flagged when the site and variable have no flagged periods."""
+        self.use_periods(monkeypatch)
+        tf = create_timeframe([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0])
+        config = create_method_config(site_id="ALIC1", network="cosmos")
+
+        result = ManualRemoval().run(tf, config)
+
+        expected = pl.Series([False, False, False, False, False, False, False])
+        assert_series_equal(result, expected, check_names=False)
+
+    def test_manual_removal_open_ended_period(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that a period with no end date runs to the end of the data."""
+        self.use_periods(monkeypatch, (datetime(2025, 1, 1, 4), None))
+        tf = create_timeframe([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0])
+        config = create_method_config(site_id="ALIC1", network="cosmos")
+
+        result = ManualRemoval().run(tf, config)
+
+        expected = pl.Series([False, False, False, False, True, True, True])
+        assert_series_equal(result, expected, check_names=False)
+
+    def test_manual_removal_single_time_period(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that a period with the same start and end date flags a single value."""
+        self.use_periods(monkeypatch, (datetime(2025, 1, 1, 3), datetime(2025, 1, 1, 3)))
+        tf = create_timeframe([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0])
+        config = create_method_config(site_id="ALIC1", network="cosmos")
+
+        result = ManualRemoval().run(tf, config)
+
+        expected = pl.Series([False, False, False, True, False, False, False])
+        assert_series_equal(result, expected, check_names=False)
+
+    def test_manual_removal_multiple_periods(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that values in any of several flagged periods are flagged."""
+        self.use_periods(
+            monkeypatch,
+            (datetime(2025, 1, 1, 1), datetime(2025, 1, 1, 2)),
+            (datetime(2025, 1, 1, 5), datetime(2025, 1, 1, 6)),
+        )
+        tf = create_timeframe([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0])
+        config = create_method_config(site_id="ALIC1", network="cosmos")
+
+        result = ManualRemoval().run(tf, config)
+
+        expected = pl.Series([False, True, True, False, False, True, True])
+        assert_series_equal(result, expected, check_names=False)
+
+    def test_splits_variables_on_semicolon(self, tmp_path: Path) -> None:
+        """Test that a ";" separated variable list becomes one entry per variable."""
+        file_path = self.write_manual_flags(tmp_path, '"ALIC1","TA;RH",2015-03-06 00:00:00,2015-03-06 12:00:00')
+
+        result = ManualRemoval.load_manual_flags(file_path)
+
+        period = (datetime(2015, 3, 6), datetime(2015, 3, 6, 12))
+        assert result == {("ALIC1", "TA"): [period], ("ALIC1", "RH"): [period]}
+
+    def test_empty_end_datetime_is_open_ended(self, tmp_path: Path) -> None:
+        """Test that a row with no end date gives a period with no end."""
+        file_path = self.write_manual_flags(tmp_path, '"ALIC1","TA",2015-03-06 00:00:00,')
+
+        result = ManualRemoval.load_manual_flags(file_path)
+
+        assert result == {("ALIC1", "TA"): [(datetime(2015, 3, 6), None)]}
+
+    def test_empty_variable_is_ignored(self, tmp_path: Path) -> None:
+        """Test that a trailing ";" in the variable list does not create an empty entry."""
+        file_path = self.write_manual_flags(tmp_path, '"ALIC1","TA;",2015-03-06 00:00:00,2015-03-06 12:00:00')
+
+        result = ManualRemoval.load_manual_flags(file_path)
+
+        assert list(result) == [("ALIC1", "TA")]
+
+    def test_periods_for_same_variable_are_collected(self, tmp_path: Path) -> None:
+        """Test that several rows for one site and variable give several periods."""
+        file_path = self.write_manual_flags(
+            tmp_path,
+            '"ALIC1","TA",2015-03-06 00:00:00,2015-03-06 12:00:00',
+            '"ALIC1","TA",2016-01-01 00:00:00,2016-01-02 00:00:00',
+        )
+
+        result = ManualRemoval.load_manual_flags(file_path)
+
+        assert result == {
+            ("ALIC1", "TA"): [
+                (datetime(2015, 3, 6), datetime(2015, 3, 6, 12)),
+                (datetime(2016, 1, 1), datetime(2016, 1, 2)),
+            ]
+        }
+
+    def test_flag_periods_includes_all_variables(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that periods recorded against "ALL" are returned alongside a variable's own periods."""
+        self.use_loaded_flags(
+            monkeypatch,
+            {
+                ("ALIC1", "TA"): [(datetime(2015, 3, 6), datetime(2015, 3, 6, 12))],
+                ("ALIC1", "ALL"): [(datetime(2021, 4, 13), None)],
+            },
+        )
+
+        assert ManualRemoval().flag_periods("cosmos", "ALIC1", "TA") == [
+            (datetime(2015, 3, 6), datetime(2015, 3, 6, 12)),
+            (datetime(2021, 4, 13), None),
+        ]
+        assert ManualRemoval().flag_periods("cosmos", "ALIC1", "SWIN") == [(datetime(2021, 4, 13), None)]
+
+    def test_flag_periods_unknown_site_returns_nothing(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that a site or variable that is not in the file returns no periods."""
+        self.use_loaded_flags(monkeypatch, {("BUNNY", "TA"): [(datetime(2016, 1, 1), datetime(2016, 1, 2))]})
+
+        assert ManualRemoval().flag_periods("cosmos", "NOSUCH", "TA") == []
+        assert ManualRemoval().flag_periods("cosmos", "BUNNY", "SWIN") == []

tests/dritimeseriesprocessor/operations/quality_control/test_qc_pipeline.py

@@ -156,6 +156,25 @@ def test_keeps_flagged_data_when_remove_flagged_false(self, monkeypatch: pytest.
         pipeline.remove_flagged_data.assert_not_called()
         assert result is qc_result
 
+    def test_injects_container_site_into_method_configs(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Test that the network and site code are put into every method config before the checks run."""
+        monkeypatch.setattr(OperationPipeline, "run", lambda self, container, repo, config: MagicMock())
+
+        container = self._make_container()
+        container.network = "cosmos"
+        container.source_site_identifier = "ALIC1"
+        method_configs = [
+            DataProcessingMethodConfig(method="range", params={"gt": 1}),
+            DataProcessingMethodConfig(method="manual_removal", params={}),
+        ]
+        config = MagicMock(method_configs=method_configs)
+
+        QCPipeline({}).run(container, {}, config, remove_flagged=False)
+
+        for method_config in method_configs:
+            assert method_config.params["network"] == "cosmos"
+            assert method_config.params["site_id"] == "ALIC1"
+
 
 class TestCoreFlagUpdater:
     def test_calls_update_qc_core_flags(self, mock_timeframe: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: