"""Inference core (no FastAPI): OpenAI-compatible client, METADATA-35 image sizing, closed taxonomy lists, the METADATA-45 confidence gate, and the METADATA-66 log redaction. Routers in inference.py compose these. """ from __future__ import annotations import base64 import io import json import time from typing import Any import httpx from PIL import Image from .crypto import decrypt _INFER_MAX_LONG_EDGE = 1024 # METADATA-45: confidence gate threshold. _THRESHOLD = 0.7 # METADATA-66 / #87: inference_log payload cap (safety net after redaction). _LOG_PAYLOAD_CAP = 200_000 def fit_long_edge_png(png_bytes: bytes, max_long_edge: int = _INFER_MAX_LONG_EDGE) -> bytes: """Downscale a PNG so max(width, height) <= max_long_edge. Never upscale. Used only for the inference request payload; stored processed_blob / Woo media stay full size. Unreadable bytes are returned unchanged so a bad blob cannot crash Call 1. """ if not png_bytes or max_long_edge < 1: return png_bytes try: im = Image.open(io.BytesIO(png_bytes)) im.load() except Exception: return png_bytes w, h = im.size long_edge = max(w, h) if long_edge <= max_long_edge: return png_bytes scale = max_long_edge / float(long_edge) nw = max(1, int(round(w * scale))) nh = max(1, int(round(h * scale))) if im.mode not in ("RGB", "RGBA", "L", "LA"): im = im.convert("RGBA") out = im.resize((nw, nh), Image.Resampling.LANCZOS) buf = io.BytesIO() out.save(buf, format="PNG", optimize=True) return buf.getvalue() def _image_part_from_processed(png_bytes: bytes) -> dict: """One OpenAI-style image_url part from a processed PNG (METADATA-35 sized).""" fitted = fit_long_edge_png(png_bytes) b64 = base64.b64encode(fitted).decode() return {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}} class InferenceError(Exception): pass class InferenceConnectionError(InferenceError): pass class InferenceClient: """Minimal OpenAI-compatible chat client. `chat` is the mock seam.""" def __init__(self, base_url: str, api_key: str, model: str, timeout: float = 60.0): self.base_url = base_url.rstrip("/") self.api_key = api_key self.model = model self.timeout = timeout def chat(self, messages: list[dict]) -> str: url = f"{self.base_url}/chat/completions" body = json.dumps({"model": self.model, "messages": messages}).encode("utf-8") try: with httpx.Client(timeout=self.timeout) as client: resp = client.post( url, content=body, headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}, ) except (httpx.TimeoutException, httpx.TransportError) as e: raise InferenceConnectionError(f"inference unreachable: {e}") from e if resp.status_code >= 400: raise InferenceError(f"inference HTTP {resp.status_code}") data = json.loads(resp.read().decode("utf-8")) return data["choices"][0]["message"]["content"] def _inference_creds(conn) -> tuple[str, str, str]: b = conn.execute("SELECT value_enc FROM settings WHERE key='inference_base_url'").fetchone() k = conn.execute("SELECT value_enc FROM settings WHERE key='inference_api_key'").fetchone() m = conn.execute("SELECT value FROM settings WHERE key='inference_model'").fetchone() base = decrypt(b[0]) if b and b[0] else "" key = decrypt(k[0]) if k and k[0] else "" model = m[0] if m and m[0] else "" return base, key, model def _enabled(conn, key: str) -> bool: row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone() return bool(row and row[0] in ("true", "True", "1")) def _taxonomy_lists(conn) -> dict: """Closed Call 1 lists only (#86 / METADATA-40). Keys match the seed prompt. No grade/thickness/moisture (owner fields). Groupers resolved by slug only (QUALITY-105 / METADATA-5): 'species' and 'wcat'. Feature tag names are shown without the feat- prefix; ids are Woo/local tag ids. """ out: dict[str, list] = { "species": [], "wood_categories": [], "edge_terms": [], "figure_terms": [], "feature_tags": [], } sp = conn.execute("SELECT id FROM woo_categories WHERE slug='species'").fetchone() if sp: for r in conn.execute("SELECT id,name FROM woo_categories WHERE parent_id=?", (sp[0],)).fetchall(): out["species"].append({"id": r[0], "name": r[1]}) wc = conn.execute("SELECT id FROM woo_categories WHERE slug='wcat'").fetchone() if wc: kids = [r[0] for r in conn.execute("SELECT id FROM woo_categories WHERE parent_id=?", (wc[0],)).fetchall()] leaf_ids = [] for cid in kids: has_kids = conn.execute("SELECT 1 FROM woo_categories WHERE parent_id=? LIMIT 1", (cid,)).fetchone() if has_kids: leaf_ids.extend(r[0] for r in conn.execute("SELECT id FROM woo_categories WHERE parent_id=?", (cid,)).fetchall()) else: leaf_ids.append(cid) if leaf_ids: q = ",".join("?" * len(leaf_ids)) for r in conn.execute(f"SELECT id,name FROM woo_categories WHERE id IN ({q})", leaf_ids).fetchall(): out["wood_categories"].append({"id": r[0], "name": r[1]}) for attr_name, key in (("Edge type", "edge_terms"), ("Figure", "figure_terms")): rows = conn.execute( "SELECT t.id,t.name FROM woo_attribute_terms t JOIN woo_attributes a ON t.attribute_id=a.id " "WHERE lower(a.name)=lower(?)", (attr_name,), ).fetchall() out[key] = [{"id": r[0], "name": r[1]} for r in rows] for r in conn.execute("SELECT id,name FROM woo_tags WHERE slug LIKE 'feat-%'").fetchall(): out["feature_tags"].append({"id": r[0], "name": _feat_display_name(r[1])}) return out def _feat_display_name(name: str) -> str: """Strip leading feat- / feat_ for the model (#86); Woo id stays authoritative.""" n = (name or "").strip() low = n.lower() for prefix in ("feat-", "feat_"): if low.startswith(prefix): return n[len(prefix):].strip() or n return n def _parse_json(raw: str) -> dict: try: s = raw.strip() if s.startswith("```"): s = s.split("```")[1] if s.startswith("json"): s = s[4:] parsed = json.loads(s) return parsed if isinstance(parsed, dict) else {} except (ValueError, IndexError): return {} def _gate(parsed: dict) -> dict: """Apply the 0.7 confidence gate; below-threshold fields are omitted (METADATA-45). Confidence is read from flat '_confidence' keys; as a safety net it also understands nested 'confidence: {: n}' some models emit. Grade is never suggested (METADATA-40); free-text observation bags are dropped (#86) — feature tags are ids only. """ conf_key = { "species_id": "species_confidence", "wood_category_ids": "wood_category_confidence", "edge_type_term_id": "edge_confidence", "figure_term_ids": "figure_confidence", "feature_tag_ids": "feature_tag_confidence", } conf_obj = parsed.get("confidence") if isinstance(parsed.get("confidence"), dict) else {} def _conf(field: str) -> Any: k = conf_key.get(field, f"{field}_confidence") return conf_obj[k] if k in conf_obj else parsed.get(k) def keep(field: str) -> Any: conf = _conf(field) if isinstance(conf, (int, float)) and conf >= _THRESHOLD and field in parsed: return parsed[field] return None out: dict[str, Any] = {} for field in conf_key: if v := keep(field): out[field] = v return out def _photo_meta_row(photo_id: str, seq: int, blob: bytes) -> dict[str, Any]: """Metadata for one photo slot in the inference log (#87).""" edge = None try: im = Image.open(io.BytesIO(blob)) im.load() edge = max(im.size) except Exception: edge = None return { "photo_id": photo_id, "seq": seq, "bytes": len(blob) if blob is not None else 0, "infer_long_edge_px": edge, "label": f"photo seq={seq} id={photo_id}", } def _slab_photos(conn, slab_id: str) -> tuple[list[dict], list[dict]]: rows = conn.execute( "SELECT id, seq, processed_blob FROM slab_photos " "WHERE slab_id=? AND processed_blob IS NOT NULL ORDER BY seq LIMIT 8", (slab_id,), ).fetchall() image_parts = [_image_part_from_processed(p["processed_blob"]) for p in rows] photo_meta = [_photo_meta_row(p["id"], p["seq"], p["processed_blob"]) for p in rows] return image_parts, photo_meta def _redact_request_for_log(messages: list, photo_meta: list[dict] | None = None) -> list: """Replace image data-URLs with photo refs for inference_log (#87 / METADATA-66). Wire messages keep base64 for the model; only the logged copy is redacted. Text parts are kept in full. Image parts become type=image_ref using photo_meta by image-part index. """ meta = list(photo_meta or []) out: list = [] for msg in messages: if not isinstance(msg, dict) or not isinstance(msg.get("content"), list): out.append(msg) continue new_parts: list = [] img_i = 0 for part in msg["content"]: if not isinstance(part, dict): new_parts.append(part) continue ptype = part.get("type") if ptype == "text": new_parts.append({"type": "text", "text": part.get("text", "")}) continue if ptype == "image_url": url = "" iu = part.get("image_url") if isinstance(iu, dict): url = str(iu.get("url") or "") if url.startswith("data:image") or "base64," in url: ref: dict[str, Any] = { "type": "image_ref", "photo_id": None, "seq": None, "bytes": None, "infer_long_edge_px": None, "label": f"photo slot={img_i}", } if img_i < len(meta): ref.update(meta[img_i]) new_parts.append(ref) img_i += 1 else: new_parts.append({"type": "image_ref", "url": url[:200], "label": f"external image slot={img_i}"}) img_i += 1 continue new_parts.append(part) out.append({**msg, "content": new_parts}) return out def _log_call(conn, slab_id: str, call: int, rev_id: int | None, model: str, request: list, response: str, latency_ms: int, photo_meta: list[dict] | None = None) -> None: """Append inference_log with redacted request (no image bytes) — #87 / METADATA-66.""" payload = json.dumps(_redact_request_for_log(request, photo_meta), ensure_ascii=False) if len(payload) > _LOG_PAYLOAD_CAP: payload = payload[: _LOG_PAYLOAD_CAP - 20] + "\n/* truncated */" conn.execute( "INSERT INTO inference_log(slab_id, call, prompt_revision_id, model, request_payload, response_payload, latency_ms, created_at) " "VALUES (?,?,?,?,?,?,?,?)", (slab_id, call, rev_id, model, payload, response, latency_ms, time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())), ) conn.commit()