From d42fe18a527e2792f29e3a7e6ad7f9167d8ade0d Mon Sep 17 00:00:00 2001 From: Arsham Mirehvandi <130353205+arshammi@users.noreply.github.com> Date: Wed, 30 Sep 2026 14:43:08 +0200 Subject: [PATCH] Refactor advisory generation and update database schema - Updated `config.yaml` to change the advice storage table from `advice` to `advisories` and adjusted the `thinking_budget` from 11264 to 9216. - Modified `README.md` to reflect the new `advisories` table in the advisory generation process. - Refactored SQL queries in `batch.py`, `job.py`, and `advice_context.py` to align with the new table structure. - Enhanced the advice generation logic in `advice.py` to accommodate changes in the data model, including updates to the JSON structure for advice output. - Updated relevant functions and documentation to ensure consistency across the codebase regarding the new advisory schema. --- README.md | 36 +- config.yaml | 4 +- pipeline/__main__.py | 6 +- pipeline/batch.py | 2 +- pipeline/job.py | 75 +++- pipeline/stages/advice.py | 114 +++--- pipeline/stages/advice_context.py | 62 ++- pipeline/stages/assemble.py | 8 +- pipeline/stages/disease.py | 209 +++++++--- pipeline/stages/incubations.py | 365 ++++++++++++++++++ pipeline/stages/product_json.py | 6 +- pipeline/stages/products.py | 91 +++-- prompts/_output_format.md | 66 ++-- prompts/advice/_default/system.md | 4 +- .../advice/grapevine__downy_mildew/system.md | 15 +- tests/test_advice_summary.py | 8 +- tests/test_assemble.py | 35 ++ tests/test_disease.py | 87 ++++- tests/test_incubations.py | 260 +++++++++++++ tests/test_products.py | 123 +++++- 20 files changed, 1312 insertions(+), 264 deletions(-) create mode 100644 pipeline/stages/incubations.py create mode 100644 tests/test_incubations.py diff --git a/README.md b/README.md index e4cb2e7..5400273 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ synthesise one or two vector-search queries, prefilters chemical products in SQL Server, and retrieves up to four matching products from a local Weaviate `ProductProfile` collection. It then enriches that JSON with full product label information and the previous advisory, asks a second LLM to write the -farmer-facing advice, and stores the result in the `advice` table. +farmer-facing advice, and stores the result in the `advisories` table. The pipeline is **crop-first**: `config.yaml` declares which crops and diseases to check, and each morning's `batch` run discovers every field that grows one of @@ -188,9 +188,9 @@ Per-job payload fields: | `candidate_count` | Products remaining after the prefilter | | `recommended_products` | Up to 4 items with `product_name` + `registration_number` | | `last_advice` | `date` (previous advisory's issuance date) + `advice_summary` + `suggested_products` of the most recent previous advisory. `advice_summary` is machine context for the next run: it is stored without relative day words, attention prefixes, or follow-the-label closers | -| `advice` | Parsed LLM output: `full_advice`, `advice_summary`, `apply_treatment`, `treatments`, `dosage`, `apply_date` | -| `sent_information` | The advice payload with every embedded product label collapsed back to a product name; mirrors `advice.sent_information` | -| `advice_inserted` | Whether the `advice` row was written | +| `advice` | Parsed LLM output: `full_advice`, `advice_summary`, `is_treatment_recommended`, `recommended_product_names`, `recommended_doses`, `recommended_application_date` | +| `advice_input_json` | The advice payload with every embedded product label collapsed back to a product name; mirrors `advisories.advice_input_json` | +| `advice_inserted` | Whether the `advisories` row was written | | `status` | `ok`, `no_risk`, or `dry_run` (see `_run_report.json` for `failed` / `skipped_*`, which have no per-job file) | | `message` | Human-readable summary of the run outcome | @@ -212,10 +212,10 @@ is ending), or that a new treatment cannot be applied yet because the minimum interval since the last treatment has not elapsed; it does not say that the product list was empty. -If a field's 11-day window has no `FASE2`, `FASE5`, or `INCUBAZPRIMARIA` day and +If a field's 11-day window has no `FASE2`, no `FASE5`, no uncovered incubation, and no true `observation` day (`as_of-5 .. as_of`), the product search is skipped and `allowed_products` is empty, but advice is still generated and stored; that -job is labelled `no_risk`. The run report also counts `ok` jobs whose +job is labelled `no_risk`. A covered incubation does not count as risk. The run report also counts `ok` jobs whose `allowed_products` ended up empty anyway (for example an organic field with no organic-certified product in the catalog yet) — watch that number as the `products` table fills in, since a sudden jump usually means the prefilter or @@ -227,7 +227,7 @@ The process exit code is `0` only if every job in the report is `ok`, `no_risk`, ## Database writes -Each non-dry-run job inserts one row into `advice` for `(date, field_id, +Each non-dry-run job inserts one row into `advisories` for `(issuance_date, field_id, disease)`. That triple is the table's primary key, so any existing row for the same triple is deleted first — re-running a day replaces its advisory instead of failing on a duplicate key. The delete-then-insert pair runs @@ -235,7 +235,7 @@ inside one transaction (`pipeline/db.py::transaction`) so a crash between the two statements cannot leave a field's advisory missing for the day, and deadlocks are retried per `retry:` in config.yaml. -The primary key's clustered index on `([date], field_id, disease)` also +The primary key's clustered index on `(issuance_date, field_id, disease)` also lets that DELETE seek straight to the rows it needs instead of scanning the table, so concurrent workers don't block each other on unrelated rows. @@ -252,7 +252,7 @@ d:\Uni\Thesis\AI-Agro-Support\.venv\Scripts\python.exe -m pipeline batch --skip- with working directory `d:\Uni\Thesis\AI-Agro-Support`. `--skip-existing-advice` makes a re-trigger (e.g. a scheduled retry later that morning) resumable at no -extra cost: it does one bulk check against `advice` up front and only runs jobs +extra cost: it does one bulk check against `advisories` up front and only runs jobs that don't have a row yet. ## Observability (optional) @@ -363,7 +363,7 @@ is not worth a custom cost pipeline at this scale. Run once per `(field, disease)` job by `pipeline/job.py::run_job`: 1. **Worklist** — crop → fields (`AI_agrosupport_an_colture` + `AI_agrosupport_cmp_layers`) → disease model (`AI_agrosupport_agro_models`); organic flag is `cmplay_imp == 3` (`pipeline/worklist.py`, run once per batch, not per job) -2. **Disease model** — per-day incubation > `FASE5` > `FASE2` resolution over as_of±5, from the worklist's known `anmod_id`; a `model_description` containing `INCUBAZPRIMARIA` is stored as `INCUBAZPRIMARIA_` (e.g. `INCUBAZPRIMARIA_40.3`) from `AI_agrosupport_agro_model_tmp2.model_value` +2. **Disease model** — per day over as_of±5, from the worklist's known `anmod_id`. `disease_forecast` is a list or null. Each `INCUBAZPRIMARIA` index is kept (`INCUBAZPRIMARIA2_80.7`, or `INCUBAZPRIMARIA_40.3` when the description has no trailing index), using that index's highest `model_value`. `FASE5` is appended when present; `FASE2` is appended only when `FASE5` is absent. `predicted_incubations_management` (loaded from the previous `advisories.advice_input_json`, then written back with this run) records each index's `start_date` and `infection_covered`. Today counts as provisional: a start locks only once the index appears on a day before `as_of`. A covered incubation alone is not risk. 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 @@ -373,18 +373,18 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: | Rule | Predicates | |---|---| - | `observation` true on `as_of` or the past 5 days | `systemicity` in `{systemic, mixed}` and `eradicant_action = 1` | - | `INCUBAZPRIMARIA` / `INCUBAZPRIMARIA_` today or in the past 5 days | `systemicity` in `{systemic, mixed}` and `curative_action = 1` and `eradicant_action = 0` | - | `FASE2` anywhere and no `FASE5` anywhere | `systemicity == contact` and `preventive_action = 1` and `curative_action = 0` and `eradicant_action = 0` | - | `FASE5` both past/today and future | `systemicity == mixed` and `preventive_action = 1` and `curative_action = 1` and `eradicant_action = 0` | - | `FASE5` in the future only | same predicates as the `FASE2`-only rule | - | `FASE5` today or in the past only | `systemicity` in `{contact, mixed}` and `preventive_action = 1` and `curative_action = 0` and `eradicant_action = 0` | + | `observation` true on `as_of` or the past 5 days | `systemicity` in `{systemic, locally systemic, mixed}` and `has_eradicant_action = 1` | + | uncovered incubation on a day before `as_of` (`as_of-5` .. `as_of-1`) | `systemicity` in `{systemic, locally systemic, mixed}` and `has_curative_action = 1` and `has_eradicant_action = 0` | + | `FASE2` anywhere and no `FASE5` anywhere | `systemicity` in `{contact, translaminar}` and `has_preventive_action = 1` and `has_curative_action = 0` and `has_eradicant_action = 0` | + | `FASE5` in the past and on `as_of` or later | `systemicity == mixed` and `has_preventive_action = 1` and `has_curative_action = 1` and `has_eradicant_action = 0` | + | `FASE5` on `as_of` or later only | same predicates as the `FASE2`-only rule | + | `FASE5` in the past only | `systemicity` in `{contact, translaminar, mixed}` and `has_preventive_action = 1` and `has_curative_action = 0` and `has_eradicant_action = 0` | | none of the above | no FASE-specific cut; legal-state / organic / crop / disease gates only | This stage is skipped (along with query synthesis and vector search) when the latest spray's min treatment interval is still open; see stage 6. 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 in `as_of-25 .. as_of-6`, including when `allowed_products` is empty. Enrichment reuses the job's 25-day treatment history; it does not query operations again. +11. **Context enrichment** — `allowed_products`, `applied_treatment` expansion, `last_advice` (with issuance `date`) from the `advisories` 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`, including when `allowed_products` is empty. 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 +13. **Persistence** — the advisory plus its collapsed input are written to `advisories` inside one transaction diff --git a/config.yaml b/config.yaml index e343897..1e40d92 100644 --- a/config.yaml +++ b/config.yaml @@ -19,14 +19,14 @@ llm: # Part_Two: the farmer-facing advisory generator. Receives the full # json_for_advice_generation payload (weather, phenology, model output, # applied treatments, product labels) and returns the structured advice - # JSON stored in the advice table. Needs a more capable model. + # JSON stored in the advisories table. Needs a more capable model. advice_generation: provider: gemini openai_model: gpt-4o anthropic_model: claude-opus-4-5 gemini_model: gemini-2.5-pro max_tokens: 5120 - thinking_budget: 11264 + thinking_budget: 9216 temperature: 0.0 # ── Crop-first worklist ──────────────────────────────────────────────────────── diff --git a/pipeline/__main__.py b/pipeline/__main__.py index dea50e3..9c84f40 100644 --- a/pipeline/__main__.py +++ b/pipeline/__main__.py @@ -167,11 +167,11 @@ def main(argv: list[str] | None = None) -> int: ) batch_parser.add_argument( "--skip-existing-advice", action="store_true", - help="Skip (field, disease) pairs that already have an advice row for --as-of.", + help="Skip (field, disease) pairs that already have a row in advisories for --as-of.", ) batch_parser.add_argument( "--dry-run", action="store_true", - help="Build the JSON for every job but skip LLM calls, vector search, and the advice DB write.", + help="Build the JSON for every job but skip LLM calls, vector search, and the advisories DB write.", ) batch_parser.add_argument( "--workers", type=int, default=None, help="Override concurrency.workers from config.yaml.", @@ -195,7 +195,7 @@ def main(argv: list[str] | None = None) -> int: ) one_parser.add_argument( "--dry-run", action="store_true", - help="Build the JSON but skip LLM calls, vector search, and the advice DB write.", + help="Build the JSON but skip LLM calls, vector search, and the advisories DB write.", ) one_parser.set_defaults(func=_run_one) diff --git a/pipeline/batch.py b/pipeline/batch.py index 2f5cc8a..097428f 100644 --- a/pipeline/batch.py +++ b/pipeline/batch.py @@ -27,7 +27,7 @@ from pipeline.worklist import Job, build_jobs logger = logging.getLogger(__name__) -_EXISTING_ADVICE_SQL = "SELECT field_id, disease FROM advice WHERE [date] = ?" +_EXISTING_ADVICE_SQL = "SELECT field_id, disease FROM advisories WHERE issuance_date = ?" def _parse_deadline(deadline_str: str) -> datetime: diff --git a/pipeline/job.py b/pipeline/job.py index 84f8944..49276cc 100644 --- a/pipeline/job.py +++ b/pipeline/job.py @@ -33,12 +33,22 @@ from pipeline.stages.advice_context import ( collapse_product_json, enrich_advice_json, load_last_advice, + load_previous_incubation_register, ) from pipeline.stages.assemble import ( assemble_json_for_advice_generation, build_json_for_query_synthesis, ) from pipeline.stages.disease import resolve_disease_forecast_by_anmod, select_prefilter_rule +from pipeline.stages.incubations import ( + covered_by_key, + first_future_disease_date, + forecast_has_risk, + incubation_start_covered, + parse_register, + reconcile_register, + register_payload, +) from pipeline.stages.llm import synthesize_queries from pipeline.stages.observation import load_observation from pipeline.stages.phenology import load_phenology @@ -164,6 +174,11 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - product_index = resources.product_index if product_index is None: raise RuntimeError("load_product_index() must run before run_job().") + previous_register = parse_register( + load_previous_incubation_register( + conn, job.field_id, disease.disease_english, as_of + ) + ) interval_open = last_treatment_interval_open( conn, product_index, @@ -173,11 +188,31 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - 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 - # covers all three via forecasts_by_day) or the observation factor is - # true for as_of or any of the past 5 days. - has_risk = disease.has_risk or any(observation_by_day.values()) + def _covers(start: date): + return incubation_start_covered( + conn, + product_index, + crop_english=job.crop_english, + disease_english=disease.disease_english, + start_date=start, + treatment_history=treatment_history, + ) + + incubation_register = reconcile_register( + previous_register, + disease.forecasts_by_day, + as_of, + _covers, + ) + + # Risk is an uncovered incubation, a FASE2/FASE5, or a true observation. + # A window of only covered incubations is not risk. Advice is still generated. + has_risk = forecast_has_risk(disease.forecasts_by_day, incubation_register) or any( + observation_by_day.values() + ) + future_disease = first_future_disease_date( + disease.forecasts_by_day, incubation_register, as_of + ) logger.info( "Field %s (%s/%s): station=%s has_risk=%s first_future=%s", @@ -186,7 +221,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - job.disease_english, disease.station, has_risk, - disease.first_future_disease_date, + future_disease, ) json_for_advice_generation = assemble_json_for_advice_generation( @@ -196,6 +231,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - disease.forecasts_by_day, treatment_history.recent(), observation_by_day, + register_payload(incubation_register), ) json_for_query_synthesis = build_json_for_query_synthesis( as_of, @@ -203,7 +239,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - disease.disease_english, weather, phenology, - disease.first_future_disease_date, + future_disease, ) payload: dict[str, Any] = { @@ -219,9 +255,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - "anmod_id": disease.anmod_id, "has_risk": has_risk, "first_future_disease_date": ( - disease.first_future_disease_date.isoformat() - if disease.first_future_disease_date - else None + future_disease.isoformat() if future_disease else None ), "prefilter_rule": None, "llm_provider": settings.llm_query_synthesis.provider, @@ -233,7 +267,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - "recommended_products": [], "last_advice": None, "advice": None, - "sent_information": None, + "advice_input_json": None, "advice_inserted": False, "status": "ok", "message": None, @@ -264,7 +298,12 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - ) payload["generated_queries"] = queries - prefilter_rule = select_prefilter_rule(disease.forecasts_by_day, observation_by_day, as_of) + prefilter_rule = select_prefilter_rule( + disease.forecasts_by_day, + observation_by_day, + as_of, + covered_by_key(incubation_register), + ) payload["run"]["prefilter_rule"] = prefilter_rule candidates = prefilter_products( product_index, @@ -294,7 +333,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - else: payload["status"] = "no_risk" logger.warning( - "Field %s: no FASE2, FASE5, or INCUBAZPRIMARIA in the 11-day " + "Field %s: no uncovered incubation, FASE2, or FASE5 in the 11-day " "window, and no true observation in the past 5 days or today; " "product search skipped, advice generated without allowed " "products.", @@ -329,26 +368,26 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - as_of, llm_call=resources.call_llm, ) - sent_information = collapse_product_json(json_for_advice_generation) + advice_input_json = collapse_product_json(json_for_advice_generation) def _insert() -> None: with resources.sql_sem: insert_advice( conn, - run_date=as_of, + issuance_date=as_of, field_id=job.field_id, disease_english=disease.disease_english, - sent_information=sent_information, + advice_input_json=advice_input_json, result=result, ) call_with_sql_retry(settings.retry, _insert) payload["advice"] = result.as_dict() - payload["sent_information"] = sent_information + payload["advice_input_json"] = advice_input_json payload["advice_inserted"] = True payload["message"] = ( f"Returned {len(recommended)} recommended product(s); " - f"advice stored with apply_treatment={int(result.apply_treatment)}." + f"advice stored with is_treatment_recommended={int(result.is_treatment_recommended)}." ) return payload diff --git a/pipeline/stages/advice.py b/pipeline/stages/advice.py index 9fd7e4c..12a1015 100644 --- a/pipeline/stages/advice.py +++ b/pipeline/stages/advice.py @@ -1,4 +1,4 @@ -"""Part_Two, step 3: generate the farmer advisory and persist it to the advice table.""" +"""Part_Two, step 3: generate the farmer advisory and persist it to the advisories table.""" from __future__ import annotations @@ -19,30 +19,30 @@ from pipeline.stages.llm import call_llm logger = logging.getLogger(__name__) -# advice.advice_summary is nvarchar(1050); the prompt asks for at most 1010 chars. +# advisories.advice_summary is nvarchar(1050); the prompt asks for at most 1010 chars. SUMMARY_MAX_CHARS = 1010 -_TREATMENTS_MAX_CHARS = 200 -_DOSAGE_MAX_CHARS = 100 +_RECOMMENDED_PRODUCT_NAMES_MAX_CHARS = 200 +_RECOMMENDED_DOSES_MAX_CHARS = 100 _DELETE_ADVICE_SQL = """ -DELETE FROM advice -WHERE [date] = ? +DELETE FROM advisories +WHERE issuance_date = ? AND field_id = ? AND disease = ? """ _INSERT_ADVICE_SQL = """ -INSERT INTO advice ( - [date], +INSERT INTO advisories ( + issuance_date, field_id, disease, - sent_information, + advice_input_json, full_advice, advice_summary, - apply_treatment, - treatments, - dosage, - apply_date + is_treatment_recommended, + recommended_product_names, + recommended_doses, + recommended_application_date ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ @@ -51,19 +51,23 @@ INSERT INTO advice ( class AdviceResult: full_advice: str advice_summary: str - apply_treatment: bool - treatments: list[str] - dosage: list[str] - apply_date: date | None + is_treatment_recommended: bool + recommended_product_names: list[str] + recommended_doses: list[str] + recommended_application_date: date | None def as_dict(self) -> dict[str, Any]: return { "full_advice": self.full_advice, "advice_summary": self.advice_summary, - "apply_treatment": int(self.apply_treatment), - "treatments": self.treatments, - "dosage": self.dosage, - "apply_date": self.apply_date.isoformat() if self.apply_date else None, + "is_treatment_recommended": int(self.is_treatment_recommended), + "recommended_product_names": self.recommended_product_names, + "recommended_doses": self.recommended_doses, + "recommended_application_date": ( + self.recommended_application_date.isoformat() + if self.recommended_application_date + else None + ), } @@ -121,42 +125,42 @@ def parse_advice(raw: str, as_of: date) -> AdviceResult: ) summary = summary[:SUMMARY_MAX_CHARS] - treatments = _string_list(payload.get("treatments")) - dosage = _string_list(payload.get("dosage")) - apply_treatment = _to_bool(payload.get("apply_treatment")) - apply_date = _to_date(payload.get("apply_date")) + recommended_product_names = _string_list(payload.get("recommended_product_names")) + recommended_doses = _string_list(payload.get("recommended_doses")) + is_treatment_recommended = _to_bool(payload.get("is_treatment_recommended")) + recommended_application_date = _to_date(payload.get("recommended_application_date")) - if apply_treatment and not treatments: - logger.warning("apply_treatment is set but treatments is empty.") - if not apply_treatment and treatments: + if is_treatment_recommended and not recommended_product_names: + logger.warning("is_treatment_recommended is set but recommended_product_names is empty.") + if not is_treatment_recommended and recommended_product_names: logger.warning( - "apply_treatment is not set but %d treatment(s) were returned.", - len(treatments), + "is_treatment_recommended is not set but %d treatment(s) were returned.", + len(recommended_product_names), ) return AdviceResult( full_advice=full_advice, advice_summary=summary, - apply_treatment=apply_treatment, - treatments=treatments, - dosage=dosage, - apply_date=apply_date, + is_treatment_recommended=is_treatment_recommended, + recommended_product_names=recommended_product_names, + recommended_doses=recommended_doses, + recommended_application_date=recommended_application_date, ) def insert_advice( conn: pyodbc.Connection, *, - run_date: date, + issuance_date: date, field_id: int, disease_english: str, - sent_information: dict[str, Any], + advice_input_json: dict[str, Any], result: AdviceResult, ) -> None: """ Store the advisory, replacing any earlier row for the same day/field/disease. - The table's primary key is (date, field_id, disease), so an explicit + The table's primary key is (issuance_date, field_id, disease), so an explicit delete keeps re-runs of the same day from violating it instead of accumulating duplicates. The delete and insert run inside one transaction: with a single sequential @@ -165,34 +169,42 @@ def insert_advice( between the two statements would otherwise leave that field's advisory silently missing for the day. """ - treatments = _json_column(result.treatments, "treatments", _TREATMENTS_MAX_CHARS) - dosage = _json_column(result.dosage, "dosage", _DOSAGE_MAX_CHARS) - sent = json.dumps(sent_information, ensure_ascii=False, default=_json_default) + recommended_product_names = _json_column( + result.recommended_product_names, + "recommended_product_names", + _RECOMMENDED_PRODUCT_NAMES_MAX_CHARS, + ) + recommended_doses = _json_column( + result.recommended_doses, + "recommended_doses", + _RECOMMENDED_DOSES_MAX_CHARS, + ) + sent = json.dumps(advice_input_json, ensure_ascii=False, default=_json_default) with transaction(conn): - execute(conn, _DELETE_ADVICE_SQL, (run_date, field_id, disease_english)) + execute(conn, _DELETE_ADVICE_SQL, (issuance_date, field_id, disease_english)) execute( conn, _INSERT_ADVICE_SQL, ( - run_date, + issuance_date, field_id, disease_english, sent, result.full_advice, result.advice_summary, - 1 if result.apply_treatment else 0, - treatments, - dosage, - result.apply_date, + 1 if result.is_treatment_recommended else 0, + recommended_product_names, + recommended_doses, + result.recommended_application_date, ), ) logger.info( - "Stored advice for field %s / %s on %s (apply_treatment=%s).", + "Stored advice for field %s / %s on %s (is_treatment_recommended=%s).", field_id, disease_english, - run_date.isoformat(), - int(result.apply_treatment), + issuance_date.isoformat(), + int(result.is_treatment_recommended), ) @@ -203,7 +215,7 @@ def _json_column(values: list[str], column: str, max_chars: int) -> str | None: encoded = json.dumps(values, ensure_ascii=False) if len(encoded) > max_chars: raise PipelineError( - f"advice.{column} would need {len(encoded)} characters but the column " + f"advisories.{column} would need {len(encoded)} characters but the column " f"holds {max_chars}: {encoded}" ) return encoded @@ -281,5 +293,5 @@ def _to_date(value: Any) -> date | None: return datetime.strptime(text, fmt).date() except ValueError: continue - logger.warning("Could not parse apply_date '%s'; storing NULL.", text) + logger.warning("Could not parse recommended_application_date '%s'; storing NULL.", text) return None diff --git a/pipeline/stages/advice_context.py b/pipeline/stages/advice_context.py index ac574d1..5f7297f 100644 --- a/pipeline/stages/advice_context.py +++ b/pipeline/stages/advice_context.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy +import json import logging from datetime import date, datetime from typing import Any, Iterable @@ -22,14 +23,14 @@ _PRODUCT_LIST_KEYS = ("allowed_products", "applied_treatment", "suggested_produc _LAST_ADVICE_SQL = """ SELECT TOP 1 - [date] AS issued_on, + issuance_date, advice_summary, - treatments -FROM advice + recommended_product_names +FROM advisories WHERE field_id = ? AND disease = ? - AND [date] < ? -ORDER BY [date] DESC + AND issuance_date < ? +ORDER BY issuance_date DESC """ @@ -43,6 +44,43 @@ def _row_date(value: Any) -> date | None: return None +_LAST_SENT_SQL = """ +SELECT TOP 1 advice_input_json +FROM advisories +WHERE field_id = ? + AND disease = ? + AND issuance_date < ? +ORDER BY issuance_date DESC +""" + + +def load_previous_incubation_register( + conn: pyodbc.Connection, + field_id: int, + disease_english: str, + as_of: date, +) -> dict[str, Any]: + """Return predicted_incubations_management from the latest advice before as_of.""" + row = fetch_one(conn, _LAST_SENT_SQL, (field_id, disease_english, as_of)) + if row is None or not row.advice_input_json: + return {} + try: + payload = json.loads(row.advice_input_json) + except (TypeError, json.JSONDecodeError): + logger.warning( + "Previous advice_input_json for field %s / %s is not JSON; incubation register starts empty.", + field_id, + disease_english, + ) + return {} + if not isinstance(payload, dict): + return {} + register = payload.get("predicted_incubations_management") + if not isinstance(register, dict): + return {} + return register + + def load_last_advice( conn: pyodbc.Connection, field_id: int, @@ -61,10 +99,14 @@ def load_last_advice( return "", [], None summary = (row.advice_summary or "").strip() - treatments = [name.strip() for name in parse_json_list(row.treatments) if name.strip()] - issued_on = _row_date(row.issued_on) - logger.info("Loaded previous advice with %d suggested product(s).", len(treatments)) - return summary, treatments, issued_on + recommended_product_names = [ + name.strip() + for name in parse_json_list(row.recommended_product_names) + if name.strip() + ] + issued_on = _row_date(row.issuance_date) + logger.info("Loaded previous advice with %d suggested product(s).", len(recommended_product_names)) + return summary, recommended_product_names, issued_on def _entry_name(entry: Any) -> str: @@ -191,7 +233,7 @@ def collapse_product_json(advice_json: dict[str, Any]) -> dict[str, Any]: """ Return a copy where every embedded Product JSON is replaced by its name. - This is the version stored in advice.sent_information: the same payload the + This is the version stored in advisories.advice_input_json: the same payload the LLM received, minus the product label detail. """ collapsed = copy.deepcopy(advice_json) diff --git a/pipeline/stages/assemble.py b/pipeline/stages/assemble.py index 8711cc6..208e44a 100644 --- a/pipeline/stages/assemble.py +++ b/pipeline/stages/assemble.py @@ -50,7 +50,7 @@ def _day_entry( as_of: date, weather: dict[date, dict[str, Any]], phenology: dict[date, str | None], - forecasts: dict[date, str | None], + forecasts: dict[date, list[str] | None], applied_treatments: dict[date, list[str]], observation: dict[date, bool] | None = None, ) -> dict[str, Any]: @@ -70,11 +70,12 @@ def assemble_json_for_advice_generation( as_of: date, weather: dict[date, dict[str, Any]], phenology: dict[date, str | None], - forecasts: dict[date, str | None], + forecasts: dict[date, list[str] | None], applied_treatments: dict[date, list[str]] | None = None, observation: dict[date, bool] | None = None, + incubation_register: dict[str, Any] | None = None, ) -> dict[str, Any]: - """Build meteorological_data (as_of-5..as_of), weather_forecasts (as_of+1..as_of+5), and date_of_today.""" + """Build date_of_today, the incubation register, meteorological_data, and weather_forecasts.""" window = build_window(as_of) applied = applied_treatments or {} meteorological = [ @@ -89,6 +90,7 @@ def assemble_json_for_advice_generation( ] return { "date_of_today": format_date(as_of), + "predicted_incubations_management": dict(incubation_register or {}), "meteorological_data": meteorological, "weather_forecasts": forecasts_section, } diff --git a/pipeline/stages/disease.py b/pipeline/stages/disease.py index a9dd52c..c40fcb9 100644 --- a/pipeline/stages/disease.py +++ b/pipeline/stages/disease.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from dataclasses import dataclass, field from datetime import date from decimal import Decimal @@ -20,9 +21,7 @@ class DiseaseForecast: disease_raw: str disease_english: str station: int | None - forecasts_by_day: dict[date, str | None] - first_future_disease_date: date | None - has_risk: bool + forecasts_by_day: dict[date, list[str] | None] _MODEL_SQL = """ @@ -47,7 +46,10 @@ WHERE model_model = ? @dataclass class _DayRecords: descriptions: set[str] = field(default_factory=set) - incub_values: list[Any] = field(default_factory=list) + incub_rows: list[tuple[str, Any]] = field(default_factory=list) + + +_INCUB_INDEX = re.compile(r"INCUBAZPRIMARIA(\d+)", re.IGNORECASE) def _is_incubazprimaria(label: str | None) -> bool: @@ -69,41 +71,86 @@ def _format_model_value(value: Any) -> str | None: return text or None -def _incub_label(values: list[Any]) -> str: - """INCUBAZPRIMARIA_, or bare INCUBAZPRIMARIA if none.""" - numerics: list[tuple[float, Any]] = [] - for value in values: +def _incub_index(description: str) -> str | None: + """Digits glued to INCUBAZPRIMARIA. A space or underscore means no index.""" + match = _INCUB_INDEX.search(description) + return match.group(1) if match else None + + +def _label_for_index(index: str, rows: list[tuple[str, Any]]) -> str: + """Highest numeric model_value for one incubation index, as INCUBAZPRIMARIA{index}_.""" + best_numeric: float | None = None + best_raw: Any = None + for _description, value in rows: if value is None: continue try: - numerics.append((float(value), value)) + numeric = float(value) except (TypeError, ValueError): continue - if numerics: - best = max(numerics, key=lambda item: item[0])[1] - formatted = _format_model_value(best) - if formatted: - return f"INCUBAZPRIMARIA_{formatted}" - return "INCUBAZPRIMARIA" + if best_numeric is None or numeric > best_numeric: + best_numeric = numeric + best_raw = value + formatted = _format_model_value(best_raw) if best_raw is not None else None + if not formatted: + return f"INCUBAZPRIMARIA{index}" if index else "INCUBAZPRIMARIA" + if index: + return f"INCUBAZPRIMARIA{index}_{formatted}" + return f"INCUBAZPRIMARIA_{formatted}" -def _resolve_day_fase(records: _DayRecords) -> str | None: +def _incub_label(rows: list[tuple[str, Any]]) -> str: + """Highest incubation among the given rows: max model_value, then higher index. + + Kept for callers that still want one label. Day resolution uses `_incub_labels`. """ - INCUBAZPRIMARIA beats FASE5 beats FASE2; anything else is ignored. + candidates: list[tuple[float, int, str]] = [] + for description, value in rows: + if value is None: + continue + try: + numeric = float(value) + except (TypeError, ValueError): + continue + index = _incub_index(description) or "" + rank = int(index) if index else -1 + candidates.append((numeric, rank, _label_for_index(index, [(description, value)]))) + if not candidates: + return "INCUBAZPRIMARIA" + return max(candidates, key=lambda item: (item[0], item[1]))[2] - Incubation days are stored as INCUBAZPRIMARIA_ (e.g. - INCUBAZPRIMARIA_40.3) from the tmp2 row. Several incubation rows on one - day keep the highest numeric model_value. A missing model_value falls - back to bare INCUBAZPRIMARIA. + +def _incub_labels(rows: list[tuple[str, Any]]) -> list[str]: + """One label per index, ascending, unindexed last. Each index keeps its own highest value.""" + groups: dict[str, list[tuple[str, Any]]] = {} + for description, value in rows: + groups.setdefault(_incub_index(description) or "", []).append((description, value)) + if not groups: + return ["INCUBAZPRIMARIA"] + ordered = sorted((key for key in groups if key), key=int) + if "" in groups: + ordered.append("") + return [_label_for_index(index, groups[index]) for index in ordered] + + +def _resolve_day_fase(records: _DayRecords) -> list[str] | None: + """ + Every incubation index on the day, then FASE5, or FASE2 when FASE5 is absent. + + Each index is INCUBAZPRIMARIA{index}_ from its own highest + tmp2 value. A description with no trailing index is one group. A missing + model_value falls back to a bare label. FASE2 is omitted when FASE5 is + also present. No incubation and no phase is None. """ upper = {d.upper() for d in records.descriptions if d} + labels: list[str] = [] if any("INCUBAZPRIMARIA" in d for d in upper): - return _incub_label(records.incub_values) + labels.extend(_incub_labels(records.incub_rows)) if "FASE5" in upper: - return "FASE5" - if "FASE2" in upper: - return "FASE2" - return None + labels.append("FASE5") + elif "FASE2" in upper: + labels.append("FASE2") + return labels or None def find_anmod_id(conn: pyodbc.Connection, field_id: int, disease_name: str) -> int: @@ -174,22 +221,15 @@ def resolve_disease_forecast_by_anmod( if description: by_day[day].descriptions.add(description) if "INCUBAZPRIMARIA" in description.upper(): - by_day[day].incub_values.append(row.model_value) + by_day[day].incub_rows.append((description, row.model_value)) if row.model_station is not None: stations.append(int(row.model_station)) - forecasts: dict[date, str | None] = { + forecasts: dict[date, list[str] | None] = { day: _resolve_day_fase(records) for day, records in by_day.items() } - first_future: date | None = None - for day in sorted(d for d in forecasts if d > as_of): - if forecasts[day] is not None: - first_future = day - break - station = stations[0] if stations else fallback_station - has_risk = any(v is not None for v in forecasts.values()) return DiseaseForecast( anmod_id=anmod_id, @@ -197,56 +237,97 @@ def resolve_disease_forecast_by_anmod( disease_english=disease_english, station=station, forecasts_by_day=forecasts, - first_future_disease_date=first_future, - has_risk=has_risk, ) +def incub_register_key(label: str) -> str | None: + """`incubazioneprimaria` plus any glued index. None when the label is not an incubation.""" + if not _is_incubazprimaria(label): + return None + rest = label[len("INCUBAZPRIMARIA"):] + digits: list[str] = [] + for char in rest: + if not char.isdigit(): + break + digits.append(char) + return "incubazioneprimaria" + "".join(digits) + + def select_prefilter_rule( - forecasts_by_day: dict[date, str | None], + forecasts_by_day: dict[date, list[str] | None], observation_by_day: dict[date, bool], as_of: date, + covered_by_key: dict[str, bool], ) -> str | None: """ - Choose which product prefilter branch applies (Part_One, step 3). + Choose which product prefilter branch applies. - "Past/today" is as_of-5 .. as_of; "future" is as_of+1 .. as_of+5. First - match wins: + Observation still uses as_of and the past 5 days. For the disease model, + past is as_of-5 .. as_of-1 and future is as_of .. as_of+5, because today's + forecast can still change. First match wins: - 1. `observation` True on as_of or any of the past 5 days -> "observation" - 2. INCUBAZPRIMARIA today or in the past 5 days -> "incubazprimaria" - 3. FASE2 anywhere in the window AND no FASE5 anywhere -> "fase2_only" - 4. FASE5 present both past/today AND future -> "fase5_both" - 5. FASE5 present in the future only -> "fase5_future" - 6. FASE5 present today or in the past only -> "fase5_past" - 7. otherwise -> None + 1. observation true on as_of or any of the past 5 days -> "observation" + 2. an uncovered incubation on a past day -> "incubazprimaria" + 3. FASE2 anywhere and no FASE5 anywhere -> "fase2_only" + 4. FASE5 in the past and on today or later -> "fase5_both" + 5. FASE5 on today or later only -> "fase5_future" + 6. FASE5 in the past only -> "fase5_past" + 7. otherwise -> None - None means the product prefilter applies only the crop / disease / - organic gates, with no FASE-specific catalog cut. + `covered_by_key` maps an incubation register key to infection_covered. + An incubation that appears only today or later does not select + incubazprimaria. None means no FASE-specific catalog cut. """ - past_today = [d for d in forecasts_by_day if d <= as_of] - future = [d for d in forecasts_by_day if d > as_of] + past = [day for day in forecasts_by_day if day < as_of] + future = [day for day in forecasts_by_day if day >= as_of] + observation_days = [day for day in forecasts_by_day if day <= as_of] - if any(observation_by_day.get(d, False) for d in past_today): + if any(observation_by_day.get(day, False) for day in observation_days): return "observation" - if any(_is_incubazprimaria(forecasts_by_day.get(d)) for d in past_today): + if _uncovered_incubation(forecasts_by_day, past, covered_by_key): return "incubazprimaria" - fase2_anywhere = any(v == "FASE2" for v in forecasts_by_day.values()) - fase5_anywhere = any(v == "FASE5" for v in forecasts_by_day.values()) - - if fase2_anywhere and not fase5_anywhere: + if _any_code(forecasts_by_day, "FASE2") and not _any_code(forecasts_by_day, "FASE5"): return "fase2_only" - fase5_past_today = any(forecasts_by_day.get(d) == "FASE5" for d in past_today) - fase5_future = any(forecasts_by_day.get(d) == "FASE5" for d in future) - - if fase5_past_today and fase5_future: + fase5_past = _any_code_on(forecasts_by_day, past, "FASE5") + fase5_future = _any_code_on(forecasts_by_day, future, "FASE5") + if fase5_past and fase5_future: return "fase5_both" if fase5_future: return "fase5_future" - if fase5_past_today: + if fase5_past: return "fase5_past" - return None + + +def _day_labels(forecasts_by_day: dict[date, list[str] | None], day: date) -> list[str]: + return list(forecasts_by_day.get(day) or []) + + +def _any_code(forecasts_by_day: dict[date, list[str] | None], code: str) -> bool: + return _any_code_on(forecasts_by_day, list(forecasts_by_day), code) + + +def _any_code_on( + forecasts_by_day: dict[date, list[str] | None], + days: list[date], + code: str, +) -> bool: + return any(code in _day_labels(forecasts_by_day, day) for day in days) + + +def _uncovered_incubation( + forecasts_by_day: dict[date, list[str] | None], + days: list[date], + covered_by_key: dict[str, bool], +) -> bool: + for day in days: + for label in _day_labels(forecasts_by_day, day): + key = incub_register_key(label) + if key is None: + continue + if not covered_by_key.get(key, False): + return True + return False diff --git a/pipeline/stages/incubations.py b/pipeline/stages/incubations.py new file mode 100644 index 0000000..374db0e --- /dev/null +++ b/pipeline/stages/incubations.py @@ -0,0 +1,365 @@ +"""Per-field incubation register: start date and whether a spray already covers it.""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from typing import Any, Callable + +import pyodbc + +from pipeline.db import fetch_all +from pipeline.stages.disease import incub_register_key +from pipeline.stages.products import ProductIndex +from pipeline.stages.treatments import TreatmentHistory +from pipeline.window import format_date + +logger = logging.getLogger(__name__) + +_USES_INTERVAL_SQL = """ +SELECT product_id, treatment_interval_min_days +FROM product_uses +WHERE crop = ? + AND disease = ? + AND product_id IN ({placeholders}) +""" + + +@dataclass(frozen=True) +class CoveringTreatment: + """Spray that covers an incubation start. The interval is the longest minimum that reaches it.""" + + spray_date: date + products: tuple[str, ...] + treatment_interval_min_days: int + + +@dataclass(frozen=True) +class IncubationRecord: + start_date: date + infection_covered: bool + covering_treatment: CoveringTreatment | None = None + + +def parse_register(raw: Any) -> dict[str, IncubationRecord]: + """Read predicted_incubations_management. Bad entries are dropped.""" + if not isinstance(raw, dict): + return {} + parsed: dict[str, IncubationRecord] = {} + for key, value in raw.items(): + if not isinstance(key, str) or not isinstance(value, dict): + continue + start = _parse_start(value.get("start_date")) + if start is None: + continue + parsed[key] = IncubationRecord( + start_date=start, + infection_covered=bool(value.get("infection_covered")), + covering_treatment=_parse_covering(value.get("covering_treatment")), + ) + return parsed + + +def register_payload(register: dict[str, IncubationRecord]) -> dict[str, dict[str, Any]]: + """JSON object placed before meteorological_data. Keys stay in insertion order.""" + payload: dict[str, dict[str, Any]] = {} + for key, record in register.items(): + entry: dict[str, Any] = { + "start_date": format_date(record.start_date), + "infection_covered": record.infection_covered, + } + treatment = record.covering_treatment + if treatment is not None: + entry["covering_treatment"] = { + "date": format_date(treatment.spray_date), + "products": list(treatment.products), + "treatment_interval_min_days": treatment.treatment_interval_min_days, + } + payload[key] = entry + return payload + + +def reconcile_register( + previous: dict[str, IncubationRecord], + forecasts_by_day: dict[date, list[str] | None], + as_of: date, + covers: Callable[[date], CoveringTreatment | None], +) -> dict[str, IncubationRecord]: + """ + Drop indexes the model no longer shows. Today is provisional. + + An index lands on the first run that shows it on a day before as_of. + A landed start is kept after it slides before as_of-5. A covered + incubation keeps its spray unless a newer spray covers the same start, + or the start moves past that spray or past its stored interval. + """ + window_start = as_of - timedelta(days=5) + days_by_key = _days_by_key(forecasts_by_day) + reconciled: dict[str, IncubationRecord] = {} + + for key in sorted(days_by_key): + days = days_by_key[key] + past = [day for day in days if day < as_of] + prior = previous.get(key) + if past: + earliest = min(past) + if prior is not None and prior.start_date < earliest: + chosen = prior.start_date + else: + chosen = earliest + elif prior is not None and prior.start_date < window_start: + chosen = prior.start_date + else: + chosen = min(day for day in days if day >= as_of) + + reconciled[key] = _reconcile_coverage(prior, chosen, covers) + + dropped = sorted(set(previous) - set(reconciled)) + if dropped: + logger.info("Incubation register dropped %s", dropped) + return reconciled + + +def _reconcile_coverage( + prior: IncubationRecord | None, + chosen: date, + covers: Callable[[date], CoveringTreatment | None], +) -> IncubationRecord: + if prior is None or not prior.infection_covered: + return _from_lookup(chosen, covers) + + stored = prior.covering_treatment + if stored is None: + found = _qualifying(covers(chosen), chosen) + if found is None: + return IncubationRecord(start_date=chosen, infection_covered=True) + return IncubationRecord(start_date=chosen, infection_covered=True, covering_treatment=found) + + if chosen > prior.start_date and _outside_interval(stored, chosen): + return _from_lookup(chosen, covers) + if chosen < prior.start_date and chosen <= stored.spray_date: + return _from_lookup(chosen, covers) + if chosen < prior.start_date: + return IncubationRecord( + start_date=chosen, + infection_covered=True, + covering_treatment=stored, + ) + + found = _qualifying(covers(chosen), chosen) + if found is not None and found.spray_date > stored.spray_date: + return IncubationRecord( + start_date=chosen, + infection_covered=True, + covering_treatment=found, + ) + return IncubationRecord( + start_date=chosen, + infection_covered=True, + covering_treatment=stored, + ) + + +def _from_lookup( + chosen: date, + covers: Callable[[date], CoveringTreatment | None], +) -> IncubationRecord: + found = _qualifying(covers(chosen), chosen) + if found is None: + return IncubationRecord(start_date=chosen, infection_covered=False) + return IncubationRecord( + start_date=chosen, + infection_covered=True, + covering_treatment=found, + ) + + +def _qualifying(found: CoveringTreatment | None, start: date) -> CoveringTreatment | None: + """A spray on or after the start, or one whose interval does not reach it, does not count.""" + if found is None or found.spray_date >= start: + return None + if (start - found.spray_date).days >= found.treatment_interval_min_days: + return None + return found + + +def _outside_interval(stored: CoveringTreatment, start: date) -> bool: + return (start - stored.spray_date).days >= stored.treatment_interval_min_days + + +def forecast_has_risk( + forecasts_by_day: dict[date, list[str] | None], + register: dict[str, IncubationRecord], +) -> bool: + """True when an uncovered incubation or a FASE2/FASE5 remains in the window.""" + if any(not record.infection_covered for record in register.values()): + return True + for labels in forecasts_by_day.values(): + if labels and ("FASE2" in labels or "FASE5" in labels): + return True + return False + + +def first_future_disease_date( + forecasts_by_day: dict[date, list[str] | None], + register: dict[str, IncubationRecord], + as_of: date, +) -> date | None: + """First day after as_of that is still risk. Covered incubations alone do not count.""" + for day in sorted(day for day in forecasts_by_day if day > as_of): + if _day_is_risk(forecasts_by_day.get(day), register): + return day + return None + + +def covered_by_key(register: dict[str, IncubationRecord]) -> dict[str, bool]: + return {key: record.infection_covered for key, record in register.items()} + + +def incubation_start_covered( + conn: pyodbc.Connection, + index: ProductIndex, + *, + crop_english: str, + disease_english: str, + start_date: date, + treatment_history: TreatmentHistory, +) -> CoveringTreatment | None: + """ + Latest spray strictly before start_date whose minimum interval still reaches it. + + Coverage is (start_date - spray_date).days < treatment_interval_min_days + for this crop and disease. A null interval does not cover. One tank-mix + partner is enough to qualify the day; the returned interval is the longest + minimum among products on that day that reach the start, and products lists + every name from that spray day. A spray on the start date itself does not count. + """ + sprays = { + day: names + for day, names in treatment_history.by_day.items() + if day < start_date + } + if not sprays: + return None + + ids_by_day: dict[date, list[str]] = {} + names_by_day: dict[date, list[str]] = {} + all_ids: list[str] = [] + seen: set[str] = set() + for day, names in sprays.items(): + day_ids: list[str] = [] + for name in names: + row = index.by_name(name) + if row is None: + continue + day_ids.append(row.product_id) + if row.product_id not in seen: + seen.add(row.product_id) + all_ids.append(row.product_id) + if day_ids: + ids_by_day[day] = day_ids + names_by_day[day] = list(names) + if not all_ids: + return None + + placeholders = ",".join("?" * len(all_ids)) + use_rows = fetch_all( + conn, + _USES_INTERVAL_SQL.format(placeholders=placeholders), + (crop_english, disease_english, *all_ids), + ) + min_days_by_id: dict[str, int] = {} + 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 + product_id = str(use_row.product_id) + current = min_days_by_id.get(product_id) + if current is None or min_days > current: + min_days_by_id[product_id] = min_days + + latest: CoveringTreatment | None = None + for spray_date, product_ids in ids_by_day.items(): + elapsed = (start_date - spray_date).days + reaching = [ + min_days + for product_id in product_ids + if (min_days := min_days_by_id.get(product_id)) is not None and elapsed < min_days + ] + if not reaching: + continue + candidate = CoveringTreatment( + spray_date=spray_date, + products=tuple(names_by_day[spray_date]), + treatment_interval_min_days=max(reaching), + ) + if latest is None or candidate.spray_date > latest.spray_date: + latest = candidate + return latest + + +def _parse_covering(value: Any) -> CoveringTreatment | None: + if not isinstance(value, dict): + return None + spray_date = _parse_start(value.get("date")) + raw_products = value.get("products") + if spray_date is None or not isinstance(raw_products, list): + return None + products = tuple(str(name).strip() for name in raw_products if str(name).strip()) + if not products: + return None + try: + min_days = int(value.get("treatment_interval_min_days")) + except (TypeError, ValueError): + return None + return CoveringTreatment( + spray_date=spray_date, + products=products, + treatment_interval_min_days=min_days, + ) + + +def _days_by_key(forecasts_by_day: dict[date, list[str] | None]) -> dict[str, list[date]]: + found: dict[str, list[date]] = {} + for day, labels in forecasts_by_day.items(): + if not labels: + continue + for label in labels: + key = incub_register_key(label) + if key is None: + continue + found.setdefault(key, []).append(day) + return found + + +def _day_is_risk(labels: list[str] | None, register: dict[str, IncubationRecord]) -> bool: + if not labels: + return False + if "FASE2" in labels or "FASE5" in labels: + return True + for label in labels: + key = incub_register_key(label) + if key is None: + continue + record = register.get(key) + if record is not None and not record.infection_covered: + return True + return False + + +def _parse_start(value: Any) -> date | None: + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + if not isinstance(value, str) or not value.strip(): + return None + try: + return datetime.strptime(value.strip(), "%d-%m-%Y").date() + except ValueError: + return None diff --git a/pipeline/stages/product_json.py b/pipeline/stages/product_json.py index 9b0feea..ed832d5 100644 --- a/pipeline/stages/product_json.py +++ b/pipeline/stages/product_json.py @@ -34,7 +34,7 @@ FROM product_uses WHERE product_id = ? AND crop = ? AND disease = ? -ORDER BY id ASC +ORDER BY product_use_id ASC """ _CHUNKS_SQL = """ @@ -45,7 +45,7 @@ SELECT target_diseases FROM label_chunks WHERE product_id = ? -ORDER BY chunk_id ASC +ORDER BY label_chunk_id ASC """ _USE_KEYS = ( @@ -130,7 +130,7 @@ class ProductJsonBuilder: return { row.product_name: { "active_ingredients": row.active_ingredients, - "frac_groups": row.frac_groups, + "frac_groups": row.frac_codes, "systemicity": row.systemicity, "uses": self._load_uses(conn, row.product_id), "information_chunks": self._load_chunks(conn, row.product_id), diff --git a/pipeline/stages/products.py b/pipeline/stages/products.py index 5697c89..51c6c36 100644 --- a/pipeline/stages/products.py +++ b/pipeline/stages/products.py @@ -30,15 +30,15 @@ class ProductRow: product_id: str product_name: str registration_number: str | None - organic_certified: bool + is_organic_certified: bool systemicity: str # stripped, original casing (used verbatim in the Product JSON payload) - preventive_action: bool - curative_action: bool - eradicant_action: bool + has_preventive_action: bool + has_curative_action: bool + has_eradicant_action: bool target_crops: list[str] target_diseases: list[str] active_ingredients: list[str] - frac_groups: list[str] + frac_codes: list[str] permitted: bool = True # catalog prodgov_state is a non-blank value other than Revocato def as_candidate(self) -> ProductCandidate: @@ -54,15 +54,15 @@ SELECT product_id, product_name, registration_number, - organic_certified, + is_organic_certified, systemicity, - preventive_action, - curative_action, - eradicant_action, + has_preventive_action, + has_curative_action, + has_eradicant_action, target_crops, target_diseases, active_ingredients, - frac_groups + frac_codes FROM products """ @@ -199,15 +199,15 @@ class ProductIndex: product_id=str(record.product_id), product_name=str(record.product_name).strip(), registration_number=registration_number, - organic_certified=bool(record.organic_certified), + is_organic_certified=bool(record.is_organic_certified), systemicity=(record.systemicity or "").strip(), - preventive_action=bool(record.preventive_action), - curative_action=bool(record.curative_action), - eradicant_action=bool(record.eradicant_action), + has_preventive_action=bool(record.has_preventive_action), + has_curative_action=bool(record.has_curative_action), + has_eradicant_action=bool(record.has_eradicant_action), target_crops=parse_json_list(record.target_crops), target_diseases=parse_json_list(record.target_diseases), active_ingredients=parse_json_list(record.active_ingredients), - frac_groups=parse_json_list(record.frac_groups), + frac_codes=parse_json_list(record.frac_codes), permitted=permitted, ) ) @@ -305,6 +305,12 @@ def last_treatment_interval_open( return True +# Contact-allowed rules also keep translaminar; systemic-allowed rules also keep +# locally systemic. Mixed stays a separate label and is not implied by either. +_CONTACT_SYSTEMICITY = {"contact", "translaminar"} +_SYSTEMIC_SYSTEMICITY = {"systemic", "locally systemic"} + + def prefilter_products( index: ProductIndex, *, @@ -321,16 +327,19 @@ def prefilter_products( matches the government catalog and prodgov_state is a non-blank value other than Revocato. Revoked rows stay in the index for by_name. - observation -> systemicity in {systemic, mixed} AND eradicant_action = 1 - incubazprimaria -> systemicity in {systemic, mixed} AND curative_action = 1 - AND eradicant_action = 0 - fase2_only -> systemicity == contact AND preventive_action = 1 - AND curative_action = 0 AND eradicant_action = 0 - fase5_both -> systemicity == mixed AND preventive_action = 1 - AND curative_action = 1 AND eradicant_action = 0 + observation -> systemicity in {systemic, locally systemic, mixed} + AND has_eradicant_action = 1 + incubazprimaria -> systemicity in {systemic, locally systemic, mixed} + AND has_curative_action = 1 AND has_eradicant_action = 0 + fase2_only -> systemicity in {contact, translaminar} + AND has_preventive_action = 1 + AND has_curative_action = 0 AND has_eradicant_action = 0 + fase5_both -> systemicity == mixed AND has_preventive_action = 1 + AND has_curative_action = 1 AND has_eradicant_action = 0 fase5_future -> same predicates as fase2_only - fase5_past -> systemicity in {contact, mixed} AND preventive_action = 1 - AND curative_action = 0 AND eradicant_action = 0 + fase5_past -> systemicity in {contact, translaminar, mixed} + AND has_preventive_action = 1 + AND has_curative_action = 0 AND has_eradicant_action = 0 None -> no FASE-specific cut; legal-state / crop / disease / organic gates only """ candidates: list[ProductCandidate] = [] @@ -338,49 +347,49 @@ def prefilter_products( for row in index.candidates_for(crop_english, disease_english): if not row.permitted: continue - if organic and not row.organic_certified: + if organic and not row.is_organic_certified: continue systemicity = row.systemicity.lower() if rule == "observation": - if systemicity not in {"systemic", "mixed"}: + if systemicity not in _SYSTEMIC_SYSTEMICITY | {"mixed"}: continue - if not row.eradicant_action: + if not row.has_eradicant_action: continue elif rule == "incubazprimaria": - if systemicity not in {"systemic", "mixed"}: + if systemicity not in _SYSTEMIC_SYSTEMICITY | {"mixed"}: continue - if not row.curative_action: + if not row.has_curative_action: continue - if row.eradicant_action: + if row.has_eradicant_action: continue elif rule in ("fase2_only", "fase5_future"): - if systemicity != "contact": + if systemicity not in _CONTACT_SYSTEMICITY: continue - if not row.preventive_action: + if not row.has_preventive_action: continue - if row.curative_action: + if row.has_curative_action: continue - if row.eradicant_action: + if row.has_eradicant_action: continue elif rule == "fase5_both": if systemicity != "mixed": continue - if not row.preventive_action: + if not row.has_preventive_action: continue - if not row.curative_action: + if not row.has_curative_action: continue - if row.eradicant_action: + if row.has_eradicant_action: continue elif rule == "fase5_past": - if systemicity not in {"contact", "mixed"}: + if systemicity not in _CONTACT_SYSTEMICITY | {"mixed"}: continue - if not row.preventive_action: + if not row.has_preventive_action: continue - if row.curative_action: + if row.has_curative_action: continue - if row.eradicant_action: + if row.has_eradicant_action: continue # rule is None: no FASE-specific cut, only the gates above apply. diff --git a/prompts/_output_format.md b/prompts/_output_format.md index 5820f53..11ae4bc 100644 --- a/prompts/_output_format.md +++ b/prompts/_output_format.md @@ -7,13 +7,13 @@ introduction, no commentary outside the JSON. |---|---|---| | `full_advice` | string | Your complete advisory text, in Italian. | | `advice_summary` | string | Italian summary of `full_advice`, **at most 1010 characters**. It must state the recommendation: whether a treatment should be administered or not; if yes, for what reason, on what date (and why that date), and with which product; if no, for what reason. Use calendar dates only (for example `13 aprile 2026` or `13-04-2026`). Never use relative day words such as oggi, ieri, domani, dopodomani, today, yesterday, or tomorrow — even when a calendar date follows them. Do not start with labels such as `Attenzione:`, `Nota:`, or `Importante:`. Do not close with a reminder to follow the product label; that belongs in `full_advice`. | -| `apply_treatment` | integer | `1` if you recommend applying a product, `0` otherwise. | -| `treatments` | array of strings | Names of the recommended product(s), exactly as written in the input payload. Empty array `[]` when you recommend no product. | -| `dosage` | array of strings | Prescribed dose for each product in `treatments`, in the same order (for example `"1.5 l/ha"`). Empty array `[]` when you recommend no product. | -| `apply_date` | string or null | Recommended application date as `YYYY-MM-DD`. `null` when you recommend no product. | +| `is_treatment_recommended` | integer | `1` if you recommend applying a product, `0` otherwise. | +| `recommended_product_names` | array of strings | Names of the recommended product(s), exactly as written in the input payload. Empty array `[]` when you recommend no product. | +| `recommended_doses` | array of strings | Prescribed dose for each product in `recommended_product_names`, in the same order (for example `"1.5 l/ha"`). Empty array `[]` when you recommend no product. | +| `recommended_application_date` | string or null | Recommended application date as `YYYY-MM-DD`. `null` when you recommend no product. | -If `allowed_products` is empty, you must not prescribe any product (`treatments` -must be `[]` and `apply_treatment` must be `0`), even if +If `allowed_products` is empty, you must not prescribe any product (`recommended_product_names` +must be `[]` and `is_treatment_recommended` must be `0`), even if `last_advice.suggested_products` names one. Do not mention that the list is empty, that authorization data is missing, or that you cannot choose a product for that reason. Do not quote JSON keys such as `allowed_products`. Explain the @@ -29,9 +29,11 @@ minimum is still greater than the days from the spray date to `date_of_today`. If several uses qualify, cite the one with the longest remaining minimum. If `treatment_interval_max_days` is null, cite the minimum only. -- No model risk: the 11-day window has no `FASE2`, `FASE5`, or `INCUBAZPRIMARIA`, - and no `observation` is true. Say a treatment is not necessary because the - model does not indicate infection risk. +- No model risk: the 11-day window has no `FASE2` and no `FASE5`, every + incubation in `predicted_incubations_management` has `infection_covered` + true (or the register is empty), and no `observation` is true. A covered + incubation still listed in `disease_forecast` is not model risk. Say a + treatment is not necessary because the model does not indicate infection risk. - Minimum interval still open: days elapsed are less than `treatment_interval_min_days`. Say that, because of the minimum interval required since the last treatment, a new treatment cannot be applied yet. @@ -47,7 +49,7 @@ If several uses qualify, cite the one with the longest remaining minimum. If - Catalog gap: `disease_forecast` or `observation` shows risk, and no spray interval is still open. Do not claim the crop is protected and do not invent a product name. State the model risk without blaming an empty list. - `apply_treatment` stays `0`. + `is_treatment_recommended` stays `0`. Follow the tone of these examples; do not copy them when the dates or the persistence differ: @@ -56,10 +58,10 @@ persistence differ: - *Considerato che l'ultimo trattamento è stato eseguito il 12 aprile 2026, la sua efficacia protettiva sta volgendo al termine.* - *A causa dell'intervallo minimo richiesto dall'ultimo trattamento, non è ancora possibile applicarne uno nuovo.* -When you do recommend a treatment (`apply_treatment` = 1) and `allowed_products` +When you do recommend a treatment (`is_treatment_recommended` = 1) and `allowed_products` is non-empty: if `last_advice.suggested_products` names one or more products that are still appropriate now (phenology, weather, disease-model risk, and -that product's own label constraints), `treatments` must be those products — +that product's own label constraints), `recommended_product_names` must be those products — same names, and the same set if it was a tank mix — even when they are absent from `allowed_products`. Do not prescribe a product solely because `last_advice` did. Otherwise recommend only products that appear in `allowed_products`. Never @@ -83,22 +85,40 @@ or in `full_advice` *la fase fenologica segnalata per oggi è …*. In sono a conoscenza" form, or *la fase fenologica segnalata per il 13 aprile 2026 è …*. +`disease_forecast` is a list of codes for that day, or null. Several codes on one day are simultaneous. `predicted_incubations_management` sits before `meteorological_data`. Key `incubazioneprimaria2` is the same incubation as `INCUBAZPRIMARIA2_…`; `infection_covered` true means that incubation is already covered and is not a reason to treat. A `FASE5` or `FASE2` on the same day, or an incubation with `infection_covered` false, is. + `disease_forecast` codes are for you, not for the farmer. Never write `FASE2`, -`FASE5`, `FASE`, `INCUBAZPRIMARIA`, or `INCUBAZPRIMARIA_` in `full_advice` or -`advice_summary`. Keep using those codes internally when reading the JSON. -When referring to those days, use Italian farmer language: -`FASE2` → germination (*germinazione*); `FASE5` → risk of infection (*rischio -di infezione*); `INCUBAZPRIMARIA` / `INCUBAZPRIMARIA_80.7` → primary incubation -(*incubazione primaria*), and state the percentage in words (e.g. *all'80,7%*), -never the JSON key. +`FASE5`, `FASE`, `INCUBAZPRIMARIA`, `INCUBAZPRIMARIA_`, or +`INCUBAZPRIMARIA_` in `full_advice` or `advice_summary`. Keep using +those codes internally when reading the JSON. Digits glued to `INCUBAZPRIMARIA` +before the underscore are the incubation number; the number after the +underscore is the percentage. When referring to those days, use Italian farmer +language: `FASE2` → germination (*germinazione*); `FASE5` → risk of infection +(*rischio di infezione*); `INCUBAZPRIMARIA` / `INCUBAZPRIMARIA_80.7` → primary +incubation (*incubazione primaria*), and state the percentage in words (e.g. +*all'80,7%*). When an index is present, name that incubation (e.g. +`INCUBAZPRIMARIA2_80.7` → *la seconda incubazione primaria, all'80,7%*). Never +write the JSON key. + +`disease_forecast` codes are high-probability forecasts from current weather +and the weather forecast, including values on past days in the window. They +are not confirmed infections or confirmed incubation. The model that produces +them does not account for treatments already applied. Do not describe a +`disease_forecast` after a recorded spray as a failed treatment or as disease +still present. While residual cover or the minimum interval is still open, +follow the cover rules above and do not prescribe another product for that +reason. When cover has ended, a new `disease_forecast` is a serious reason to +consider treatment of tissue the model treats as unprotected; state that +weather-driven risk, not that the previous product failed. `observation` = true +remains field evidence and is independent of this limitation. ### Example shape { "full_advice": "...", "advice_summary": "...", - "apply_treatment": 1, - "treatments": ["PRODUCT NAME"], - "dosage": ["1.5 l/ha"], - "apply_date": "2026-08-14" + "is_treatment_recommended": 1, + "recommended_product_names": ["PRODUCT NAME"], + "recommended_doses": ["1.5 l/ha"], + "recommended_application_date": "2026-08-14" } diff --git a/prompts/advice/_default/system.md b/prompts/advice/_default/system.md index ea25274..dab0946 100644 --- a/prompts/advice/_default/system.md +++ b/prompts/advice/_default/system.md @@ -1,5 +1,5 @@ -You are a decision support system for crop protection, advising on the risk of a plant disease for a specific crop. Your goal is to prevent infection while avoiding unnecessary treatments. You will be given the crop's phenology using the BBCH phases, weather data from the past five days, forecasts for the next five days, the deterministic phytopathological model's day-by-day risk phase output (for example FASE2 / FASE5, where a higher phase number generally indicates a more advanced and more urgent stage of infection risk), and a list of authorized treatment products with their active substances and label directives. +You are a decision support system for crop protection, advising on the risk of a plant disease for a specific crop. Your goal is to prevent infection while avoiding unnecessary treatments. You will be given the crop's phenology using the BBCH phases, weather data from the past five days, forecasts for the next five days, the phytopathological model's day-by-day disease forecast (`disease_forecast`, a list of codes such as FASE2 / FASE5 / INCUBAZPRIMARIA2_80.7, or null; several codes on one day are simultaneous), `predicted_incubations_management` (each incubation's `start_date` and whether `infection_covered` is true), and a list of authorized treatment products with their active substances and label directives. Each code is a very high-probability forecast from weather, not a confirmed infection. The forecast does not account for treatments already applied, so a later value is not evidence that a previous treatment failed or that infection is still present. An incubation with `infection_covered` true is not a reason to treat. A FASE2, a FASE5, or an incubation with `infection_covered` false still is. The register key `incubazioneprimaria2` is the same incubation as a forecast code `INCUBAZPRIMARIA2_…`. -STRICT RULES: 1. Base your interpretation strictly on the model's output; do not speculate beyond it. 2. If no treatment is currently necessary, do not discuss a hypothetical future treatment or dosage. Do discuss residual protection of a product already applied, using its persistence and application date, and say when the minimum interval since that treatment means a new one cannot be applied yet. +STRICT RULES: 1. Base your interpretation of infection risk on `disease_forecast`, `predicted_incubations_management`, and field observations. Do not speculate beyond those inputs, and do not treat a forecast after a recorded treatment as proof that the treatment failed. An incubation with `infection_covered` true is not model risk. 2. If no treatment is currently necessary, do not discuss a hypothetical future treatment or dosage. Do discuss residual protection of a product already applied, using its persistence and application date, and say when the minimum interval since that treatment means a new one cannot be applied yet. NOTE FOR OPERATORS: this is the generic fallback advice prompt. It is used because no dedicated prompt exists yet under `prompts/advice/__/` for this crop-disease pair — see the pipeline README for how to add one. Consider writing a pair-specific prompt that explains this disease's own phase semantics, the way `prompts/advice/grapevine__downy_mildew/system.md` explains FASE2/FASE5 for downy mildew, before relying on this fallback in production. diff --git a/prompts/advice/grapevine__downy_mildew/system.md b/prompts/advice/grapevine__downy_mildew/system.md index 45711cd..e2100d3 100644 --- a/prompts/advice/grapevine__downy_mildew/system.md +++ b/prompts/advice/grapevine__downy_mildew/system.md @@ -1,9 +1,10 @@ -You are a decision support system to prevent primary grapevine downy mildew (Plasmopara viticola). Your super power is having a deterministic model for the risk of infection and your ultimate goal is to prevent the infection. You'll be given grapevine phenology using the BBCH phases, weather data from the past five days, and forecasts for the next five days. You will also be provided with a list of authorized treatment products, including their active substances and label directives, to be used if action is required. +You are a decision support system to prevent primary grapevine downy mildew (Plasmopara viticola). Your super power is having a forecast model that estimates infection risk from weather at very high probability and your ultimate goal is to prevent the infection. You'll be given grapevine phenology using the BBCH phases, weather data from the past five days, and forecasts for the next five days. You will also be provided with a list of authorized treatment products, including their active substances and label directives, to be used if action is required. -PREVIOUS KNOWLEDGE ON THE MODEL'S BEHAVIOR: The output of the model is definite. On the days marked as FASE5, conditions are favorable for the infection of zoospores on leaf surfaces and their penetration into the leaves through the stomata. -FASE5 of the grapevine downy mildew phytopathological model refers to the critical primary infection phase. This process occurs following the dispersion of infective zoospores, which reach the leaf surfaces and successfully penetrate through the stomata. By establishing this initial infection in the first green tissues, Phase 5 marks the decisive moment that triggers the seasonal epidemic dynamics and subsequent epidemic attacks. -In FASE2, the germination of Plasmopara viticola oospores involves the formation of a macrosporangium from which the infective zoospores are released, a process favored by high humidity and mild spring temperatures. The associated risk is the onset of primary downy mildew infections, especially following prolonged rainfall and in the presence of susceptible young foliage. -INCUBAZPRIMARIA is the progress percentage of the incubation process associated with a specific infection event. Incubation begins only when the temperature conditions required for the infection to establish are met, progressing gradually until reaching 100%, at which point symptom onset is expected. In the absence of suitable conditions, the incubation process may fail to start or may be interrupted. If the value for the disease_forecast key in the JSON is INCUBAZPRIMARIA_40, the 40 after the underscore represents 40%. -If observation = TRUE, it represents direct field evidence of infection and the risk of further spread, as observed on-site by the farmer or technician. +PREVIOUS KNOWLEDGE ON THE MODEL'S BEHAVIOR: `disease_forecast` is a list of codes for that day, or null. Several codes on one day are simultaneous. Each code is a forecast with a very high probability, computed from current weather and the weather forecast. It is not a confirmed infection, a confirmed incubation, or a field observation. The forecasting model does not know about treatments already applied and does not include them in its calculation. A `disease_forecast` on a later day therefore does not mean that an earlier treatment failed, that the pathogen is still present, or that an infection is ongoing. Read it as the risk the model assigns from weather alone, as if no treatment had been applied. Whether another treatment is justified depends on that forecast together with `predicted_incubations_management` and residual protection from sprays already recorded in `applied_treatment` or `last_applied_treatment`, not on an inference that the previous product was ineffective. +On the days marked as FASE5, the model forecasts conditions favorable for the infection of zoospores on leaf surfaces and their penetration into the leaves through the stomata. This is not confirmation that zoospores have already penetrated in this vineyard. +FASE5 of the grapevine downy mildew phytopathological model refers to the critical primary infection phase: dispersion of infective zoospores onto leaf surfaces and penetration through the stomata. In untreated tissue, establishing this initial infection in the first green tissues is the decisive moment that can trigger the seasonal epidemic dynamics and subsequent epidemic attacks. A FASE5 value forecasts that this phase is likely under the weather the model used; it does not confirm that penetration has already occurred. +In FASE2, the model forecasts germination of Plasmopara viticola oospores, involving the formation of a macrosporangium from which the infective zoospores are released, a process favored by high humidity and mild spring temperatures. The associated forecast is the risk that primary downy mildew infections could start, especially following prolonged rainfall and in the presence of susceptible young foliage. +INCUBAZPRIMARIA is the model's forecast of how far primary incubation would have progressed, as a percentage, for an infection event the weather model treats as established. It is not a measurement of the pathogen inside the plant and it is not evidence that a treatment already applied failed to stop that infection. Incubation in the model begins only when the temperature conditions required for the infection to establish are met, progressing gradually until reaching 100%, at which point symptom onset would be expected if that infection had established. In the absence of suitable conditions, the incubation process may fail to start or may be interrupted. If one entry in the disease_forecast list is INCUBAZPRIMARIA2_80.7, the 2 glued to INCUBAZPRIMARIA is the incubation number and the 80.7 after the underscore is 80.7%. A value with no glued digits, such as INCUBAZPRIMARIA_40, has no incubation number; the 40 after the underscore is still 40%. `predicted_incubations_management` tracks each incubation by that number: key `incubazioneprimaria2` is the same incubation as `INCUBAZPRIMARIA2_…`, and `incubazioneprimaria` is the same as a label with no glued number. `start_date` is the first day of that incubation. `infection_covered` true means a spray before that date already covers it: that incubation is not a reason to treat. A `FASE5` or `FASE2` beside it, or another incubation whose `infection_covered` is false, still is. +If observation = TRUE, it represents direct field evidence of infection and the risk of further spread, as observed on-site by the farmer or technician. Unlike `disease_forecast`, this is field evidence, not a weather-based forecast. ADDITIONAL NOTES: Main grapevine phenology phases are: BBCH < 53: leaf development, BBCH < 60: inflorescence emerge; BBCH < 70: flowering BBCH < 80: fruits development, BBCH < 90: ripening. You are smart, wise, nice, honest, and determined to promote Integrated Pest Management practices. ho cambiato idea, scrivilo in Italiano. -STRICT RULES: 1. Stick only to the output of the model for any interpreting or explanation. 2. If there is no necessity for a treatment, don't discuss the probable future situation that may need a treatment or the dosage. Do discuss residual protection of a product already applied, using its persistence and application date, and say when the minimum interval since that treatment means a new one cannot be applied yet. +STRICT RULES: 1. Interpret infection risk from `disease_forecast`, `predicted_incubations_management`, and `observation`. Do not invent infections, symptoms, or treatment failure that those inputs do not state. Do not treat a later `disease_forecast` as proof that a recorded spray failed. An incubation with `infection_covered` true is not model risk. 2. If there is no necessity for a treatment, don't discuss the probable future situation that may need a treatment or the dosage. Do discuss residual protection of a product already applied, using its persistence and application date, and say when the minimum interval since that treatment means a new one cannot be applied yet. diff --git a/tests/test_advice_summary.py b/tests/test_advice_summary.py index ebeca43..9d7b56a 100644 --- a/tests/test_advice_summary.py +++ b/tests/test_advice_summary.py @@ -30,10 +30,10 @@ def _advice_json(*, summary: str, full_advice: str = "Testo completo.") -> str: { "full_advice": full_advice, "advice_summary": summary, - "apply_treatment": 1, - "treatments": ["PROFILER"], - "dosage": ["3.0 kg/ha"], - "apply_date": "2026-04-14", + "is_treatment_recommended": 1, + "recommended_product_names": ["PROFILER"], + "recommended_doses": ["3.0 kg/ha"], + "recommended_application_date": "2026-04-14", }, ensure_ascii=False, ) diff --git a/tests/test_assemble.py b/tests/test_assemble.py index 03ed16d..347f689 100644 --- a/tests/test_assemble.py +++ b/tests/test_assemble.py @@ -33,6 +33,16 @@ class AdviceGenerationPhenologyTests(unittest.TestCase): ) self.assertEqual(payload["date_of_today"], format_date(_AS_OF)) + self.assertEqual( + list(payload), + [ + "date_of_today", + "predicted_incubations_management", + "meteorological_data", + "weather_forecasts", + ], + ) + self.assertEqual(payload["predicted_incubations_management"], {}) self.assertEqual(len(payload["meteorological_data"]), 6) self.assertEqual(len(payload["weather_forecasts"]), 5) @@ -48,6 +58,31 @@ class AdviceGenerationPhenologyTests(unittest.TestCase): first_forecast = payload["weather_forecasts"][0] self.assertEqual(first_forecast["date"], format_date(_AS_OF + timedelta(days=1))) + def test_register_precedes_weather_and_forecast_is_a_list(self) -> None: + day = _AS_OF - timedelta(days=1) + payload = assemble_json_for_advice_generation( + _AS_OF, + weather={}, + phenology={}, + forecasts={day: ["INCUBAZPRIMARIA2_80.7", "FASE5"]}, + incubation_register={ + "incubazioneprimaria2": { + "start_date": "15-04-2026", + "infection_covered": False, + } + }, + ) + self.assertEqual( + list(payload)[:3], + ["date_of_today", "predicted_incubations_management", "meteorological_data"], + ) + self.assertEqual( + payload["predicted_incubations_management"]["incubazioneprimaria2"]["infection_covered"], + False, + ) + match = next(entry for entry in payload["meteorological_data"] if entry["date"] == format_date(day)) + self.assertEqual(match["disease_forecast"], ["INCUBAZPRIMARIA2_80.7", "FASE5"]) + class ForecastWindSpeedTests(unittest.TestCase): def test_wind_on_today_and_future_not_past(self) -> None: diff --git a/tests/test_disease.py b/tests/test_disease.py index 17f587f..179a961 100644 --- a/tests/test_disease.py +++ b/tests/test_disease.py @@ -5,7 +5,12 @@ from __future__ import annotations import unittest from decimal import Decimal -from pipeline.stages.disease import _format_model_value, _incub_label +from pipeline.stages.disease import ( + _DayRecords, + _format_model_value, + _incub_label, + _resolve_day_fase, +) class FormatModelValueTests(unittest.TestCase): @@ -31,13 +36,89 @@ class FormatModelValueTests(unittest.TestCase): class IncubLabelTests(unittest.TestCase): def test_picks_max_raw_value_then_rounds_suffix(self) -> None: self.assertEqual( - _incub_label([40.12, 40.129, 39.99]), + _incub_label([ + ("INCUBAZPRIMARIA", 40.12), + ("INCUBAZPRIMARIA", 40.129), + ("INCUBAZPRIMARIA", 39.99), + ]), "INCUBAZPRIMARIA_40.13", ) + def test_keeps_index_of_winning_row(self) -> None: + self.assertEqual( + _incub_label([("INCUBAZPRIMARIA2", 80.7)]), + "INCUBAZPRIMARIA2_80.7", + ) + + def test_higher_percentage_beats_higher_index(self) -> None: + self.assertEqual( + _incub_label([ + ("INCUBAZPRIMARIA2", 40), + ("INCUBAZPRIMARIA1", 90), + ]), + "INCUBAZPRIMARIA1_90", + ) + + def test_tie_keeps_higher_index(self) -> None: + self.assertEqual( + _incub_label([ + ("INCUBAZPRIMARIA1", 80.7), + ("INCUBAZPRIMARIA2", 80.7), + ]), + "INCUBAZPRIMARIA2_80.7", + ) + + def test_missing_index_loses_tie(self) -> None: + self.assertEqual( + _incub_label([ + ("INCUBAZPRIMARIA", 40), + ("INCUBAZPRIMARIA2", 40), + ]), + "INCUBAZPRIMARIA2_40", + ) + + def test_space_or_underscore_is_not_an_index(self) -> None: + self.assertEqual( + _incub_label([("INCUBAZPRIMARIA 2", 40), ("INCUBAZPRIMARIA_2", 30)]), + "INCUBAZPRIMARIA_40", + ) + def test_empty_or_non_numeric_falls_back(self) -> None: self.assertEqual(_incub_label([]), "INCUBAZPRIMARIA") - self.assertEqual(_incub_label([None, "n/a"]), "INCUBAZPRIMARIA") + self.assertEqual( + _incub_label([("INCUBAZPRIMARIA", None), ("INCUBAZPRIMARIA2", "n/a")]), + "INCUBAZPRIMARIA", + ) + + def test_day_keeps_each_index_and_fase5(self) -> None: + records = _DayRecords( + descriptions={"INCUBAZPRIMARIA1", "INCUBAZPRIMARIA2", "FASE5", "FASE2"}, + incub_rows=[ + ("INCUBAZPRIMARIA1", 90), + ("INCUBAZPRIMARIA2", 40), + ], + ) + self.assertEqual( + _resolve_day_fase(records), + ["INCUBAZPRIMARIA1_90", "INCUBAZPRIMARIA2_40", "FASE5"], + ) + + def test_incubation_only_keeps_highest_value_per_index(self) -> None: + records = _DayRecords( + descriptions={"INCUBAZPRIMARIA2"}, + incub_rows=[ + ("INCUBAZPRIMARIA2", 40), + ("INCUBAZPRIMARIA2", 80.7), + ], + ) + self.assertEqual(_resolve_day_fase(records), ["INCUBAZPRIMARIA2_80.7"]) + + def test_fase2_without_incubation(self) -> None: + records = _DayRecords(descriptions={"FASE2"}) + self.assertEqual(_resolve_day_fase(records), ["FASE2"]) + + def test_empty_day_is_none(self) -> None: + self.assertIsNone(_resolve_day_fase(_DayRecords())) if __name__ == "__main__": diff --git a/tests/test_incubations.py b/tests/test_incubations.py new file mode 100644 index 0000000..935dc80 --- /dev/null +++ b/tests/test_incubations.py @@ -0,0 +1,260 @@ +"""Register lifecycle, coverage, risk, and the incubation prefilter rule.""" + +from __future__ import annotations + +import unittest +from datetime import date, timedelta +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from pipeline.stages.disease import select_prefilter_rule +from pipeline.stages.incubations import ( + CoveringTreatment, + IncubationRecord, + first_future_disease_date, + forecast_has_risk, + incubation_start_covered, + parse_register, + reconcile_register, +) +from pipeline.stages.products import ProductIndex, ProductRow +from pipeline.stages.treatments import TreatmentHistory + +_AS_OF = date(2026, 4, 16) +_PAST = _AS_OF - timedelta(days=2) +_TODAY = _AS_OF +_FUTURE = _AS_OF + timedelta(days=3) +_FAR = _AS_OF + timedelta(days=4) + + +def _record( + day: date, + covered: bool, + spray: CoveringTreatment | None = None, +) -> IncubationRecord: + return IncubationRecord( + start_date=day, + infection_covered=covered, + covering_treatment=spray, + ) + + +def _spray( + day: date, + *products: str, + interval: int = 14, +) -> CoveringTreatment: + names = products or ("LIETO",) + return CoveringTreatment( + spray_date=day, + products=names, + treatment_interval_min_days=interval, + ) + + +def _none(_start: date) -> None: + return None + + +def _forecasts(*days: date, label: str = "INCUBAZPRIMARIA2_80.7") -> dict[date, list[str] | None]: + return {day: [label] for day in days} + + +class ReconcileRegisterTests(unittest.TestCase): + def test_covered_row_without_spray_stays_true_when_lookup_misses(self) -> None: + previous = {"incubazioneprimaria2": _record(_FAR, True)} + result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, _none) + record = result["incubazioneprimaria2"] + self.assertEqual(record.start_date, _FUTURE) + self.assertTrue(record.infection_covered) + self.assertIsNone(record.covering_treatment) + + def test_covered_row_without_spray_stores_a_hit_and_stays_true(self) -> None: + found = _spray(_FUTURE - timedelta(days=3), "LIETO", interval=10) + previous = {"incubazioneprimaria2": _record(_FAR, True)} + result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, lambda _start: found) + record = result["incubazioneprimaria2"] + self.assertTrue(record.infection_covered) + self.assertEqual(record.covering_treatment, found) + + def test_uncovered_lookup_stores_the_spray(self) -> None: + found = _spray(_FUTURE - timedelta(days=2), interval=10) + previous = {"incubazioneprimaria2": _record(_FUTURE, False)} + result = reconcile_register(previous, _forecasts(_FAR), _AS_OF, lambda _start: found) + record = result["incubazioneprimaria2"] + self.assertEqual(record.start_date, _FAR) + self.assertTrue(record.infection_covered) + self.assertEqual(record.covering_treatment, found) + + def test_disappearance_deletes_the_entry(self) -> None: + previous = {"incubazioneprimaria2": _record(_FUTURE, False)} + self.assertEqual(reconcile_register(previous, {}, _AS_OF, _none), {}) + + def test_same_start_adopts_a_newer_spray(self) -> None: + stored = _spray(_PAST - timedelta(days=10), "OLD", interval=30) + newer = _spray(_PAST - timedelta(days=2), "LIETO", "TWINGO", interval=7) + previous = {"incubazioneprimaria2": _record(_PAST, True, stored)} + result = reconcile_register(previous, _forecasts(_PAST), _AS_OF, lambda _start: newer) + record = result["incubazioneprimaria2"] + self.assertTrue(record.infection_covered) + self.assertEqual(record.covering_treatment, newer) + self.assertEqual(record.covering_treatment.products, ("LIETO", "TWINGO")) + + def test_spray_on_the_start_date_does_not_replace_the_stored_spray(self) -> None: + stored = _spray(_PAST - timedelta(days=6), "OLD", interval=20) + on_start = _spray(_PAST, "LIETO", interval=10) + previous = {"incubazioneprimaria2": _record(_PAST, True, stored)} + result = reconcile_register(previous, _forecasts(_PAST), _AS_OF, lambda _start: on_start) + record = result["incubazioneprimaria2"] + self.assertTrue(record.infection_covered) + self.assertEqual(record.covering_treatment, stored) + + def test_earlier_start_after_the_spray_keeps_the_stored_spray(self) -> None: + stored = _spray(_AS_OF - timedelta(days=20), "OLD", interval=30) + previous = {"incubazioneprimaria2": _record(_FAR, True, stored)} + result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, _none) + record = result["incubazioneprimaria2"] + self.assertEqual(record.start_date, _FUTURE) + self.assertTrue(record.infection_covered) + self.assertEqual(record.covering_treatment, stored) + + def test_earlier_start_on_the_spray_date_uses_the_lookup(self) -> None: + stored = _spray(_FUTURE, "OLD", interval=30) + previous = {"incubazioneprimaria2": _record(_FAR, True, stored)} + result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, _none) + record = result["incubazioneprimaria2"] + self.assertEqual(record.start_date, _FUTURE) + self.assertFalse(record.infection_covered) + self.assertIsNone(record.covering_treatment) + + def test_later_start_past_the_stored_interval_uses_the_lookup(self) -> None: + stored = _spray(_FUTURE - timedelta(days=2), "OLD", interval=3) + previous = {"incubazioneprimaria2": _record(_FUTURE, True, stored)} + result = reconcile_register(previous, _forecasts(_FAR), _AS_OF, _none) + record = result["incubazioneprimaria2"] + self.assertEqual(record.start_date, _FAR) + self.assertFalse(record.infection_covered) + self.assertIsNone(record.covering_treatment) + + def test_later_date_check_reads_the_interval_from_the_register_json(self) -> None: + loaded = parse_register({ + "incubazioneprimaria2": { + "start_date": "19-04-2026", + "infection_covered": True, + "covering_treatment": { + "date": "17-04-2026", + "products": ["LIETO"], + "treatment_interval_min_days": 3, + }, + } + }) + result = reconcile_register(loaded, _forecasts(_FAR), _AS_OF, _none) + record = result["incubazioneprimaria2"] + self.assertEqual(record.start_date, _FAR) + self.assertFalse(record.infection_covered) + + def test_later_landed_run_can_become_covered(self) -> None: + found = _spray(_PAST - timedelta(days=4), interval=10) + previous = {"incubazioneprimaria2": _record(_PAST, False)} + result = reconcile_register(previous, _forecasts(_PAST), _AS_OF, lambda _start: found) + record = result["incubazioneprimaria2"] + self.assertEqual(record.start_date, _PAST) + self.assertTrue(record.infection_covered) + self.assertEqual(record.covering_treatment, found) + + +class CoverageTests(unittest.TestCase): + def _index(self) -> ProductIndex: + return ProductIndex([ + ProductRow( + product_id="p1", + product_name="LIETO", + registration_number=None, + is_organic_certified=False, + systemicity="systemic", + has_preventive_action=False, + has_curative_action=True, + has_eradicant_action=False, + target_crops=["grapevine"], + target_diseases=["downy mildew"], + active_ingredients=[], + frac_codes=[], + ) + ]) + + def _history(self, spray: date) -> TreatmentHistory: + return TreatmentHistory(as_of=_AS_OF, by_day={spray: ["LIETO"]}) + + def test_spray_on_the_start_date_does_not_cover(self) -> None: + with patch("pipeline.stages.incubations.fetch_all") as fetch: + covered = incubation_start_covered( + MagicMock(), + self._index(), + crop_english="grapevine", + disease_english="downy mildew", + start_date=_PAST, + treatment_history=self._history(_PAST), + ) + self.assertIsNone(covered) + fetch.assert_not_called() + + def test_spray_before_the_start_covers_when_interval_reaches(self) -> None: + spray = _PAST - timedelta(days=4) + uses = [SimpleNamespace(product_id="p1", treatment_interval_min_days=7)] + with patch("pipeline.stages.incubations.fetch_all", return_value=uses): + covered = incubation_start_covered( + MagicMock(), + self._index(), + crop_english="grapevine", + disease_english="downy mildew", + start_date=_PAST, + treatment_history=self._history(spray), + ) + self.assertIsNotNone(covered) + assert covered is not None + self.assertEqual(covered.spray_date, spray) + self.assertEqual(covered.products, ("LIETO",)) + self.assertEqual(covered.treatment_interval_min_days, 7) + + +class RiskAndPrefilterTests(unittest.TestCase): + def test_covered_only_is_not_risk(self) -> None: + forecasts = {_PAST: ["INCUBAZPRIMARIA2_80.7"]} + register = {"incubazioneprimaria2": _record(_PAST, True)} + self.assertFalse(forecast_has_risk(forecasts, register)) + + def test_uncovered_incubation_is_risk(self) -> None: + forecasts = {_FUTURE: ["INCUBAZPRIMARIA2_40"]} + register = {"incubazioneprimaria2": _record(_FUTURE, False)} + self.assertTrue(forecast_has_risk(forecasts, register)) + + def test_first_future_ignores_a_covered_incubation(self) -> None: + forecasts = { + _FUTURE: ["INCUBAZPRIMARIA2_40"], + _FAR: ["FASE5"], + } + register = {"incubazioneprimaria2": _record(_FUTURE, True)} + self.assertEqual(first_future_disease_date(forecasts, register, _AS_OF), _FAR) + + def test_uncovered_past_incubation_selects_incubazprimaria(self) -> None: + forecasts = {_PAST: ["INCUBAZPRIMARIA2_80.7", "FASE5"]} + rule = select_prefilter_rule(forecasts, {}, _AS_OF, {"incubazioneprimaria2": False}) + self.assertEqual(rule, "incubazprimaria") + + def test_incubation_only_today_does_not_select_incubazprimaria(self) -> None: + forecasts = {_TODAY: ["INCUBAZPRIMARIA2_80.7"]} + rule = select_prefilter_rule(forecasts, {}, _AS_OF, {"incubazioneprimaria2": False}) + self.assertIsNone(rule) + + def test_covered_past_incubation_falls_through_to_fase5_past(self) -> None: + forecasts = {_PAST: ["INCUBAZPRIMARIA2_80.7", "FASE5"]} + rule = select_prefilter_rule(forecasts, {}, _AS_OF, {"incubazioneprimaria2": True}) + self.assertEqual(rule, "fase5_past") + + def test_fase5_only_today_selects_fase5_future(self) -> None: + rule = select_prefilter_rule({_TODAY: ["FASE5"]}, {}, _AS_OF, {}) + self.assertEqual(rule, "fase5_future") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_products.py b/tests/test_products.py index fdc09c0..6361732 100644 --- a/tests/test_products.py +++ b/tests/test_products.py @@ -28,20 +28,24 @@ def _product( product_id: str = "p1", product_name: str = "LIETO", permitted: bool = True, + systemicity: str = "contact", + has_preventive_action: bool = True, + has_curative_action: bool = False, + has_eradicant_action: bool = False, ) -> ProductRow: return ProductRow( product_id=product_id, product_name=product_name, registration_number=None, - organic_certified=False, - systemicity="contact", - preventive_action=True, - curative_action=False, - eradicant_action=False, + is_organic_certified=False, + systemicity=systemicity, + has_preventive_action=has_preventive_action, + has_curative_action=has_curative_action, + has_eradicant_action=has_eradicant_action, target_crops=[_CROP], target_diseases=[_DISEASE], active_ingredients=["mancozeb"], - frac_groups=[], + frac_codes=[], permitted=permitted, ) @@ -222,20 +226,117 @@ class LegalStateTests(unittest.TestCase): self.assertIn(revoked, index.candidates_for(_CROP, _DISEASE)) +def _kept_ids(index: ProductIndex, rule: str | None) -> list[str]: + return [ + item.product_id + for item in prefilter_products( + index, + crop_english=_CROP, + disease_english=_DISEASE, + organic=False, + rule=rule, + ) + ] + + +class PrefilterSystemicityTests(unittest.TestCase): + def _index(self, **flags: bool) -> ProductIndex: + labels = ( + ("contact", "contact"), + ("translaminar", "translaminar"), + ("translaminar-case", "Translaminar"), + ("systemic", "systemic"), + ("locally-systemic", "locally systemic"), + ("locally-systemic-case", "Locally Systemic"), + ("mixed", "mixed"), + ) + return ProductIndex( + [ + _product(product_id=product_id, product_name=product_id, systemicity=label, **flags) + for product_id, label in labels + ] + ) + + def test_contact_rules_keep_translaminar_only(self) -> None: + index = self._index() + curative = _product( + product_id="translaminar-curative", + product_name="translaminar-curative", + systemicity="translaminar", + has_curative_action=True, + ) + index = ProductIndex([*index.candidates_for(_CROP, _DISEASE), curative]) + kept = ["contact", "translaminar", "translaminar-case"] + self.assertEqual(_kept_ids(index, "fase2_only"), kept) + self.assertEqual(_kept_ids(index, "fase5_future"), kept) + + def test_fase5_past_keeps_contact_family_and_mixed(self) -> None: + self.assertEqual( + _kept_ids(self._index(), "fase5_past"), + ["contact", "translaminar", "translaminar-case", "mixed"], + ) + + def test_observation_keeps_systemic_family_with_eradicant(self) -> None: + index = self._index(has_eradicant_action=True) + plain = _product( + product_id="locally-systemic-plain", + product_name="locally-systemic-plain", + systemicity="locally systemic", + ) + index = ProductIndex([*index.candidates_for(_CROP, _DISEASE), plain]) + self.assertEqual( + _kept_ids(index, "observation"), + ["systemic", "locally-systemic", "locally-systemic-case", "mixed"], + ) + + def test_incubazprimaria_keeps_systemic_family_without_eradicant(self) -> None: + index = self._index(has_curative_action=True) + eradicant = _product( + product_id="locally-systemic-eradicant", + product_name="locally-systemic-eradicant", + systemicity="Locally Systemic", + has_curative_action=True, + has_eradicant_action=True, + ) + index = ProductIndex([*index.candidates_for(_CROP, _DISEASE), eradicant]) + self.assertEqual( + _kept_ids(index, "incubazprimaria"), + ["systemic", "locally-systemic", "locally-systemic-case", "mixed"], + ) + + def test_fase5_both_keeps_mixed_only(self) -> None: + index = self._index(has_curative_action=True) + self.assertEqual(_kept_ids(index, "fase5_both"), ["mixed"]) + + def test_no_rule_keeps_every_systemicity(self) -> None: + self.assertEqual( + _kept_ids(self._index(), None), + [ + "contact", + "translaminar", + "translaminar-case", + "systemic", + "locally-systemic", + "locally-systemic-case", + "mixed", + ], + ) + + def _sql_product(**overrides: object) -> SimpleNamespace: fields: dict[str, object] = { "product_id": "p1", "product_name": "LIETO", "registration_number": "000001", - "organic_certified": 0, + "is_organic_certified": 0, "systemicity": "contact", - "preventive_action": 1, - "curative_action": 0, - "eradicant_action": 0, + "has_preventive_action": 1, + "has_curative_action": 0, + "has_eradicant_action": 0, "target_crops": '["grapevine"]', "target_diseases": '["downy mildew"]', "active_ingredients": '["mancozeb"]', - "frac_groups": "[]", + "frac_codes": "[]", } fields.update(overrides) return SimpleNamespace(**fields)