98afec550d
- Updated README.md to clarify the handling of historical/test dates and the conditions for generating advice. - Modified job.py to incorporate observation data into risk assessment and product prefiltering. - Introduced a new function in disease.py to select the appropriate prefilter rule based on observation and FASE data. - Enhanced product filtering logic in products.py to accommodate new prefilter rules. - Updated assemble.py to include observation data in JSON assembly for advice generation. - Improved documentation in the grapevine downy mildew system prompt to explain the significance of INCUBAZPRIMARIA and observation data.
335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""Run the full single-field pipeline for one (field, crop, disease) 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,
|
|
)
|
|
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.llm import synthesize_queries
|
|
from pipeline.stages.observation import load_observation
|
|
from pipeline.stages.phenology import load_phenology
|
|
from pipeline.stages.products import prefilter_products
|
|
from pipeline.stages.treatments import load_applied_treatments
|
|
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`, `skipped_deadline`)."""
|
|
|
|
job: Job
|
|
status: str # "ok" | "no_risk" | "dry_run" | "failed" | "skipped_existing" | "skipped_deadline"
|
|
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) for this job's per-field output JSON."""
|
|
return f"field{job.field_id}__{pair_slug(job.crop_english, disease_english)}"
|
|
|
|
|
|
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,
|
|
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",
|
|
job.field_id,
|
|
job.crop_english,
|
|
job.disease_english,
|
|
)
|
|
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}: 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)
|
|
applied_treatments = load_applied_treatments(conn, job.field_id, as_of)
|
|
observation_by_day = load_observation(conn, job.field_id, as_of)
|
|
|
|
# has_risk gates query synthesis and the product prefilter: it is true if
|
|
# any day is FASE2 / FASE5 / INCUBAZPRIMARIA (disease.has_risk, already
|
|
# covers all three via forecasts_by_day) or the observation factor is
|
|
# true for as_of or any of the past 5 days.
|
|
has_risk = disease.has_risk or any(observation_by_day.values())
|
|
|
|
logger.info(
|
|
"Field %s (%s/%s): station=%s has_risk=%s first_future=%s",
|
|
job.field_id,
|
|
job.crop_english,
|
|
job.disease_english,
|
|
disease.station,
|
|
has_risk,
|
|
disease.first_future_disease_date,
|
|
)
|
|
|
|
json_for_advice_generation = assemble_json_for_advice_generation(
|
|
as_of,
|
|
weather,
|
|
phenology,
|
|
disease.forecasts_by_day,
|
|
applied_treatments,
|
|
observation_by_day,
|
|
)
|
|
json_for_query_synthesis = build_json_for_query_synthesis(
|
|
as_of,
|
|
job.crop_english,
|
|
disease.disease_english,
|
|
weather,
|
|
phenology,
|
|
disease.first_future_disease_date,
|
|
)
|
|
|
|
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": (
|
|
disease.first_future_disease_date.isoformat()
|
|
if disease.first_future_disease_date
|
|
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,
|
|
"sent_information": 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:
|
|
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)
|
|
payload["run"]["prefilter_rule"] = prefilter_rule
|
|
candidates = prefilter_products(
|
|
resources.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: recommended products: %s",
|
|
job.field_id,
|
|
[p["product_name"] for p in payload["recommended_products"]],
|
|
)
|
|
else:
|
|
payload["status"] = "no_risk"
|
|
logger.warning(
|
|
"Field %s: no FASE2, FASE5, or INCUBAZPRIMARIA 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,
|
|
)
|
|
|
|
builder = resources.product_json_builder(job.crop_english, disease.disease_english)
|
|
with resources.sql_sem:
|
|
last_summary, last_treatments = load_last_advice(
|
|
conn, job.field_id, disease.disease_english, as_of
|
|
)
|
|
enrich_advice_json(
|
|
json_for_advice_generation,
|
|
conn=conn,
|
|
builder=builder,
|
|
recommended=recommended,
|
|
last_advice_summary=last_summary,
|
|
last_advice_treatments=last_treatments,
|
|
)
|
|
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,
|
|
llm_call=resources.call_llm,
|
|
)
|
|
sent_information = collapse_product_json(json_for_advice_generation)
|
|
|
|
def _insert() -> None:
|
|
with resources.sql_sem:
|
|
insert_advice(
|
|
conn,
|
|
run_date=as_of,
|
|
field_id=job.field_id,
|
|
disease_english=disease.disease_english,
|
|
sent_information=sent_information,
|
|
result=result,
|
|
)
|
|
|
|
call_with_sql_retry(settings.retry, _insert)
|
|
|
|
payload["advice"] = result.as_dict()
|
|
payload["sent_information"] = sent_information
|
|
payload["advice_inserted"] = True
|
|
payload["message"] = (
|
|
f"Returned {len(recommended)} recommended product(s); "
|
|
f"advice stored with apply_treatment={int(result.apply_treatment)}."
|
|
)
|
|
return payload
|