""" ChillTrack API — Gemini vision extraction for chiller/freezer gauge readings. Follows the google-genai SDK usage pattern from NeoFix-API/main.py and iristrack-api/main.py: `from google import genai`, `genai.Client(api_key=...)`, `types.Part.from_bytes(...)` for image bytes, `response_mime_type='application/json'` plus `response_schema=...` for structured output, and parsing the response by joining `resp.candidates[0].content.parts`. The Gemini API key env var is literally named `Gemini` (not GEMINI_API_KEY), matching every sibling repo. """ import os import json from google import genai from google.genai import types MODEL_NAME = "gemini-3.1-flash-lite" PROMPT_VERSION = "v1" # response_schema is an OpenAPI 3.0 subset — no additionalProperties, no $ref. READING_SCHEMA = { "type": "object", "properties": { "value_c": { "type": "number", "description": "The air/chamber temperature reading in Celsius, read from the gauge or LCD.", }, "confidence": { "type": "number", "description": "Confidence in the reading, 0.0 to 1.0.", }, "detected_display_type": { "type": "string", "enum": ["analog_dial", "digital_lcd", "unclear"], }, "raw_text_seen": { "type": "string", "description": "Any raw text/digits actually visible on the display, verbatim.", }, "setpoint_text_seen": { "type": "string", "nullable": True, "description": "The SETPOINT/Code field text if a digital controller shows one, else null.", }, "image_quality": { "type": "string", "enum": ["ok", "blurry", "glare", "dusty", "too_dark", "obstructed"], }, "needle_angle_estimate_deg": { "type": "number", "nullable": True, "description": "Estimated needle angle in degrees for an analog dial, else null.", }, "notes": { "type": "string", "description": "Any other observations useful for a human reviewing this reading.", }, }, "required": [ "value_c", "confidence", "detected_display_type", "raw_text_seen", "image_quality", "notes", ], } _client = None def get_client(): """Returns a cached google-genai Client, built from the `Gemini` env var.""" global _client if _client is not None: return _client api_key = os.environ.get("Gemini") if not api_key: print("FATAL: The 'Gemini' environment variable for the API key is not set.") exit(1) try: _client = genai.Client(api_key=api_key) print("Google GenAI Client initialized successfully.") except Exception as e: print(f"FATAL: Error initializing GenAI Client: {e}") exit(1) return _client def build_extraction_prompt(unit: dict) -> str: """Builds the vision-extraction prompt for one unit's photo. `unit` is a ctk_units doc: {name, setpoint_c, tolerance_c, display_type, ...}. """ setpoint_c = unit.get("setpoint_c") tolerance_c = unit.get("tolerance_c") name = unit.get("name", "this unit") return f"""You are reading a temperature gauge photographed by a meat-plant technician for cold-chain compliance logging. The unit being photographed is: "{name}". This plant has two kinds of displays. Identify which one is in the photo and read it according to its type: 1. ANALOG DIAL GAUGE: a round dial with a needle pointing at a number on a -40 to +40 degrees Celsius scale. Read the exact number the needle points at (interpolate between marked graduations if needed). Report your estimate of the needle's angle in degrees if you can. 2. DIGITAL LCD REEFER CONTROLLER: an electronic panel with TWO separate fields — a "SETPOINT" or "Code" field on the LEFT, and an "AIR TEMPERATURE" or "Data" field on the RIGHT. You must extract ONLY the right-hand air-temperature field as the reading. Do NOT report the setpoint/code field as the reading — but do report what it shows in `setpoint_text_seen` so a human can cross-check it. This unit's expected setpoint is around {setpoint_c} degrees Celsius, +/- {tolerance_c} degrees Celsius tolerance. Use this ONLY as a sanity check on which field/needle you are reading — report what you actually see even if it differs from this expectation. Do not force the reading to match the expected setpoint. If the image is dusty, glared, poorly lit, or partially obscured, still give your best numeric estimate. Do not refuse to answer. Instead, lower `confidence` accordingly and set `image_quality` to describe the problem (blurry / glare / dusty / too_dark / obstructed). Only use `image_quality: "ok"` when the display is clearly and fully readable. Return your reading as structured JSON matching the provided schema.""" # response_schema for a whole handwritten paper form: one photo can cover dozens # of units across all 12 slot columns. Handwriting OCR at that density is # inherently less reliable than a single close-up gauge photo, which is exactly # why every extracted row goes through an admin review/correct screen before # anything is written to ctk_readings — this schema is a first-pass draft, not # a final record. FORM_SCHEMA = { "type": "object", "properties": { "shift_date_seen": { "type": "string", "nullable": True, "description": "A date written on the form header, as YYYY-MM-DD if you can determine it, else null.", }, "rows": { "type": "array", "description": "One entry per legible handwritten cell. Skip cells that are blank or fully illegible.", "items": { "type": "object", "properties": { "unit_name_seen": { "type": "string", "description": "The row's unit label exactly as written or printed on the form, verbatim.", }, "matched_unit_name": { "type": "string", "nullable": True, "description": "The single best match for this row from the provided list of known unit names, or null if none plausibly matches.", }, "slot_index": { "type": "integer", "nullable": True, "description": "0-11 index into the 12 time-slot columns (08:00,10:00,12:00,14:00,16:00,18:00,20:00,22:00,00:00,02:00,04:00,06:00), or null if the column is unclear.", }, "value_c": { "type": "number", "nullable": True, "description": "The handwritten numeric temperature in that cell, or null if it can't be read as a number.", }, "confidence": {"type": "number", "description": "0.0 to 1.0."}, "raw_text_seen": {"type": "string", "description": "The literal characters seen in the cell."}, }, "required": ["unit_name_seen", "slot_index", "value_c", "confidence", "raw_text_seen"], }, }, }, "required": ["rows"], } def build_form_extraction_prompt(unit_names: list) -> str: """Builds the prompt for bulk-extracting a photographed paper form. `unit_names` is the list of known active ctk_units names, given as context so the model can match handwritten row labels against real units instead of inventing names. """ names_block = "\n".join(f"- {n}" for n in unit_names) return f"""You are transcribing a photographed paper "ROOM TEMPERATURE RECORD" form from a meat-processing plant's cold-chain log. The form is a grid: each row is a chiller, freezer, or container, and there are 12 fixed time-slot columns across the top, in this exact order: 08:00, 10:00, 12:00, 14:00, 16:00, 18:00, 20:00, 22:00, 00:00, 02:00, 04:00, 06:00 (an 8am-to-8am shift). Entries are handwritten in pen or pencil. The plant's known unit names are: {names_block} For every cell that has a legible handwritten number in it, return one row with: the row's label exactly as written (unit_name_seen), your best match for that label against the list above (matched_unit_name, or null if nothing plausibly matches), which of the 12 columns it's in (slot_index, 0-11, matching the order above), and the numeric value (value_c). Skip cells that are blank or you truly cannot read. If the form's header has a date written on it, report your best reading of it as shift_date_seen (YYYY-MM-DD), else null. Handwriting varies a lot in legibility. Do your best on every cell rather than skipping anything you can make a reasonable guess at, but lower confidence accordingly rather than guessing wildly. A human reviews every row you return before anything is saved, so it's fine to be uncertain as long as it's reflected in confidence. Return your transcription as structured JSON matching the provided schema.""" def extract_form(image_bytes: bytes, mime_type: str, unit_names: list) -> dict: """Calls Gemini vision on a photographed paper form and returns the parsed {shift_date_seen, rows: [...]} dict. See FORM_SCHEMA for the row shape.""" client = get_client() prompt = build_form_extraction_prompt(unit_names) resp = client.models.generate_content( model=MODEL_NAME, contents=[ prompt, types.Part.from_bytes(data=image_bytes, mime_type=mime_type), ], config=types.GenerateContentConfig( response_mime_type="application/json", response_schema=FORM_SCHEMA, ), ) raw_json = "".join(part.text or "" for part in resp.candidates[0].content.parts) parsed = json.loads(raw_json) parsed["model"] = MODEL_NAME parsed["prompt_version"] = PROMPT_VERSION return parsed def extract_reading(image_bytes: bytes, mime_type: str, unit: dict) -> dict: """Calls Gemini vision on the photo and returns the parsed reading dict.""" client = get_client() prompt = build_extraction_prompt(unit) resp = client.models.generate_content( model=MODEL_NAME, contents=[ prompt, types.Part.from_bytes(data=image_bytes, mime_type=mime_type), ], config=types.GenerateContentConfig( response_mime_type="application/json", response_schema=READING_SCHEMA, ), ) raw_json = "".join(part.text or "" for part in resp.candidates[0].content.parts) parsed = json.loads(raw_json) parsed["model"] = MODEL_NAME parsed["prompt_version"] = PROMPT_VERSION return parsed