ebc8ad26b6
- 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.
416 lines
15 KiB
Python
416 lines
15 KiB
Python
"""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
|
|
single-field CLI ran (resolve disease model -> weather/phenology/treatments ->
|
|
query synthesis -> product prefilter -> vector search -> advice generation ->
|
|
persistence), restructured to take a `worklist.Job` (crop, organic flag,
|
|
station, and anmod_id already resolved by the worklist join) plus shared
|
|
`Resources`, instead of reading `field_id` / `disease_name` from config and
|
|
opening its own SQL/Weaviate connections.
|
|
|
|
`run_job` never raises: any failure — from either this module or anything it
|
|
calls — comes back as a `failed` JobResult, so one field's bad data or a
|
|
transient API error cannot stop the batch from processing every other field.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from pipeline.errors import PipelineError
|
|
from pipeline.observability import job_span
|
|
from pipeline.prompts import pair_slug
|
|
from pipeline.resources import Resources
|
|
from pipeline.retry import call_with_sql_retry
|
|
from pipeline.stages.advice import generate_advice, insert_advice
|
|
from pipeline.stages.advice_context import (
|
|
collapse_product_json,
|
|
enrich_advice_json,
|
|
load_last_advice,
|
|
load_previous_incubation_register,
|
|
)
|
|
from pipeline.stages.assemble import (
|
|
assemble_json_for_advice_generation,
|
|
build_json_for_query_synthesis,
|
|
)
|
|
from pipeline.stages.disease import resolve_disease_forecast_by_anmod, select_prefilter_rule
|
|
from pipeline.stages.incubations import (
|
|
covered_by_key,
|
|
first_future_disease_date,
|
|
forecast_has_risk,
|
|
incubation_start_covered,
|
|
parse_register,
|
|
reconcile_register,
|
|
register_payload,
|
|
)
|
|
from pipeline.stages.llm import synthesize_queries
|
|
from pipeline.stages.observation import load_observation
|
|
from pipeline.stages.phenology import load_phenology
|
|
from pipeline.stages.products import last_treatment_interval_open, prefilter_products
|
|
from pipeline.stages.treatments import load_treatment_history
|
|
from pipeline.stages.vector import RecommendedProduct
|
|
from pipeline.stages.weather import load_weather
|
|
from pipeline.worklist import Job
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class JobResult:
|
|
"""Outcome of one job. Constructed either by `run_job` or, for jobs the
|
|
batch orchestrator decides not to run at all, directly by `pipeline.batch`
|
|
(`skipped_existing`)."""
|
|
|
|
job: Job
|
|
status: str # "ok" | "no_risk" | "dry_run" | "failed" | "skipped_existing"
|
|
duration_s: float = 0.0
|
|
error_type: str | None = None
|
|
error_message: str | None = None
|
|
recommended_count: int = 0
|
|
allowed_products_empty: bool = False
|
|
message: str | None = None
|
|
payload: dict[str, Any] | None = None
|
|
|
|
|
|
def json_default(obj: Any) -> Any:
|
|
"""`json.dump(..., default=...)` hook for the `date`/`Decimal` values in a job payload."""
|
|
if isinstance(obj, date):
|
|
return obj.isoformat()
|
|
if isinstance(obj, Decimal):
|
|
return float(obj)
|
|
raise TypeError(f"Object of type {type(obj)!r} is not JSON serializable")
|
|
|
|
|
|
def output_stem(job: Job, disease_english: str) -> str:
|
|
"""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:
|
|
"""
|
|
Run the full pipeline for one job.
|
|
|
|
Wrapped in `job_span`, the one root span per `(field, crop, disease,
|
|
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,
|
|
tracing or no tracing.
|
|
"""
|
|
started = time.monotonic()
|
|
with job_span(job, as_of, dry_run) as span:
|
|
try:
|
|
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 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__)
|
|
return JobResult(
|
|
job=job,
|
|
status="failed",
|
|
duration_s=time.monotonic() - started,
|
|
error_type=type(exc).__name__,
|
|
error_message=str(exc),
|
|
)
|
|
|
|
status = payload["status"]
|
|
allowed_products_empty = False
|
|
if status in ("ok", "no_risk"):
|
|
allowed = (payload.get("json_for_advice_generation") or {}).get("allowed_products")
|
|
allowed_products_empty = not allowed
|
|
|
|
span.set_attribute("job.status", status)
|
|
span.set_attribute(
|
|
"job.recommended_count", len(payload.get("recommended_products") or [])
|
|
)
|
|
span.set_output(payload.get("message") or "")
|
|
|
|
return JobResult(
|
|
job=job,
|
|
status=status,
|
|
duration_s=time.monotonic() - started,
|
|
recommended_count=len(payload.get("recommended_products") or []),
|
|
allowed_products_empty=allowed_products_empty,
|
|
message=payload.get("message"),
|
|
payload=payload,
|
|
)
|
|
|
|
|
|
def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -> dict[str, Any]:
|
|
settings = resources.settings
|
|
conn = resources.sql_connection()
|
|
|
|
with resources.sql_sem:
|
|
disease = resolve_disease_forecast_by_anmod(
|
|
conn,
|
|
anmod_id=job.anmod_id,
|
|
disease_name=job.model_name,
|
|
disease_english=job.disease_english,
|
|
as_of=as_of,
|
|
fallback_station=job.station,
|
|
)
|
|
|
|
if disease.station is None:
|
|
raise PipelineError(
|
|
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)
|
|
phenology = load_phenology(conn, job.field_id, as_of)
|
|
treatment_history = load_treatment_history(conn, job.field_id, as_of)
|
|
observation_by_day = load_observation(conn, job.field_id, as_of)
|
|
product_index = resources.product_index
|
|
if product_index is None:
|
|
raise RuntimeError("load_product_index() must run before run_job().")
|
|
previous_register = parse_register(
|
|
load_previous_incubation_register(
|
|
conn,
|
|
job.field_id,
|
|
disease.disease_english,
|
|
job.anmod_id,
|
|
as_of,
|
|
settings.sql,
|
|
)
|
|
)
|
|
interval_open = last_treatment_interval_open(
|
|
conn,
|
|
product_index,
|
|
crop_english=job.crop_english,
|
|
disease_english=disease.disease_english,
|
|
latest_spray=treatment_history.latest(),
|
|
as_of=as_of,
|
|
sql=settings.sql,
|
|
)
|
|
|
|
def _covers(start: date):
|
|
return incubation_start_covered(
|
|
conn,
|
|
product_index,
|
|
crop_english=job.crop_english,
|
|
disease_english=disease.disease_english,
|
|
start_date=start,
|
|
treatment_history=treatment_history,
|
|
sql=settings.sql,
|
|
)
|
|
|
|
incubation_register = reconcile_register(
|
|
previous_register,
|
|
disease.forecasts_by_day,
|
|
as_of,
|
|
_covers,
|
|
)
|
|
|
|
# Risk is an uncovered incubation, a FASE2/FASE5, or a true observation.
|
|
# A window of only covered incubations is not risk. Advice is still generated.
|
|
has_risk = forecast_has_risk(disease.forecasts_by_day, incubation_register) or any(
|
|
observation_by_day.values()
|
|
)
|
|
future_disease = first_future_disease_date(
|
|
disease.forecasts_by_day, incubation_register, as_of
|
|
)
|
|
|
|
logger.info(
|
|
"Field %s (%s/%s) 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,
|
|
)
|
|
|
|
json_for_advice_generation = assemble_json_for_advice_generation(
|
|
as_of,
|
|
weather,
|
|
phenology,
|
|
disease.forecasts_by_day,
|
|
treatment_history.recent(),
|
|
observation_by_day,
|
|
register_payload(incubation_register),
|
|
)
|
|
json_for_query_synthesis = build_json_for_query_synthesis(
|
|
as_of,
|
|
job.crop_english,
|
|
disease.disease_english,
|
|
weather,
|
|
phenology,
|
|
future_disease,
|
|
)
|
|
|
|
payload: dict[str, Any] = {
|
|
"run": {
|
|
"as_of": as_of.isoformat(),
|
|
"field_id": job.field_id,
|
|
"crop": job.crop_english,
|
|
"crop_italian": job.crop_italian,
|
|
"disease": disease.disease_english,
|
|
"disease_raw": disease.disease_raw,
|
|
"organic": job.organic,
|
|
"station": disease.station,
|
|
"anmod_id": disease.anmod_id,
|
|
"has_risk": has_risk,
|
|
"first_future_disease_date": (
|
|
future_disease.isoformat() if future_disease else None
|
|
),
|
|
"prefilter_rule": None,
|
|
"llm_provider": settings.llm_query_synthesis.provider,
|
|
},
|
|
"json_for_advice_generation": json_for_advice_generation,
|
|
"json_for_query_synthesis": json_for_query_synthesis,
|
|
"generated_queries": [],
|
|
"candidate_count": 0,
|
|
"recommended_products": [],
|
|
"last_advice": None,
|
|
"advice": None,
|
|
"advice_input_json": None,
|
|
"advice_inserted": False,
|
|
"status": "ok",
|
|
"message": None,
|
|
"output_stem": output_stem(job, disease.disease_english),
|
|
}
|
|
|
|
if dry_run:
|
|
payload["status"] = "dry_run"
|
|
payload["message"] = (
|
|
"Dry run: skipped LLM, vector search, advice generation, and DB write."
|
|
)
|
|
return payload
|
|
|
|
recommended: list[RecommendedProduct] = []
|
|
if has_risk and interval_open:
|
|
logger.info(
|
|
"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)
|
|
queries = synthesize_queries(
|
|
settings,
|
|
query_system_prompt,
|
|
json_for_query_synthesis,
|
|
llm_call=resources.call_llm,
|
|
)
|
|
payload["generated_queries"] = queries
|
|
|
|
prefilter_rule = select_prefilter_rule(
|
|
disease.forecasts_by_day,
|
|
observation_by_day,
|
|
as_of,
|
|
covered_by_key(incubation_register),
|
|
)
|
|
payload["run"]["prefilter_rule"] = prefilter_rule
|
|
candidates = prefilter_products(
|
|
product_index,
|
|
crop_english=job.crop_english,
|
|
disease_english=disease.disease_english,
|
|
organic=job.organic,
|
|
rule=prefilter_rule,
|
|
)
|
|
payload["candidate_count"] = len(candidates)
|
|
|
|
recommended = resources.search_products(queries, candidates)
|
|
payload["recommended_products"] = [
|
|
{
|
|
"product_name": item.product_name,
|
|
"registration_number": item.registration_number,
|
|
"distance": item.distance,
|
|
"query_index": item.query_index,
|
|
"product_id": item.product_id,
|
|
}
|
|
for item in recommended
|
|
]
|
|
logger.info(
|
|
"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 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,
|
|
job.anmod_id,
|
|
as_of,
|
|
settings.sql,
|
|
)
|
|
enrich_advice_json(
|
|
json_for_advice_generation,
|
|
conn=conn,
|
|
builder=builder,
|
|
recommended=recommended,
|
|
last_advice_summary=last_summary,
|
|
last_advice_treatments=last_treatments,
|
|
last_advice_date=last_date,
|
|
treatment_history=treatment_history,
|
|
)
|
|
payload["last_advice"] = json_for_advice_generation["last_advice"]
|
|
|
|
advice_system_prompt, advice_user_prompt = resources.prompts.advice_prompts(
|
|
job.crop_english, disease.disease_english
|
|
)
|
|
result = generate_advice(
|
|
settings,
|
|
advice_system_prompt,
|
|
advice_user_prompt,
|
|
json_for_advice_generation,
|
|
as_of,
|
|
llm_call=resources.call_llm,
|
|
)
|
|
advice_input_json = collapse_product_json(json_for_advice_generation)
|
|
|
|
def _insert() -> None:
|
|
with resources.sql_sem:
|
|
insert_advice(
|
|
conn,
|
|
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,
|
|
)
|
|
|
|
call_with_sql_retry(settings.retry, _insert)
|
|
|
|
payload["advice"] = result.as_dict()
|
|
payload["advice_input_json"] = advice_input_json
|
|
payload["advice_inserted"] = True
|
|
payload["message"] = (
|
|
f"Returned {len(recommended)} recommended product(s); "
|
|
f"advice stored with is_treatment_recommended={int(result.is_treatment_recommended)}."
|
|
)
|
|
return payload
|