From d1569a301f71fe75e5a24f7ca87e278ff38dc480 Mon Sep 17 00:00:00 2001 From: Arsham Mirehvandi <130353205+arshammi@users.noreply.github.com> Date: Sun, 20 Sep 2026 23:38:25 +0200 Subject: [PATCH] Refactor treatment history handling and enhance product filtering logic - Updated README.md to clarify the conditions for `last_applied_treatment` and its integration into the advice JSON structure. - Refactored job.py to utilize a new `TreatmentHistory` class for managing treatment data, improving clarity and efficiency in treatment retrieval. - Enhanced advice_context.py to conditionally include `last_applied_treatment` based on the treatment history, ensuring no redundant queries are made. - Modified products.py to implement logic for excluding active ingredients based on the latest treatment's interval, refining the product prefiltering process. - Updated vector.py to ensure the SQL allowlist is applied correctly, improving the efficiency of product searches. - Added unit tests to validate the new treatment history functionality and product filtering logic. --- README.md | 8 +- pipeline/job.py | 28 +++- pipeline/stages/advice_context.py | 15 +- pipeline/stages/products.py | 120 +++++++++++++++- pipeline/stages/treatments.py | 120 +++++++++------- pipeline/stages/vector.py | 39 +----- tests/test_advice_context.py | 159 ++++++++++----------- tests/test_products.py | 225 ++++++++++++++++++++++++++++++ tests/test_treatments.py | 96 +++++++++++++ tests/test_vector.py | 30 +++- 10 files changed, 643 insertions(+), 197 deletions(-) create mode 100644 tests/test_products.py create mode 100644 tests/test_treatments.py diff --git a/README.md b/README.md index 8c63108..3a533d8 100644 --- a/README.md +++ b/README.md @@ -363,7 +363,7 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: 3. **Observation** — `AI_agrosupport_agro_ril_pathogen` (`rilpato_layer`/`rilpato_date`/`rilpato_diffusion`); `observation` is true for `as_of` or any of the past 5 days if a row exists there with `rilpato_diffusion` other than `32` (including `NULL`) 4. **Weather** — `TDatiMeteo_D` for past days, `TDatiMeteo_D_FRC` for today/future 5. **Phenology** — carry-forward of latest observation ≤ day; future days null; if today's phase is missing, use yesterday's -6. **Applied treatments** — products sprayed over `as_of-5 .. as_of`, from `AI_agrosupport_agro_ril_operations*` with `rilop_operation = 10`. A separate lookback over `as_of-25 .. as_of-6` is used only for `last_applied_treatment` during context enrichment. +6. **Applied treatments** — one query loads crop-protection sprays over `as_of-25 .. as_of` from `AI_agrosupport_agro_ril_operations*` with `rilop_operation = 10`. Days in `as_of-5 .. as_of` fill meteorological `applied_treatment`. The latest spray in the full 25-day window drives the prefilter rotation gate. `last_applied_treatment` still uses only `as_of-25 .. as_of-6`, and only when the recent window is empty. 7. **LLM** — synthesise 1–2 anonymised search queries from the subset 8. **Product prefilter** — organic / crop / disease gates, then one prefilter rule against the batch-wide, in-memory `products` index (`pipeline/stages/disease.py::select_prefilter_rule`, first match wins): @@ -377,8 +377,10 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: | `FASE5` today or in the past only | `systemicity` in `{contact, mixed}` and `preventive_action = 1` and `curative_action = 0` and `eradicant_action = 0` | | none of the above | no FASE-specific cut; organic/crop/disease gates only | -9. **Vector search** — Weaviate `near_text` on `ProductProfile`, filtered server-side to the allowlisted `product_id`s. One synthesized query retrieves the top 4; two queries retrieve the top 2 unique each (the second search over-fetches so overlaps can still fill two products) + After those gates, if the latest spray in `as_of-25 .. as_of` still has a non-NULL `product_uses.treatment_interval_min_days` strictly greater than days elapsed, candidates whose `active_ingredients` overlap that product's (tank-mix partners evaluated per product) are dropped. NULL or elapsed interval leaves the list unchanged. + +9. **Vector search** — Weaviate `near_text` on `ProductProfile`, always filtered with `contains_any` on the SQL-allowlisted `product_id`s (`product_id` must use `tokenization: field`). One synthesized query retrieves the top 4; two queries retrieve the top 2 unique each (the second search over-fetches so overlaps can still fill two products) 10. **Product JSON** — label details per product from `products`, `product_uses`, `label_chunks`, cached per `(crop, disease)` -11. **Context enrichment** — `allowed_products`, `applied_treatment` expansion, `last_advice` (with issuance `date`) from the `advice` table, and optional `last_applied_treatment` (spray date + products) when the recent window is empty but a spray exists within 25 days +11. **Context enrichment** — `allowed_products`, `applied_treatment` expansion, `last_advice` (with issuance `date`) from the `advice` table, and optional `last_applied_treatment` (spray date + products) when the recent window is empty but a spray exists in `as_of-25 .. as_of-6`. Enrichment reuses the job's 25-day treatment history; it does not query operations again. 12. **Advice generation** — second LLM profile returns the structured advisory 13. **Persistence** — the advisory plus its collapsed input are written to `advice` inside one transaction diff --git a/pipeline/job.py b/pipeline/job.py index e30aaab..1259052 100644 --- a/pipeline/job.py +++ b/pipeline/job.py @@ -42,8 +42,11 @@ from pipeline.stages.disease import resolve_disease_forecast_by_anmod, select_pr from pipeline.stages.llm import synthesize_queries from pipeline.stages.observation import load_observation from pipeline.stages.phenology import load_phenology -from pipeline.stages.products import prefilter_products -from pipeline.stages.treatments import load_applied_treatments +from pipeline.stages.products import ( + excluded_active_ingredients_for_last_treatment, + prefilter_products, +) +from pipeline.stages.treatments import load_treatment_history from pipeline.stages.vector import RecommendedProduct from pipeline.stages.weather import load_weather from pipeline.worklist import Job @@ -159,8 +162,19 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - weather = load_weather(conn, disease.station, as_of) phenology = load_phenology(conn, job.field_id, as_of) - applied_treatments = load_applied_treatments(conn, job.field_id, as_of) + treatment_history = load_treatment_history(conn, job.field_id, as_of) observation_by_day = load_observation(conn, job.field_id, as_of) + product_index = resources.product_index + if product_index is None: + raise RuntimeError("load_product_index() must run before run_job().") + excluded_ais = excluded_active_ingredients_for_last_treatment( + conn, + product_index, + crop_english=job.crop_english, + disease_english=disease.disease_english, + latest_spray=treatment_history.latest(), + as_of=as_of, + ) # has_risk gates query synthesis and the product prefilter: it is true if # any day is FASE2 / FASE5 / INCUBAZPRIMARIA (disease.has_risk, already @@ -183,7 +197,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - weather, phenology, disease.forecasts_by_day, - applied_treatments, + treatment_history.recent(), observation_by_day, ) json_for_query_synthesis = build_json_for_query_synthesis( @@ -250,11 +264,12 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - prefilter_rule = select_prefilter_rule(disease.forecasts_by_day, observation_by_day, as_of) payload["run"]["prefilter_rule"] = prefilter_rule candidates = prefilter_products( - resources.product_index, + product_index, crop_english=job.crop_english, disease_english=disease.disease_english, organic=job.organic, rule=prefilter_rule, + excluded_active_ingredients=excluded_ais, ) payload["candidate_count"] = len(candidates) @@ -297,8 +312,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - last_advice_summary=last_summary, last_advice_treatments=last_treatments, last_advice_date=last_date, - as_of=as_of, - field_id=job.field_id, + treatment_history=treatment_history, ) payload["last_advice"] = json_for_advice_generation["last_advice"] diff --git a/pipeline/stages/advice_context.py b/pipeline/stages/advice_context.py index bf4e43d..61c39fe 100644 --- a/pipeline/stages/advice_context.py +++ b/pipeline/stages/advice_context.py @@ -12,7 +12,7 @@ import pyodbc from pipeline.db import fetch_one from pipeline.jsonutils import parse_json_list from pipeline.stages.product_json import ProductJsonBuilder, normalize_name -from pipeline.stages.treatments import load_last_treatment_before_recent_window +from pipeline.stages.treatments import TreatmentHistory from pipeline.stages.vector import RecommendedProduct from pipeline.window import format_date @@ -98,8 +98,7 @@ def enrich_advice_json( recommended: Iterable[RecommendedProduct], last_advice_summary: str, last_advice_treatments: Iterable[str], - as_of: date, - field_id: int, + treatment_history: TreatmentHistory, last_advice_date: date | None = None, ) -> None: """ @@ -115,6 +114,8 @@ def enrich_advice_json( When allowed_products is non-empty, the meteorological window has no sprays, and a spray exists in as_of-25 .. as_of-6, append last_applied_treatment as the last key (date + products). Otherwise that key is omitted. + `treatment_history` is the job's already-loaded 25-day operations window; + this function does not query treatments again. """ recommended = list(recommended) @@ -146,8 +147,7 @@ def enrich_advice_json( advice_json, conn=conn, builder=builder, - as_of=as_of, - field_id=field_id, + treatment_history=treatment_history, ) @@ -156,15 +156,14 @@ def _maybe_add_last_applied_treatment( *, conn: pyodbc.Connection, builder: ProductJsonBuilder, - as_of: date, - field_id: int, + treatment_history: TreatmentHistory, ) -> None: if not advice_json.get("allowed_products"): return if _has_recent_applied_treatment(advice_json): return - last = load_last_treatment_before_recent_window(conn, field_id, as_of) + last = treatment_history.latest_before_recent() if last is None: return diff --git a/pipeline/stages/products.py b/pipeline/stages/products.py index 1dfb028..2d74080 100644 --- a/pipeline/stages/products.py +++ b/pipeline/stages/products.py @@ -4,11 +4,14 @@ from __future__ import annotations import logging from dataclasses import dataclass +from datetime import date +from typing import Iterable import pyodbc from pipeline.db import fetch_all from pipeline.jsonutils import normalize_name, parse_json_list +from pipeline.stages.treatments import LAST_TREATMENT_MAX_AGE_DAYS logger = logging.getLogger(__name__) @@ -62,6 +65,14 @@ SELECT FROM products """ +_USES_INTERVAL_SQL = """ +SELECT product_id, treatment_interval_min_days +FROM product_uses +WHERE crop = ? + AND disease = ? + AND product_id IN ({placeholders}) +""" + class ProductIndex: """ @@ -125,6 +136,97 @@ class ProductIndex: return self._by_name.get(normalize_name(product_name)) +def _normalize_ingredient(value: str) -> str: + return value.strip().casefold() + + +def excluded_active_ingredients_for_last_treatment( + conn: pyodbc.Connection, + index: ProductIndex, + *, + crop_english: str, + disease_english: str, + latest_spray: tuple[date, list[str]] | None, + as_of: date, +) -> frozenset[str]: + """ + Active ingredients still locked by the latest spray's treatment interval. + + For each product in `latest_spray`, look up `product_uses.treatment_interval_min_days` + for this crop/disease. A product is locked if any matching use row has a + non-NULL min interval strictly greater than days elapsed since the spray. + Only locked products contribute ingredients. Tank-mix partners are + evaluated independently. Names missing from `products` are skipped. + """ + if latest_spray is None: + return frozenset() + + spray_date, product_names = latest_spray + days_elapsed = (as_of - spray_date).days + if days_elapsed < 0 or days_elapsed > LAST_TREATMENT_MAX_AGE_DAYS: + return frozenset() + + resolved: list[ProductRow] = [] + seen_ids: set[str] = set() + for name in product_names: + row = index.by_name(name) + if row is None: + logger.warning( + "Last applied product '%s' has no row in products; " + "it cannot contribute to the rotation gate.", + name, + ) + continue + if row.product_id in seen_ids: + continue + seen_ids.add(row.product_id) + resolved.append(row) + + if not resolved: + return frozenset() + + product_ids = [row.product_id for row in resolved] + placeholders = ",".join("?" * len(product_ids)) + use_rows = fetch_all( + conn, + _USES_INTERVAL_SQL.format(placeholders=placeholders), + (crop_english, disease_english, *product_ids), + ) + + locked_ids: set[str] = set() + for use_row in use_rows: + interval = use_row.treatment_interval_min_days + if interval is None: + continue + try: + min_days = int(interval) + except (TypeError, ValueError): + continue + if min_days > days_elapsed: + locked_ids.add(str(use_row.product_id)) + + excluded: set[str] = set() + locked_names: list[str] = [] + for row in resolved: + if row.product_id not in locked_ids: + continue + locked_names.append(row.product_name) + for ingredient in row.active_ingredients: + key = _normalize_ingredient(str(ingredient)) + if key: + excluded.add(key) + + if locked_names: + logger.info( + "Rotation gate: spray_date=%s days_elapsed=%s locked=%s excluded_ingredients=%d", + spray_date.isoformat(), + days_elapsed, + locked_names, + len(excluded), + ) + return frozenset(excluded) + + def prefilter_products( index: ProductIndex, *, @@ -132,6 +234,7 @@ def prefilter_products( disease_english: str, organic: bool, rule: str | None, + excluded_active_ingredients: Iterable[str] | None = None, ) -> list[ProductCandidate]: """ Filter the (crop, disease) bucket by organic flag and the selected @@ -148,7 +251,15 @@ def prefilter_products( fase5_past -> systemicity in {contact, mixed} AND preventive_action = 1 AND curative_action = 0 AND eradicant_action = 0 None -> no FASE-specific cut; crop/disease/organic gates only + + When `excluded_active_ingredients` is non-empty, a candidate is also + dropped if any of its `active_ingredients` matches (case-insensitive). """ + blocked = { + _normalize_ingredient(str(item)) + for item in (excluded_active_ingredients or []) + if item is not None and str(item).strip() + } candidates: list[ProductCandidate] = [] for row in index.candidates_for(crop_english, disease_english): @@ -198,14 +309,21 @@ def prefilter_products( continue # rule is None: no FASE-specific cut, only the gates above apply. + if blocked and any( + _normalize_ingredient(str(ingredient)) in blocked + for ingredient in row.active_ingredients + ): + continue + candidates.append(row.as_candidate()) logger.info( - "Product prefilter: crop=%s disease=%s organic=%s rule=%s -> %d candidates", + "Product prefilter: crop=%s disease=%s organic=%s rule=%s excluded_ais=%d -> %d candidates", crop_english, disease_english, organic, rule, + len(blocked), len(candidates), ) return candidates diff --git a/pipeline/stages/treatments.py b/pipeline/stages/treatments.py index 7252cd9..7dac85e 100644 --- a/pipeline/stages/treatments.py +++ b/pipeline/stages/treatments.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from dataclasses import dataclass, field from datetime import date, timedelta from typing import Any, Iterable @@ -16,7 +17,7 @@ logger = logging.getLogger(__name__) _TREATMENT_OPERATION = 10 # Meteorological applied_treatment covers as_of-RECENT .. as_of. RECENT_TREATMENT_DAYS = 5 -# last_applied_treatment looks at sprays strictly before that window, back to as_of-25. +# Rotation gate and last_applied_treatment look back to as_of-25. LAST_TREATMENT_MAX_AGE_DAYS = 25 _APPLIED_TREATMENTS_SQL = """ @@ -59,79 +60,92 @@ def _group_treatments_by_day(rows: Iterable[Any]) -> dict[date, list[str]]: return by_day -def load_applied_treatments( +@dataclass +class TreatmentHistory: + """Crop-protection sprays on one field over as_of-25 .. as_of, loaded once.""" + + as_of: date + by_day: dict[date, list[str]] = field(default_factory=dict) + + def _window(self, start: date, end: date) -> dict[date, list[str]]: + return {day: names for day, names in self.by_day.items() if start <= day <= end} + + def recent(self) -> dict[date, list[str]]: + """Product names per day over as_of-5 .. as_of (meteorological applied_treatment).""" + start = self.as_of - timedelta(days=RECENT_TREATMENT_DAYS) + return self._window(start, self.as_of) + + def latest(self) -> tuple[date, list[str]] | None: + """Most recent spray in as_of-25 .. as_of, or None. Used by the rotation gate.""" + start = self.as_of - timedelta(days=LAST_TREATMENT_MAX_AGE_DAYS) + window = self._window(start, self.as_of) + if not window: + return None + day = max(window) + return day, list(window[day]) + + def latest_before_recent(self) -> tuple[date, list[str]] | None: + """Most recent spray in as_of-25 .. as_of-6, or None (last_applied_treatment).""" + start = self.as_of - timedelta(days=LAST_TREATMENT_MAX_AGE_DAYS) + end = self.as_of - timedelta(days=RECENT_TREATMENT_DAYS + 1) + window = self._window(start, end) + if not window: + return None + day = max(window) + return day, list(window[day]) + + +def load_treatment_history( conn: pyodbc.Connection, field_id: int, as_of: date, -) -> dict[date, list[str]]: +) -> TreatmentHistory: """ - Return the product names applied on the field per day over as_of-5 .. as_of. + Load every crop-protection spray on the field over as_of-25 .. as_of. - Days without a treatment are absent from the mapping. An empty result means - no operation matched the field, the date window, or the treatment operation - type, in which case every applied_treatment stays empty downstream. + Callers slice the result: recent() for meteorological applied_treatment, + latest() for the rotation gate, latest_before_recent() for + last_applied_treatment. Days without a treatment are absent from by_day. """ - start = as_of - timedelta(days=RECENT_TREATMENT_DAYS) + start = as_of - timedelta(days=LAST_TREATMENT_MAX_AGE_DAYS) rows = fetch_all( conn, _APPLIED_TREATMENTS_SQL, (field_id, start, as_of, _TREATMENT_OPERATION), ) + by_day = _group_treatments_by_day(rows) + history = TreatmentHistory(as_of=as_of, by_day=by_day) - if not rows: + recent = history.recent() + if not recent: logger.warning( "No crop protection operations for field %s between %s and %s; " "applied_treatment stays empty.", field_id, - start.isoformat(), + (as_of - timedelta(days=RECENT_TREATMENT_DAYS)).isoformat(), as_of.isoformat(), ) - return {} + else: + logger.info( + "Applied treatments for field %s: %d day(s), %d product mention(s)", + field_id, + len(recent), + sum(len(names) for names in recent.values()), + ) - by_day = _group_treatments_by_day(rows) - logger.info( - "Applied treatments for field %s: %d day(s), %d product mention(s)", - field_id, - len(by_day), - sum(len(v) for v in by_day.values()), - ) - return by_day - - -def load_last_treatment_before_recent_window( - conn: pyodbc.Connection, - field_id: int, - as_of: date, -) -> tuple[date, list[str]] | None: - """ - Return the most recent spray in as_of-25 .. as_of-6, or None. - - The meteorological window already covers as_of-5 .. as_of; this lookback - is only used for last_applied_treatment when that recent window is empty. - """ - start = as_of - timedelta(days=LAST_TREATMENT_MAX_AGE_DAYS) - end = as_of - timedelta(days=RECENT_TREATMENT_DAYS + 1) - rows = fetch_all( - conn, - _APPLIED_TREATMENTS_SQL, - (field_id, start, end, _TREATMENT_OPERATION), - ) - by_day = _group_treatments_by_day(rows) - if not by_day: + latest = history.latest() + if latest is None: logger.info( "No crop protection operations for field %s between %s and %s.", field_id, start.isoformat(), - end.isoformat(), + as_of.isoformat(), ) - return None - - latest = max(by_day) - names = by_day[latest] - logger.info( - "Last applied treatment for field %s on %s: %s", - field_id, - latest.isoformat(), - names, - ) - return latest, names + else: + logger.info( + "Latest treatment for field %s on %s: %s", + field_id, + latest[0].isoformat(), + latest[1], + ) + return history diff --git a/pipeline/stages/vector.py b/pipeline/stages/vector.py index 4bcc16b..3f66fa4 100644 --- a/pipeline/stages/vector.py +++ b/pipeline/stages/vector.py @@ -14,13 +14,6 @@ logger = logging.getLogger(__name__) COLLECTION = "ProductProfile" -# Above this many allowlisted product_ids, encoding every one of them as an -# OR filter term gets expensive to build and transmit. Past that size the -# allowlist is a large fraction of the whole catalog anyway, so falling back -# to a global over-fetch + Python filter (the original approach) is cheaper -# and just as correct. -_MAX_FILTERED_IDS = 200 - @dataclass(frozen=True) class RecommendedProduct: @@ -78,15 +71,10 @@ def search_products( `pipeline.resources.Resources`); this function does not open or close a connection itself, so it is safe to call from many jobs in a row. - The allowlist is applied as a `product_id` OR filter before ranking, - rather than over-fetching globally and intersecting in Python afterwards. - The original code fetched a flat `limit=30` and only worked because the - whole `ProductProfile` collection had ~30 objects, so the top 30 by raw - similarity were, by construction, the entire catalog. Once the catalog - grows past that, a selective allowlist (organic + crop + disease + FASE - can easily narrow it to a handful of products) would often be missing - entirely from an unfiltered top-30, silently returning zero or too few - recommendations instead of raising an error. + The SQL allowlist is always applied as a `product_id` `contains_any` + filter before ranking, so `limit` is the nearest neighbours among + allowed products even when the shortlist is large. This assumes + `product_id` is filterable with `tokenization: field` (exact ID match). """ if not candidates: logger.warning("No product candidates after prefilter; skipping vector search.") @@ -98,22 +86,7 @@ def search_products( allowlist = {c.product_id: c for c in candidates} per_query_limit = 4 if len(queries) == 1 else 2 - - query_filter = None - unfiltered_overfetch = False - if len(allowlist) <= _MAX_FILTERED_IDS: - query_filter = Filter.any_of( - [Filter.by_property("product_id").equal(pid) for pid in allowlist] - ) - else: - unfiltered_overfetch = True - logger.warning( - "Allowlist has %d product(s), above the %d-id server-side filter " - "threshold; falling back to a global over-fetch + Python filter " - "for this query.", - len(allowlist), - _MAX_FILTERED_IDS, - ) + query_filter = Filter.by_property("product_id").contains_any(list(allowlist)) collection = client.collections.get(COLLECTION) results: list[RecommendedProduct] = [] @@ -121,8 +94,6 @@ def search_products( for query_index, query in enumerate(queries): fetch_limit = per_query_limit * (query_index + 1) - if unfiltered_overfetch: - fetch_limit = min(200, max(fetch_limit, per_query_limit * 5, 15)) response = collection.query.near_text( query=query, limit=fetch_limit, diff --git a/tests/test_advice_context.py b/tests/test_advice_context.py index 8ea241b..d92932a 100644 --- a/tests/test_advice_context.py +++ b/tests/test_advice_context.py @@ -4,15 +4,13 @@ from __future__ import annotations import unittest from datetime import date -from types import SimpleNamespace -from unittest.mock import ANY, MagicMock, patch +from unittest.mock import MagicMock, patch from pipeline.stages.advice_context import collapse_product_json, enrich_advice_json -from pipeline.stages.treatments import load_last_treatment_before_recent_window +from pipeline.stages.treatments import TreatmentHistory from pipeline.stages.vector import RecommendedProduct _AS_OF = date(2026, 4, 16) -_FIELD_ID = 7216 _SPRAY_DATE = date(2026, 4, 10) @@ -41,12 +39,18 @@ def _payload(*, applied: list[str] | None = None) -> dict: } +def _history(*product_names: str, spray_date: date = _SPRAY_DATE) -> TreatmentHistory: + by_day = {spray_date: list(product_names)} if product_names else {} + return TreatmentHistory(as_of=_AS_OF, by_day=by_day) + + def _enrich( advice_json: dict, *, recommended: list[RecommendedProduct] | None = None, last_advice_treatments: list[str] | None = None, last_advice_date: date | None = None, + treatment_history: TreatmentHistory | None = None, ) -> None: enrich_advice_json( advice_json, @@ -55,49 +59,53 @@ def _enrich( recommended=recommended or [], last_advice_summary="", last_advice_treatments=last_advice_treatments or [], - as_of=_AS_OF, - field_id=_FIELD_ID, last_advice_date=last_advice_date, + treatment_history=treatment_history if treatment_history is not None else _history(), ) class LastAppliedTreatmentTests(unittest.TestCase): def test_omits_when_allowed_products_empty_and_does_not_query(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window" - ) as loader: - _enrich(advice_json, recommended=[]) - loader.assert_not_called() + with patch("pipeline.stages.treatments.fetch_all") as fetch: + _enrich( + advice_json, + recommended=[], + treatment_history=_history("LIETO"), + ) + fetch.assert_not_called() self.assertNotIn("last_applied_treatment", advice_json) self.assertEqual(advice_json["allowed_products"], []) def test_omits_when_recent_applied_treatment_present(self) -> None: advice_json = _payload(applied=["LIETO"]) - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window" - ) as loader: - _enrich(advice_json, recommended=[_recommended("LIETO")]) - loader.assert_not_called() + with patch("pipeline.stages.treatments.fetch_all") as fetch: + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + treatment_history=_history("LIETO"), + ) + fetch.assert_not_called() self.assertNotIn("last_applied_treatment", advice_json) - def test_omits_when_loader_returns_none(self) -> None: + def test_omits_when_history_has_no_older_spray(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window", - return_value=None, - ) as loader: - _enrich(advice_json, recommended=[_recommended("LIETO")]) - loader.assert_called_once_with(ANY, _FIELD_ID, _AS_OF) + with patch("pipeline.stages.treatments.fetch_all") as fetch: + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + treatment_history=TreatmentHistory(as_of=_AS_OF, by_day={}), + ) + fetch.assert_not_called() self.assertNotIn("last_applied_treatment", advice_json) def test_includes_at_end_when_all_gates_pass(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window", - return_value=(_SPRAY_DATE, ["LIETO"]), - ): - _enrich(advice_json, recommended=[_recommended("LIETO")]) + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + treatment_history=_history("LIETO"), + ) self.assertEqual(list(advice_json)[-1], "last_applied_treatment") self.assertEqual( advice_json["last_applied_treatment"], @@ -106,38 +114,32 @@ class LastAppliedTreatmentTests(unittest.TestCase): def test_name_only_when_product_in_allowed_products(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window", - return_value=(_SPRAY_DATE, ["LIETO"]), - ): - _enrich(advice_json, recommended=[_recommended("LIETO")]) + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + treatment_history=_history("LIETO"), + ) self.assertEqual(advice_json["last_applied_treatment"]["products"], ["LIETO"]) def test_name_only_when_product_in_suggested_products_as_string(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window", - return_value=(_SPRAY_DATE, ["LIETO"]), - ): - _enrich( - advice_json, - recommended=[_recommended("LIETO")], - last_advice_treatments=["LIETO"], - ) + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + last_advice_treatments=["LIETO"], + treatment_history=_history("LIETO"), + ) self.assertEqual(advice_json["last_advice"]["suggested_products"], ["LIETO"]) self.assertEqual(advice_json["last_applied_treatment"]["products"], ["LIETO"]) def test_name_only_when_product_in_suggested_products_as_object(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window", - return_value=(_SPRAY_DATE, ["TWINGO"]), - ): - _enrich( - advice_json, - recommended=[_recommended("LIETO")], - last_advice_treatments=["TWINGO"], - ) + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + last_advice_treatments=["TWINGO"], + treatment_history=_history("TWINGO"), + ) self.assertEqual( advice_json["last_advice"]["suggested_products"], [{"TWINGO": {"uses": []}}], @@ -146,11 +148,11 @@ class LastAppliedTreatmentTests(unittest.TestCase): def test_full_json_when_product_in_neither_list(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window", - return_value=(_SPRAY_DATE, ["UNKNOWN"]), - ): - _enrich(advice_json, recommended=[_recommended("LIETO")]) + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + treatment_history=_history("UNKNOWN"), + ) self.assertEqual( advice_json["last_applied_treatment"]["products"], [{"UNKNOWN": {"uses": []}}], @@ -158,16 +160,27 @@ class LastAppliedTreatmentTests(unittest.TestCase): def test_tank_mix_mixes_name_and_full_json(self) -> None: advice_json = _payload() - with patch( - "pipeline.stages.advice_context.load_last_treatment_before_recent_window", - return_value=(_SPRAY_DATE, ["LIETO", "UNKNOWN"]), - ): - _enrich(advice_json, recommended=[_recommended("LIETO")]) + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + treatment_history=_history("LIETO", "UNKNOWN"), + ) self.assertEqual( advice_json["last_applied_treatment"]["products"], ["LIETO", {"UNKNOWN": {"uses": []}}], ) + def test_does_not_query_operations_when_including_last_applied_treatment(self) -> None: + advice_json = _payload() + with patch("pipeline.stages.treatments.fetch_all") as fetch: + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + treatment_history=_history("LIETO"), + ) + fetch.assert_not_called() + self.assertIn("last_applied_treatment", advice_json) + class CollapseProductJsonTests(unittest.TestCase): def test_collapses_last_applied_treatment_products(self) -> None: @@ -193,29 +206,5 @@ class CollapseProductJsonTests(unittest.TestCase): ) -class LoadLastTreatmentBeforeRecentWindowTests(unittest.TestCase): - def test_returns_latest_day_products(self) -> None: - rows = [ - SimpleNamespace(rilop_date=date(2026, 3, 22), prodgov_name="OLD"), - SimpleNamespace(rilop_date=date(2026, 4, 10), prodgov_name="LIETO"), - SimpleNamespace(rilop_date=date(2026, 4, 10), prodgov_name="TWINGO"), - ] - with patch("pipeline.stages.treatments.fetch_all", return_value=rows) as fetch: - result = load_last_treatment_before_recent_window( - MagicMock(), _FIELD_ID, _AS_OF - ) - params = fetch.call_args[0][2] - self.assertEqual(params[1], date(2026, 3, 22)) # as_of-25 - self.assertEqual(params[2], date(2026, 4, 10)) # as_of-6 - self.assertEqual(result, (_SPRAY_DATE, ["LIETO", "TWINGO"])) - - def test_returns_none_when_no_rows(self) -> None: - with patch("pipeline.stages.treatments.fetch_all", return_value=[]): - result = load_last_treatment_before_recent_window( - MagicMock(), _FIELD_ID, _AS_OF - ) - self.assertIsNone(result) - - if __name__ == "__main__": unittest.main() diff --git a/tests/test_products.py b/tests/test_products.py new file mode 100644 index 0000000..6035ed0 --- /dev/null +++ b/tests/test_products.py @@ -0,0 +1,225 @@ +"""Unit tests for the treatment-interval active-ingredient rotation gate.""" + +from __future__ import annotations + +import unittest +from datetime import date +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from pipeline.stages.products import ( + ProductIndex, + ProductRow, + excluded_active_ingredients_for_last_treatment, + prefilter_products, +) + +_AS_OF = date(2026, 4, 16) +_SPRAY_DATE = date(2026, 4, 6) # 10 days before as_of +_CROP = "grapevine" +_DISEASE = "downy mildew" + + +def _product( + *, + product_id: str = "p1", + product_name: str = "LIETO", + active_ingredients: list[str] | None = None, + systemicity: str = "contact", + preventive_action: bool = True, + curative_action: bool = False, + eradicant_action: bool = False, + organic_certified: bool = False, +) -> ProductRow: + return ProductRow( + product_id=product_id, + product_name=product_name, + registration_number=None, + organic_certified=organic_certified, + systemicity=systemicity, + preventive_action=preventive_action, + curative_action=curative_action, + eradicant_action=eradicant_action, + target_crops=[_CROP], + target_diseases=[_DISEASE], + active_ingredients=active_ingredients or ["mancozeb"], + frac_groups=[], + ) + + +def _excluded( + index: ProductIndex, + *, + names: list[str], + uses: list[SimpleNamespace], + spray_date: date = _SPRAY_DATE, + as_of: date = _AS_OF, +) -> frozenset[str]: + with patch("pipeline.stages.products.fetch_all", return_value=uses) as fetch: + result = excluded_active_ingredients_for_last_treatment( + MagicMock(), + index, + crop_english=_CROP, + disease_english=_DISEASE, + latest_spray=(spray_date, names), + as_of=as_of, + ) + fetch.assert_called_once() + return result + + +class ExcludedActiveIngredientsTests(unittest.TestCase): + def test_none_spray_returns_empty_and_does_not_query(self) -> None: + index = ProductIndex([_product()]) + with patch("pipeline.stages.products.fetch_all") as fetch: + result = excluded_active_ingredients_for_last_treatment( + MagicMock(), + index, + crop_english=_CROP, + disease_english=_DISEASE, + latest_spray=None, + as_of=_AS_OF, + ) + fetch.assert_not_called() + self.assertEqual(result, frozenset()) + + def test_null_interval_returns_empty(self) -> None: + index = ProductIndex([_product()]) + result = _excluded( + index, + names=["LIETO"], + uses=[SimpleNamespace(product_id="p1", treatment_interval_min_days=None)], + ) + self.assertEqual(result, frozenset()) + + def test_interval_less_or_equal_days_elapsed_does_not_lock(self) -> None: + index = ProductIndex([_product()]) + equal = _excluded( + index, + names=["LIETO"], + uses=[SimpleNamespace(product_id="p1", treatment_interval_min_days=10)], + ) + shorter = _excluded( + index, + names=["LIETO"], + uses=[SimpleNamespace(product_id="p1", treatment_interval_min_days=7)], + ) + self.assertEqual(equal, frozenset()) + self.assertEqual(shorter, frozenset()) + + def test_interval_greater_than_days_elapsed_locks_ingredients(self) -> None: + index = ProductIndex([_product(active_ingredients=["Mancozeb", "copper"])]) + result = _excluded( + index, + names=["LIETO"], + uses=[SimpleNamespace(product_id="p1", treatment_interval_min_days=14)], + ) + self.assertEqual(result, frozenset({"mancozeb", "copper"})) + + def test_any_qualifying_use_row_locks_the_product(self) -> None: + index = ProductIndex([_product()]) + result = _excluded( + index, + names=["LIETO"], + uses=[ + SimpleNamespace(product_id="p1", treatment_interval_min_days=None), + SimpleNamespace(product_id="p1", treatment_interval_min_days=21), + ], + ) + self.assertEqual(result, frozenset({"mancozeb"})) + + def test_tank_mix_only_locked_partner_contributes_ingredients(self) -> None: + index = ProductIndex( + [ + _product( + product_id="p1", + product_name="LIETO", + active_ingredients=["mancozeb"], + ), + _product( + product_id="p2", + product_name="TWINGO", + active_ingredients=["metalaxyl"], + ), + ] + ) + result = _excluded( + index, + names=["LIETO", "TWINGO"], + uses=[ + SimpleNamespace(product_id="p1", treatment_interval_min_days=14), + SimpleNamespace(product_id="p2", treatment_interval_min_days=None), + ], + ) + self.assertEqual(result, frozenset({"mancozeb"})) + + def test_catalog_miss_is_ignored_other_partner_still_locks(self) -> None: + index = ProductIndex([_product()]) + result = _excluded( + index, + names=["UNKNOWN", "LIETO"], + uses=[SimpleNamespace(product_id="p1", treatment_interval_min_days=14)], + ) + self.assertEqual(result, frozenset({"mancozeb"})) + + def test_catalog_miss_only_does_not_query_uses(self) -> None: + index = ProductIndex([_product()]) + with patch("pipeline.stages.products.fetch_all") as fetch: + result = excluded_active_ingredients_for_last_treatment( + MagicMock(), + index, + crop_english=_CROP, + disease_english=_DISEASE, + latest_spray=(_SPRAY_DATE, ["UNKNOWN"]), + as_of=_AS_OF, + ) + fetch.assert_not_called() + self.assertEqual(result, frozenset()) + + +class PrefilterRotationGateTests(unittest.TestCase): + def test_drops_candidates_sharing_an_excluded_ingredient(self) -> None: + index = ProductIndex( + [ + _product( + product_id="a", + product_name="A", + active_ingredients=["mancozeb"], + ), + _product( + product_id="b", + product_name="B", + active_ingredients=["copper hydroxide"], + ), + ] + ) + candidates = prefilter_products( + index, + crop_english=_CROP, + disease_english=_DISEASE, + organic=False, + rule=None, + excluded_active_ingredients={"MANCOZEB"}, + ) + self.assertEqual([item.product_name for item in candidates], ["B"]) + + def test_empty_exclusion_set_leaves_candidates_unchanged(self) -> None: + index = ProductIndex( + [ + _product(product_id="a", product_name="A"), + _product(product_id="b", product_name="B", active_ingredients=["copper"]), + ] + ) + candidates = prefilter_products( + index, + crop_english=_CROP, + disease_english=_DISEASE, + organic=False, + rule=None, + excluded_active_ingredients=frozenset(), + ) + self.assertEqual([item.product_name for item in candidates], ["A", "B"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_treatments.py b/tests/test_treatments.py new file mode 100644 index 0000000..45c0a2d --- /dev/null +++ b/tests/test_treatments.py @@ -0,0 +1,96 @@ +"""Unit tests for TreatmentHistory window slicing and the single 25-day load.""" + +from __future__ import annotations + +import unittest +from datetime import date +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from pipeline.stages.treatments import TreatmentHistory, load_treatment_history + +_AS_OF = date(2026, 4, 16) +_FIELD_ID = 7216 +_SPRAY_DATE = date(2026, 4, 10) # as_of-6 + + +class TreatmentHistorySliceTests(unittest.TestCase): + def test_latest_picks_newest_day_and_keeps_tank_mix_order(self) -> None: + history = TreatmentHistory( + as_of=_AS_OF, + by_day={ + date(2026, 3, 22): ["OLD"], + _SPRAY_DATE: ["LIETO", "TWINGO"], + }, + ) + self.assertEqual(history.latest(), (_SPRAY_DATE, ["LIETO", "TWINGO"])) + + def test_latest_prefers_a_spray_inside_the_recent_window(self) -> None: + history = TreatmentHistory( + as_of=_AS_OF, + by_day={ + _SPRAY_DATE: ["OLD"], + _AS_OF: ["TODAY"], + }, + ) + self.assertEqual(history.latest(), (_AS_OF, ["TODAY"])) + self.assertEqual(history.latest_before_recent(), (_SPRAY_DATE, ["OLD"])) + + def test_recent_drops_days_older_than_five(self) -> None: + history = TreatmentHistory( + as_of=_AS_OF, + by_day={ + _SPRAY_DATE: ["OLD"], + date(2026, 4, 11): ["RECENT"], + _AS_OF: ["TODAY"], + }, + ) + self.assertEqual( + history.recent(), + {date(2026, 4, 11): ["RECENT"], _AS_OF: ["TODAY"]}, + ) + + def test_latest_before_recent_matches_as_of_minus_25_to_minus_6(self) -> None: + history = TreatmentHistory( + as_of=_AS_OF, + by_day={ + date(2026, 3, 22): ["OLD"], + _SPRAY_DATE: ["LIETO"], + date(2026, 4, 11): ["RECENT"], + }, + ) + self.assertEqual(history.latest_before_recent(), (_SPRAY_DATE, ["LIETO"])) + + def test_latest_and_latest_before_recent_return_none_when_empty(self) -> None: + history = TreatmentHistory(as_of=_AS_OF, by_day={}) + self.assertIsNone(history.latest()) + self.assertIsNone(history.latest_before_recent()) + self.assertEqual(history.recent(), {}) + + +class LoadTreatmentHistoryTests(unittest.TestCase): + def test_queries_as_of_minus_25_through_as_of(self) -> None: + rows = [ + SimpleNamespace(rilop_date=date(2026, 3, 22), prodgov_name="OLD"), + SimpleNamespace(rilop_date=_SPRAY_DATE, prodgov_name="LIETO"), + SimpleNamespace(rilop_date=_SPRAY_DATE, prodgov_name="TWINGO"), + SimpleNamespace(rilop_date=_SPRAY_DATE, prodgov_name="LIETO"), + ] + with patch("pipeline.stages.treatments.fetch_all", return_value=rows) as fetch: + history = load_treatment_history(MagicMock(), _FIELD_ID, _AS_OF) + params = fetch.call_args[0][2] + self.assertEqual(params[1], date(2026, 3, 22)) # as_of-25 + self.assertEqual(params[2], _AS_OF) + self.assertEqual(history.latest(), (_SPRAY_DATE, ["LIETO", "TWINGO"])) + self.assertEqual(history.latest_before_recent(), (_SPRAY_DATE, ["LIETO", "TWINGO"])) + self.assertEqual(history.recent(), {}) + + def test_returns_empty_history_when_no_rows(self) -> None: + with patch("pipeline.stages.treatments.fetch_all", return_value=[]): + history = load_treatment_history(MagicMock(), _FIELD_ID, _AS_OF) + self.assertEqual(history.by_day, {}) + self.assertIsNone(history.latest()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_vector.py b/tests/test_vector.py index 077ad76..337fb12 100644 --- a/tests/test_vector.py +++ b/tests/test_vector.py @@ -38,7 +38,7 @@ class FakeNearTextQuery: return_metadata: Any, return_properties: list[str], ) -> SimpleNamespace: - self.calls.append({"query": query, "limit": limit}) + self.calls.append({"query": query, "limit": limit, "filters": filters}) hits = list(self.hits_by_query.get(query, [])) return SimpleNamespace(objects=hits[:limit]) @@ -61,7 +61,11 @@ class SearchProductsLimitTests(unittest.TestCase): self.assertEqual([item.product_id for item in results], ["p1", "p2", "p3", "p4"]) self.assertEqual([item.query_index for item in results], [0, 0, 0, 0]) - self.assertEqual(client.query.calls, [{"query": "q1", "limit": 4}]) + self.assertEqual( + [(c["query"], c["limit"]) for c in client.query.calls], + [("q1", 4)], + ) + self.assertTrue(all(c["filters"] is not None for c in client.query.calls)) def test_two_queries_keep_two_each(self) -> None: candidates = [_candidate(f"p{i}") for i in range(1, 7)] @@ -79,9 +83,10 @@ class SearchProductsLimitTests(unittest.TestCase): [("p1", 0), ("p2", 0), ("p4", 1), ("p5", 1)], ) self.assertEqual( - client.query.calls, - [{"query": "q1", "limit": 2}, {"query": "q2", "limit": 4}], + [(c["query"], c["limit"]) for c in client.query.calls], + [("q1", 2), ("q2", 4)], ) + self.assertTrue(all(c["filters"] is not None for c in client.query.calls)) def test_overlapping_hits_keep_first_occurrence(self) -> None: candidates = [_candidate(f"p{i}") for i in range(1, 6)] @@ -99,9 +104,22 @@ class SearchProductsLimitTests(unittest.TestCase): [("p1", 0), ("p2", 0), ("p4", 1), ("p5", 1)], ) self.assertEqual( - client.query.calls, - [{"query": "q1", "limit": 2}, {"query": "q2", "limit": 4}], + [(c["query"], c["limit"]) for c in client.query.calls], + [("q1", 2), ("q2", 4)], ) + self.assertTrue(all(c["filters"] is not None for c in client.query.calls)) + + def test_large_allowlist_still_filters_and_keeps_top_four(self) -> None: + candidates = [_candidate(f"p{i}") for i in range(1, 202)] + hits = [_hit(f"p{i}") for i in range(1, 10)] + client = FakeClient({"q1": hits}) + + results = search_products(client, ["q1"], candidates) + + self.assertEqual([item.product_id for item in results], ["p1", "p2", "p3", "p4"]) + self.assertEqual(len(client.query.calls), 1) + self.assertEqual(client.query.calls[0]["limit"], 4) + self.assertIsNotNone(client.query.calls[0]["filters"]) if __name__ == "__main__":