From f77c0abebdd30621956c90de72c7ce7eedda13f1 Mon Sep 17 00:00:00 2001 From: Arsham Mirehvandi <130353205+arshammi@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:57:45 +0200 Subject: [PATCH] Update database handling and configuration for advisory generation - Enhanced `.env.example` to include `SQL_APP_DATABASE` for managing additional advisory-related tables. - Refactored `config.yaml` to remove the `crops` block, shifting to a database-driven approach for job discovery. - Updated `README.md` to clarify the new database structure and job discovery process, emphasizing the role of the `AI_agrosupport_an_pathogen_models` and `AI_agrosupport_agro_models` tables. - Modified SQL queries across multiple files to accommodate the new database schema, ensuring correct table references and improving overall data handling. - Adjusted `pipeline` logic to utilize the new configuration, enhancing the efficiency of the advisory generation process. --- .env.example | 5 + README.md | 85 ++++++----- config.yaml | 33 ++--- pipeline/__main__.py | 13 +- pipeline/batch.py | 44 ++++-- pipeline/config.py | 42 +----- pipeline/db.py | 56 ++++++- pipeline/job.py | 7 +- pipeline/resources.py | 9 +- pipeline/stages/advice.py | 17 ++- pipeline/stages/advice_context.py | 21 ++- pipeline/stages/disease.py | 5 +- pipeline/stages/incubations.py | 8 +- pipeline/stages/product_json.py | 17 ++- pipeline/stages/products.py | 14 +- pipeline/worklist.py | 197 +++++++++++-------------- tests/test_db.py | 235 ++++++++++++++++++++++++++++++ tests/test_incubations.py | 11 ++ tests/test_products.py | 14 +- tests/test_worklist.py | 107 ++++++++++++++ 20 files changed, 678 insertions(+), 262 deletions(-) create mode 100644 tests/test_db.py create mode 100644 tests/test_worklist.py diff --git a/.env.example b/.env.example index 85020c6..6110c9d 100644 --- a/.env.example +++ b/.env.example @@ -12,8 +12,13 @@ WEAVIATE_GRPC_PORT=50051 # Microsoft SQL Server connection SQL_DRIVER=ODBC Driver 17 for SQL Server +# host, or host,port. ODBC uses a comma before the port, not a colon. SQL_SERVER=localhost +# Field, weather, and disease-model tables (AI_agrosupport_*). SQL_DATABASE=your-database-name +# advisories, products, product_uses, and label_chunks, on the same instance. +# Unset, or equal to SQL_DATABASE, when those four tables live in SQL_DATABASE. +SQL_APP_DATABASE=your-app-database-name SQL_USERNAME=your-sql-username SQL_PASSWORD=your-sql-password diff --git a/README.md b/README.md index 439c88f..5359913 100644 --- a/README.md +++ b/README.md @@ -1,23 +1,25 @@ # AI Agro Support — Daily Pipeline -Daily agronomic advisory pipeline. For every field growing a configured crop, it -builds an 11-day weather / phenology / disease-forecast JSON, asks an LLM to -synthesise one or two vector-search queries, prefilters chemical products in SQL -Server, and retrieves up to four matching products from a local Weaviate -`ProductProfile` collection. It then enriches that JSON with full product label -information and the previous advisory, asks a second LLM to write the -farmer-facing advice, and stores the result in the `advisories` table. +Daily agronomic advisory pipeline. For every field whose disease model is +enabled for advice, it builds an 11-day weather / phenology / disease-forecast +JSON, asks an LLM to synthesise one or two vector-search queries, prefilters +chemical products in SQL Server, and retrieves up to four matching products from +a local Weaviate `ProductProfile` collection. It then enriches that JSON with +full product label information and the previous advisory, asks a second LLM to +write the farmer-facing advice, and stores the result in the `advisories` table. -The pipeline is **crop-first**: `config.yaml` declares which crops and diseases -to check, and each morning's `batch` run discovers every field that grows one of -those crops and has a matching disease-forecast model configured, then runs the -full sequence above for each `(field, disease)` pair — in parallel, with one -field's failure never blocking another's. +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 +`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. ## Prerequisites - Python 3.12+ -- Microsoft SQL Server (`DBMeteo`) reachable with ODBC Driver 17 (or 18) +- Microsoft SQL Server (`DBMeteo`, and `SQL_APP_DATABASE` on the same instance) reachable with ODBC Driver 17 (or 18) - Weaviate running locally (see `docker-compose.yml`) with a populated `ProductProfile` collection - API key for the chosen LLM provider (Gemini by default) @@ -40,35 +42,40 @@ docker compose up -d ## Configuration -- [`config.yaml`](config.yaml) — the `crops:` worklist, concurrency/rate limits, retry policy, - the 09:00 SLA deadline, LLM provider/model, and vocab paths +- [`config.yaml`](config.yaml) — concurrency/rate limits, retry policy, the 09:00 SLA + deadline, LLM provider/model, vocab paths, and an optional field allowlist. + Advice targets come from the database (see below) - [`.env`](.env) — SQL Server, Weaviate, and API keys - [`vocab/crops.yaml`](vocab/crops.yaml) / [`vocab/diseases.yaml`](vocab/diseases.yaml) — Italian → English maps - [`prompts/`](prompts) — the prompt registry (see below) -### `crops:` worklist +### SQL databases -```yaml -crops: - - crop: grapevine # canonical English, matched via vocab/crops.yaml - diseases: - - model_name: PERONOSPORA # matched against AI_agrosupport_agro_models.anmod_model - disease: downy mildew # canonical English, used for products/advice/prompts -``` +The pipeline opens one connection to `SQL_DATABASE`. Field, weather, and disease-model tables (`AI_agrosupport_*`) stay in that database (`DBMeteo`). `advisories`, `products`, `product_uses`, and `label_chunks` are read through the same connection from `SQL_APP_DATABASE` when that names a different database on the same instance. Leave `SQL_APP_DATABASE` unset, or set it equal to `SQL_DATABASE`, when those four tables live in the connected database. `SQL_SERVER` is `host` or `host,port`; the ODBC driver uses a comma before the port. -Every morning, `batch` maps each `crop` to the fields growing it -(`AI_agrosupport_an_colture` → `AI_agrosupport_cmp_layers`), joins to -`AI_agrosupport_agro_models` for each listed `model_name`, and runs one job per -matching `(field, disease)` pair. To add a crop or disease: add an entry here, add -the Italian/English mapping to `vocab/crops.yaml` / `vocab/diseases.yaml` if it -doesn't already exist, and add a prompt directory (see below) — otherwise the -pair silently uses the generic fallback prompt. +### Advice targets + +`batch` does not read a crop or disease list from `config.yaml`. It loads +diseases with `pat_mod_AI_model = 1` from `AI_agrosupport_an_pathogen_models` +(`pat_mod_desc`), then fields from `AI_agrosupport_agro_models` where +`anmod_AI_advice = 1` and `anmod_model` matches one of those names. The field +id is `anmod_cmplay`. The crop is `cmplay_colture` on `AI_agrosupport_cmp_layers`, +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. + +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), +and optionally add a prompt directory (see below). An unmapped disease or crop +name is skipped and logged. A pair without a dedicated prompt uses the generic +fallback and is logged at startup. ### `worklist:`, `concurrency:`, `retry:`, `schedule:` ```yaml worklist: - require_enabled_model: true # AI_agrosupport_agro_models.anmod_enabled must be 1 field_allowlist: [] # restrict to specific field IDs while testing concurrency: @@ -90,7 +97,7 @@ schedule: Each crop-disease pair needs its own advice prompt (its phytopathological model's phase semantics, phenology notes, etc.). Prompts are plain Markdown files resolved by `pipeline/prompts.py::PromptRegistry`, with a shared fallback -so a newly configured pair never fails outright — it just logs a warning and +so a newly enabled pair never fails outright — it just logs a warning and uses a generic prompt until a dedicated one is written. ``` @@ -102,18 +109,18 @@ prompts/ advice/__/{system.md,user.md} per-pair advice prompts ``` -The `__` slug is built from the same canonical English names used -in `config.yaml`'s `crops:` block (lower-cased, non-alphanumeric runs collapsed -to `_`) — e.g. `grapevine` + `downy mildew` → `grapevine__downy_mildew`. To add a -prompt for a new pair, create `prompts/advice//system.md` and `user.md`; -`batch` logs every configured pair still falling back to `_default` at startup. +The `__` slug is built from the canonical English crop and disease +names (lower-cased, non-alphanumeric runs collapsed to `_`) — e.g. `grapevine` + +`downy mildew` → `grapevine__downy_mildew`. To add a prompt for a new pair, create +`prompts/advice//system.md` and `user.md`; `batch` logs every discovered +pair still falling back to `_default` at startup. ## Run ```powershell .\.venv\Scripts\Activate.ps1 -# Daily entry point: every crop/disease pair in config.yaml, across all matching fields +# Daily entry point: every field/disease pair enabled for advice in the database python -m pipeline batch # Historical / test dates (useful when the live window has no FASE2/FASE5/INCUBAZPRIMARIA/observation) @@ -372,7 +379,7 @@ is not worth a custom cost pipeline at this scale. Run once per `(field, disease)` job by `pipeline/job.py::run_job`: -1. **Worklist** — crop → fields (`AI_agrosupport_an_colture` + `AI_agrosupport_cmp_layers`) → disease model (`AI_agrosupport_agro_models`); organic flag is `cmplay_imp == 3` (`pipeline/worklist.py`, run once per batch, not per job) +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. 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 diff --git a/config.yaml b/config.yaml index 054a9d1..4433d82 100644 --- a/config.yaml +++ b/config.yaml @@ -29,31 +29,20 @@ llm: thinking_level: medium temperature: 0.0 -# ── Crop-first worklist ──────────────────────────────────────────────────────── -# `python -m pipeline batch` discovers jobs from this list: for every crop, it -# finds all fields growing it (AI_agrosupport_cmp_layers ⨝ an_colture) that also -# have a matching row in AI_agrosupport_agro_models for each listed disease's -# model_name, and runs the full single-field pipeline for every resulting -# (field, disease) pair. `disease` is the canonical English name used for the -# product prefilter, the vector search, and the advice/prompt lookup; it must -# exist in vocab/diseases.yaml. `model_name` is matched (case-insensitively) -# against AI_agrosupport_agro_models.anmod_model. +# ── Worklist filters ─────────────────────────────────────────────────────────── +# `python -m pipeline batch` discovers jobs from the database, not from this +# file. A disease can generate advice when +# AI_agrosupport_an_pathogen_models.pat_mod_AI_model = 1; its name is +# pat_mod_desc. A field is included when AI_agrosupport_agro_models.anmod_AI_advice +# = 1 and anmod_model matches that name. The cultivated crop comes from +# AI_agrosupport_cmp_layers / AI_agrosupport_an_colture and vocab/crops.yaml. +# The English disease name comes from vocab/diseases.yaml. # # Each crop-disease pair needs a matching prompt directory under -# prompts/advice/__/ (see prompts/README or the pipeline -# README for the exact slug rules); pairs without one fall back to -# prompts/advice/_default/ and are logged at startup. -crops: - - crop: grapevine - diseases: - - model_name: PERONOSPORA - disease: downy mildew - -# ── Worklist filters ─────────────────────────────────────────────────────────── +# prompts/advice/__/ (see the pipeline README for the slug +# rules); pairs without one fall back to prompts/advice/_default/ and are +# logged at startup. worklist: - # AI_agrosupport_agro_models.anmod_enabled is True for every model row in the - # current database; keep this on so a disabled model never generates advice. - require_enabled_model: true # Restrict the batch to specific field IDs while testing; empty = no filter. field_allowlist: [] diff --git a/pipeline/__main__.py b/pipeline/__main__.py index 9c84f40..ce79ae1 100644 --- a/pipeline/__main__.py +++ b/pipeline/__main__.py @@ -7,9 +7,10 @@ python -m pipeline one --field-id N --disease NAME [--as-of DATE] [--dry-run] `batch` is the daily entry point: it discovers every (field, disease) job -implied by config.yaml's `crops:` list and the database, and runs them -through a bounded thread pool (see pipeline/batch.py). `one` keeps the -original single-field behaviour for ad-hoc debugging. +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. """ from __future__ import annotations @@ -108,10 +109,11 @@ def _run_one(args: argparse.Namespace) -> int: def _run_batch(args: argparse.Namespace) -> int: - """`python -m pipeline batch`: the daily crop-first multi-field run.""" + """`python -m pipeline batch`: every database-enabled field/disease pair.""" settings = load_settings() setup_tracing(settings) crop_vocab = load_crop_vocab(settings.crops_vocab) + disease_vocab = load_disease_vocab(settings.diseases_vocab) logger.info( "Starting batch as_of=%s crop_filter=%s disease_filter=%s field_filter=%s " @@ -128,6 +130,7 @@ def _run_batch(args: argparse.Namespace) -> int: return run_batch( settings, crop_vocab, + disease_vocab, args.as_of, crop=args.crop, disease=args.disease, @@ -145,7 +148,7 @@ def main(argv: list[str] | None = None) -> int: batch_parser = subparsers.add_parser( "batch", - help="Run every configured crop/disease pair across all matching fields.", + help="Run every field/disease pair enabled for advice in the database.", ) batch_parser.add_argument( "--as-of", type=_parse_as_of, default=None, diff --git a/pipeline/batch.py b/pipeline/batch.py index 097428f..b41f673 100644 --- a/pipeline/batch.py +++ b/pipeline/batch.py @@ -17,7 +17,7 @@ from datetime import date, datetime from datetime import time as dtime from pipeline.config import Settings -from pipeline.db import fetch_all +from pipeline.db import fetch_all, qualify_app_sql from pipeline.job import JobResult, run_job from pipeline.output import write_job_output, write_run_report from pipeline.prompts import PromptRegistry @@ -27,7 +27,7 @@ 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 field_id, disease FROM {advisories} WHERE issuance_date = ?" def _parse_deadline(deadline_str: str) -> datetime: @@ -69,13 +69,18 @@ def filter_jobs( def _existing_advice_keys(resources: Resources, as_of: date) -> set[tuple[int, str]]: """One bulk SELECT instead of one per job, for --skip-existing-advice.""" conn = resources.sql_connection() - rows = fetch_all(conn, _EXISTING_ADVICE_SQL, (as_of,)) + rows = fetch_all( + conn, + qualify_app_sql(resources.settings.sql, _EXISTING_ADVICE_SQL), + (as_of,), + ) return {(int(row.field_id), str(row.disease)) for row in rows} def run_batch( settings: Settings, crop_vocab: Vocabulary, + disease_vocab: Vocabulary, as_of: date, *, crop: str | None = None, @@ -88,26 +93,33 @@ def run_batch( ) -> int: """Run every discovered job and write the run report. Returns the process exit code.""" prompts = PromptRegistry(settings.prompts_dir) - configured_pairs = [(cp.crop, dp.disease) for cp in settings.crops for dp in cp.diseases] - fallbacks = prompts.validate(configured_pairs) - if fallbacks: - logger.warning( - "%d crop/disease pair(s) have no dedicated advice prompt and will fall back " - "to prompts/advice/_default/: %s", - len(fallbacks), - fallbacks, - ) - resources = Resources(settings, prompts) try: - jobs = build_jobs(resources.sql_connection(), settings.crops, settings.worklist, crop_vocab) + jobs = build_jobs( + resources.sql_connection(), + settings.worklist, + crop_vocab, + disease_vocab, + ) jobs = filter_jobs(jobs, crop=crop, disease=disease, field_ids=field_ids) logger.info("Worklist discovered %d job(s) for as_of=%s", len(jobs), as_of.isoformat()) + # Unique pairs only: one warning per crop/disease, not one per field. + # An empty job list validates nothing. + prompt_pairs = list(dict.fromkeys((job.crop_english, job.disease_english) for job in jobs)) + fallbacks = prompts.validate(prompt_pairs) + if fallbacks: + logger.warning( + "%d crop/disease pair(s) have no dedicated advice prompt and will fall back " + "to prompts/advice/_default/: %s", + len(fallbacks), + fallbacks, + ) + if not jobs: logger.warning( - "No jobs to run for as_of=%s; nothing matched the configured " - "crops/diseases (or the CLI filters).", + "No jobs to run for as_of=%s; nothing matched pat_mod_AI_model = 1 " + "and anmod_AI_advice = 1 (or the CLI filters).", as_of.isoformat(), ) write_run_report(settings.output_dir, as_of, []) diff --git a/pipeline/config.py b/pipeline/config.py index 5ac57a6..aac3c0a 100644 --- a/pipeline/config.py +++ b/pipeline/config.py @@ -28,6 +28,7 @@ class SqlSettings: driver: str server: str database: str + app_database: str username: str password: str @@ -39,25 +40,8 @@ class WeaviateSettings: grpc_port: int -@dataclass(frozen=True) -class DiseasePlan: - """One crop-disease pair to run the batch for.""" - - model_name: str # matched against AI_agrosupport_agro_models.anmod_model - disease: str # canonical English disease name (vocab/diseases.yaml) - - -@dataclass(frozen=True) -class CropPlan: - """A crop and the diseases to check for every field growing it.""" - - crop: str # canonical English crop name (vocab/crops.yaml) - diseases: tuple[DiseasePlan, ...] - - @dataclass(frozen=True) class WorklistSettings: - require_enabled_model: bool field_allowlist: tuple[int, ...] @@ -99,7 +83,6 @@ class Settings: root: Path field_id: int | None disease_name: str | None - crops: tuple[CropPlan, ...] worklist: WorklistSettings concurrency: ConcurrencySettings retry: RetrySettings @@ -142,24 +125,8 @@ def _parse_llm_settings(raw: dict, defaults: dict) -> LlmSettings: ) -def _parse_crops(raw: list[dict] | None) -> tuple[CropPlan, ...]: - plans: list[CropPlan] = [] - for entry in raw or []: - crop = str(entry["crop"]).strip() - diseases = tuple( - DiseasePlan( - model_name=str(d["model_name"]).strip(), - disease=str(d["disease"]).strip(), - ) - for d in entry.get("diseases", []) - ) - plans.append(CropPlan(crop=crop, diseases=diseases)) - return tuple(plans) - - def _parse_worklist(raw: dict) -> WorklistSettings: return WorklistSettings( - require_enabled_model=bool(raw.get("require_enabled_model", True)), field_allowlist=tuple(int(v) for v in raw.get("field_allowlist", []) or []), ) @@ -234,8 +201,8 @@ def load_settings( llm_raw = raw.get("llm", {}) vocab = raw.get("vocab", {}) - # field_id / disease_name only matter for `python -m pipeline one`; the - # batch worklist is driven entirely by the crops: block below. + # field_id / disease_name only matter for `python -m pipeline one`. The + # batch worklist is discovered from the pathogen and agro-model tables. field_id = field_id_override if field_id_override is not None else raw.get("field_id") field_id = int(field_id) if field_id is not None else None disease_name = disease_name_override or raw.get("disease_name") @@ -264,7 +231,6 @@ def load_settings( root=ROOT, field_id=field_id, disease_name=disease_name, - crops=_parse_crops(raw.get("crops")), worklist=_parse_worklist(raw.get("worklist", {})), concurrency=_parse_concurrency(raw.get("concurrency", {})), retry=_parse_retry(raw.get("retry", {})), @@ -282,6 +248,8 @@ def load_settings( driver=os.environ.get("SQL_DRIVER", "ODBC Driver 17 for SQL Server"), server=os.environ["SQL_SERVER"], database=os.environ["SQL_DATABASE"], + app_database=os.environ.get("SQL_APP_DATABASE", "").strip() + or os.environ["SQL_DATABASE"], username=os.environ["SQL_USERNAME"], password=os.environ["SQL_PASSWORD"], ), diff --git a/pipeline/db.py b/pipeline/db.py index 34b25a0..f067823 100644 --- a/pipeline/db.py +++ b/pipeline/db.py @@ -10,10 +10,64 @@ import pyodbc from pipeline.config import SqlSettings +# Tables that live in SQL_APP_DATABASE. Every other name stays in SQL_DATABASE. +_APP_TABLES = ("advisories", "products", "product_uses", "label_chunks") + + +def _bracket(identifier: str) -> str: + """Quote a SQL Server identifier. A closing bracket inside the name is doubled.""" + return "[" + identifier.replace("]", "]]") + "]" + + +def _odbc_server(server: str) -> str: + """ + ODBC Driver expects host,port. Accept host:port and leave named instances alone. + + `192.168.220.50:1433` becomes `192.168.220.50,1433`. `localhost\\SQLEXPRESS` + has no colon, so it is returned unchanged. + """ + host, colon, port = server.rpartition(":") + if colon and port.isdigit() and ":" not in host: + return f"{host},{port}" + return server + + +def app_table(settings: SqlSettings, table: str) -> str: + """ + Name of an application table on the open connection. + + Bare when `app_database` is the connected database, so a single-database + setup keeps unqualified SQL. Otherwise a three-part name, + `[Agrosupport].dbo.[advisories]`, so one DBMeteo connection can read the + other database on the same instance. + """ + if table not in _APP_TABLES: + allowed = ", ".join(_APP_TABLES) + raise ValueError(f"Unknown application table {table!r}. Expected one of {allowed}.") + if ( + not settings.app_database + or settings.app_database.casefold() == settings.database.casefold() + ): + return table + return f"{_bracket(settings.app_database)}.dbo.{_bracket(table)}" + + +def qualify_app_sql(settings: SqlSettings, sql: str) -> str: + """ + Replace `{advisories}`, `{products}`, `{product_uses}`, and `{label_chunks}`. + + Call this before `.format(placeholders=...)`. Only those four tokens are + substituted, so a `{placeholders}` brace list is left for the later format. + """ + for table in _APP_TABLES: + sql = sql.replace("{" + table + "}", app_table(settings, table)) + return sql + + def _connection_string(settings: SqlSettings) -> str: return ( f"DRIVER={{{settings.driver}}};" - f"SERVER={settings.server};" + f"SERVER={_odbc_server(settings.server)};" f"DATABASE={settings.database};" f"UID={settings.username};" f"PWD={settings.password};" diff --git a/pipeline/job.py b/pipeline/job.py index 49276cc..446d520 100644 --- a/pipeline/job.py +++ b/pipeline/job.py @@ -176,7 +176,7 @@ 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 + conn, job.field_id, disease.disease_english, as_of, settings.sql ) ) interval_open = last_treatment_interval_open( @@ -186,6 +186,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - disease_english=disease.disease_english, latest_spray=treatment_history.latest(), as_of=as_of, + sql=settings.sql, ) def _covers(start: date): @@ -196,6 +197,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - disease_english=disease.disease_english, start_date=start, treatment_history=treatment_history, + sql=settings.sql, ) incubation_register = reconcile_register( @@ -343,7 +345,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - 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 + conn, job.field_id, disease.disease_english, as_of, settings.sql ) enrich_advice_json( json_for_advice_generation, @@ -379,6 +381,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - disease_english=disease.disease_english, advice_input_json=advice_input_json, result=result, + sql=settings.sql, ) call_with_sql_retry(settings.retry, _insert) diff --git a/pipeline/resources.py b/pipeline/resources.py index 423dead..0524b88 100644 --- a/pipeline/resources.py +++ b/pipeline/resources.py @@ -134,7 +134,7 @@ class Resources: def load_product_index(self) -> ProductIndex: """Load the whole `products` table once; call before starting the pool.""" - self.product_index = ProductIndex.load(self.sql_connection()) + self.product_index = ProductIndex.load(self.sql_connection(), self.settings.sql) return self.product_index def product_json_builder(self, crop_english: str, disease_english: str) -> ProductJsonBuilder: @@ -148,7 +148,12 @@ class Resources: with self._builder_lock: builder = self._builder_cache.get(key) if builder is None: - builder = ProductJsonBuilder(self.product_index, crop_english, disease_english) + builder = ProductJsonBuilder( + self.product_index, + crop_english, + disease_english, + self.settings.sql, + ) self._builder_cache[key] = builder return builder diff --git a/pipeline/stages/advice.py b/pipeline/stages/advice.py index e5895dd..ee3cca5 100644 --- a/pipeline/stages/advice.py +++ b/pipeline/stages/advice.py @@ -11,8 +11,8 @@ from typing import Any, Callable import pyodbc -from pipeline.config import Settings -from pipeline.db import execute, transaction +from pipeline.config import Settings, SqlSettings +from pipeline.db import execute, qualify_app_sql, transaction from pipeline.errors import PipelineError from pipeline.stages.advice_guardrails import AdviceRejected, validate_advice from pipeline.stages.advice_summary import sanitize_advice_summary @@ -26,14 +26,14 @@ _RECOMMENDED_PRODUCT_NAMES_MAX_CHARS = 200 _RECOMMENDED_DOSES_MAX_CHARS = 100 _DELETE_ADVICE_SQL = """ -DELETE FROM advisories +DELETE FROM {advisories} WHERE issuance_date = ? AND field_id = ? AND disease = ? """ _INSERT_ADVICE_SQL = """ -INSERT INTO advisories ( +INSERT INTO {advisories} ( issuance_date, field_id, disease, @@ -176,6 +176,7 @@ def insert_advice( disease_english: str, advice_input_json: dict[str, Any], result: AdviceResult, + sql: SqlSettings, ) -> None: """ Store the advisory, replacing any earlier row for the same day/field/disease. @@ -202,10 +203,14 @@ def insert_advice( sent = json.dumps(advice_input_json, ensure_ascii=False, default=_json_default) with transaction(conn): - execute(conn, _DELETE_ADVICE_SQL, (issuance_date, field_id, disease_english)) execute( conn, - _INSERT_ADVICE_SQL, + qualify_app_sql(sql, _DELETE_ADVICE_SQL), + (issuance_date, field_id, disease_english), + ) + execute( + conn, + qualify_app_sql(sql, _INSERT_ADVICE_SQL), ( issuance_date, field_id, diff --git a/pipeline/stages/advice_context.py b/pipeline/stages/advice_context.py index 5f7297f..6a81b51 100644 --- a/pipeline/stages/advice_context.py +++ b/pipeline/stages/advice_context.py @@ -10,7 +10,8 @@ from typing import Any, Iterable import pyodbc -from pipeline.db import fetch_one +from pipeline.config import SqlSettings +from pipeline.db import fetch_one, qualify_app_sql from pipeline.jsonutils import parse_json_list from pipeline.stages.product_json import ProductJsonBuilder, normalize_name from pipeline.stages.treatments import TreatmentHistory @@ -26,7 +27,7 @@ SELECT TOP 1 issuance_date, advice_summary, recommended_product_names -FROM advisories +FROM {advisories} WHERE field_id = ? AND disease = ? AND issuance_date < ? @@ -46,7 +47,7 @@ def _row_date(value: Any) -> date | None: _LAST_SENT_SQL = """ SELECT TOP 1 advice_input_json -FROM advisories +FROM {advisories} WHERE field_id = ? AND disease = ? AND issuance_date < ? @@ -59,9 +60,14 @@ def load_previous_incubation_register( field_id: int, disease_english: str, as_of: date, + sql: SqlSettings, ) -> 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)) + row = fetch_one( + conn, + qualify_app_sql(sql, _LAST_SENT_SQL), + (field_id, disease_english, as_of), + ) if row is None or not row.advice_input_json: return {} try: @@ -86,9 +92,14 @@ def load_last_advice( field_id: int, disease_english: str, 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.""" - row = fetch_one(conn, _LAST_ADVICE_SQL, (field_id, disease_english, as_of)) + row = fetch_one( + conn, + qualify_app_sql(sql, _LAST_ADVICE_SQL), + (field_id, disease_english, as_of), + ) if row is None: logger.info( "No previous advice for field %s / %s before %s.", diff --git a/pipeline/stages/disease.py b/pipeline/stages/disease.py index c40fcb9..8d3688a 100644 --- a/pipeline/stages/disease.py +++ b/pipeline/stages/disease.py @@ -158,9 +158,8 @@ def find_anmod_id(conn: pyodbc.Connection, field_id: int, disease_name: str) -> Look up the anmod_id for a field's disease model by name. Used by the single-field `one` CLI path, where `anmod_id` is not known - ahead of time. The batch worklist resolves it directly with one join per - crop-disease pair instead (see `pipeline.worklist.build_jobs`), so it - never calls this per field. + ahead of time. The batch worklist resolves it in one database join + (see `pipeline.worklist.build_jobs`), so it never calls this per field. """ models = fetch_all(conn, _MODEL_SQL, (field_id,)) target = disease_name.strip().upper() diff --git a/pipeline/stages/incubations.py b/pipeline/stages/incubations.py index 374db0e..37cd312 100644 --- a/pipeline/stages/incubations.py +++ b/pipeline/stages/incubations.py @@ -9,7 +9,8 @@ from typing import Any, Callable import pyodbc -from pipeline.db import fetch_all +from pipeline.config import SqlSettings +from pipeline.db import fetch_all, qualify_app_sql from pipeline.stages.disease import incub_register_key from pipeline.stages.products import ProductIndex from pipeline.stages.treatments import TreatmentHistory @@ -19,7 +20,7 @@ logger = logging.getLogger(__name__) _USES_INTERVAL_SQL = """ SELECT product_id, treatment_interval_min_days -FROM product_uses +FROM {product_uses} WHERE crop = ? AND disease = ? AND product_id IN ({placeholders}) @@ -225,6 +226,7 @@ def incubation_start_covered( disease_english: str, start_date: date, treatment_history: TreatmentHistory, + sql: SqlSettings, ) -> CoveringTreatment | None: """ Latest spray strictly before start_date whose minimum interval still reaches it. @@ -266,7 +268,7 @@ def incubation_start_covered( placeholders = ",".join("?" * len(all_ids)) use_rows = fetch_all( conn, - _USES_INTERVAL_SQL.format(placeholders=placeholders), + qualify_app_sql(sql, _USES_INTERVAL_SQL).format(placeholders=placeholders), (crop_english, disease_english, *all_ids), ) min_days_by_id: dict[str, int] = {} diff --git a/pipeline/stages/product_json.py b/pipeline/stages/product_json.py index ed832d5..d600243 100644 --- a/pipeline/stages/product_json.py +++ b/pipeline/stages/product_json.py @@ -9,7 +9,8 @@ from typing import Any import pyodbc -from pipeline.db import fetch_all +from pipeline.config import SqlSettings +from pipeline.db import fetch_all, qualify_app_sql from pipeline.jsonutils import normalize_name, parse_json_list from pipeline.stages.products import ProductIndex @@ -30,7 +31,7 @@ SELECT pre_harvest_interval_days, growth_stage_start, growth_stage_end -FROM product_uses +FROM {product_uses} WHERE product_id = ? AND crop = ? AND disease = ? @@ -43,7 +44,7 @@ SELECT chunk_text, target_crops, target_diseases -FROM label_chunks +FROM {label_chunks} WHERE product_id = ? ORDER BY label_chunk_id ASC """ @@ -91,10 +92,12 @@ class ProductJsonBuilder: product_index: ProductIndex, crop_english: str, disease_english: str, + sql: SqlSettings, ) -> None: self._products = product_index self._crop = crop_english self._disease = disease_english + self._sql = sql self._cache: dict[str, dict[str, Any] | str] = {} self._cache_lock = threading.Lock() @@ -138,14 +141,18 @@ class ProductJsonBuilder: } def _load_uses(self, conn: pyodbc.Connection, product_id: str) -> list[dict[str, Any]]: - rows = fetch_all(conn, _USES_SQL, (product_id, self._crop, self._disease)) + rows = fetch_all( + conn, + qualify_app_sql(self._sql, _USES_SQL), + (product_id, self._crop, self._disease), + ) return [ {key: _scalar(getattr(row, key)) for key in _USE_KEYS} for row in rows ] def _load_chunks(self, conn: pyodbc.Connection, product_id: str) -> list[dict[str, Any]]: - rows = fetch_all(conn, _CHUNKS_SQL, (product_id,)) + rows = fetch_all(conn, qualify_app_sql(self._sql, _CHUNKS_SQL), (product_id,)) chunks: list[dict[str, Any]] = [] for row in rows: if self._crop not in parse_json_list(row.target_crops): diff --git a/pipeline/stages/products.py b/pipeline/stages/products.py index 51c6c36..ab6bd2d 100644 --- a/pipeline/stages/products.py +++ b/pipeline/stages/products.py @@ -9,7 +9,8 @@ from typing import Any import pyodbc -from pipeline.db import fetch_all +from pipeline.config import SqlSettings +from pipeline.db import fetch_all, qualify_app_sql from pipeline.jsonutils import normalize_name, parse_json_list from pipeline.stages.treatments import LAST_TREATMENT_MAX_AGE_DAYS @@ -63,12 +64,12 @@ SELECT target_diseases, active_ingredients, frac_codes -FROM products +FROM {products} """ _USES_INTERVAL_SQL = """ SELECT product_id, treatment_interval_min_days -FROM product_uses +FROM {product_uses} WHERE crop = ? AND disease = ? AND product_id IN ({placeholders}) @@ -175,8 +176,8 @@ class ProductIndex: ) @classmethod - def load(cls, conn: pyodbc.Connection) -> "ProductIndex": - product_rows = fetch_all(conn, _PRODUCTS_SQL) + def load(cls, conn: pyodbc.Connection, sql: SqlSettings) -> "ProductIndex": + product_rows = fetch_all(conn, qualify_app_sql(sql, _PRODUCTS_SQL)) catalog_rows = fetch_all(conn, _CATALOG_STATE_SQL) permitted_by_registration = _permitted_by_registration(catalog_rows) @@ -235,6 +236,7 @@ def last_treatment_interval_open( disease_english: str, latest_spray: tuple[date, list[str]] | None, as_of: date, + sql: SqlSettings, ) -> bool: """ True if the latest spray is still inside its treatment interval. @@ -276,7 +278,7 @@ def last_treatment_interval_open( placeholders = ",".join("?" * len(product_ids)) use_rows = fetch_all( conn, - _USES_INTERVAL_SQL.format(placeholders=placeholders), + qualify_app_sql(sql, _USES_INTERVAL_SQL).format(placeholders=placeholders), (crop_english, disease_english, *product_ids), ) diff --git a/pipeline/worklist.py b/pipeline/worklist.py index 33a72d8..10e278f 100644 --- a/pipeline/worklist.py +++ b/pipeline/worklist.py @@ -1,20 +1,23 @@ -"""Crop-first job discovery: find every (field, disease) pair to run today. +"""Database-driven job discovery: find every (field, disease) pair to run today. -Unlike the original single-field pipeline, the batch does not start from a -field ID. It starts from `config.yaml`'s `crops:` list, resolves which fields -grow each crop, and for each of that crop's configured diseases finds the -fields that also have a matching disease-forecast model configured -(`AI_agrosupport_agro_models`). Each match becomes one `Job`. +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 +(`AI_agrosupport_an_pathogen_models.pat_mod_AI_model = 1`, named by +`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`. """ from __future__ import annotations import logging +from collections import Counter from dataclasses import dataclass import pyodbc -from pipeline.config import CropPlan, WorklistSettings +from pipeline.config import WorklistSettings from pipeline.db import fetch_all from pipeline.errors import PipelineError from pipeline.vocab import Vocabulary @@ -36,128 +39,104 @@ class Job: anmod_id: int -_COLTURE_SQL = "SELECT colture_id, colture_name FROM AI_agrosupport_an_colture" - -_JOBS_SQL_TEMPLATE = """ -SELECT +_JOBS_SQL = """ +SELECT DISTINCT l.cmplay_id, l.cmplay_colture, l.cmplay_imp, l.cmplay_station, + c.colture_name, m.anmod_id, - m.anmod_enabled -FROM AI_agrosupport_cmp_layers l -JOIN AI_agrosupport_agro_models m ON m.anmod_cmplay = l.cmplay_id -WHERE l.cmplay_colture IN ({placeholders}) - AND UPPER(LTRIM(RTRIM(m.anmod_model))) = ? + m.anmod_model +FROM AI_agrosupport_agro_models m +INNER JOIN ( + SELECT DISTINCT LTRIM(RTRIM(pat_mod_desc)) AS model_name + FROM AI_agrosupport_an_pathogen_models + WHERE pat_mod_AI_model = 1 + AND pat_mod_desc IS NOT NULL + AND LTRIM(RTRIM(pat_mod_desc)) <> '' +) p + ON UPPER(LTRIM(RTRIM(m.anmod_model))) = UPPER(p.model_name) +INNER JOIN AI_agrosupport_cmp_layers l + ON l.cmplay_id = m.anmod_cmplay +LEFT JOIN AI_agrosupport_an_colture c + ON c.colture_id = l.cmplay_colture +WHERE m.anmod_AI_advice = 1 """ -def _load_crop_to_colture_ids( - conn: pyodbc.Connection, crop_vocab: Vocabulary -) -> dict[str, list[int]]: - """ - Map every canonical English crop name to the colture_ids that resolve to it. - - `Vocabulary.to_english` already strips whitespace and lower-cases before - matching, so `AI_agrosupport_an_colture` rows like `"Vite "` (trailing - space) resolve correctly without extra cleanup here. Colture names with no - vocabulary entry are logged once and skipped, so a single unmapped crop - cannot break discovery for every other crop. - """ - mapping: dict[str, list[int]] = {} - unmapped: list[str] = [] - for row in fetch_all(conn, _COLTURE_SQL): - italian = (row.colture_name or "").strip() - if not italian: - continue - try: - english = crop_vocab.to_english(italian, kind="crop") - except PipelineError: - unmapped.append(italian) - continue - mapping.setdefault(english, []).append(int(row.colture_id)) - - if unmapped: - logger.warning( - "%d colture name(s) have no crop vocabulary mapping and will be " - "ignored by the worklist: %s", - len(unmapped), - sorted(set(unmapped)), - ) - return mapping - - -def _colture_names_by_id(conn: pyodbc.Connection) -> dict[int, str]: - return { - int(row.colture_id): (row.colture_name or "").strip() - for row in fetch_all(conn, _COLTURE_SQL) - } - - def build_jobs( conn: pyodbc.Connection, - crops: tuple[CropPlan, ...], worklist: WorklistSettings, crop_vocab: Vocabulary, + disease_vocab: Vocabulary, ) -> list[Job]: - """Discover every (field, disease) job implied by `crops` and `worklist`.""" - crop_to_colture = _load_crop_to_colture_ids(conn, crop_vocab) - colture_names = _colture_names_by_id(conn) + """Discover every (field, disease) job enabled in the database.""" allowlist = set(worklist.field_allowlist) + rows = fetch_all(conn, _JOBS_SQL) jobs: list[Job] = [] - for crop_plan in crops: - colture_ids = crop_to_colture.get(crop_plan.crop, []) - if not colture_ids: - logger.warning( - "No field layer maps to crop '%s' (check vocab/crops.yaml against " - "AI_agrosupport_an_colture); skipping its %d configured disease(s).", - crop_plan.crop, - len(crop_plan.diseases), - ) + unmapped_diseases: list[str] = [] + unmapped_colture: list[str] = [] + + for row in rows: + if allowlist and int(row.cmplay_id) not in allowlist: continue - placeholders = ",".join("?" for _ in colture_ids) - sql = _JOBS_SQL_TEMPLATE.format(placeholders=placeholders) + model_name = (row.anmod_model or "").strip() + try: + disease_english = disease_vocab.to_english(model_name, kind="disease") + except PipelineError: + unmapped_diseases.append(model_name or "(blank)") + continue - for disease_plan in crop_plan.diseases: - params = (*colture_ids, disease_plan.model_name.strip().upper()) - rows = fetch_all(conn, sql, params) + colture_name = (row.colture_name or "").strip() + try: + crop_english = crop_vocab.to_english(colture_name, kind="crop") + except PipelineError: + unmapped_colture.append(colture_name or "(blank)") + continue - kept = 0 - for row in rows: - if allowlist and int(row.cmplay_id) not in allowlist: - continue - if worklist.require_enabled_model and not bool(row.anmod_enabled): - continue - - jobs.append( - Job( - field_id=int(row.cmplay_id), - crop_english=crop_plan.crop, - crop_italian=colture_names.get(int(row.cmplay_colture), ""), - model_name=disease_plan.model_name, - disease_english=disease_plan.disease, - organic=row.cmplay_imp == 3, - station=( - int(row.cmplay_station) - if row.cmplay_station is not None - else None - ), - anmod_id=int(row.anmod_id), - ) - ) - kept += 1 - - logger.info( - "Worklist: crop=%s disease=%s (model=%s) -> %d field(s) " - "(%d row(s) matched before allowlist/enabled filters)", - crop_plan.crop, - disease_plan.disease, - disease_plan.model_name, - kept, - len(rows), + jobs.append( + Job( + field_id=int(row.cmplay_id), + crop_english=crop_english, + crop_italian=colture_name, + model_name=model_name, + disease_english=disease_english, + organic=row.cmplay_imp == 3, + station=( + int(row.cmplay_station) if row.cmplay_station is not None else None + ), + anmod_id=int(row.anmod_id), ) + ) + + jobs.sort(key=lambda job: (job.field_id, job.model_name, job.anmod_id)) + + if unmapped_diseases: + logger.warning( + "%d disease model name(s) have no disease vocabulary mapping and will be " + "ignored by the worklist: %s", + len(set(unmapped_diseases)), + sorted(set(unmapped_diseases)), + ) + if unmapped_colture: + logger.warning( + "%d colture name(s) have no crop vocabulary mapping and will be " + "ignored by the worklist: %s", + len(set(unmapped_colture)), + sorted(set(unmapped_colture)), + ) + + counts = Counter((job.crop_english, job.disease_english, job.model_name) for job in jobs) + for crop, disease, model_name in sorted(counts): + logger.info( + "Worklist: crop=%s disease=%s (model=%s) -> %d job(s)", + crop, + disease, + model_name, + counts[(crop, disease, model_name)], + ) return jobs diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..e14e809 --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,235 @@ +"""Unit tests for the application-database qualifier and ODBC server form.""" + +from __future__ import annotations + +import ast +import unittest +from datetime import date, timedelta +from pathlib import Path +from unittest.mock import MagicMock, patch + +from pipeline.config import SqlSettings +from pipeline.db import _connection_string, app_table, qualify_app_sql +from pipeline.stages.advice import AdviceResult, insert_advice +from pipeline.stages.advice_context import load_last_advice, load_previous_incubation_register +from pipeline.stages.incubations import incubation_start_covered +from pipeline.stages.product_json import ProductJsonBuilder +from pipeline.stages.products import ProductIndex, ProductRow, last_treatment_interval_open +from pipeline.stages.treatments import TreatmentHistory + +_ROOT = Path(__file__).resolve().parent.parent + + +def _sql(**overrides: str) -> SqlSettings: + values = { + "driver": "ODBC Driver 17 for SQL Server", + "server": "localhost", + "database": "DBMeteo", + "app_database": "DBMeteo", + "username": "user", + "password": "secret", + } + values.update(overrides) + return SqlSettings(**values) + + +class AppTableTests(unittest.TestCase): + def test_same_database_stays_bare(self) -> None: + self.assertEqual(app_table(_sql(), "advisories"), "advisories") + self.assertEqual(app_table(_sql(app_database="dbmeteo"), "products"), "products") + + def test_different_database_is_three_part(self) -> None: + settings = _sql(app_database="Agrosupport") + self.assertEqual(app_table(settings, "advisories"), "[Agrosupport].dbo.[advisories]") + + def test_qualify_leaves_placeholders(self) -> None: + sql = "SELECT product_id FROM {product_uses} WHERE product_id IN ({placeholders})" + qualified = qualify_app_sql(_sql(app_database="Agrosupport"), sql) + self.assertEqual( + qualified, + "SELECT product_id FROM [Agrosupport].dbo.[product_uses] " + "WHERE product_id IN ({placeholders})", + ) + + def test_unknown_table_rejected(self) -> None: + with self.assertRaises(ValueError): + app_table(_sql(), "AI_agrosupport_cmp_layers") + + +class ConnectionStringTests(unittest.TestCase): + def test_colon_port_becomes_comma(self) -> None: + text = _connection_string(_sql(server="192.168.220.50:1433")) + self.assertIn("SERVER=192.168.220.50,1433;", text) + + def test_named_instance_unchanged(self) -> None: + text = _connection_string(_sql(server=r"localhost\SQLEXPRESS")) + self.assertIn(r"SERVER=localhost\SQLEXPRESS;", text) + + def test_comma_port_unchanged(self) -> None: + text = _connection_string(_sql(server="192.168.220.50,1433")) + self.assertIn("SERVER=192.168.220.50,1433;", text) + + +def _product() -> ProductRow: + return ProductRow( + product_id="p1", + product_name="LIETO", + registration_number=None, + is_organic_certified=False, + systemicity="contact", + has_preventive_action=True, + has_curative_action=False, + has_eradicant_action=False, + target_crops=["grapevine"], + target_diseases=["downy mildew"], + active_ingredients=[], + frac_codes=[], + ) + + +class QualifiedSqlTests(unittest.TestCase): + def setUp(self) -> None: + self.split = _sql(app_database="Agrosupport") + + def test_product_index_qualifies_products_and_leaves_the_catalog(self) -> None: + with patch("pipeline.stages.products.fetch_all", side_effect=[[], []]) as fetch: + ProductIndex.load(MagicMock(), self.split) + products_sql, catalog_sql = (call.args[1] for call in fetch.call_args_list) + self.assertIn("[Agrosupport].dbo.[products]", products_sql) + self.assertIn("AI_agrosupport_products_catalog", catalog_sql) + self.assertNotIn("Agrosupport", catalog_sql) + + def test_treatment_interval_qualifies_product_uses(self) -> None: + with patch("pipeline.stages.products.fetch_all", return_value=[]) as fetch: + last_treatment_interval_open( + MagicMock(), + ProductIndex([_product()]), + crop_english="grapevine", + disease_english="downy mildew", + latest_spray=(date(2026, 4, 6), ["LIETO"]), + as_of=date(2026, 4, 16), + sql=self.split, + ) + self.assertIn("[Agrosupport].dbo.[product_uses]", fetch.call_args.args[1]) + + def test_incubation_coverage_qualifies_product_uses(self) -> None: + as_of = date(2026, 4, 16) + start = as_of - timedelta(days=2) + with patch("pipeline.stages.incubations.fetch_all", return_value=[]) as fetch: + incubation_start_covered( + MagicMock(), + ProductIndex([_product()]), + crop_english="grapevine", + disease_english="downy mildew", + start_date=start, + treatment_history=TreatmentHistory(as_of=as_of, by_day={start - timedelta(days=4): ["LIETO"]}), + sql=self.split, + ) + self.assertIn("[Agrosupport].dbo.[product_uses]", fetch.call_args.args[1]) + + def test_product_json_qualifies_uses_and_chunks(self) -> None: + builder = ProductJsonBuilder(ProductIndex([_product()]), "grapevine", "downy mildew", self.split) + with patch("pipeline.stages.product_json.fetch_all", side_effect=[[], []]) as fetch: + builder.build(MagicMock(), "LIETO") + uses_sql, chunks_sql = (call.args[1] for call in fetch.call_args_list) + self.assertIn("[Agrosupport].dbo.[product_uses]", uses_sql) + self.assertIn("[Agrosupport].dbo.[label_chunks]", chunks_sql) + + 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) + for call in fetch.call_args_list: + self.assertIn("[Agrosupport].dbo.[advisories]", call.args[1]) + + def test_advice_write_qualifies_advisories(self) -> None: + result = AdviceResult( + full_advice="text", + advice_summary="summary", + is_treatment_recommended=False, + recommended_product_names=[], + recommended_doses=[], + recommended_application_date=None, + ) + with ( + patch("pipeline.stages.advice.transaction", MagicMock()), + patch("pipeline.stages.advice.execute") as execute, + ): + insert_advice( + MagicMock(), + issuance_date=date(2026, 4, 16), + field_id=4012, + disease_english="downy mildew", + advice_input_json={}, + result=result, + sql=self.split, + ) + delete_sql, insert_sql = (call.args[1] for call in execute.call_args_list) + self.assertIn("DELETE FROM [Agrosupport].dbo.[advisories]", delete_sql) + self.assertIn("INSERT INTO [Agrosupport].dbo.[advisories]", insert_sql) + + +def _called_name(node: ast.Call) -> str | None: + func = node.func + if isinstance(func, ast.Name): + return func.id + if isinstance(func, ast.Attribute): + return func.attr + return None + + +def _passes_sql(node: ast.Call) -> bool: + values = list(node.args) + [keyword.value for keyword in node.keywords] + return any(isinstance(value, ast.Attribute) and value.attr == "sql" for value in values) + + +class CallSiteTests(unittest.TestCase): + def _calls(self, relative: str) -> list[ast.Call]: + tree = ast.parse((_ROOT / relative).read_text(encoding="utf-8")) + return [node for node in ast.walk(tree) if isinstance(node, ast.Call)] + + def test_job_passes_sql_settings(self) -> None: + calls = self._calls("pipeline/job.py") + expected = { + "load_previous_incubation_register", + "last_treatment_interval_open", + "incubation_start_covered", + "load_last_advice", + "insert_advice", + } + found = {name for node in calls if (name := _called_name(node)) in expected} + self.assertEqual(found, expected) + for node in calls: + if _called_name(node) in expected: + self.assertTrue(_passes_sql(node), _called_name(node)) + + def test_resources_pass_sql_settings(self) -> None: + calls = self._calls("pipeline/resources.py") + load = [node for node in calls if _called_name(node) == "load"] + builder = [node for node in calls if _called_name(node) == "ProductJsonBuilder"] + self.assertEqual(len(load), 1) + self.assertEqual(len(builder), 1) + self.assertTrue(_passes_sql(load[0])) + self.assertTrue(_passes_sql(builder[0])) + + def test_batch_existing_advice_uses_app_database(self) -> None: + calls = self._calls("pipeline/batch.py") + qualify = [node for node in calls if _called_name(node) == "qualify_app_sql"] + self.assertEqual(len(qualify), 1) + self.assertTrue(_passes_sql(qualify[0])) + module = ast.parse((_ROOT / "pipeline/batch.py").read_text(encoding="utf-8")) + constant = next( + node.value.value + for node in module.body + if isinstance(node, ast.Assign) + and isinstance(node.targets[0], ast.Name) + and node.targets[0].id == "_EXISTING_ADVICE_SQL" + and isinstance(node.value, ast.Constant) + ) + qualified = qualify_app_sql(_sql(app_database="Agrosupport"), constant) + self.assertIn("FROM [Agrosupport].dbo.[advisories]", qualified) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_incubations.py b/tests/test_incubations.py index 935dc80..1228a77 100644 --- a/tests/test_incubations.py +++ b/tests/test_incubations.py @@ -7,6 +7,7 @@ from datetime import date, timedelta from types import SimpleNamespace from unittest.mock import MagicMock, patch +from pipeline.config import SqlSettings from pipeline.stages.disease import select_prefilter_rule from pipeline.stages.incubations import ( CoveringTreatment, @@ -20,6 +21,14 @@ from pipeline.stages.incubations import ( from pipeline.stages.products import ProductIndex, ProductRow from pipeline.stages.treatments import TreatmentHistory +_SQL = SqlSettings( + driver="ODBC Driver 17 for SQL Server", + server="localhost", + database="DBMeteo", + app_database="DBMeteo", + username="user", + password="secret", +) _AS_OF = date(2026, 4, 16) _PAST = _AS_OF - timedelta(days=2) _TODAY = _AS_OF @@ -194,6 +203,7 @@ class CoverageTests(unittest.TestCase): disease_english="downy mildew", start_date=_PAST, treatment_history=self._history(_PAST), + sql=_SQL, ) self.assertIsNone(covered) fetch.assert_not_called() @@ -209,6 +219,7 @@ class CoverageTests(unittest.TestCase): disease_english="downy mildew", start_date=_PAST, treatment_history=self._history(spray), + sql=_SQL, ) self.assertIsNotNone(covered) assert covered is not None diff --git a/tests/test_products.py b/tests/test_products.py index 6361732..6017380 100644 --- a/tests/test_products.py +++ b/tests/test_products.py @@ -7,6 +7,7 @@ from datetime import date from types import SimpleNamespace from unittest.mock import MagicMock, patch +from pipeline.config import SqlSettings from pipeline.stages.products import ( ProductIndex, ProductRow, @@ -21,6 +22,14 @@ _AS_OF = date(2026, 4, 16) _SPRAY_DATE = date(2026, 4, 6) # 10 days before as_of _CROP = "grapevine" _DISEASE = "downy mildew" +_SQL = SqlSettings( + driver="ODBC Driver 17 for SQL Server", + server="localhost", + database="DBMeteo", + app_database="DBMeteo", + username="user", + password="secret", +) def _product( @@ -66,6 +75,7 @@ def _interval_open( disease_english=_DISEASE, latest_spray=(spray_date, names), as_of=as_of, + sql=_SQL, ) fetch.assert_called_once() return result @@ -82,6 +92,7 @@ class LastTreatmentIntervalOpenTests(unittest.TestCase): disease_english=_DISEASE, latest_spray=None, as_of=_AS_OF, + sql=_SQL, ) fetch.assert_not_called() self.assertFalse(result) @@ -172,6 +183,7 @@ class LastTreatmentIntervalOpenTests(unittest.TestCase): disease_english=_DISEASE, latest_spray=(_SPRAY_DATE, ["UNKNOWN"]), as_of=_AS_OF, + sql=_SQL, ) fetch.assert_not_called() self.assertFalse(result) @@ -361,7 +373,7 @@ class ProductIndexLoadTests(unittest.TestCase): "pipeline.stages.products.fetch_all", side_effect=[products, catalog], ) as fetch: - index = ProductIndex.load(MagicMock()) + index = ProductIndex.load(MagicMock(), _SQL) self.assertEqual(fetch.call_count, 2) self.assertIn("FROM products", fetch.call_args_list[0].args[1]) diff --git a/tests/test_worklist.py b/tests/test_worklist.py new file mode 100644 index 0000000..284e67e --- /dev/null +++ b/tests/test_worklist.py @@ -0,0 +1,107 @@ +"""Unit tests for database-driven worklist discovery.""" + +from __future__ import annotations + +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from pipeline.config import WorklistSettings +from pipeline.vocab import load_crop_vocab, load_disease_vocab +from pipeline.worklist import build_jobs + +_ROOT = Path(__file__).resolve().parent.parent +_CROP_VOCAB = load_crop_vocab(_ROOT / "vocab" / "crops.yaml") +_DISEASE_VOCAB = load_disease_vocab(_ROOT / "vocab" / "diseases.yaml") + + +def _row(**overrides: object) -> SimpleNamespace: + fields = { + "cmplay_id": 4012, + "cmplay_colture": 1, + "cmplay_imp": 3, + "cmplay_station": 7, + "colture_name": "Vite", + "anmod_id": 10, + "anmod_model": "PERONOSPORA", + } + fields.update(overrides) + return SimpleNamespace(**fields) + + +def _build(rows: list[SimpleNamespace], allowlist: tuple[int, ...] = ()) -> tuple[list, str]: + worklist = WorklistSettings(field_allowlist=allowlist) + with patch("pipeline.worklist.fetch_all", return_value=rows) as fetch: + jobs = build_jobs(object(), worklist, _CROP_VOCAB, _DISEASE_VOCAB) + sql = fetch.call_args.args[1] + return jobs, sql + + +class BuildJobsTests(unittest.TestCase): + def test_peronospora_on_grapevine_becomes_one_job(self) -> None: + jobs, sql = _build([_row()]) + + self.assertEqual(len(jobs), 1) + job = jobs[0] + self.assertEqual(job.field_id, 4012) + self.assertEqual(job.model_name, "PERONOSPORA") + self.assertEqual(job.disease_english, "downy mildew") + self.assertEqual(job.crop_english, "grapevine") + self.assertEqual(job.crop_italian, "Vite") + self.assertTrue(job.organic) + self.assertEqual(job.station, 7) + self.assertEqual(job.anmod_id, 10) + self.assertIn("pat_mod_AI_model = 1", sql) + self.assertIn("anmod_AI_advice = 1", sql) + self.assertIn("anmod_cmplay", sql) + + def test_allowlist_drops_other_field_ids(self) -> None: + jobs, _sql = _build( + [_row(cmplay_id=4012), _row(cmplay_id=4085, anmod_id=11)], + allowlist=(4085,), + ) + + self.assertEqual([job.field_id for job in jobs], [4085]) + + def test_unmapped_disease_and_colture_are_skipped(self) -> None: + jobs, _sql = _build([ + _row(), + _row(anmod_id=11, anmod_model="NOT_A_DISEASE"), + _row(cmplay_id=4085, anmod_id=12, colture_name="Not A Crop"), + ]) + + self.assertEqual([(job.field_id, job.anmod_id) for job in jobs], [(4012, 10)]) + + def test_blank_and_null_colture_are_skipped(self) -> None: + jobs, _sql = _build([ + _row(cmplay_id=1, anmod_id=1, colture_name=""), + _row(cmplay_id=2, anmod_id=2, colture_name=None), + _row(cmplay_id=3, anmod_id=3), + ]) + + self.assertEqual([job.field_id for job in jobs], [3]) + + def test_same_english_disease_keeps_distinct_anmod_rows(self) -> None: + jobs, _sql = _build([ + _row(anmod_id=20, anmod_model="PERONOSPORA"), + _row(anmod_id=11, anmod_model="Peronospora"), + ]) + + self.assertEqual( + [(job.anmod_id, job.model_name, job.disease_english) for job in jobs], + [ + (20, "PERONOSPORA", "downy mildew"), + (11, "Peronospora", "downy mildew"), + ], + ) + + def test_imp_other_than_3_is_not_organic(self) -> None: + jobs, _sql = _build([_row(cmplay_imp=1, cmplay_station=None)]) + + self.assertFalse(jobs[0].organic) + self.assertIsNone(jobs[0].station) + + +if __name__ == "__main__": + unittest.main()