Arsham/AI_Agro_Support
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor advisory generation and update database schema

Browse Source 
- 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.
This commit is contained in:
Arsham Mirehvandi 2026-09-30 14:43:08 +02:00
parent ae53961fd2
commit d42fe18a52
20 changed files with 1312 additions and 264 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
README.md
View File

@ -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 Server, and retrieves up to four matching products from a local Weaviate
`ProductProfile` collection. It then enriches that JSON with full product label `ProductProfile` collection. It then enriches that JSON with full product label
information and the previous advisory, asks a second LLM to write the 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 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 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 | | `candidate_count` | Products remaining after the prefilter |
| `recommended_products` | Up to 4 items with `product_name` + `registration_number` | | `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 | | `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` | | `advice` | Parsed LLM output: `full_advice`, `advice_summary`, `is_treatment_recommended`, `recommended_product_names`, `recommended_doses`, `recommended_application_date` |
| `sent_information` | The advice payload with every embedded product label collapsed back to a product name; mirrors `advice.sent_information` | | `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 `advice` row was written | | `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) | | `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 | | `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 interval since the last treatment has not elapsed; it does not say that the
product list was empty. 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 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 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 `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 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 `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 ## 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 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 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 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 the two statements cannot leave a field's advisory missing for the day, and
deadlocks are retried per `retry:` in config.yaml. 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 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. 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` 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 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. that don't have a row yet.
## Observability (optional) ## 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`: 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) 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_<model_value>` (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`) 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 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 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 | | Rule | Predicates |
|---|---| |---|---|
| `observation` true on `as_of` or the past 5 days | `systemicity` in `{systemic, mixed}` and `eradicant_action = 1` | | `observation` true on `as_of` or the past 5 days | `systemicity` in `{systemic, locally systemic, mixed}` and `has_eradicant_action = 1` |
| `INCUBAZPRIMARIA` / `INCUBAZPRIMARIA_<value>` today or in the past 5 days | `systemicity` in `{systemic, mixed}` and `curative_action = 1` and `eradicant_action = 0` | | 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 == contact` and `preventive_action = 1` and `curative_action = 0` and `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` both past/today and future | `systemicity == mixed` and `preventive_action = 1` and `curative_action = 1` and `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` in the future only | same predicates as the `FASE2`-only rule | | `FASE5` on `as_of` or later 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` | | `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 | | 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. 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) 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)` 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 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

4
config.yaml
View File

@ -19,14 +19,14 @@ llm:
# Part_Two: the farmer-facing advisory generator. Receives the full # Part_Two: the farmer-facing advisory generator. Receives the full
# json_for_advice_generation payload (weather, phenology, model output, # json_for_advice_generation payload (weather, phenology, model output,
# applied treatments, product labels) and returns the structured advice # 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: advice_generation:
provider: gemini provider: gemini
openai_model: gpt-4o openai_model: gpt-4o
anthropic_model: claude-opus-4-5 anthropic_model: claude-opus-4-5
gemini_model: gemini-2.5-pro gemini_model: gemini-2.5-pro
max_tokens: 5120 max_tokens: 5120
thinking_budget: 11264 thinking_budget: 9216
temperature: 0.0 temperature: 0.0
# ── Crop-first worklist ──────────────────────────────────────────────────────── # ── Crop-first worklist ────────────────────────────────────────────────────────

6
pipeline/__main__.py
View File

@ -167,11 +167,11 @@ def main(argv: list[str] | None = None) -> int:
) )
batch_parser.add_argument( batch_parser.add_argument(
"--skip-existing-advice", action="store_true", "--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( batch_parser.add_argument(
"--dry-run", action="store_true", "--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( batch_parser.add_argument(
"--workers", type=int, default=None, help="Override concurrency.workers from config.yaml.", "--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( one_parser.add_argument(
"--dry-run", action="store_true", "--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) one_parser.set_defaults(func=_run_one)

2
pipeline/batch.py
View File

@ -27,7 +27,7 @@ from pipeline.worklist import Job, build_jobs
logger = logging.getLogger(__name__) 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: def _parse_deadline(deadline_str: str) -> datetime:

75
pipeline/job.py
View File

@ -33,12 +33,22 @@ from pipeline.stages.advice_context import (
collapse_product_json, collapse_product_json,
enrich_advice_json, enrich_advice_json,
load_last_advice, load_last_advice,
load_previous_incubation_register,
) )
from pipeline.stages.assemble import ( from pipeline.stages.assemble import (
assemble_json_for_advice_generation, assemble_json_for_advice_generation,
build_json_for_query_synthesis, build_json_for_query_synthesis,
) )
from pipeline.stages.disease import resolve_disease_forecast_by_anmod, select_prefilter_rule 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.llm import synthesize_queries
from pipeline.stages.observation import load_observation from pipeline.stages.observation import load_observation
from pipeline.stages.phenology import load_phenology 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 product_index = resources.product_index
if product_index is None: if product_index is None:
raise RuntimeError("load_product_index() must run before run_job().") 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( interval_open = last_treatment_interval_open(
conn, conn,
product_index, product_index,
@ -173,11 +188,31 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
as_of=as_of, as_of=as_of,
) )
# has_risk gates query synthesis and the product prefilter: it is true if def _covers(start: date):
# any day is FASE2 / FASE5 / INCUBAZPRIMARIA (disease.has_risk, already return incubation_start_covered(
# covers all three via forecasts_by_day) or the observation factor is conn,
# true for as_of or any of the past 5 days. product_index,
has_risk = disease.has_risk or any(observation_by_day.values()) 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( logger.info(
"Field %s (%s/%s): station=%s has_risk=%s first_future=%s", "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, job.disease_english,
disease.station, disease.station,
has_risk, has_risk,
disease.first_future_disease_date, future_disease,
) )
json_for_advice_generation = assemble_json_for_advice_generation( 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, disease.forecasts_by_day,
treatment_history.recent(), treatment_history.recent(),
observation_by_day, observation_by_day,
register_payload(incubation_register),
) )
json_for_query_synthesis = build_json_for_query_synthesis( json_for_query_synthesis = build_json_for_query_synthesis(
as_of, as_of,
@ -203,7 +239,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
disease.disease_english, disease.disease_english,
weather, weather,
phenology, phenology,
disease.first_future_disease_date, future_disease,
) )
payload: dict[str, Any] = { 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, "anmod_id": disease.anmod_id,
"has_risk": has_risk, "has_risk": has_risk,
"first_future_disease_date": ( "first_future_disease_date": (
disease.first_future_disease_date.isoformat() future_disease.isoformat() if future_disease else None
if disease.first_future_disease_date
else None
), ),
"prefilter_rule": None, "prefilter_rule": None,
"llm_provider": settings.llm_query_synthesis.provider, "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": [], "recommended_products": [],
"last_advice": None, "last_advice": None,
"advice": None, "advice": None,
"sent_information": None, "advice_input_json": None,
"advice_inserted": False, "advice_inserted": False,
"status": "ok", "status": "ok",
"message": None, "message": None,
@ -264,7 +298,12 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
) )
payload["generated_queries"] = queries 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 payload["run"]["prefilter_rule"] = prefilter_rule
candidates = prefilter_products( candidates = prefilter_products(
product_index, product_index,
@ -294,7 +333,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
else: else:
payload["status"] = "no_risk" payload["status"] = "no_risk"
logger.warning( 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; " "window, and no true observation in the past 5 days or today; "
"product search skipped, advice generated without allowed " "product search skipped, advice generated without allowed "
"products.", "products.",
@ -329,26 +368,26 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
as_of, as_of,
llm_call=resources.call_llm, 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: def _insert() -> None:
with resources.sql_sem: with resources.sql_sem:
insert_advice( insert_advice(
conn, conn,
run_date=as_of, issuance_date=as_of,
field_id=job.field_id, field_id=job.field_id,
disease_english=disease.disease_english, disease_english=disease.disease_english,
sent_information=sent_information, advice_input_json=advice_input_json,
result=result, result=result,
) )
call_with_sql_retry(settings.retry, _insert) call_with_sql_retry(settings.retry, _insert)
payload["advice"] = result.as_dict() payload["advice"] = result.as_dict()
payload["sent_information"] = sent_information payload["advice_input_json"] = advice_input_json
payload["advice_inserted"] = True payload["advice_inserted"] = True
payload["message"] = ( payload["message"] = (
f"Returned {len(recommended)} recommended product(s); " 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 return payload

114
pipeline/stages/advice.py
View File

@ -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 from __future__ import annotations
@ -19,30 +19,30 @@ from pipeline.stages.llm import call_llm
logger = logging.getLogger(__name__) 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 SUMMARY_MAX_CHARS = 1010
_TREATMENTS_MAX_CHARS = 200 _RECOMMENDED_PRODUCT_NAMES_MAX_CHARS = 200
_DOSAGE_MAX_CHARS = 100 _RECOMMENDED_DOSES_MAX_CHARS = 100
_DELETE_ADVICE_SQL = """ _DELETE_ADVICE_SQL = """
DELETE FROM advice DELETE FROM advisories
WHERE [date] = ? WHERE issuance_date = ?
AND field_id = ? AND field_id = ?
AND disease = ? AND disease = ?
""" """
_INSERT_ADVICE_SQL = """ _INSERT_ADVICE_SQL = """
INSERT INTO advice ( INSERT INTO advisories (
[date], issuance_date,
field_id, field_id,
disease, disease,
sent_information, advice_input_json,
full_advice, full_advice,
advice_summary, advice_summary,
apply_treatment, is_treatment_recommended,
treatments, recommended_product_names,
dosage, recommended_doses,
apply_date recommended_application_date
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""" """
@ -51,19 +51,23 @@ INSERT INTO advice (
class AdviceResult: class AdviceResult:
full_advice: str full_advice: str
advice_summary: str advice_summary: str
apply_treatment: bool is_treatment_recommended: bool
treatments: list[str] recommended_product_names: list[str]
dosage: list[str] recommended_doses: list[str]
apply_date: date | None recommended_application_date: date | None
def as_dict(self) -> dict[str, Any]: def as_dict(self) -> dict[str, Any]:
return { return {
"full_advice": self.full_advice, "full_advice": self.full_advice,
"advice_summary": self.advice_summary, "advice_summary": self.advice_summary,
"apply_treatment": int(self.apply_treatment), "is_treatment_recommended": int(self.is_treatment_recommended),
"treatments": self.treatments, "recommended_product_names": self.recommended_product_names,
"dosage": self.dosage, "recommended_doses": self.recommended_doses,
"apply_date": self.apply_date.isoformat() if self.apply_date else None, "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] summary = summary[:SUMMARY_MAX_CHARS]
treatments = _string_list(payload.get("treatments")) recommended_product_names = _string_list(payload.get("recommended_product_names"))
dosage = _string_list(payload.get("dosage")) recommended_doses = _string_list(payload.get("recommended_doses"))
apply_treatment = _to_bool(payload.get("apply_treatment")) is_treatment_recommended = _to_bool(payload.get("is_treatment_recommended"))
apply_date = _to_date(payload.get("apply_date")) recommended_application_date = _to_date(payload.get("recommended_application_date"))
if apply_treatment and not treatments: if is_treatment_recommended and not recommended_product_names:
logger.warning("apply_treatment is set but treatments is empty.") logger.warning("is_treatment_recommended is set but recommended_product_names is empty.")
if not apply_treatment and treatments: if not is_treatment_recommended and recommended_product_names:
logger.warning( logger.warning(
"apply_treatment is not set but %d treatment(s) were returned.", "is_treatment_recommended is not set but %d treatment(s) were returned.",
len(treatments), len(recommended_product_names),
) )
return AdviceResult( return AdviceResult(
full_advice=full_advice, full_advice=full_advice,
advice_summary=summary, advice_summary=summary,
apply_treatment=apply_treatment, is_treatment_recommended=is_treatment_recommended,
treatments=treatments, recommended_product_names=recommended_product_names,
dosage=dosage, recommended_doses=recommended_doses,
apply_date=apply_date, recommended_application_date=recommended_application_date,
) )
def insert_advice( def insert_advice(
conn: pyodbc.Connection, conn: pyodbc.Connection,
*, *,
run_date: date, issuance_date: date,
field_id: int, field_id: int,
disease_english: str, disease_english: str,
sent_information: dict[str, Any], advice_input_json: dict[str, Any],
result: AdviceResult, result: AdviceResult,
) -> None: ) -> None:
""" """
Store the advisory, replacing any earlier row for the same day/field/disease. 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 delete keeps re-runs of the same day from violating it instead of
accumulating duplicates. The accumulating duplicates. The
delete and insert run inside one transaction: with a single sequential 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 between the two statements would otherwise leave that field's advisory
silently missing for the day. silently missing for the day.
""" """
treatments = _json_column(result.treatments, "treatments", _TREATMENTS_MAX_CHARS) recommended_product_names = _json_column(
dosage = _json_column(result.dosage, "dosage", _DOSAGE_MAX_CHARS) result.recommended_product_names,
sent = json.dumps(sent_information, ensure_ascii=False, default=_json_default) "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): 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( execute(
conn, conn,
_INSERT_ADVICE_SQL, _INSERT_ADVICE_SQL,
( (
run_date, issuance_date,
field_id, field_id,
disease_english, disease_english,
sent, sent,
result.full_advice, result.full_advice,
result.advice_summary, result.advice_summary,
1 if result.apply_treatment else 0, 1 if result.is_treatment_recommended else 0,
treatments, recommended_product_names,
dosage, recommended_doses,
result.apply_date, result.recommended_application_date,
), ),
) )
logger.info( 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, field_id,
disease_english, disease_english,
run_date.isoformat(), issuance_date.isoformat(),
int(result.apply_treatment), 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) encoded = json.dumps(values, ensure_ascii=False)
if len(encoded) > max_chars: if len(encoded) > max_chars:
raise PipelineError( 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}" f"holds {max_chars}: {encoded}"
) )
return encoded return encoded
@ -281,5 +293,5 @@ def _to_date(value: Any) -> date | None:
return datetime.strptime(text, fmt).date() return datetime.strptime(text, fmt).date()
except ValueError: except ValueError:
continue 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 return None

62
pipeline/stages/advice_context.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import copy import copy
import json
import logging import logging
from datetime import date, datetime from datetime import date, datetime
from typing import Any, Iterable from typing import Any, Iterable
@ -22,14 +23,14 @@ _PRODUCT_LIST_KEYS = ("allowed_products", "applied_treatment", "suggested_produc
_LAST_ADVICE_SQL = """ _LAST_ADVICE_SQL = """
SELECT TOP 1 SELECT TOP 1
[date] AS issued_on, issuance_date,
advice_summary, advice_summary,
treatments recommended_product_names
FROM advice FROM advisories
WHERE field_id = ? WHERE field_id = ?
AND disease = ? AND disease = ?
AND [date] < ? AND issuance_date < ?
ORDER BY [date] DESC ORDER BY issuance_date DESC
""" """
@ -43,6 +44,43 @@ def _row_date(value: Any) -> date | None:
return 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( def load_last_advice(
conn: pyodbc.Connection, conn: pyodbc.Connection,
field_id: int, field_id: int,
@ -61,10 +99,14 @@ def load_last_advice(
return "", [], None return "", [], None
summary = (row.advice_summary or "").strip() summary = (row.advice_summary or "").strip()
treatments = [name.strip() for name in parse_json_list(row.treatments) if name.strip()] recommended_product_names = [
issued_on = _row_date(row.issued_on) name.strip()
logger.info("Loaded previous advice with %d suggested product(s).", len(treatments)) for name in parse_json_list(row.recommended_product_names)
return summary, treatments, issued_on 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: 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. 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. LLM received, minus the product label detail.
""" """
collapsed = copy.deepcopy(advice_json) collapsed = copy.deepcopy(advice_json)

8
pipeline/stages/assemble.py
View File

@ -50,7 +50,7 @@ def _day_entry(
as_of: date, as_of: date,
weather: dict[date, dict[str, Any]], weather: dict[date, dict[str, Any]],
phenology: dict[date, str | None], phenology: dict[date, str | None],
forecasts: dict[date, str | None], forecasts: dict[date, list[str] | None],
applied_treatments: dict[date, list[str]], applied_treatments: dict[date, list[str]],
observation: dict[date, bool] | None = None, observation: dict[date, bool] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
@ -70,11 +70,12 @@ def assemble_json_for_advice_generation(
as_of: date, as_of: date,
weather: dict[date, dict[str, Any]], weather: dict[date, dict[str, Any]],
phenology: dict[date, str | None], phenology: dict[date, str | None],
forecasts: dict[date, str | None], forecasts: dict[date, list[str] | None],
applied_treatments: dict[date, list[str]] | None = None, applied_treatments: dict[date, list[str]] | None = None,
observation: dict[date, bool] | None = None, observation: dict[date, bool] | None = None,
incubation_register: dict[str, Any] | None = None,
) -> dict[str, Any]: ) -> 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) window = build_window(as_of)
applied = applied_treatments or {} applied = applied_treatments or {}
meteorological = [ meteorological = [
@ -89,6 +90,7 @@ def assemble_json_for_advice_generation(
] ]
return { return {
"date_of_today": format_date(as_of), "date_of_today": format_date(as_of),
"predicted_incubations_management": dict(incubation_register or {}),
"meteorological_data": meteorological, "meteorological_data": meteorological,
"weather_forecasts": forecasts_section, "weather_forecasts": forecasts_section,
} }

205
pipeline/stages/disease.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import date from datetime import date
from decimal import Decimal from decimal import Decimal
@ -20,9 +21,7 @@ class DiseaseForecast:
disease_raw: str disease_raw: str
disease_english: str disease_english: str
station: int | None station: int | None
forecasts_by_day: dict[date, str | None] forecasts_by_day: dict[date, list[str] | None]
first_future_disease_date: date | None
has_risk: bool
_MODEL_SQL = """ _MODEL_SQL = """
@ -47,7 +46,10 @@ WHERE model_model = ?
@dataclass @dataclass
class _DayRecords: class _DayRecords:
descriptions: set[str] = field(default_factory=set) 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: def _is_incubazprimaria(label: str | None) -> bool:
@ -69,41 +71,86 @@ def _format_model_value(value: Any) -> str | None:
return text or None return text or None
def _incub_label(values: list[Any]) -> str: def _incub_index(description: str) -> str | None:
"""INCUBAZPRIMARIA_<highest model_value>, or bare INCUBAZPRIMARIA if none.""" """Digits glued to INCUBAZPRIMARIA. A space or underscore means no index."""
numerics: list[tuple[float, Any]] = [] match = _INCUB_INDEX.search(description)
for value in values: 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}_<value>."""
best_numeric: float | None = None
best_raw: Any = None
for _description, value in rows:
if value is None: if value is None:
continue continue
try: try:
numerics.append((float(value), value)) numeric = float(value)
except (TypeError, ValueError): except (TypeError, ValueError):
continue continue
if numerics: if best_numeric is None or numeric > best_numeric:
best = max(numerics, key=lambda item: item[0])[1] best_numeric = numeric
formatted = _format_model_value(best) best_raw = value
if formatted: 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}" return f"INCUBAZPRIMARIA_{formatted}"
return "INCUBAZPRIMARIA"
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_<model_value> (e.g.
INCUBAZPRIMARIA_40.3) from the tmp2 row. Several incubation rows on one def _incub_labels(rows: list[tuple[str, Any]]) -> list[str]:
day keep the highest numeric model_value. A missing model_value falls """One label per index, ascending, unindexed last. Each index keeps its own highest value."""
back to bare INCUBAZPRIMARIA. 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}_<model_value> 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} upper = {d.upper() for d in records.descriptions if d}
labels: list[str] = []
if any("INCUBAZPRIMARIA" in d for d in upper): 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: if "FASE5" in upper:
return "FASE5" labels.append("FASE5")
if "FASE2" in upper: elif "FASE2" in upper:
return "FASE2" labels.append("FASE2")
return None return labels or None
def find_anmod_id(conn: pyodbc.Connection, field_id: int, disease_name: str) -> int: 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: if description:
by_day[day].descriptions.add(description) by_day[day].descriptions.add(description)
if "INCUBAZPRIMARIA" in description.upper(): 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: if row.model_station is not None:
stations.append(int(row.model_station)) 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() 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 station = stations[0] if stations else fallback_station
has_risk = any(v is not None for v in forecasts.values())
return DiseaseForecast( return DiseaseForecast(
anmod_id=anmod_id, anmod_id=anmod_id,
@ -197,56 +237,97 @@ def resolve_disease_forecast_by_anmod(
disease_english=disease_english, disease_english=disease_english,
station=station, station=station,
forecasts_by_day=forecasts, 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( 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], observation_by_day: dict[date, bool],
as_of: date, as_of: date,
covered_by_key: dict[str, bool],
) -> str | None: ) -> 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 Observation still uses as_of and the past 5 days. For the disease model,
match wins: 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" 1. observation true on as_of or any of the past 5 days -> "observation"
2. INCUBAZPRIMARIA today or in the past 5 days -> "incubazprimaria" 2. an uncovered incubation on a past day -> "incubazprimaria"
3. FASE2 anywhere in the window AND no FASE5 anywhere -> "fase2_only" 3. FASE2 anywhere and no FASE5 anywhere -> "fase2_only"
4. FASE5 present both past/today AND future -> "fase5_both" 4. FASE5 in the past and on today or later -> "fase5_both"
5. FASE5 present in the future only -> "fase5_future" 5. FASE5 on today or later only -> "fase5_future"
6. FASE5 present today or in the past only -> "fase5_past" 6. FASE5 in the past only -> "fase5_past"
7. otherwise -> None 7. otherwise -> None
None means the product prefilter applies only the crop / disease / `covered_by_key` maps an incubation register key to infection_covered.
organic gates, with no FASE-specific catalog cut. 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] past = [day for day in forecasts_by_day if day < as_of]
future = [d for d in forecasts_by_day if d > 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" 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" return "incubazprimaria"
fase2_anywhere = any(v == "FASE2" for v in forecasts_by_day.values()) if _any_code(forecasts_by_day, "FASE2") and not _any_code(forecasts_by_day, "FASE5"):
fase5_anywhere = any(v == "FASE5" for v in forecasts_by_day.values())
if fase2_anywhere and not fase5_anywhere:
return "fase2_only" return "fase2_only"
fase5_past_today = any(forecasts_by_day.get(d) == "FASE5" for d in past_today) fase5_past = _any_code_on(forecasts_by_day, past, "FASE5")
fase5_future = any(forecasts_by_day.get(d) == "FASE5" for d in future) fase5_future = _any_code_on(forecasts_by_day, future, "FASE5")
if fase5_past and fase5_future:
if fase5_past_today and fase5_future:
return "fase5_both" return "fase5_both"
if fase5_future: if fase5_future:
return "fase5_future" return "fase5_future"
if fase5_past_today: if fase5_past:
return "fase5_past" return "fase5_past"
return None 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

365
pipeline/stages/incubations.py Normal file
View File

@ -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

6
pipeline/stages/product_json.py
View File

@ -34,7 +34,7 @@ FROM product_uses
WHERE product_id = ? WHERE product_id = ?
AND crop = ? AND crop = ?
AND disease = ? AND disease = ?
ORDER BY id ASC ORDER BY product_use_id ASC
""" """
_CHUNKS_SQL = """ _CHUNKS_SQL = """
@ -45,7 +45,7 @@ SELECT
target_diseases target_diseases
FROM label_chunks FROM label_chunks
WHERE product_id = ? WHERE product_id = ?
ORDER BY chunk_id ASC ORDER BY label_chunk_id ASC
""" """
_USE_KEYS = ( _USE_KEYS = (
@ -130,7 +130,7 @@ class ProductJsonBuilder:
return { return {
row.product_name: { row.product_name: {
"active_ingredients": row.active_ingredients, "active_ingredients": row.active_ingredients,
"frac_groups": row.frac_groups, "frac_groups": row.frac_codes,
"systemicity": row.systemicity, "systemicity": row.systemicity,
"uses": self._load_uses(conn, row.product_id), "uses": self._load_uses(conn, row.product_id),
"information_chunks": self._load_chunks(conn, row.product_id), "information_chunks": self._load_chunks(conn, row.product_id),

91
pipeline/stages/products.py
View File

@ -30,15 +30,15 @@ class ProductRow:
product_id: str product_id: str
product_name: str product_name: str
registration_number: str | None registration_number: str | None
organic_certified: bool is_organic_certified: bool
systemicity: str # stripped, original casing (used verbatim in the Product JSON payload) systemicity: str # stripped, original casing (used verbatim in the Product JSON payload)
preventive_action: bool has_preventive_action: bool
curative_action: bool has_curative_action: bool
eradicant_action: bool has_eradicant_action: bool
target_crops: list[str] target_crops: list[str]
target_diseases: list[str] target_diseases: list[str]
active_ingredients: 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 permitted: bool = True # catalog prodgov_state is a non-blank value other than Revocato
def as_candidate(self) -> ProductCandidate: def as_candidate(self) -> ProductCandidate:
@ -54,15 +54,15 @@ SELECT
product_id, product_id,
product_name, product_name,
registration_number, registration_number,
organic_certified, is_organic_certified,
systemicity, systemicity,
preventive_action, has_preventive_action,
curative_action, has_curative_action,
eradicant_action, has_eradicant_action,
target_crops, target_crops,
target_diseases, target_diseases,
active_ingredients, active_ingredients,
frac_groups frac_codes
FROM products FROM products
""" """
@ -199,15 +199,15 @@ class ProductIndex:
product_id=str(record.product_id), product_id=str(record.product_id),
product_name=str(record.product_name).strip(), product_name=str(record.product_name).strip(),
registration_number=registration_number, registration_number=registration_number,
organic_certified=bool(record.organic_certified), is_organic_certified=bool(record.is_organic_certified),
systemicity=(record.systemicity or "").strip(), systemicity=(record.systemicity or "").strip(),
preventive_action=bool(record.preventive_action), has_preventive_action=bool(record.has_preventive_action),
curative_action=bool(record.curative_action), has_curative_action=bool(record.has_curative_action),
eradicant_action=bool(record.eradicant_action), has_eradicant_action=bool(record.has_eradicant_action),
target_crops=parse_json_list(record.target_crops), target_crops=parse_json_list(record.target_crops),
target_diseases=parse_json_list(record.target_diseases), target_diseases=parse_json_list(record.target_diseases),
active_ingredients=parse_json_list(record.active_ingredients), 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, permitted=permitted,
) )
) )
@ -305,6 +305,12 @@ def last_treatment_interval_open(
return True 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( def prefilter_products(
index: ProductIndex, index: ProductIndex,
*, *,
@ -321,16 +327,19 @@ def prefilter_products(
matches the government catalog and prodgov_state is a non-blank value other matches the government catalog and prodgov_state is a non-blank value other
than Revocato. Revoked rows stay in the index for by_name. than Revocato. Revoked rows stay in the index for by_name.
observation -> systemicity in {systemic, mixed} AND eradicant_action = 1 observation -> systemicity in {systemic, locally systemic, mixed}
incubazprimaria -> systemicity in {systemic, mixed} AND curative_action = 1 AND has_eradicant_action = 1
AND eradicant_action = 0 incubazprimaria -> systemicity in {systemic, locally systemic, mixed}
fase2_only -> systemicity == contact AND preventive_action = 1 AND has_curative_action = 1 AND has_eradicant_action = 0
AND curative_action = 0 AND eradicant_action = 0 fase2_only -> systemicity in {contact, translaminar}
fase5_both -> systemicity == mixed AND preventive_action = 1 AND has_preventive_action = 1
AND curative_action = 1 AND eradicant_action = 0 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_future -> same predicates as fase2_only
fase5_past -> systemicity in {contact, mixed} AND preventive_action = 1 fase5_past -> systemicity in {contact, translaminar, mixed}
AND curative_action = 0 AND eradicant_action = 0 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 None -> no FASE-specific cut; legal-state / crop / disease / organic gates only
""" """
candidates: list[ProductCandidate] = [] candidates: list[ProductCandidate] = []
@ -338,49 +347,49 @@ def prefilter_products(
for row in index.candidates_for(crop_english, disease_english): for row in index.candidates_for(crop_english, disease_english):
if not row.permitted: if not row.permitted:
continue continue
if organic and not row.organic_certified: if organic and not row.is_organic_certified:
continue continue
systemicity = row.systemicity.lower() systemicity = row.systemicity.lower()
if rule == "observation": if rule == "observation":
if systemicity not in {"systemic", "mixed"}: if systemicity not in _SYSTEMIC_SYSTEMICITY | {"mixed"}:
continue continue
if not row.eradicant_action: if not row.has_eradicant_action:
continue continue
elif rule == "incubazprimaria": elif rule == "incubazprimaria":
if systemicity not in {"systemic", "mixed"}: if systemicity not in _SYSTEMIC_SYSTEMICITY | {"mixed"}:
continue continue
if not row.curative_action: if not row.has_curative_action:
continue continue
if row.eradicant_action: if row.has_eradicant_action:
continue continue
elif rule in ("fase2_only", "fase5_future"): elif rule in ("fase2_only", "fase5_future"):
if systemicity != "contact": if systemicity not in _CONTACT_SYSTEMICITY:
continue continue
if not row.preventive_action: if not row.has_preventive_action:
continue continue
if row.curative_action: if row.has_curative_action:
continue continue
if row.eradicant_action: if row.has_eradicant_action:
continue continue
elif rule == "fase5_both": elif rule == "fase5_both":
if systemicity != "mixed": if systemicity != "mixed":
continue continue
if not row.preventive_action: if not row.has_preventive_action:
continue continue
if not row.curative_action: if not row.has_curative_action:
continue continue
if row.eradicant_action: if row.has_eradicant_action:
continue continue
elif rule == "fase5_past": elif rule == "fase5_past":
if systemicity not in {"contact", "mixed"}: if systemicity not in _CONTACT_SYSTEMICITY | {"mixed"}:
continue continue
if not row.preventive_action: if not row.has_preventive_action:
continue continue
if row.curative_action: if row.has_curative_action:
continue continue
if row.eradicant_action: if row.has_eradicant_action:
continue continue
# rule is None: no FASE-specific cut, only the gates above apply. # rule is None: no FASE-specific cut, only the gates above apply.

66
prompts/_output_format.md
View File

@ -7,13 +7,13 @@ introduction, no commentary outside the JSON.
|---|---|---| |---|---|---|
| `full_advice` | string | Your complete advisory text, in Italian. | | `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`. | | `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. | | `is_treatment_recommended` | 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. | | `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. |
| `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. | | `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. |
| `apply_date` | string or null | Recommended application date as `YYYY-MM-DD`. `null` 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` If `allowed_products` is empty, you must not prescribe any product (`recommended_product_names`
must be `[]` and `apply_treatment` must be `0`), even if must be `[]` and `is_treatment_recommended` must be `0`), even if
`last_advice.suggested_products` names one. Do not mention that the list is `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 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 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 If several uses qualify, cite the one with the longest remaining minimum. If
`treatment_interval_max_days` is null, cite the minimum only. `treatment_interval_max_days` is null, cite the minimum only.
- No model risk: the 11-day window has no `FASE2`, `FASE5`, or `INCUBAZPRIMARIA`, - No model risk: the 11-day window has no `FASE2` and no `FASE5`, every
and no `observation` is true. Say a treatment is not necessary because the incubation in `predicted_incubations_management` has `infection_covered`
model does not indicate infection risk. 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 - Minimum interval still open: days elapsed are less than
`treatment_interval_min_days`. Say that, because of the minimum interval `treatment_interval_min_days`. Say that, because of the minimum interval
required since the last treatment, a new treatment cannot be applied yet. 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 - 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 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. 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 Follow the tone of these examples; do not copy them when the dates or the
persistence differ: 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.* - *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.* - *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 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 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 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` from `allowed_products`. Do not prescribe a product solely because `last_advice`
did. Otherwise recommend only products that appear in `allowed_products`. Never 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 sono a conoscenza" form, or *la fase fenologica segnalata per il 13 aprile
2026 è …*. 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`, `disease_forecast` codes are for you, not for the farmer. Never write `FASE2`,
`FASE5`, `FASE`, `INCUBAZPRIMARIA`, or `INCUBAZPRIMARIA_<n>` in `full_advice` or `FASE5`, `FASE`, `INCUBAZPRIMARIA`, `INCUBAZPRIMARIA_<n>`, or
`advice_summary`. Keep using those codes internally when reading the JSON. `INCUBAZPRIMARIA<index>_<n>` in `full_advice` or `advice_summary`. Keep using
When referring to those days, use Italian farmer language: those codes internally when reading the JSON. Digits glued to `INCUBAZPRIMARIA`
`FASE2` → germination (*germinazione*); `FASE5` → risk of infection (*rischio before the underscore are the incubation number; the number after the
di infezione*); `INCUBAZPRIMARIA` / `INCUBAZPRIMARIA_80.7` → primary incubation underscore is the percentage. When referring to those days, use Italian farmer
(*incubazione primaria*), and state the percentage in words (e.g. *all'80,7%*), language: `FASE2` → germination (*germinazione*); `FASE5` → risk of infection
never the JSON key. (*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 ### Example shape
{ {
"full_advice": "...", "full_advice": "...",
"advice_summary": "...", "advice_summary": "...",
"apply_treatment": 1, "is_treatment_recommended": 1,
"treatments": ["PRODUCT NAME"], "recommended_product_names": ["PRODUCT NAME"],
"dosage": ["1.5 l/ha"], "recommended_doses": ["1.5 l/ha"],
"apply_date": "2026-08-14" "recommended_application_date": "2026-08-14"
} }

4
prompts/advice/_default/system.md
View File

@ -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/<crop>__<disease>/` 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. NOTE FOR OPERATORS: this is the generic fallback advice prompt. It is used because no dedicated prompt exists yet under `prompts/advice/<crop>__<disease>/` 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.

15
prompts/advice/grapevine__downy_mildew/system.md
View File

@ -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. 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.
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. 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.
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. 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.
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%. 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.
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. 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. 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.

8
tests/test_advice_summary.py
View File

@ -30,10 +30,10 @@ def _advice_json(*, summary: str, full_advice: str = "Testo completo.") -> str:
{ {
"full_advice": full_advice, "full_advice": full_advice,
"advice_summary": summary, "advice_summary": summary,
"apply_treatment": 1, "is_treatment_recommended": 1,
"treatments": ["PROFILER"], "recommended_product_names": ["PROFILER"],
"dosage": ["3.0 kg/ha"], "recommended_doses": ["3.0 kg/ha"],
"apply_date": "2026-04-14", "recommended_application_date": "2026-04-14",
}, },
ensure_ascii=False, ensure_ascii=False,
) )

35
tests/test_assemble.py
View File

@ -33,6 +33,16 @@ class AdviceGenerationPhenologyTests(unittest.TestCase):
) )
self.assertEqual(payload["date_of_today"], format_date(_AS_OF)) 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["meteorological_data"]), 6)
self.assertEqual(len(payload["weather_forecasts"]), 5) self.assertEqual(len(payload["weather_forecasts"]), 5)
@ -48,6 +58,31 @@ class AdviceGenerationPhenologyTests(unittest.TestCase):
first_forecast = payload["weather_forecasts"][0] first_forecast = payload["weather_forecasts"][0]
self.assertEqual(first_forecast["date"], format_date(_AS_OF + timedelta(days=1))) 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): class ForecastWindSpeedTests(unittest.TestCase):
def test_wind_on_today_and_future_not_past(self) -> None: def test_wind_on_today_and_future_not_past(self) -> None:

87
tests/test_disease.py
View File

@ -5,7 +5,12 @@ from __future__ import annotations
import unittest import unittest
from decimal import Decimal 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): class FormatModelValueTests(unittest.TestCase):
@ -31,13 +36,89 @@ class FormatModelValueTests(unittest.TestCase):
class IncubLabelTests(unittest.TestCase): class IncubLabelTests(unittest.TestCase):
def test_picks_max_raw_value_then_rounds_suffix(self) -> None: def test_picks_max_raw_value_then_rounds_suffix(self) -> None:
self.assertEqual( self.assertEqual(
_incub_label([40.12, 40.129, 39.99]), _incub_label([
("INCUBAZPRIMARIA", 40.12),
("INCUBAZPRIMARIA", 40.129),
("INCUBAZPRIMARIA", 39.99),
]),
"INCUBAZPRIMARIA_40.13", "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: def test_empty_or_non_numeric_falls_back(self) -> None:
self.assertEqual(_incub_label([]), "INCUBAZPRIMARIA") 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__": if __name__ == "__main__":

260
tests/test_incubations.py Normal file
View File

@ -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()

123
tests/test_products.py
View File

@ -28,20 +28,24 @@ def _product(
product_id: str = "p1", product_id: str = "p1",
product_name: str = "LIETO", product_name: str = "LIETO",
permitted: bool = True, permitted: bool = True,
systemicity: str = "contact",
has_preventive_action: bool = True,
has_curative_action: bool = False,
has_eradicant_action: bool = False,
) -> ProductRow: ) -> ProductRow:
return ProductRow( return ProductRow(
product_id=product_id, product_id=product_id,
product_name=product_name, product_name=product_name,
registration_number=None, registration_number=None,
organic_certified=False, is_organic_certified=False,
systemicity="contact", systemicity=systemicity,
preventive_action=True, has_preventive_action=has_preventive_action,
curative_action=False, has_curative_action=has_curative_action,
eradicant_action=False, has_eradicant_action=has_eradicant_action,
target_crops=[_CROP], target_crops=[_CROP],
target_diseases=[_DISEASE], target_diseases=[_DISEASE],
active_ingredients=["mancozeb"], active_ingredients=["mancozeb"],
frac_groups=[], frac_codes=[],
permitted=permitted, permitted=permitted,
) )
@ -222,20 +226,117 @@ class LegalStateTests(unittest.TestCase):
self.assertIn(revoked, index.candidates_for(_CROP, _DISEASE)) 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: def _sql_product(**overrides: object) -> SimpleNamespace:
fields: dict[str, object] = { fields: dict[str, object] = {
"product_id": "p1", "product_id": "p1",
"product_name": "LIETO", "product_name": "LIETO",
"registration_number": "000001", "registration_number": "000001",
"organic_certified": 0, "is_organic_certified": 0,
"systemicity": "contact", "systemicity": "contact",
"preventive_action": 1, "has_preventive_action": 1,
"curative_action": 0, "has_curative_action": 0,
"eradicant_action": 0, "has_eradicant_action": 0,
"target_crops": '["grapevine"]', "target_crops": '["grapevine"]',
"target_diseases": '["downy mildew"]', "target_diseases": '["downy mildew"]',
"active_ingredients": '["mancozeb"]', "active_ingredients": '["mancozeb"]',
"frac_groups": "[]", "frac_codes": "[]",
} }
fields.update(overrides) fields.update(overrides)
return SimpleNamespace(**fields) return SimpleNamespace(**fields)