From ebc8ad26b626e2bde051736b05aa3fb6daf94bcd Mon Sep 17 00:00:00 2001 From: Arsham Mirehvandi <130353205+arshammi@users.noreply.github.com> Date: Mon, 19 Oct 2026 13:43:29 +0200 Subject: [PATCH] Refactor pipeline to support multiple advice-enabled models per field and disease - Updated the job structure to include `anmod_id`, allowing for multiple mathematical models to be processed for each field and disease combination. - Enhanced the batch processing logic to discover and execute jobs for `(field, disease, anmod_id)` instead of just `(field, disease)`. - Modified the output file naming convention to include `anmod_id`, ensuring unique output for each model. - Adjusted database interactions to accommodate the new primary key structure based on `pathogen_model_id`. - Updated relevant documentation in `README.md` to reflect changes in job discovery, execution, and output generation. - Enhanced unit tests to validate the new functionality and ensure correct behavior across various scenarios. --- README.md | 69 ++++++++++++--------- pipeline/__main__.py | 100 +++++++++++++++++++++--------- pipeline/batch.py | 21 ++++--- pipeline/job.py | 47 +++++++++----- pipeline/observability.py | 3 +- pipeline/output.py | 3 +- pipeline/stages/advice.py | 30 ++++----- pipeline/stages/advice_context.py | 23 ++++--- pipeline/stages/disease.py | 65 +++++++++++++------ pipeline/worklist.py | 7 ++- tests/test_batch.py | 47 ++++++++++++-- tests/test_db.py | 20 +++++- tests/test_disease.py | 94 ++++++++++++++++++++++++++++ 13 files changed, 390 insertions(+), 139 deletions(-) diff --git a/README.md b/README.md index d4629a1..543d179 100644 --- a/README.md +++ b/README.md @@ -8,13 +8,15 @@ a local Weaviate `ProductProfile` collection. It then enriches that JSON with full product label information and the previous advisory, asks a second LLM to write the farmer-facing advice, and stores the result in the `advisories` table. -Each morning's `batch` run discovers `(field, disease)` pairs from the database. -A disease can generate advice when `AI_agrosupport_an_pathogen_models.pat_mod_AI_model` -is 1; its name is `pat_mod_desc`. A field is included when +Each morning's `batch` run discovers `(field, disease, anmod_id)` jobs from the +database. A disease can generate advice when +`AI_agrosupport_an_pathogen_models.pat_mod_AI_model` is 1; its name is +`pat_mod_desc`. A field model is included when `AI_agrosupport_agro_models.anmod_AI_advice` is 1 and `anmod_model` matches that -name. The cultivated crop is read from the field layer. The run then executes -the sequence above for each pair — in parallel, with one field's failure never -blocking another's. +name. Several mathematical models for the same field and disease stay separate +jobs, one per `anmod_id`. The cultivated crop is read from the field layer. The +run then executes the sequence above for each job — in parallel, with one +field's failure never blocking another's. ## Prerequisites @@ -64,7 +66,11 @@ joined to `AI_agrosupport_an_colture` and mapped through [`vocab/crops.yaml`](vocab/crops.yaml) (`Vite` → `grapevine`). The English disease name is mapped through [`vocab/diseases.yaml`](vocab/diseases.yaml) (`PERONOSPORA` → `downy mildew`). Those two English names select the prompt -and the product filter. One agro-model row is one job. +and the product filter. One agro-model row is one job. Its `anmod_id` is +written to `advisories.pathogen_model_id`. The same id is `model_model` in +`AI_agrosupport_agro_model_tmp2`, and the forecast uses `model_description` +from that model only. A field with three advice-enabled models for one disease +produces three advisories. To enable another disease, set both flags in SQL. Add a `pat_mod_desc` mapping to `vocab/diseases.yaml` if it is not already there (`peronospora` already is), @@ -120,7 +126,7 @@ pair still falling back to `_default` at startup. ```powershell .\.venv\Scripts\Activate.ps1 -# Daily entry point: every field/disease pair enabled for advice in the database +# Daily entry point: every advice-enabled field, disease, and model in the database python -m pipeline batch # Historical / test dates (useful when the live window has no FASE2/FASE5/INCUBAZPRIMARIA/observation) @@ -135,8 +141,10 @@ python -m pipeline batch --skip-existing-advice # Build every job's JSON but skip LLM calls, vector search, and the DB write python -m pipeline batch --dry-run -# Single-field debugging (keeps the pre-scaling CLI behaviour) +# Single-field debugging. Without --anmod-id, every advice-enabled model for +# that field and disease is run. python -m pipeline one --field-id 4085 --disease PERONOSPORA --dry-run +python -m pipeline one --field-id 4085 --disease PERONOSPORA --anmod-id 11 --dry-run ``` `batch` accepts `--deadline HH:MM` and `--workers N` to override @@ -179,18 +187,19 @@ jitter, `retry.attempts` tries), applied automatically inside Each run writes to `output//`: -- `field____.json` — one file per job (see fields below) +- `field______model.json` — one file per job (see fields below) - `_run_report.json` — the operational artifact for the morning: totals by status, `jobs_started_after_deadline` (jobs that began after the 09:00 - target; they are still written), plus a per-job breakdown (`status`, - `duration_s`, `recommended_count`, `allowed_products_empty`, `message`, - and, on failure, `error_type` / `error_message`) + target; they are still written), plus a per-job breakdown (`field_id`, + `disease`, `anmod_id`, `status`, `duration_s`, `recommended_count`, + `allowed_products_empty`, `message`, and, on failure, `error_type` / + `error_message`) Per-job payload fields: | Field | Description | |---|---| -| `run` | Metadata (as_of, field, crop, disease, organic, station, …) | +| `run` | Metadata (as_of, field, crop, disease, anmod_id, organic, station, …) | | `json_for_advice_generation` | Full 11-day meteorological + forecast payload, plus `date_of_today`, `allowed_products`, `applied_treatment`, `last_advice`, and (when present) `last_applied_treatment`; this is what the advice LLM receives | | `json_for_query_synthesis` | Subset of the above, sent to the LLM for search query generation | | `generated_queries` | One or two natural-language search queries | @@ -247,16 +256,18 @@ The process exit code is `0` only if every job in the report is `ok`, `no_risk`, ## Database writes -Each non-dry-run job inserts one row into `advisories` for `(issuance_date, field_id, -disease)`. That triple is the table's primary key, so any existing row for -the same triple is deleted first — re-running a day replaces its advisory -instead of failing on a duplicate key. The delete-then-insert pair runs -inside one transaction (`pipeline/db.py::transaction`) so a crash between -the two statements cannot leave a field's advisory missing for the day, and -deadlocks are retried per `retry:` in config.yaml. +Each non-dry-run job inserts one row into `advisories` for `(issuance_date, +pathogen_model_id)`. `pathogen_model_id` stores that job's `anmod_id`. That pair +is the table's primary key, so any existing row for the same pair is deleted +first — re-running a day replaces that model's advisory and leaves the other +models for the same day in place. `field_id` and `disease` stay on the row: +`disease` is the English name that selects the prompt and the product filter. +The delete-then-insert pair runs inside one transaction (`pipeline/db.py::transaction`) +so a crash between the two statements cannot leave that model's advisory +missing for the day, and deadlocks are retried per `retry:` in config.yaml. -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 +The primary key's clustered index on `(issuance_date, pathogen_model_id)` also +lets that DELETE seek straight to the row it needs instead of scanning the table, so concurrent workers don't block each other on unrelated rows. ## Daily scheduling @@ -275,7 +286,7 @@ with working directory `d:\Uni\Thesis\AI-Agro-Support`. Jobs still waiting at late in `jobs_started_after_deadline`. `--skip-existing-advice` makes a re-trigger (e.g. a scheduled retry later that morning) resumable at no extra cost: it does one bulk check against `advisories` up front and only runs jobs -that don't have a row yet. +that don't have a row yet for that pathogen model. ## Observability (optional) @@ -329,7 +340,7 @@ collector cannot block a worker or push the run past `schedule.deadline`. ### What you get per job -Every `(field, crop, disease, as_of)` job produces one trace, rooted at a +Every `(field, crop, disease, anmod_id, as_of)` job produces one trace, rooted at a `run_job` span, with: - a `query_synthesis` span (gemini-3.8-flash by default) and an @@ -381,10 +392,10 @@ is not worth a custom cost pipeline at this scale. ## Pipeline stages -Run once per `(field, disease)` job by `pipeline/job.py::run_job`: +Run once per `(field, disease, anmod_id)` job by `pipeline/job.py::run_job`: 1. **Worklist** — supported diseases (`AI_agrosupport_an_pathogen_models.pat_mod_AI_model = 1`) → advice-enabled field models (`AI_agrosupport_agro_models.anmod_AI_advice = 1`) → crop from `AI_agrosupport_cmp_layers` + `AI_agrosupport_an_colture`; organic flag is `cmplay_imp == 3` (`pipeline/worklist.py`, run once per batch, not per job) -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. +2. **Disease model** — per day over as_of±5, from the worklist's known `anmod_id`. Rows come from `AI_agrosupport_agro_model_tmp2` where `model_model` equals that id. Only those rows' `model_description` values and `model_value` are used. `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` for the same `pathogen_model_id`, then written back with this run) records each index's `start_date` and `infection_covered`. Today counts as provisional: a start locks only once the index appears on a day before `as_of`. A covered incubation alone is not risk. 3. **Observation** — `AI_agrosupport_agro_ril_pathogen` (`rilpato_layer`/`rilpato_date`/`rilpato_diffusion`); `observation` is true for `as_of` or any of the past 5 days if a row exists there with `rilpato_diffusion` other than `32` (including `NULL`) 4. **Weather** — `TDatiMeteo_D` for past days, `TDatiMeteo_D_FRC` for today/future 5. **Phenology** — carry-forward of latest observation ≤ day; future days null; if today's phase is missing, use yesterday's @@ -406,6 +417,6 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: 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`), then reranked locally with `reranker-transformers` (`ms-marco-MiniLM-L-6-v2`) on `situation_fit_summary`. One synthesized query retrieves 10 and keeps the top 4 in rerank order; two queries retrieve 6 each and keep the top 2 unique each (a product already kept for the first query is skipped so the second can fill its two slots from the rest of its reranked six) 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 `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. +11. **Context enrichment** — `allowed_products`, `applied_treatment` expansion, `last_advice` (with issuance `date`) from the `advisories` row for the same `pathogen_model_id`, and optional `last_applied_treatment` (spray date + products) when the recent window is empty but a spray exists in `as_of-25 .. as_of-6`, including when `allowed_products` is empty. Enrichment reuses the job's 25-day treatment history; it does not query operations again. 12. **Advice generation** — second LLM profile returns the structured advisory -13. **Persistence** — the advisory plus its collapsed input are written to `advisories` inside one transaction +13. **Persistence** — the advisory plus its collapsed input are written to `advisories` with `pathogen_model_id` = `anmod_id`, inside one transaction diff --git a/pipeline/__main__.py b/pipeline/__main__.py index 8f5dcfb..364d71f 100644 --- a/pipeline/__main__.py +++ b/pipeline/__main__.py @@ -4,13 +4,14 @@ [--field-id N ...] [--deadline HH:MM] [--skip-existing-advice] [--dry-run] [--workers N] - python -m pipeline one --field-id N --disease NAME [--as-of DATE] [--dry-run] + python -m pipeline one --field-id N --disease NAME [--anmod-id N] + [--as-of DATE] [--dry-run] -`batch` is the daily entry point: it discovers every (field, disease) job -from AI_agrosupport_an_pathogen_models (pat_mod_AI_model = 1) and +`batch` is the daily entry point: it discovers every (field, disease, anmod_id) +job from AI_agrosupport_an_pathogen_models (pat_mod_AI_model = 1) and AI_agrosupport_agro_models (anmod_AI_advice = 1), and runs them through a -bounded thread pool (see pipeline/batch.py). `one` keeps the original -single-field behaviour for ad-hoc debugging. +bounded thread pool (see pipeline/batch.py). `one` runs every advice-enabled +model for one field and disease, or a single `--anmod-id`. """ from __future__ import annotations @@ -28,7 +29,7 @@ from pipeline.observability import setup_tracing, shutdown_tracing from pipeline.output import write_job_output from pipeline.prompts import PromptRegistry from pipeline.resources import Resources -from pipeline.stages.disease import find_anmod_id +from pipeline.stages.disease import find_advice_anmod_ids from pipeline.stages.field import resolve_field from pipeline.vocab import load_crop_vocab, load_disease_vocab from pipeline.worklist import Job @@ -77,39 +78,66 @@ def _run_one(args: argparse.Namespace) -> int: prompts = PromptRegistry(settings.prompts_dir) resources = Resources(settings, prompts) + exit_code = 0 try: conn = resources.sql_connection() field = resolve_field(conn, settings.field_id, crop_vocab) - anmod_id = find_anmod_id(conn, settings.field_id, settings.disease_name) + anmod_ids = find_advice_anmod_ids(conn, settings.field_id, settings.disease_name) + if args.anmod_id is not None: + if args.anmod_id not in anmod_ids: + raise NoModelConfiguredError( + f"anmod_id {args.anmod_id} is not an advice-enabled model " + f"for {settings.disease_name} on field {settings.field_id}." + ) + anmod_ids = [args.anmod_id] + logger.info( + "Field %s disease %s (%s): %d advice-enabled model(s) %s", + settings.field_id, + settings.disease_name, + disease_english, + len(anmod_ids), + anmod_ids, + ) resources.load_product_index() - job = Job( - field_id=settings.field_id, - crop_english=field.crop_english, - crop_italian=field.crop_italian, - model_name=settings.disease_name, - disease_english=disease_english, - organic=field.organic, - station=field.cmplay_station, - anmod_id=anmod_id, - ) - result = run_job(resources, job, args.as_of, dry_run=args.dry_run) + for anmod_id in anmod_ids: + job = Job( + field_id=settings.field_id, + crop_english=field.crop_english, + crop_italian=field.crop_italian, + model_name=settings.disease_name, + disease_english=disease_english, + organic=field.organic, + station=field.cmplay_station, + anmod_id=anmod_id, + ) + result = run_job(resources, job, args.as_of, dry_run=args.dry_run) + if result.payload is not None: + path = write_job_output(settings.output_dir, args.as_of, result.payload) + logger.info("Wrote output -> %s", path) + if result.status == "failed": + logger.error( + "anmod_id=%s %s: %s", + anmod_id, + result.error_type, + result.error_message, + ) + exit_code = 1 + else: + logger.info( + "Done: anmod_id=%s status=%s message=%s", + anmod_id, + result.status, + result.message, + ) finally: resources.close() - if result.payload is not None: - path = write_job_output(settings.output_dir, args.as_of, result.payload) - logger.info("Wrote output -> %s", path) - - if result.status == "failed": - logger.error("%s: %s", result.error_type, result.error_message) - return 1 - logger.info("Done: status=%s message=%s", result.status, result.message) - return 0 + return exit_code def _run_batch(args: argparse.Namespace) -> int: - """`python -m pipeline batch`: every database-enabled field/disease pair.""" + """`python -m pipeline batch`: every advice-enabled field, disease, and model.""" settings = load_settings() setup_tracing(settings) crop_vocab = load_crop_vocab(settings.crops_vocab) @@ -173,7 +201,9 @@ def main(argv: list[str] | None = None) -> int: ) batch_parser.add_argument( "--skip-existing-advice", action="store_true", - help="Skip (field, disease) pairs that already have a row in advisories for --as-of.", + help=( + "Skip jobs that already have an advisories row for this pathogen model and --as-of." + ), ) batch_parser.add_argument( "--dry-run", action="store_true", @@ -186,7 +216,10 @@ def main(argv: list[str] | None = None) -> int: one_parser = subparsers.add_parser( "one", - help="Run a single field/disease pair (debugging; the original single-field CLI behaviour).", + help=( + "Run one field and disease. Without --anmod-id, every advice-enabled " + "mathematical model for that pair is run." + ), ) one_parser.add_argument( "--as-of", type=_parse_as_of, default=None, @@ -199,6 +232,13 @@ def main(argv: list[str] | None = None) -> int: "--disease", default=None, help="Disease model name, e.g. PERONOSPORA (defaults to config.yaml's disease_name).", ) + one_parser.add_argument( + "--anmod-id", type=int, default=None, + help=( + "Run only this agro-model id. Default: every advice-enabled model " + "for the field and disease." + ), + ) one_parser.add_argument( "--dry-run", action="store_true", help="Build the JSON but skip LLM calls, vector search, and the advisories DB write.", diff --git a/pipeline/batch.py b/pipeline/batch.py index 43ba514..ae53cb5 100644 --- a/pipeline/batch.py +++ b/pipeline/batch.py @@ -1,7 +1,7 @@ """Batch orchestrator: discover jobs, run them across a bounded thread pool, and write one run report per day. -One `ThreadPoolExecutor` runs every discovered (field, disease) job so a slow +One `ThreadPoolExecutor` runs every discovered (field, disease, anmod_id) job so a slow crop cannot serialise the others; `pipeline.job.run_job` isolates each job's failures from the rest. This module owns everything above that: worklist discovery + CLI filters, the `--skip-existing-advice` resume check, the @@ -27,7 +27,9 @@ from pipeline.worklist import Job, build_jobs logger = logging.getLogger(__name__) -_EXISTING_ADVICE_SQL = "SELECT field_id, disease FROM {advisories} WHERE issuance_date = ?" +_EXISTING_ADVICE_SQL = ( + "SELECT pathogen_model_id FROM {advisories} WHERE issuance_date = ?" +) def _parse_deadline(deadline_str: str) -> datetime: @@ -90,7 +92,7 @@ def filter_jobs( return filtered -def _existing_advice_keys(resources: Resources, as_of: date) -> set[tuple[int, str]]: +def _existing_advice_keys(resources: Resources, as_of: date) -> set[int]: """One bulk SELECT instead of one per job, for --skip-existing-advice.""" conn = resources.sql_connection() rows = fetch_all( @@ -98,7 +100,7 @@ def _existing_advice_keys(resources: Resources, as_of: date) -> set[tuple[int, s qualify_app_sql(resources.settings.sql, _EXISTING_ADVICE_SQL), (as_of,), ) - return {(int(row.field_id), str(row.disease)) for row in rows} + return {int(row.pathogen_model_id) for row in rows} def run_batch( @@ -151,11 +153,11 @@ def run_batch( resources.load_product_index() - existing: set[tuple[int, str]] = set() + existing: set[int] = set() if skip_existing_advice: existing = _existing_advice_keys(resources, as_of) logger.info( - "%d field/disease pair(s) already have advice for %s and will be skipped", + "%d model row(s) already have advice for %s and will be skipped", len(existing), as_of.isoformat(), ) @@ -195,7 +197,7 @@ def run_batch( deadline_str, ) - if (job.field_id, job.disease_english) in existing: + if job.anmod_id in existing: return JobResult(job=job, status="skipped_existing") if started_late: @@ -206,7 +208,7 @@ def run_batch( # Worker threads start with an empty OpenTelemetry context # (`contextvars` do not cross `ThreadPoolExecutor.submit`). That is # intentional: `job_span` opens inside the worker and becomes a root - # span, one independent trace per (field, crop, disease, as_of). Do + # span, one independent trace per (field, crop, disease, anmod_id, as_of). Do # not wrap this submit in a tracing context manager from this thread # — it would not reach the workers. See pipeline/observability.py. workers = workers_override or settings.concurrency.workers @@ -219,10 +221,11 @@ def run_batch( result = future.result() except BaseException as exc: # noqa: BLE001 - defense in depth; run_job already isolates failures logger.exception( - "Unexpected error scheduling job field=%s crop=%s disease=%s", + "Unexpected error scheduling job field=%s crop=%s disease=%s anmod_id=%s", job.field_id, job.crop_english, job.disease_english, + job.anmod_id, ) result = JobResult( job=job, diff --git a/pipeline/job.py b/pipeline/job.py index 7417b24..e71400c 100644 --- a/pipeline/job.py +++ b/pipeline/job.py @@ -1,4 +1,4 @@ -"""Run the full single-field pipeline for one (field, crop, disease) job. +"""Run the full pipeline for one (field, crop, disease, anmod_id) job. This is the per-job unit of work the batch orchestrator (`pipeline.batch`) dispatches to its thread pool. It is the same sequence the original @@ -88,8 +88,11 @@ def json_default(obj: Any) -> Any: def output_stem(job: Job, disease_english: str) -> str: - """File name stem (no extension) for this job's per-field output JSON.""" - return f"field{job.field_id}__{pair_slug(job.crop_english, disease_english)}" + """File name stem (no extension). Includes anmod_id so each mathematical model has its own file.""" + return ( + f"field{job.field_id}__{pair_slug(job.crop_english, disease_english)}" + f"__model{job.anmod_id}" + ) def run_job(resources: Resources, job: Job, as_of: date, dry_run: bool = False) -> JobResult: @@ -97,7 +100,7 @@ def run_job(resources: Resources, job: Job, as_of: date, dry_run: bool = False) Run the full pipeline for one job. Wrapped in `job_span`, the one root span per `(field, crop, disease, - as_of)` job (a no-op when observability is disabled/unreachable -- see + anmod_id, as_of)` job (a no-op when observability is disabled/unreachable -- see `pipeline/observability.py`). The span records the final `status` (and, on failure, the error type) but never changes the fault-isolation behaviour below: a failure always comes back as a `failed` JobResult, @@ -109,10 +112,11 @@ def run_job(resources: Resources, job: Job, as_of: date, dry_run: bool = False) payload = _run_job_inner(resources, job, as_of, dry_run) except BaseException as exc: # noqa: BLE001 - fault isolation is the whole point of this wrapper logger.exception( - "Job failed: field=%s crop=%s disease=%s", + "Job failed: field=%s crop=%s disease=%s anmod_id=%s", job.field_id, job.crop_english, job.disease_english, + job.anmod_id, ) span.set_attribute("job.status", "failed") span.set_attribute("job.error_type", type(exc).__name__) @@ -163,8 +167,8 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - if disease.station is None: raise PipelineError( - f"Field {job.field_id}: unable to determine weather station " - "(model_station / cmplay_station both missing)." + f"Field {job.field_id} anmod_id={job.anmod_id}: unable to determine " + "weather station (model_station / cmplay_station both missing)." ) weather = load_weather(conn, disease.station, as_of) @@ -176,7 +180,12 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - 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, settings.sql + conn, + job.field_id, + disease.disease_english, + job.anmod_id, + as_of, + settings.sql, ) ) interval_open = last_treatment_interval_open( @@ -217,10 +226,11 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - ) logger.info( - "Field %s (%s/%s): station=%s has_risk=%s first_future=%s", + "Field %s (%s/%s) anmod_id=%s: station=%s has_risk=%s first_future=%s", job.field_id, job.crop_english, job.disease_english, + job.anmod_id, disease.station, has_risk, future_disease, @@ -286,9 +296,10 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - recommended: list[RecommendedProduct] = [] if has_risk and interval_open: logger.info( - "Field %s: last treatment interval still open; product search " + "Field %s anmod_id=%s: last treatment interval still open; product search " "skipped, advice generated without allowed products.", job.field_id, + job.anmod_id, ) elif has_risk: query_system_prompt = resources.prompts.query_prompt(job.crop_english, disease.disease_english) @@ -328,24 +339,31 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - for item in recommended ] logger.info( - "Field %s: recommended products: %s", + "Field %s anmod_id=%s: recommended products: %s", job.field_id, + job.anmod_id, [p["product_name"] for p in payload["recommended_products"]], ) else: payload["status"] = "no_risk" logger.warning( - "Field %s: no uncovered incubation, FASE2, or FASE5 in the 11-day " - "window, and no true observation in the past 5 days or today; " + "Field %s anmod_id=%s: no uncovered incubation, FASE2, or FASE5 in the " + "11-day window, and no true observation in the past 5 days or today; " "product search skipped, advice generated without allowed " "products.", job.field_id, + job.anmod_id, ) builder = resources.product_json_builder(job.crop_english, disease.disease_english) with resources.sql_sem: last_summary, last_treatments, last_date = load_last_advice( - conn, job.field_id, disease.disease_english, as_of, settings.sql + conn, + job.field_id, + disease.disease_english, + job.anmod_id, + as_of, + settings.sql, ) enrich_advice_json( json_for_advice_generation, @@ -379,6 +397,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - issuance_date=as_of, field_id=job.field_id, disease_english=disease.disease_english, + pathogen_model_id=job.anmod_id, advice_input_json=advice_input_json, result=result, sql=settings.sql, diff --git a/pipeline/observability.py b/pipeline/observability.py index c31e6d0..c0b923b 100644 --- a/pipeline/observability.py +++ b/pipeline/observability.py @@ -266,7 +266,7 @@ def _span(name: str, kind: str, metadata: dict[str, Any] | None = None) -> Itera def job_span(job: "Job", as_of: date, dry_run: bool) -> contextlib.AbstractContextManager: """ - One root CHAIN span per `(field, crop, disease, as_of)` job. + One root CHAIN span per `(field, crop, disease, anmod_id, as_of)` job. `run_job` sets the final `job.status` (and, on failure, `job.error_type`) on the yielded span itself before the `with` block exits; this function @@ -276,6 +276,7 @@ def job_span(job: "Job", as_of: date, dry_run: bool) -> contextlib.AbstractConte "field_id": job.field_id, "crop": job.crop_english, "disease": job.disease_english, + "anmod_id": job.anmod_id, "as_of": as_of.isoformat(), "organic": job.organic, "station": job.station, diff --git a/pipeline/output.py b/pipeline/output.py index 943e0bc..ef1875a 100644 --- a/pipeline/output.py +++ b/pipeline/output.py @@ -2,7 +2,7 @@ The single-field CLI wrote `output/_field.json`, which collides as soon as one field has two diseases. The batch instead writes one file per -(field, disease) pair under a per-day directory, plus a `_run_report.json` +(field, disease, anmod_id) job under a per-day directory, plus a `_run_report.json` that is the actual operational artifact: it says whether the morning's run succeeded, and if not, exactly which jobs failed and why. """ @@ -50,6 +50,7 @@ def _job_report_entry(result: JobResult) -> dict[str, Any]: "field_id": job.field_id, "crop": job.crop_english, "disease": job.disease_english, + "anmod_id": job.anmod_id, "organic": job.organic, "station": job.station, "status": result.status, diff --git a/pipeline/stages/advice.py b/pipeline/stages/advice.py index ee3cca5..e21e6e1 100644 --- a/pipeline/stages/advice.py +++ b/pipeline/stages/advice.py @@ -28,8 +28,7 @@ _RECOMMENDED_DOSES_MAX_CHARS = 100 _DELETE_ADVICE_SQL = """ DELETE FROM {advisories} WHERE issuance_date = ? - AND field_id = ? - AND disease = ? + AND pathogen_model_id = ? """ _INSERT_ADVICE_SQL = """ @@ -37,6 +36,7 @@ INSERT INTO {advisories} ( issuance_date, field_id, disease, + pathogen_model_id, advice_input_json, full_advice, advice_summary, @@ -44,7 +44,7 @@ INSERT INTO {advisories} ( recommended_product_names, recommended_doses, recommended_application_date -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ @@ -174,21 +174,21 @@ def insert_advice( issuance_date: date, field_id: int, disease_english: str, + pathogen_model_id: int, advice_input_json: dict[str, Any], result: AdviceResult, sql: SqlSettings, ) -> 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 and model. - The table's primary key is (issuance_date, field_id, disease), so an explicit - delete keeps re-runs of the same day from violating it instead of - accumulating duplicates. The - delete and insert run inside one transaction: with a single sequential - runner, autocommitting each statement separately was harmless, but the - batch pipeline writes from several worker threads at once, and a crash - between the two statements would otherwise leave that field's advisory - silently missing for the day. + The table's primary key is (issuance_date, pathogen_model_id). + pathogen_model_id is the agro-model anmod_id. An explicit delete keeps a + re-run of that model from violating the key, and leaves other models for + the same day in place. The delete and insert run inside one transaction: + the batch pipeline writes from several worker threads at once, and a crash + between the two statements would otherwise leave that model's advisory + missing for the day. """ recommended_product_names = _json_column( result.recommended_product_names, @@ -206,7 +206,7 @@ def insert_advice( execute( conn, qualify_app_sql(sql, _DELETE_ADVICE_SQL), - (issuance_date, field_id, disease_english), + (issuance_date, pathogen_model_id), ) execute( conn, @@ -215,6 +215,7 @@ def insert_advice( issuance_date, field_id, disease_english, + pathogen_model_id, sent, result.full_advice, result.advice_summary, @@ -225,9 +226,10 @@ def insert_advice( ), ) logger.info( - "Stored advice for field %s / %s on %s (is_treatment_recommended=%s).", + "Stored advice for field %s / %s / model %s on %s (is_treatment_recommended=%s).", field_id, disease_english, + pathogen_model_id, issuance_date.isoformat(), int(result.is_treatment_recommended), ) diff --git a/pipeline/stages/advice_context.py b/pipeline/stages/advice_context.py index 6a81b51..62c1351 100644 --- a/pipeline/stages/advice_context.py +++ b/pipeline/stages/advice_context.py @@ -28,8 +28,7 @@ SELECT TOP 1 advice_summary, recommended_product_names FROM {advisories} -WHERE field_id = ? - AND disease = ? +WHERE pathogen_model_id = ? AND issuance_date < ? ORDER BY issuance_date DESC """ @@ -48,8 +47,7 @@ def _row_date(value: Any) -> date | None: _LAST_SENT_SQL = """ SELECT TOP 1 advice_input_json FROM {advisories} -WHERE field_id = ? - AND disease = ? +WHERE pathogen_model_id = ? AND issuance_date < ? ORDER BY issuance_date DESC """ @@ -59,14 +57,15 @@ def load_previous_incubation_register( conn: pyodbc.Connection, field_id: int, disease_english: str, + pathogen_model_id: int, as_of: date, sql: SqlSettings, ) -> dict[str, Any]: - """Return predicted_incubations_management from the latest advice before as_of.""" + """Return predicted_incubations_management from this model's latest advice before as_of.""" row = fetch_one( conn, qualify_app_sql(sql, _LAST_SENT_SQL), - (field_id, disease_english, as_of), + (pathogen_model_id, as_of), ) if row is None or not row.advice_input_json: return {} @@ -74,9 +73,11 @@ def load_previous_incubation_register( 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.", + "Previous advice_input_json for field %s / %s / model %s is not JSON; " + "incubation register starts empty.", field_id, disease_english, + pathogen_model_id, ) return {} if not isinstance(payload, dict): @@ -91,20 +92,22 @@ def load_last_advice( conn: pyodbc.Connection, field_id: int, disease_english: str, + pathogen_model_id: int, as_of: date, sql: SqlSettings, ) -> tuple[str, list[str], date | None]: - """Return (advice_summary, treatment names, issuance date) of the most recent advice before as_of.""" + """Return (advice_summary, treatment names, issuance date) of this model's latest advice before as_of.""" row = fetch_one( conn, qualify_app_sql(sql, _LAST_ADVICE_SQL), - (field_id, disease_english, as_of), + (pathogen_model_id, as_of), ) if row is None: logger.info( - "No previous advice for field %s / %s before %s.", + "No previous advice for field %s / %s / model %s before %s.", field_id, disease_english, + pathogen_model_id, as_of.isoformat(), ) return "", [], None diff --git a/pipeline/stages/disease.py b/pipeline/stages/disease.py index 8d3688a..632f7e6 100644 --- a/pipeline/stages/disease.py +++ b/pipeline/stages/disease.py @@ -11,7 +11,7 @@ from typing import Any import pyodbc from pipeline.db import fetch_all -from pipeline.errors import NoModelConfiguredError +from pipeline.errors import NoModelConfiguredError, PipelineError from pipeline.window import build_window @@ -25,11 +25,12 @@ class DiseaseForecast: _MODEL_SQL = """ -SELECT anmod_id, anmod_model, anmod_enabled +SELECT anmod_id, anmod_model, anmod_AI_advice FROM AI_agrosupport_agro_models WHERE anmod_cmplay = ? """ +# model_model is anmod_id. model_description belongs to that mathematical model. _TMP2_SQL = """ SELECT CONVERT(date, model_date) AS model_day, @@ -153,25 +154,41 @@ def _resolve_day_fase(records: _DayRecords) -> list[str] | None: return labels or None -def find_anmod_id(conn: pyodbc.Connection, field_id: int, disease_name: str) -> int: - """ - Look up the anmod_id for a field's disease model by name. +def _advice_enabled(value: Any) -> bool: + if value is None or isinstance(value, bool): + return bool(value) + try: + return int(value) == 1 + except (TypeError, ValueError): + return str(value).strip() == "1" - Used by the single-field `one` CLI path, where `anmod_id` is not known - ahead of time. The batch worklist resolves it in one database join - (see `pipeline.worklist.build_jobs`), so it never calls this per field. + +def find_advice_anmod_ids( + conn: pyodbc.Connection, + field_id: int, + disease_name: str, +) -> list[int]: + """ + Every advice-enabled anmod_id for this field and disease name. + + A field can store several mathematical models for one disease. Each row + with anmod_AI_advice = 1 is its own id. The batch worklist resolves the + same rows in one join (see `pipeline.worklist.build_jobs`); `one` calls + this when the id is not already known. """ models = fetch_all(conn, _MODEL_SQL, (field_id,)) target = disease_name.strip().upper() - match = next( - (row for row in models if (row.anmod_model or "").strip().upper() == target), - None, - ) - if match is None: + ids = { + int(row.anmod_id) + for row in models + if (row.anmod_model or "").strip().upper() == target + and _advice_enabled(row.anmod_AI_advice) + } + if not ids: raise NoModelConfiguredError( - f"No disease forecast model is configured for {disease_name}." + f"No advice-enabled disease forecast model is configured for {disease_name}." ) - return int(match.anmod_id) + return sorted(ids) def resolve_disease_forecast( @@ -182,11 +199,16 @@ def resolve_disease_forecast( as_of: date, fallback_station: int | None, ) -> DiseaseForecast: - """Find the disease model for the field and build the 11-day forecast list.""" - anmod_id = find_anmod_id(conn, field_id, disease_name) + """Build the 11-day forecast when the field has one advice-enabled model.""" + ids = find_advice_anmod_ids(conn, field_id, disease_name) + if len(ids) != 1: + raise PipelineError( + f"Field {field_id} has {len(ids)} advice-enabled models for {disease_name}. " + "Call resolve_disease_forecast_by_anmod with the anmod_id to use." + ) return resolve_disease_forecast_by_anmod( conn, - anmod_id=anmod_id, + anmod_id=ids[0], disease_name=disease_name, disease_english=disease_english, as_of=as_of, @@ -202,7 +224,12 @@ def resolve_disease_forecast_by_anmod( as_of: date, fallback_station: int | None, ) -> DiseaseForecast: - """Build the 11-day forecast list for an already-known anmod_id.""" + """ + Build the 11-day forecast list for an already-known anmod_id. + + tmp2 rows are limited to model_model = anmod_id. The forecast is built + from those rows' model_description and model_value. + """ window = build_window(as_of) start, end = window[0], window[-1] rows = fetch_all(conn, _TMP2_SQL, (anmod_id, start, end)) diff --git a/pipeline/worklist.py b/pipeline/worklist.py index 10e278f..ac164f0 100644 --- a/pipeline/worklist.py +++ b/pipeline/worklist.py @@ -1,4 +1,4 @@ -"""Database-driven job discovery: find every (field, disease) pair to run today. +"""Database-driven job discovery: find every advice-enabled agro-model row to run today. The batch does not start from a field ID or from a crop list in config.yaml. It reads the diseases that can generate advice @@ -6,7 +6,8 @@ It reads the diseases that can generate advice `pat_mod_desc`) and the fields that have advice enabled for one of those diseases (`AI_agrosupport_agro_models.anmod_AI_advice = 1`). The cultivated crop comes from the field layer. Each matching agro-model row becomes one -`Job`. +`Job`, including several mathematical models (`anmod_id`) for the same field +and disease. """ from __future__ import annotations @@ -27,7 +28,7 @@ logger = logging.getLogger(__name__) @dataclass(frozen=True) class Job: - """One field/disease pair to run the pipeline for.""" + """One field, disease, and mathematical model (`anmod_id`) to run.""" field_id: int crop_english: str diff --git a/tests/test_batch.py b/tests/test_batch.py index ba5161e..2ddcd2b 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -21,7 +21,7 @@ from pipeline.config import ( WeaviateSettings, WorklistSettings, ) -from pipeline.job import JobResult +from pipeline.job import JobResult, output_stem from pipeline.output import write_run_report from pipeline.worklist import Job @@ -30,7 +30,7 @@ def _at(hour: int, minute: int = 0, second: int = 0) -> datetime: return datetime.combine(date.today(), dtime(hour, minute, second)) -def _job(field_id: int = 4012) -> Job: +def _job(field_id: int = 4012, anmod_id: int = 10) -> Job: return Job( field_id=field_id, crop_english="grapevine", @@ -39,7 +39,7 @@ def _job(field_id: int = 4012) -> Job: disease_english="downy mildew", organic=False, station=7, - anmod_id=10, + anmod_id=anmod_id, ) @@ -133,7 +133,8 @@ class RunBatchDeadlineTests(unittest.TestCase): *, now, skip_existing: bool = False, - existing: set[tuple[int, str]] | None = None, + existing: set[int] | None = None, + log_level: str = "WARNING", ) -> dict: captured: dict = {} prompts = MagicMock() @@ -157,7 +158,7 @@ class RunBatchDeadlineTests(unittest.TestCase): patch("pipeline.batch.write_run_report", side_effect=_write_report), patch("pipeline.batch._existing_advice_keys", return_value=existing or set()), patch("pipeline.batch.datetime") as frozen, - self.assertLogs("pipeline.batch", level="WARNING") as logs, + self.assertLogs("pipeline.batch", level=log_level) as logs, ): frozen.now.side_effect = now_fn frozen.combine = datetime.combine @@ -196,7 +197,7 @@ class RunBatchDeadlineTests(unittest.TestCase): [job], now=_at(10, 0), skip_existing=True, - existing={(job.field_id, job.disease_english)}, + existing={job.anmod_id}, ) self.assertEqual(captured["run_job"].call_count, 0) @@ -215,6 +216,39 @@ class RunBatchDeadlineTests(unittest.TestCase): self.assertEqual(sum("80%" in message for message in captured["logs"]), 1) self.assertFalse(any("has passed" in message for message in captured["logs"])) + def test_skip_existing_is_per_pathogen_model(self) -> None: + stored = _job(anmod_id=20) + other = _job(anmod_id=11) + captured = self._run( + [stored, other], + now=_at(8, 0), + skip_existing=True, + existing={stored.anmod_id}, + log_level="INFO", + ) + + self.assertEqual(captured["run_job"].call_count, 1) + self.assertEqual(captured["run_job"].call_args.args[1].anmod_id, 11) + self.assertEqual( + sorted(result.status for result in captured["results"]), + ["ok", "skipped_existing"], + ) + + +class OutputStemTests(unittest.TestCase): + def test_each_model_has_its_own_file_stem(self) -> None: + stems = { + output_stem(_job(anmod_id=11), "downy mildew"), + output_stem(_job(anmod_id=20), "downy mildew"), + } + self.assertEqual( + stems, + { + "field4012__grapevine__downy_mildew__model11", + "field4012__grapevine__downy_mildew__model20", + }, + ) + class RunReportTests(unittest.TestCase): def test_report_records_late_starts_without_skipped_deadline(self) -> None: @@ -229,6 +263,7 @@ class RunReportTests(unittest.TestCase): empty = write_run_report(Path(tmp), date(2026, 10, 14), []) self.assertEqual(report["jobs_started_after_deadline"], 2) + self.assertEqual(report["jobs"][0]["anmod_id"], 10) self.assertNotIn("skipped_deadline", report["totals"]) self.assertIn("skipped_existing", report["totals"]) self.assertEqual(empty["jobs_started_after_deadline"], 0) diff --git a/tests/test_db.py b/tests/test_db.py index e14e809..8873773 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -138,10 +138,14 @@ class QualifiedSqlTests(unittest.TestCase): def test_advice_reads_qualify_advisories(self) -> None: as_of = date(2026, 4, 16) with patch("pipeline.stages.advice_context.fetch_one", return_value=None) as fetch: - load_last_advice(MagicMock(), 4012, "downy mildew", as_of, self.split) - load_previous_incubation_register(MagicMock(), 4012, "downy mildew", as_of, self.split) + load_last_advice(MagicMock(), 4012, "downy mildew", 11, as_of, self.split) + load_previous_incubation_register( + MagicMock(), 4012, "downy mildew", 11, as_of, self.split + ) for call in fetch.call_args_list: self.assertIn("[Agrosupport].dbo.[advisories]", call.args[1]) + self.assertIn("pathogen_model_id = ?", call.args[1]) + self.assertEqual(call.args[2], (11, as_of)) def test_advice_write_qualifies_advisories(self) -> None: result = AdviceResult( @@ -161,13 +165,23 @@ class QualifiedSqlTests(unittest.TestCase): issuance_date=date(2026, 4, 16), field_id=4012, disease_english="downy mildew", + pathogen_model_id=11, advice_input_json={}, result=result, sql=self.split, ) - delete_sql, insert_sql = (call.args[1] for call in execute.call_args_list) + delete_call, insert_call = execute.call_args_list + delete_sql, insert_sql = delete_call.args[1], insert_call.args[1] self.assertIn("DELETE FROM [Agrosupport].dbo.[advisories]", delete_sql) + self.assertIn("AND pathogen_model_id = ?", delete_sql) + self.assertNotIn("field_id", delete_sql) + self.assertNotIn("disease", delete_sql) + self.assertEqual(delete_call.args[2], (date(2026, 4, 16), 11)) self.assertIn("INSERT INTO [Agrosupport].dbo.[advisories]", insert_sql) + self.assertIn("field_id", insert_sql) + self.assertIn("disease", insert_sql) + self.assertIn("pathogen_model_id", insert_sql) + self.assertEqual(insert_call.args[2][3], 11) def _called_name(node: ast.Call) -> str | None: diff --git a/tests/test_disease.py b/tests/test_disease.py index 179a961..ea0114a 100644 --- a/tests/test_disease.py +++ b/tests/test_disease.py @@ -3,13 +3,20 @@ from __future__ import annotations import unittest +from datetime import date from decimal import Decimal +from types import SimpleNamespace +from unittest.mock import patch +from pipeline.errors import NoModelConfiguredError, PipelineError from pipeline.stages.disease import ( _DayRecords, _format_model_value, _incub_label, _resolve_day_fase, + find_advice_anmod_ids, + resolve_disease_forecast, + resolve_disease_forecast_by_anmod, ) @@ -121,5 +128,92 @@ class IncubLabelTests(unittest.TestCase): self.assertIsNone(_resolve_day_fase(_DayRecords())) +def _model(anmod_id: int, name: str = "PERONOSPORA", advice: int = 1) -> SimpleNamespace: + return SimpleNamespace(anmod_id=anmod_id, anmod_model=name, anmod_AI_advice=advice) + + +def _tmp_row(day: date, description: str, value: object, station: int = 7) -> SimpleNamespace: + return SimpleNamespace( + model_day=day, + model_description=description, + model_value=value, + model_station=station, + ) + + +class AdviceModelLookupTests(unittest.TestCase): + def test_keeps_every_advice_enabled_id_for_the_disease(self) -> None: + rows = [ + _model(20), + _model(11, name="Peronospora"), + _model(12, advice=0), + _model(13, name="OIDIO"), + ] + with patch("pipeline.stages.disease.fetch_all", return_value=rows) as fetch: + ids = find_advice_anmod_ids(object(), 4012, "PERONOSPORA") + + self.assertEqual(ids, [11, 20]) + self.assertIn("anmod_AI_advice", fetch.call_args.args[1]) + + def test_disabled_only_match_is_not_configured(self) -> None: + with patch("pipeline.stages.disease.fetch_all", return_value=[_model(12, advice=0)]): + with self.assertRaises(NoModelConfiguredError): + find_advice_anmod_ids(object(), 4012, "PERONOSPORA") + + def test_forecast_reads_model_description_for_that_anmod_id(self) -> None: + as_of = date(2026, 6, 15) + with patch( + "pipeline.stages.disease.fetch_all", + return_value=[_tmp_row(as_of, "INCUBAZPRIMARIA2", 80.7)], + ) as fetch: + forecast = resolve_disease_forecast_by_anmod( + object(), + anmod_id=11, + disease_name="PERONOSPORA", + disease_english="downy mildew", + as_of=as_of, + fallback_station=None, + ) + + sql, params = fetch.call_args.args[1], fetch.call_args.args[2] + self.assertIn("WHERE model_model = ?", sql) + self.assertIn("model_description", sql) + self.assertEqual(params[0], 11) + self.assertEqual(forecast.anmod_id, 11) + self.assertEqual(forecast.forecasts_by_day[as_of], ["INCUBAZPRIMARIA2_80.7"]) + + def test_single_enabled_model_is_used_for_the_forecast(self) -> None: + as_of = date(2026, 6, 15) + with patch( + "pipeline.stages.disease.fetch_all", + side_effect=[[_model(11)], [_tmp_row(as_of, "FASE5", None, station=4)]], + ) as fetch: + forecast = resolve_disease_forecast( + object(), + 4012, + "PERONOSPORA", + "downy mildew", + as_of, + None, + ) + + self.assertEqual(forecast.anmod_id, 11) + self.assertEqual(forecast.forecasts_by_day[as_of], ["FASE5"]) + self.assertEqual(fetch.call_args_list[1].args[2][0], 11) + + def test_several_enabled_models_are_not_collapsed_to_the_first(self) -> None: + rows = [_model(11), _model(20)] + with patch("pipeline.stages.disease.fetch_all", return_value=rows): + with self.assertRaises(PipelineError): + resolve_disease_forecast( + object(), + 4012, + "PERONOSPORA", + "downy mildew", + date(2026, 6, 15), + None, + ) + + if __name__ == "__main__": unittest.main()