r""" extraction.py — single source of truth for the AIM Composites extraction pipeline. This module consolidates the Gemini prompt/schema and all post-extraction processing that used to live (and drift) inside ``batch_ingest.py`` and the Streamlit app repo's ``page_files/categorized/page6.py`` / ``Backend/Pdf_DataExtraction.py``. Pipeline shape:: pdf_bytes -> extract_from_pdf() # Gemini structured output -> Extraction -> verify_against_text() # ground every value in the PDF text (Task 1) -> to_rows() # Extraction -> list[PropertyRow] (Task 2/3/5/6) -> (caller inserts rows, carrying `status`/`flag_reason`) Design notes ------------ * The model is asked for a *list* of materials, each with structured, unit-aware property values plus a verbatim ``source_quote`` and ``page``. * Numbers are parsed into ``value_num`` / ``value_min`` / ``value_max`` / ``qualifier`` and normalized to a canonical unit (``unit_canonical`` + ``value_si``) with :mod:`pint`. Plausibility is range-checked *after* conversion, so a GPa-vs-MPa mix no longer false-flags. * Nothing is silently dropped: every row carries a ``status`` and, when not ``ok``, a ``flag_reason``. Keep ``temperature=0`` and bump :data:`PROMPT_VERSION` on any prompt change. """ from __future__ import annotations import base64 import dataclasses import json import logging import os import re import time import unicodedata from typing import Any, Optional import requests try: # PyMuPDF — used for text grounding (Task 1) and scanned detection (Task 10) import fitz # type: ignore except Exception: # pragma: no cover - import guard fitz = None # type: ignore try: import pint # unit normalization (Task 3) except Exception: # pragma: no cover - import guard pint = None # type: ignore log = logging.getLogger("extraction") # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- # Stable release only — the previous pin ("gemini-2.5-flash-preview-09-2025", # a *preview*) was shut down by Google on 2026-02-17 and silently killed every # extraction call. Previews get retired with ~2 weeks notice; stable models # don't. Overridable via env so a future swap needs no code change. GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "gemini-3.5-flash") # Model for the figure (vision) calls: classify + mining. Empty = same as # GEMINI_MODEL. Classification is an easy task (one call per PDF picking a # figure kind), so a cheaper model here costs little quality. The agent sets # both from Run Control each cycle; read them at call time via text_model() / # vision_model(), never through a name imported at module load. GEMINI_VISION_MODEL = os.environ.get("GEMINI_VISION_MODEL", "").strip() PROMPT_VERSION = "2.0" def text_model() -> str: """The model the text extraction call uses right now.""" return (GEMINI_MODEL or "gemini-3.5-flash").strip() def vision_model() -> str: """The model the figure classify / mining calls use right now.""" return (GEMINI_VISION_MODEL or "").strip() or text_model() # How much the model may "think" per call. Thinking tokens are billed as # output (6x the input price on gemini-3.5-flash): on 28 Sep 2026 a 25-paper # cycle cost ~$7.50, $6.69 of it output tokens, ~30k per PDF where the extraction JSON itself # is ~4k. Values: "dynamic" (the API default: unbounded), "low" / "minimal" / # "high" (Gemini 3 thinking levels), "off" (budget 0), or an integer token # budget (Gemini 2.x style). The agent sets this from its config each cycle # (Run Control); GEMINI_THINKING is the first-boot default and what # batch_ingest.py uses. If the model rejects the option (400), the call is # repeated without it and the option is dropped for the rest of the process. THINKING = os.environ.get("GEMINI_THINKING", "low").strip().lower() or "dynamic" _THINKING_UNSUPPORTED = False GEMINI_URL_TEMPLATE = ( "https://generativelanguage.googleapis.com/v1beta/models/" "{model}:generateContent?key={key}" ) GEMINI_UPLOAD_URL = ( "https://generativelanguage.googleapis.com/upload/v1beta/files?key={key}" ) GEMINI_FILE_STATUS_URL = ( "https://generativelanguage.googleapis.com/v1beta/{name}?key={key}" ) REQUEST_TIMEOUT_S = 300 # Transient-failure retry policy, mirrored from pdf_crawler.http_get(). The # Gemini free tier returns 429s under load; a request that fails once is often # fine moments later. Retry connection errors and 429/5xx, honoring Retry-After. MAX_RETRIES = 3 RETRY_STATUS = frozenset({429, 500, 502, 503, 504}) BACKOFF_BASE = 2.0 # seconds; exponential per attempt # Inline a PDF up to this size; above it, upload via the Gemini File API so the # base64-inflated request stays under the generateContent size limit (Task 10). INLINE_PDF_LIMIT_BYTES = 15 * 1024 * 1024 INLINE_PAGE_LIMIT = 80 # Average extractable characters/page below this => treat as scanned/image-only. SCANNED_TEXT_PER_PAGE = 50 # --- enums (Task 8) -------------------------------------------------------- # Aligned with the app repo's page6.py PROPERTY_CATEGORIES. SECTION_ENUM = [ "Mechanical", "Thermal", "Electrical", "Physical", "Optical", "Rheological", "Processing", "Descriptive", "Composition/Reinforcement", "Architecture/Structure", ] MATERIAL_CLASS_ENUM = ["Polymer", "Fiber", "Composite"] # --- schema (Tasks 2/3/8) -------------------------------------------------- _PROPERTY_SCHEMA: dict[str, Any] = { "type": "OBJECT", "properties": { "section": {"type": "STRING", "enum": SECTION_ENUM}, "property_name": {"type": "STRING"}, "value_raw": {"type": "STRING"}, "value_num": {"type": "NUMBER", "nullable": True}, "value_min": {"type": "NUMBER", "nullable": True}, "value_max": {"type": "NUMBER", "nullable": True}, "qualifier": {"type": "STRING"}, "unit": {"type": "STRING"}, "test_condition": {"type": "STRING"}, "comments": {"type": "STRING"}, "source_quote": {"type": "STRING"}, "page": {"type": "INTEGER"}, }, "required": ["section", "property_name", "value_raw", "unit", "source_quote", "page"], } EXTRACTION_SCHEMA: dict[str, Any] = { "type": "OBJECT", "properties": { "materials": { "type": "ARRAY", "items": { "type": "OBJECT", "properties": { "material_name": {"type": "STRING"}, "material_abbreviation": {"type": "STRING"}, "material_class": {"type": "STRING", "enum": MATERIAL_CLASS_ENUM}, "trade_grade": {"type": "STRING"}, "manufacturer": {"type": "STRING"}, "matrix": {"type": "STRING"}, "fiber": {"type": "STRING"}, "fiber_volume_fraction": {"type": "STRING"}, "properties": {"type": "ARRAY", "items": _PROPERTY_SCHEMA}, }, "required": ["material_name", "material_class", "properties"], }, } }, "required": ["materials"], } EXTRACTION_PROMPT = ( "You are an expert materials scientist. From the attached PDF, extract every " "distinct material it characterizes as a list under `materials`. A datasheet " "or paper may describe MORE THAN ONE material (e.g. a neat resin and its " "composite, or several grades) — return each as its own entry with only its " "own properties. Do NOT merge two materials into one.\n\n" "For each material provide:\n" "- material_name (generic material, e.g. 'isotactic polypropylene')\n" "- material_abbreviation\n" "- material_class: one of Polymer | Fiber | Composite. Choose Composite when a " "matrix is reinforced with fibers (laminate, prepreg, CF/PEEK, glass-filled, a " "reported fiber volume fraction); Fiber for a bare fiber/yarn/tow datasheet; " "Polymer otherwise.\n" "- trade_grade (commercial/trade name; '' if absent)\n" "- manufacturer (company; '' if absent)\n" "- matrix, fiber, fiber_volume_fraction (e.g. '55%') — for composites; '' otherwise\n\n" "Extract ALL numeric and descriptive properties across every category. For " "each property return:\n" "- section: one of " + ", ".join(SECTION_ENUM) + "\n" "- property_name\n" "- value_raw: the value EXACTLY as printed, including ranges and qualifiers " "(e.g. '100-120', '≤ -18.0', '~3.5')\n" "- value_num: the single numeric value, or null if it is a range/non-numeric\n" "- value_min, value_max: the low/high of a range, else null\n" "- qualifier: one of '', '<', '<=', '>', '>=', '~', '±'\n" "- unit: the unit exactly as printed ('' if dimensionless)\n" "- test_condition (e.g. '23 °C, 50% RH'; '' if none)\n" "- comments ('' if none)\n" "- source_quote: the VERBATIM sentence or table cell from the PDF that states " "this value. Copy it character-for-character; do not paraphrase.\n" "- page: the 1-based PDF page number the value appears on.\n\n" "Never invent values. If a value is not in the document, do not include it. " "Respond ONLY with valid JSON following the schema." ) # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclasses.dataclass class Property: section: str property_name: str value_raw: str unit: str = "" value_num: Optional[float] = None value_min: Optional[float] = None value_max: Optional[float] = None qualifier: str = "" test_condition: str = "" comments: str = "" source_quote: str = "" page: Optional[int] = None # filled by verify_against_text() / canonicalization: unit_canonical: str = "" value_si: Optional[float] = None status: str = "ok" flag_reason: str = "" @dataclasses.dataclass class Material: material_name: str material_abbreviation: str = "" material_class: str = "" trade_grade: str = "" manufacturer: str = "" matrix: str = "" fiber: str = "" fiber_volume_fraction: str = "" properties: list[Property] = dataclasses.field(default_factory=list) @dataclasses.dataclass class Extraction: materials: list[Material] = dataclasses.field(default_factory=list) model: str = dataclasses.field(default_factory=text_model) prompt_version: str = PROMPT_VERSION doc_status: str = "ok" # "ok" | "scanned_no_text" | "empty_extraction" # Gemini usage for the text call (0 when no call was made). tokens_out # includes the model's thinking tokens, which are billed as output; # tokens_thinking is that thinking share on its own (for cost analysis). tokens_in: int = 0 tokens_out: int = 0 tokens_thinking: int = 0 def usage_detail(resp: Any) -> tuple[int, int, int]: """(tokens_in, tokens_out, tokens_thinking) from a generateContent response's ``usageMetadata``. tokens_out is billed output (candidates + thinking); tokens_thinking is the thinking part alone. (0, 0, 0) when absent or unparseable; never raises.""" try: data = resp.json() if hasattr(resp, "json") else (resp or {}) u = (data or {}).get("usageMetadata") or {} tin = int(u.get("promptTokenCount") or 0) think = int(u.get("thoughtsTokenCount") or 0) tout = int(u.get("candidatesTokenCount") or 0) + think return max(0, tin), max(0, tout), max(0, think) except Exception: return 0, 0, 0 def usage_from_response(resp: Any) -> tuple[int, int]: """(tokens_in, tokens_out) from a generateContent response's ``usageMetadata``; (0, 0) when absent or unparseable. Output counts thinking tokens (``thoughtsTokenCount``) with the candidate tokens, which is how they are billed. Never raises: cost accounting must not be able to fail an extraction.""" try: data = resp.json() if hasattr(resp, "json") else (resp or {}) u = (data or {}).get("usageMetadata") or {} tin = int(u.get("promptTokenCount") or 0) tout = int(u.get("candidatesTokenCount") or 0) + int(u.get("thoughtsTokenCount") or 0) return max(0, tin), max(0, tout) except Exception: return 0, 0 @dataclasses.dataclass class PropertyRow: """One flattened, fully-resolved row ready for the SQLite mirror. Carries the legacy columns (`value`, `unit`, `english`, ...) so the CSV export and page1.py keep working, plus the new structured/provenance columns added in this phase. """ # identity / routing material_name: str material_abbreviation: str material_key: str material_class: str # legacy columns (kept populated for backward compat) section: str property_name: str value: str unit: str english: str test_condition: str comments: str # composite descriptors promoted to real columns (Task 2) trade_grade: str = "" manufacturer: str = "" matrix: str = "" fiber: str = "" fiber_volume_fraction: str = "" # structured numeric value (Task 3) value_raw: str = "" value_num: Optional[float] = None value_min: Optional[float] = None value_max: Optional[float] = None qualifier: str = "" unit_canonical: str = "" value_si: Optional[float] = None # provenance (Tasks 1/6) source_pdf: str = "" source_sha1: str = "" page: Optional[int] = None source_quote: str = "" # status (Tasks 1/3/4) status: str = "ok" flag_reason: str = "" # bookkeeping model: str = dataclasses.field(default_factory=text_model) prompt_version: str = PROMPT_VERSION # provenance kind (figure-mining phase): 'text' (grounded in the PDF # text) or 'figure' (read off a plot/table image — an estimate; see # figures.py). figure_id points at the harvested PNG. origin: str = "text" figure_id: str = "" # figure-linking phase (figure_links.py, ported from the InDeS mapper): # for a TEXT row, figure_id may instead point at the harvested figure the # row's evidence CITES ("see Fig. 3"). The link is grounding evidence, # never a value — status is untouched. image_url / image are the columns # the shared Space reads to show a row's plot (S3 ref / PNG bytes). figure_ref: str = "" # citation as parsed, e.g. "Fig. 3(b)" figure_link_score: Optional[float] = None # mapper5 score: 0.90-1.00 citation, <0.9 fallback figure_link_signals: str = "" # "figure_citation, page_confirm" / "page, tokens" image_url: str = "" image: bytes = dataclasses.field(default=b"", repr=False, compare=False) # --------------------------------------------------------------------------- # Gemini request with retry/backoff (Task 7) # --------------------------------------------------------------------------- _KEY_PARAM_RE = re.compile(r"([?&](?:key|api_key|apikey|token|access_token)=)[^&\s'\")\]]+", re.I) def redact_secrets(text: Any) -> str: """Strip credential query parameters from error text. requests' HTTPError and urllib3's connection errors quote the full request URL — for Gemini that URL carries `?key=` — so any error message that is logged or stored has to pass through here first.""" return _KEY_PARAM_RE.sub(r"\1", str(text)) class GeminiHTTPError(requests.HTTPError): """HTTPError for a Gemini endpoint whose message names the status and the API's own error detail (e.g. RESOURCE_EXHAUSTED: "Your project has exceeded its monthly spending cap") instead of the keyed URL.""" def __init__(self, message: str, response: requests.Response, status: str = "", detail: str = "") -> None: super().__init__(message, response=response) self.status = status # Gemini "status" field, e.g. RESOURCE_EXHAUSTED self.detail = detail # Gemini "message" field, human-readable @property def quota(self) -> bool: """True when the API refused for budget/quota reasons (spending cap, daily quota): a condition that will not clear by retrying now.""" return (self.status == "RESOURCE_EXHAUSTED" or "spending cap" in self.detail.lower() or "quota" in self.detail.lower()) def _gemini_error_fields(resp: requests.Response) -> tuple[str, str]: """(status, message) from a Gemini error body, or ('', '') if not JSON.""" try: body = resp.json() except ValueError: return "", "" err = (body or {}).get("error") if isinstance(body, dict) else None if not isinstance(err, dict): return "", "" return str(err.get("status") or ""), str(err.get("message") or "")[:300] def _retry_after(resp: requests.Response) -> Optional[float]: val = resp.headers.get("Retry-After") if not val: return None try: # Cap: a hostile/buggy server sending Retry-After: 86400 must not # stall a cycle for hours (Space anti-hang hardening, commit 3f67f54). return max(0.0, min(float(val), 300.0)) except ValueError: return None def _request_with_retry( method: str, url: str, *, what: str = "Gemini", timeout: int = REQUEST_TIMEOUT_S, _sleep=time.sleep, **kw: Any, ) -> Optional[requests.Response]: """Issue one HTTP request, retrying 429/5xx and connection errors. Shared policy for every Gemini endpoint (generateContent, File API start / upload / status poll): bounded retries with exponential backoff, honoring a Retry-After header. Returns the 200 Response, or None if retries are exhausted / the status is non-retryable (after raise_for_status). """ for attempt in range(MAX_RETRIES + 1): try: resp = requests.request(method, url, timeout=timeout, **kw) except requests.RequestException as exc: if attempt < MAX_RETRIES: delay = BACKOFF_BASE * (2 ** attempt) log.warning("%s request failed: %s; retry %d/%d in %.1fs", what, redact_secrets(exc), attempt + 1, MAX_RETRIES, delay) _sleep(delay) continue # Connection errors quote the URL (with the key) — re-raise clean. raise requests.ConnectionError( f"{what} request failed: {redact_secrets(exc)}") from None if resp.status_code == 200: return resp status, detail = _gemini_error_fields(resp) if resp.status_code == 429 and (status == "RESOURCE_EXHAUSTED" or "spending cap" in detail.lower()): # A spent budget (monthly spending cap, daily quota) is not a # momentary rate limit: retrying it in 2/4/8 s only burns time, # 14 s per PDF, and on 28 Sep 2026 the retries' error text put # the API key into the public event stream. Fail at once. log.error("%s -> 429 %s: %s", what, status, detail) raise GeminiHTTPError(f"429 {status or 'RESOURCE_EXHAUSTED'} from {what}: {detail}", resp, status, detail) if resp.status_code in RETRY_STATUS and attempt < MAX_RETRIES: delay = _retry_after(resp) or BACKOFF_BASE * (2 ** attempt) log.warning( "%s -> %s; retry %d/%d in %.1fs", what, resp.status_code, attempt + 1, MAX_RETRIES, delay, ) _sleep(delay) continue # Non-retryable (4xx) or retries exhausted: fail loudly, once. This # raise deliberately sits OUTSIDE the try above — HTTPError is a # RequestException, and raising it inside used to be caught by the # connection-error branch and retried, so a 400/401/403 (bad key, # invalid schema) burned MAX_RETRIES backoffs before surfacing. # The message names the endpoint and the API's detail, never the # keyed URL (requests' own raise_for_status() message does). log.error("%s -> %s: %s", what, resp.status_code, resp.text[:300]) raise GeminiHTTPError(f"{resp.status_code} {resp.reason or 'error'} from {what}" + (f": {status} {detail}".rstrip() if (status or detail) else ""), resp, status, detail) return None def thinking_config(setting: Optional[str] = None) -> Optional[dict]: """The generationConfig.thinkingConfig for the current THINKING setting, or None for "dynamic" (send nothing: the API default) and after the model rejected the option once.""" if _THINKING_UNSUPPORTED: return None v = (setting if setting is not None else THINKING or "").strip().lower() if not v or v == "dynamic": return None if v == "off": return {"thinkingBudget": 0} if v in ("minimal", "low", "medium", "high"): return {"thinkingLevel": v} try: return {"thinkingBudget": max(0, int(v))} except ValueError: log.warning("GEMINI_THINKING=%r not understood; using the API default", v) return None def apply_thinking(payload: dict) -> dict: """Add the thinking option to a generateContent payload (in place).""" tc = thinking_config() gc = payload.setdefault("generationConfig", {}) if tc is None: gc.pop("thinkingConfig", None) else: gc["thinkingConfig"] = tc return payload def gemini_request( url: str, payload: dict[str, Any], *, timeout: int = REQUEST_TIMEOUT_S, _sleep=time.sleep, ) -> Optional[requests.Response]: """POST `payload` to `url`, retrying 429/5xx and connection errors. Mirrors pdf_crawler.http_get()'s policy: bounded retries with exponential backoff, honoring a Retry-After header. Returns the 200 Response, or None if retries are exhausted / the status is non-retryable. The thinking option (THINKING) is applied here, so every Gemini call in extraction.py and figures.py gets it. A 400 that names the option (a model without that field, an out-of-range value) is retried once without it and disables it for the process: cost tuning must never stop an extraction. """ global _THINKING_UNSUPPORTED apply_thinking(payload) try: return _request_with_retry("POST", url, json=payload, timeout=timeout, _sleep=_sleep) except GeminiHTTPError as exc: gc = payload.get("generationConfig") or {} if (exc.response is not None and exc.response.status_code == 400 and "thinkingConfig" in gc and "think" in (exc.detail or "").lower()): log.warning("model rejected thinkingConfig=%s (%s); retrying without it and " "leaving thinking at the API default from now on", gc.get("thinkingConfig"), exc.detail[:120]) _THINKING_UNSUPPORTED = True gc.pop("thinkingConfig", None) return _request_with_retry("POST", url, json=payload, timeout=timeout, _sleep=_sleep) raise # File API processing wait: exponential poll (1, 2, 4, 8, ... capped at 15 s) # up to this many seconds. Was a fixed 30 x 1 s, so a large PDF that took # longer than 30 s to process came back None -> empty_extraction and was # re-uploaded from scratch on every run. FILE_ACTIVE_WAIT_S = 180 FILE_POLL_CAP_S = 15.0 def _upload_pdf_file( pdf_bytes: bytes, filename: str, api_key: str, *, _sleep=time.sleep ) -> Optional[str]: """Upload a PDF via the Gemini File API (resumable protocol); return file URI. Used for large/long PDFs (Task 10) where base64 inlining would blow the generateContent request-size limit. Every hop (start, upload, status poll) goes through the same retry/backoff policy as generateContent — a single free-tier 429 used to fail the whole PDF. """ start_url = GEMINI_UPLOAD_URL.format(key=api_key) start = _request_with_retry( "POST", start_url, what="File API start", headers={ "X-Goog-Upload-Protocol": "resumable", "X-Goog-Upload-Command": "start", "X-Goog-Upload-Header-Content-Length": str(len(pdf_bytes)), "X-Goog-Upload-Header-Content-Type": "application/pdf", "Content-Type": "application/json", }, json={"file": {"display_name": filename}}, _sleep=_sleep, ) if start is None: return None upload_url = start.headers.get("X-Goog-Upload-URL") if not upload_url: log.error("File API did not return an upload URL") return None up = _request_with_retry( "POST", upload_url, what="File API upload", headers={ "X-Goog-Upload-Offset": "0", "X-Goog-Upload-Command": "upload, finalize", "Content-Length": str(len(pdf_bytes)), }, data=pdf_bytes, _sleep=_sleep, ) if up is None: return None info = up.json().get("file", {}) name = info.get("name") uri = info.get("uri") state = info.get("state") # Wait for the file to become ACTIVE before referencing it. waited = 0.0 delay = 1.0 while True: if state == "ACTIVE": return uri if state == "FAILED": log.error("File API processing failed for %s", filename) return None if waited >= FILE_ACTIVE_WAIT_S or not name: log.error("File API: %s not ACTIVE after %.0fs (state=%s)", filename, waited, state) return None _sleep(delay) waited += delay delay = min(delay * 2, FILE_POLL_CAP_S) poll = _request_with_retry( "GET", GEMINI_FILE_STATUS_URL.format(name=name, key=api_key), what="File API status", _sleep=_sleep, ) if poll is None: return None info = poll.json() state = info.get("state") uri = info.get("uri", uri) def _parse_extraction_json(resp: Any) -> Optional[dict[str, Any]]: data = resp.json() if hasattr(resp, "json") else (resp or {}) candidates = data.get("candidates", []) if not candidates: return None parts = candidates[0].get("content", {}).get("parts", []) for part in parts: text = (part.get("text") or "").strip() if text.startswith("{"): try: return json.loads(text) except json.JSONDecodeError: log.error("Gemini returned non-JSON text") return None return None # --------------------------------------------------------------------------- # PDF text (grounding + scanned detection) # --------------------------------------------------------------------------- def pdf_page_texts(pdf_bytes: bytes) -> list[str]: """Return per-page extractable text. Empty list if PyMuPDF is unavailable.""" if fitz is None: return [] try: doc = fitz.open(stream=pdf_bytes, filetype="pdf") except Exception as exc: # pragma: no cover log.warning("Could not open PDF for text extraction: %s", exc) return [] try: return [p.get_text() for p in doc] finally: doc.close() def _is_scanned(page_texts: list[str]) -> bool: if not page_texts: return False # can't tell without PyMuPDF; don't falsely flag total = sum(len(t.strip()) for t in page_texts) return (total / max(1, len(page_texts))) < SCANNED_TEXT_PER_PAGE # --------------------------------------------------------------------------- # Extraction entry point # --------------------------------------------------------------------------- def extract_from_pdf(pdf_bytes: bytes, filename: str, api_key: str) -> Extraction: """Extract structured materials from a PDF (Tasks 2/3/8/10). Returns an Extraction whose ``doc_status`` is ``scanned_no_text`` (and ``materials`` empty) when the PDF has no extractable text, so the caller can record the source without fabricating rows. """ page_texts = pdf_page_texts(pdf_bytes) if _is_scanned(page_texts): log.info("%s looks scanned/image-only; skipping extraction", filename) return Extraction(doc_status="scanned_no_text") parts: list[dict[str, Any]] = [{"text": EXTRACTION_PROMPT}] use_file_api = ( len(pdf_bytes) > INLINE_PDF_LIMIT_BYTES or len(page_texts) > INLINE_PAGE_LIMIT ) if use_file_api: uri = _upload_pdf_file(pdf_bytes, filename, api_key) if not uri: return Extraction(doc_status="empty_extraction") parts.append({"fileData": {"mimeType": "application/pdf", "fileUri": uri}}) else: parts.append( { "inlineData": { "mimeType": "application/pdf", "data": base64.b64encode(pdf_bytes).decode("utf-8"), } } ) payload = { "contents": [{"parts": parts}], "generationConfig": { "temperature": 0, "responseMimeType": "application/json", "responseSchema": EXTRACTION_SCHEMA, }, } model = text_model() url = GEMINI_URL_TEMPLATE.format(model=model, key=api_key) resp = gemini_request(url, payload) if resp is None: return Extraction(doc_status="empty_extraction", model=model) return extraction_from_response(resp, model) def extraction_from_response(resp: Any, model: Optional[str] = None) -> Extraction: """Extraction (with usage) from a generateContent response object or its parsed JSON dict (the Batch API hands back dicts).""" model = model or text_model() tokens_in, tokens_out, tokens_thinking = usage_detail(resp) raw = _parse_extraction_json(resp) if not raw: # The call was made and billed even though nothing usable came back. return Extraction(doc_status="empty_extraction", model=model, tokens_in=tokens_in, tokens_out=tokens_out, tokens_thinking=tokens_thinking) out = _extraction_from_json(raw) out.model = model out.tokens_in, out.tokens_out, out.tokens_thinking = tokens_in, tokens_out, tokens_thinking return out def _extraction_from_json(raw: dict[str, Any]) -> Extraction: """Coerce raw model JSON into a typed Extraction (tolerates the old shape).""" materials_json = raw.get("materials") if materials_json is None: # Back-compat: a single-material response with a flat property list. flat = raw.get("mechanical_properties") or raw.get("properties") or [] materials_json = [ { "material_name": raw.get("material_name", ""), "material_abbreviation": raw.get("material_abbreviation", ""), "material_class": raw.get("material_class", ""), "trade_grade": raw.get("trade_grade", ""), "manufacturer": raw.get("manufacturer", ""), "properties": flat, } ] materials: list[Material] = [] for mj in materials_json or []: props: list[Property] = [] for pj in mj.get("properties") or []: value_raw = (pj.get("value_raw") or pj.get("value") or "").strip() prop = Property( section=_norm_section(pj.get("section")), property_name=(pj.get("property_name") or "").strip() or "Unknown property", value_raw=value_raw, unit=(pj.get("unit") or "").strip(), value_num=_as_float(pj.get("value_num")), value_min=_as_float(pj.get("value_min")), value_max=_as_float(pj.get("value_max")), qualifier=(pj.get("qualifier") or "").strip(), test_condition=(pj.get("test_condition") or "").strip(), comments=(pj.get("comments") or "").strip(), source_quote=(pj.get("source_quote") or "").strip(), page=_as_int(pj.get("page")), ) # Fill structured numeric fields from value_raw if the model omitted them. _fill_numeric(prop) props.append(prop) materials.append( Material( material_name=(mj.get("material_name") or "").strip(), material_abbreviation=(mj.get("material_abbreviation") or "").strip(), material_class=(mj.get("material_class") or "").strip(), trade_grade=(mj.get("trade_grade") or "").strip(), manufacturer=(mj.get("manufacturer") or "").strip(), matrix=(mj.get("matrix") or "").strip(), fiber=(mj.get("fiber") or "").strip(), fiber_volume_fraction=(mj.get("fiber_volume_fraction") or "").strip(), properties=props, ) ) return Extraction(materials=materials) def _as_float(v: Any) -> Optional[float]: if v is None or v == "": return None try: return float(v) except (TypeError, ValueError): return None def _as_int(v: Any) -> Optional[int]: if v is None or v == "": return None try: return int(v) except (TypeError, ValueError): return None def _norm_section(section: Optional[str]) -> str: """Normalize free-text section drift to the enum (Task 8).""" s = (section or "").strip() if not s: return "Descriptive" low = s.lower() for canon in SECTION_ENUM: if low == canon.lower(): return canon # substring / synonym normalization, e.g. "Mechanical Properties" -> "Mechanical" synonyms = { "mechanical": "Mechanical", "thermal": "Thermal", "electrical": "Electrical", "dielectric": "Electrical", "physical": "Physical", "optical": "Optical", "rheolog": "Rheological", "viscosity": "Rheological", "processing": "Processing", "descriptive": "Descriptive", "composition": "Composition/Reinforcement", "reinforcement": "Composition/Reinforcement", "architecture": "Architecture/Structure", "structure": "Architecture/Structure", } for key, canon in synonyms.items(): if key in low: return canon return "Descriptive" # --------------------------------------------------------------------------- # Value parsing (Task 3) # --------------------------------------------------------------------------- _QUALIFIER_MAP = [ ("≤", "<="), ("≥", ">="), ("≈", "~"), ("<=", "<="), (">=", ">="), ("<", "<"), (">", ">"), ("~", "~"), ("±", "±"), ] _NUM_RE = re.compile(r"[-+]?\d{1,3}(?:,\d{3})+(?:\.\d+)?|[-+]?\d*\.?\d+(?:[eE][-+]?\d+)?") # Plus/minus built from _NUM_RE so both operands accept thousands separators and # scientific notation ('1,200 ± 100', '1e-3 ± 2e-4') — a plain \d*\.?\d+ operand # used to anchor after the comma and read '1,200 ± 100' as 200 ± 100. _PM_RE = re.compile(rf"({_NUM_RE.pattern})\s*(?:±|\+/-|\+-)\s*({_NUM_RE.pattern})") def _clean_number(tok: str) -> Optional[float]: try: return float(tok.replace(",", "")) except ValueError: return None def parse_value_raw(value_raw: str) -> tuple[Optional[float], Optional[float], Optional[float], str]: """Parse a printed value into (value_num, value_min, value_max, qualifier). Handles ranges ('100-120', '100 to 120'), qualifiers ('<= -18', '~3.5'), plus-minus ('5.0 +/- 0.2'), thousands separators and scientific notation. """ if not value_raw: return None, None, None, "" s = unicodedata.normalize("NFKC", value_raw).strip() # NFKC does NOT fold the typographic minus (U+2212) or dashes to '-'. PDFs # typeset negatives and ranges with them, so without this a Tg of '−60 °C' # parses as +60 and '20−30' loses its range. Same folding as _normalize_text. s = s.replace("−", "-").replace("–", "-").replace("—", "-") qualifier = "" for needle, canon in _QUALIFIER_MAP: if needle in s: qualifier = canon break # plus/minus -> midpoint value with min/max pm = _PM_RE.search(s) if pm: base = _clean_number(pm.group(1)) delta = _clean_number(pm.group(2)) if base is not None and delta is not None: delta = abs(delta) return base, base - delta, base + delta, "±" # Range: two numbers separated by a dash/"to" (all dash variants were folded # to '-' above). Guard against a leading sign being misread as a separator # by working on the sign-stripped remainder. body = s lead_sign = "" if body[:1] in "+-": lead_sign, body = body[0], body[1:] range_match = re.match( r"\s*(\d{1,3}(?:,\d{3})+(?:\.\d+)?|\d*\.?\d+(?:[eE][-+]?\d+)?)" r"\s*(?:-|to|\.\.\.|…)\s*" r"([-+]?\d{1,3}(?:,\d{3})+(?:\.\d+)?|[-+]?\d*\.?\d+(?:[eE][-+]?\d+)?)", body, ) if range_match: lo = _clean_number(lead_sign + range_match.group(1)) hi = _clean_number(range_match.group(2)) if lo is not None and hi is not None: if lo > hi: lo, hi = hi, lo return None, lo, hi, qualifier nums = _NUM_RE.findall(s) if not nums: return None, None, None, qualifier val = _clean_number(nums[0]) return val, None, None, qualifier def _fill_numeric(prop: Property) -> None: """Backfill value_num/min/max/qualifier from value_raw when model omitted them.""" has_any = ( prop.value_num is not None or prop.value_min is not None or prop.value_max is not None ) if has_any and prop.qualifier: return num, lo, hi, qual = parse_value_raw(prop.value_raw) if prop.value_num is None and prop.value_min is None and prop.value_max is None: prop.value_num, prop.value_min, prop.value_max = num, lo, hi if not prop.qualifier: prop.qualifier = qual def _representative_value(prop: Property) -> Optional[float]: if prop.value_num is not None: return prop.value_num if prop.value_min is not None and prop.value_max is not None: return (prop.value_min + prop.value_max) / 2.0 return prop.value_min if prop.value_min is not None else prop.value_max # --------------------------------------------------------------------------- # Unit normalization + plausibility (Task 3) # --------------------------------------------------------------------------- if pint is not None: # pragma: no branch _UREG = pint.UnitRegistry() _UREG.define("ksi = 1000 * psi") _UREG.define("Msi = 1000000 * psi") else: # pragma: no cover _UREG = None @dataclasses.dataclass class _Family: name: str keywords: tuple[str, ...] canonical: str # pint unit string (or display for non-pint families) display: str # human-facing unit label lo: float # plausibility low, in `display` units hi: float # plausibility high, in `display` units kind: str # "physical" | "temperature" | "raw" | "passthrough" si_factor: float = 1.0 # for "raw" families: value_si = value_in_display * si_factor # For "raw" families: accepted printed-unit spellings (normalized via # _norm_unit_key) -> factor that converts the printed value into `display` # units. '' may be present to accept a bare number. Anything else is a # unit_review, never a silent default. Populated below. unit_map: dict[str, float] = dataclasses.field(default_factory=dict) # Keyword matching. Multi-word keywords match as substrings ("tensile modulus" # in "Tensile modulus (0°)"). SHORT tokens (tg, tm, hdt, cte) must match on # token boundaries — bare substring matching made 'tm' hit "ASTM" (so every # property citing an ASTM standard became a melting temperature) and 'tg' hit # "outgassing". Left boundary = not a letter/digit; right boundary = not a # LETTER (a trailing digit is allowed: 'Tg2', 'CTE1', 'HDT1.8' are how DSC/DMA # reports and laminate datasheets label second-heating Tg and the two CTE # regimes). _SHORT_TOKENS = frozenset({"tg", "tm", "tc", "hdt", "cte", "clte", "cai", "sbs", "young", "young's", "youngs"}) _KEYWORD_RE_CACHE: dict[str, "re.Pattern[str]"] = {} def _keyword_matches(kw: str, name: str) -> bool: if kw not in _SHORT_TOKENS: return kw in name pat = _KEYWORD_RE_CACHE.get(kw) if pat is None: pat = re.compile(r"(? ppm/°C "": 1.0, # bare: see _bare_value_factor (ppm vs 1/K by magnitude) "ppm/c": 1.0, "ppm/k": 1.0, "ppm/degc": 1.0, "ppm/°c": 1.0, "ppm": 1.0, "um/m/c": 1.0, "um/m/k": 1.0, "um/m/degc": 1.0, "um/m/°c": 1.0, "um/(m*c)": 1.0, "um/(m*k)": 1.0, "um/(m·c)": 1.0, "um/(m·k)": 1.0, "um/m·c": 1.0, "um/m·k": 1.0, "um/m°c": 1.0, "um/mk": 1.0, "um/m*c": 1.0, "um/m*k": 1.0, "um·/m/k": 1.0, "um/m/m/k": 1.0, "10^-6/c": 1.0, "10^-6/k": 1.0, "10^-6/degc": 1.0, "10^-6/°c": 1.0, "10-6/c": 1.0, "10-6/k": 1.0, "10-6/°c": 1.0, "x10^-6/c": 1.0, "x10^-6/k": 1.0, "x10^-6/°c": 1.0, "x10-6/c": 1.0, "x10-6/k": 1.0, "x10-6/°c": 1.0, "e-6/c": 1.0, "e-6/k": 1.0, "e-6/°c": 1.0, "1e-6/c": 1.0, "1e-6/k": 1.0, "1e-6/°c": 1.0, "uin/in/f": 1.8, "uin/in/°f": 1.8, "uin/in/degf": 1.8, "ppm/f": 1.8, "ppm/degf": 1.8, "ppm/°f": 1.8, "10^-6/f": 1.8, "10^-6/°f": 1.8, "10-6/f": 1.8, "10-6/°f": 1.8, "x10^-6/f": 1.8, "x10^-6/°f": 1.8, "10^-6in/in/f": 1.8, "10^-6in/in/°f": 1.8, "10-6in/in/f": 1.8, "10-6in/in/°f": 1.8, "1/c": 1e6, "1/k": 1e6, "1/degc": 1e6, "1/°c": 1e6, "/c": 1e6, "/k": 1e6, "/degc": 1e6, "/°c": 1e6, "m/m/c": 1e6, "m/m/k": 1e6, "m/m/°c": 1e6, "m/(m*k)": 1e6, "m/(m·k)": 1e6, "mm/mm/c": 1e6, "mm/mm/k": 1e6, "mm/mm/°c": 1e6, "1/f": 1.8e6, "1/degf": 1.8e6, "1/°f": 1.8e6, "/f": 1.8e6, "/°f": 1.8e6, "in/in/f": 1.8e6, "in/in/°f": 1.8e6, } _ELONGATION_UNIT_MAP: dict[str, float] = { # -> % "%": 1.0, "percent": 1.0, "pct": 1.0, # bare / ratio: see _bare_value_factor — a strain FRACTION (0.024) or a # percent printed without its unit (2.4), decided by magnitude. "": 1.0, "-": 1.0, "mm/mm": 100.0, "m/m": 100.0, "in/in": 100.0, "ratio": 100.0, "strain": 100.0, "fraction": 100.0, } _SPECIFIC_GRAVITY_UNIT_MAP: dict[str, float] = { # dimensionless by definition; tolerate g/cm³ spellings (SG ≈ density in g/cm³) "": 1.0, "-": 1.0, "g/cm3": 1.0, "g/cm^3": 1.0, "g/cc": 1.0, "g/ml": 1.0, "g/l": 0.001, "kg/m3": 0.001, "kg/m^3": 0.001, "kg/dm3": 1.0, "kg/dm^3": 1.0, "kg/l": 1.0, } def _bare_value_factor(fam: _Family, rep: float, value_raw: str) -> Optional[float]: """Factor for a raw-family value printed with NO unit (or a lone dash). The bare case is genuinely ambiguous, so decide by what the number can be: * elongation: '%' inside value_raw wins (a datasheet cell "2.4 %" the model copied whole) -> percent. Otherwise |v| < 0.5 is a strain fraction (0.024 = 2.4 %; no thermoplastic composite fails below 0.5 % strain); |v| >= 0.5 is a percent printed without its unit (2.4 -> 2.4 %). The x100 rule used to apply to every bare number and turned "2.4 %"/'' into 240 % with status ok. * cte: |v| < 1e-3 is a 1/K fraction (2.3e-5 -> 23 ppm/°C), else ppm/°C. * specific gravity: dimensionless, factor 1. Returns None to say "don't know" -> unit_review. """ if fam.name == "elongation": if "%" in value_raw: return 1.0 return 100.0 if abs(rep) < 0.5 else 1.0 if fam.name == "cte": return 1e6 if 0 < abs(rep) < 1e-3 else 1.0 if fam.name == "specific_gravity": return 1.0 return None # Ordered: specific keywords before generic ones (first match wins). # "passthrough" families exist to *shield* generic keywords: e.g. dielectric, # impact and tear strength are not pressures, so they must not fall into the # MPa families below them. They carry no unit check and no plausibility range. # Compression-after-impact (CAI) is a real MPa strength and must be caught # BEFORE the impact shield. PROPERTY_FAMILIES: list[_Family] = [ _Family("cai_strength", ("compression after impact", "compression-after-impact", "cai"), "MPa", "MPa", 0.5, 6000.0, "physical"), _Family("dielectric_strength", ("dielectric strength", "breakdown strength", "breakdown voltage"), "", "", 0.0, 0.0, "passthrough"), _Family("impact_strength", ("impact strength", "impact resistance", "izod", "charpy", "impact energy"), "", "", 0.0, 0.0, "passthrough"), _Family("tear_strength", ("tear strength", "tear resistance"), "", "", 0.0, 0.0, "passthrough"), _Family("peel_strength", ("peel strength", "peel resistance", "bond strength", "adhesion strength", "adhesive strength", "weld strength"), "", "", 0.0, 0.0, "passthrough"), _Family("tensile_modulus", ("tensile modulus", "modulus of elasticity", "young", "young's", "youngs", "elastic modulus"), "GPa", "GPa", 0.01, 1000.0, "physical"), _Family("flexural_modulus", ("flexural modulus", "bending modulus"), "GPa", "GPa", 0.01, 800.0, "physical"), _Family("shear_modulus", ("shear modulus", "modulus of rigidity"), "GPa", "GPa", 0.005, 500.0, "physical"), _Family("storage_modulus", ("storage modulus", "compressive modulus", "modulus"), "GPa", "GPa", 0.001, 1000.0, "physical"), _Family("tensile_strength", ("tensile strength", "ultimate tensile", "strength at break", "yield strength", "tensile stress"), "MPa", "MPa", 0.5, 10000.0, "physical"), _Family("flexural_strength", ("flexural strength", "bending strength"), "MPa", "MPa", 0.5, 4000.0, "physical"), _Family("compressive_strength", ("compressive strength", "compression strength"), "MPa", "MPa", 0.5, 6000.0, "physical"), # Bare "strength" was removed from this family: it captured dielectric / # impact / tear strength (see passthrough guards above) into an MPa check. _Family("shear_strength", ("shear strength", "ilss", "interlaminar shear"), "MPa", "MPa", 0.5, 4000.0, "physical"), _Family("glass_transition", ("glass transition", "tg"), "degC", "°C", -150.0, 600.0, "temperature"), _Family("melting", ("melting", "melt temperature", "tm"), "degC", "°C", 50.0, 500.0, "temperature"), _Family("crystallization", ("crystallization",), "degC", "°C", 0.0, 500.0, "temperature"), _Family("decomposition", ("decomposition", "degradation temperature"), "degC", "°C", 100.0, 1200.0, "temperature"), _Family("hdt", ("heat deflection", "deflection temperature", "hdt", "heat distortion"), "degC", "°C", 0.0, 600.0, "temperature"), _Family("cte", ("thermal expansion", "cte", "clte", "expansion coefficient"), "ppm/degC", "ppm/°C", -50.0, 500.0, "raw", si_factor=1e-6, unit_map=_CTE_UNIT_MAP), # Specific gravity is dimensionless by definition; keep it OUT of the pint # density family or every SG row is false-flagged missing_unit. _Family("specific_gravity", ("specific gravity", "relative density"), "", "", 0.1, 12.0, "raw", si_factor=1.0, unit_map=_SPECIFIC_GRAVITY_UNIT_MAP), _Family("density", ("density",), "g/cm**3", "g/cm³", 0.1, 12.0, "physical"), _Family("elongation", ("elongation", "strain at break", "strain to failure", "failure strain", "ultimate strain"), "%", "%", 0.001, 2000.0, "raw", si_factor=0.01, unit_map=_ELONGATION_UNIT_MAP), # Generic composite strengths that are real pressures (MPa). Kept LAST so # every shield / specific family above wins first; the bare "strength" # keyword is safe here because dielectric / impact / tear / peel are # already routed to passthrough families. _Family("other_strength", ("short beam", "short-beam", "sbs", "bearing strength", "open hole", "open-hole", "filled hole", "filled-hole", "notched tensile", "unnotched", "interlaminar strength", "transverse strength", "strength"), "MPa", "MPa", 0.5, 10000.0, "physical"), ] def _match_family(prop: Property) -> Optional[_Family]: name = prop.property_name.lower() for fam in PROPERTY_FAMILIES: for kw in fam.keywords: if _keyword_matches(kw, name): return fam return None # 'X⁻¹' / 'X^-1' / 'X-1' suffix -> '/X' so 'K⁻¹', '10⁻⁶ K⁻¹', 'µm·m⁻¹·K⁻¹' # collapse onto the '/k' spellings already in the maps. _INVERSE_SUFFIX_RE = re.compile(r"([a-z°]+)\^?-1") def _norm_unit_key(unit: str) -> str: """Normalize a printed unit into a lookup key for the raw-family unit maps.""" u = unit.strip() # BEFORE NFKC: it folds 'º' (masculine ordinal, often typed for a degree # sign) to 'o' and superscripts to plain digits, so handle those first. u = u.replace("º", "°").replace("˚", "°") u = u.replace("⁻", "-").replace("⁺", "+") u = u.translate(str.maketrans("⁰¹²³⁴⁵⁶⁷⁸⁹", "0123456789")) u = unicodedata.normalize("NFKC", u).lower() u = u.replace("µ", "u").replace("μ", "u") u = u.replace("−", "-").replace("–", "-").replace("—", "-") u = u.replace(" ", "").replace("(", "").replace(")", "") u = u.replace("**", "^").replace("×", "x").replace("*10", "x10").replace("e-06", "e-6") u = u.replace("·", "*") # 'x⁻¹' style: 'k^-1' -> '/k'; 'um*m-1*k-1' -> 'um/m/k' u = _INVERSE_SUFFIX_RE.sub(r"/\1", u) u = u.replace("*/", "/").replace("//", "/") if u.startswith("*"): u = u[1:] return u def _raw_factor(fam: _Family, unit: str, rep: Optional[float] = None, value_raw: str = "") -> Optional[float]: """Factor converting a printed raw-family value into `fam.display` units, or None.""" key = _norm_unit_key(unit) if key in ("", "-") and rep is not None: return _bare_value_factor(fam, rep, value_raw) if key in fam.unit_map: return fam.unit_map[key] # tolerate '°c' vs 'c' vs 'degc' interchangeably for a, b in (("°c", "c"), ("degc", "c"), ("°f", "f"), ("degf", "f"), ("°k", "k"), ("degk", "k")): alt = key.replace(a, b) if alt in fam.unit_map: return fam.unit_map[alt] return None # A letter followed by 2/3 is a power (cm3, mm2, in2, ft3, kJ/m2 ...) — but NOT # the exponent marker of scientific notation ('1e3 psi', '10E3 psi'). _UNIT_POWER_RE = re.compile(r"(?<=[A-Za-z])(? str: # Superscripts FIRST: NFKC folds '³' to a plain '3', so '10³ psi' became # '103 psi' (a silent 10x error on the standard US spelling of ksi) and # the later '³' replace was dead code. u = _SUPERSCRIPT_RE.sub(lambda m: "**" + m.group(1).translate(_SUP_TRANS), unit.strip()) u = unicodedata.normalize("NFKC", u) u = u.replace("·", "*").replace("−", "-") u = u.replace("µ", "u").replace("μ", "u") u = u.replace("^", "**") u = u.replace("g/cc", "g/cm**3") # Any letter immediately followed by 2/3 is a power: cm3, mm2, m3, in2, ft3… # (was a hardcoded list that missed 'mm2' — the standard European MPa # spelling 'N/mm2' failed to parse and landed in unit_review). u = _UNIT_POWER_RE.sub(r"**\1", u) return u _TEMP_UNITS = { "": "degC", "c": "degC", "degc": "degC", "°c": "degC", "celsius": "degC", "k": "kelvin", "kelvin": "kelvin", "f": "degF", "degf": "degF", "°f": "degF", "fahrenheit": "degF", } def canonicalize(prop: Property) -> tuple[str, Optional[float], Optional[str]]: """Return (unit_canonical, value_si, problem). `problem` is None on success, or a short reason ("unit_review:...") when the unit is missing/dimensionally wrong for the property family. """ fam = _match_family(prop) rep = _representative_value(prop) if fam is None or fam.kind == "passthrough": # No known family (or a shield family): pass the unit through, no SI # conversion, no check. return (prop.unit, None, None) if rep is None: return (fam.display, None, None) if fam.kind == "raw": # Never apply si_factor blindly: CTE '2.3e-5 1/K' is 23 ppm/°C, not # 2.3e-11; elongation '0.024' (a strain fraction) is 2.4 %, not 0.024 %. factor = _raw_factor(fam, prop.unit, rep, prop.value_raw) if factor is None: return (fam.display, None, f"unit_review:unexpected_unit:{prop.unit}!~{fam.display}") return (fam.display, rep * factor * fam.si_factor, None) if _UREG is None: # pragma: no cover return (fam.display, None, None) try: if fam.kind == "temperature": key = unicodedata.normalize("NFKC", prop.unit).strip().lower() src = _TEMP_UNITS.get(key) if src is None: return (fam.display, None, f"unit_review:bad_temp_unit:{prop.unit}") q = _UREG.Quantity(rep, src) value_si = q.to("kelvin").magnitude return (fam.display, value_si, None) # physical (pressure, density, ...) pre = _preprocess_unit(prop.unit) if not pre: return (fam.display, None, "unit_review:missing_unit") q = rep * _UREG(pre) target = _UREG(fam.canonical) if q.dimensionality != target.dimensionality: return (fam.display, None, f"unit_review:dim_mismatch:{prop.unit}!~{fam.display}") value_si = q.to_base_units().magnitude return (fam.display, value_si, None) except Exception as exc: # pint parse failure, undefined unit, etc. return (fam.display, None, f"unit_review:unparseable:{prop.unit}:{exc}") def _canonical_value(prop: Property, fam: _Family) -> Optional[float]: """Representative value expressed in the family's `display` unit (for range check).""" rep = _representative_value(prop) if rep is None: return None if fam.kind == "passthrough": return None if fam.kind == "raw": factor = _raw_factor(fam, prop.unit, rep, prop.value_raw) return None if factor is None else rep * factor if _UREG is None: # pragma: no cover return rep try: if fam.kind == "temperature": key = unicodedata.normalize("NFKC", prop.unit).strip().lower() src = _TEMP_UNITS.get(key) if src is None: return None return _UREG.Quantity(rep, src).to(fam.canonical).magnitude pre = _preprocess_unit(prop.unit) if not pre: return None q = rep * _UREG(pre) if q.dimensionality != _UREG(fam.canonical).dimensionality: return None return q.to(fam.canonical).magnitude except Exception: return None def plausibility_problem(prop: Property) -> Optional[str]: """Range-check the value *after* unit conversion. Returns reason or None.""" fam = _match_family(prop) if fam is None or fam.kind == "passthrough": return None cval = _canonical_value(prop, fam) if cval is None: return None # couldn't convert -> handled as unit_review elsewhere if not (fam.lo <= cval <= fam.hi): return f"out_of_range[{fam.lo},{fam.hi}{fam.display}]:{cval:.4g}" return None _PLACEHOLDER_VALUES = {"", "n/a", "na", "-", "--", "n.a.", "none", "tbd", "…", "..."} def _empty_value(prop: Property) -> bool: # Fold the typographic dashes a table cell is actually printed with ('–', # '—', '−') before the lookup — a verbatim '–' cell used to pass as a # value and even ground (its folded '-' matched any hyphen on the page). v = prop.value_raw.strip().lower() v = v.replace("–", "-").replace("—", "-").replace("−", "-") return v in _PLACEHOLDER_VALUES # --------------------------------------------------------------------------- # Text grounding (Task 1) # --------------------------------------------------------------------------- _SUPERSCRIPT_CHARS_RE = re.compile(r"[⁰¹²³⁴⁵⁶⁷⁸⁹⁺⁻₀₁₂₃₄₅₆₇₈₉]+") def _normalize_text(s: str) -> str: # Superscript/subscript digits are footnote markers or degree signs glued # to a value ('776¹', '0⁰/45⁰/90⁰'); NFKC would fold them into the number # ('7761'). Replace them with a space BEFORE normalizing. s = _SUPERSCRIPT_CHARS_RE.sub(" ", s) s = unicodedata.normalize("NFKC", s) s = s.replace("–", "-").replace("—", "-").replace("−", "-") s = re.sub(r"\s+", " ", s) return s.lower().strip() _NUMERIC_NEEDLE_RE = re.compile(r"^[-+~<>=≤≥≈±.,\d\s/e]+$", re.IGNORECASE) def _grounded(needle: str, haystacks: list[str]) -> bool: """True if `needle` occurs in any (already-normalized) haystack. Purely numeric needles are matched on **digit boundaries**: a value of "3" must not be "verified" by the "3" inside "ISO 527-3" or "23 °C", "1.2" must not match "11.25", and "200" must not match "1,200". Text needles (source_quote chunks) keep plain substring matching. Haystacks are expected to be `_normalize_text` output. """ n = _normalize_text(needle) if not n: return False if _NUMERIC_NEEDLE_RE.match(n): # Strip qualifiers/whitespace so "<= -18.0" grounds on "-18.0" (the # sign is part of the number; the qualifier may be typeset elsewhere). core = re.sub(r"^[~<>=≤≥≈±\s]+", "", n).strip() if not core or not re.search(r"\d", core): return False # '-', '.', 'e', '/' alone can never ground # Tighten the needle's own range dash ('70 - 75' -> '70-75'); the # haystack side is handled by the tolerant body pattern below. core = re.sub(r"(?<=\d) ?- ?(?=\d)", "-", core) # Left guard: no digit/dot/thousands-comma, and — for an unsigned # needle — no '-' either, so '3' cannot match the tail of 'ISO 527-3'. # A signed needle ('-18.0') legitimately starts with the '-'. left = r"(? Extraction: """Ground each property's value in the PDF text (Task 1). For every property: if `value_raw` (or a chunk of `source_quote`) appears on the cited page (falling back to any page), mark `status='ok'`; otherwise `status='unverified'`. Also folds in empty-value, unit, and range checks, using this precedence: empty_value > unverified > unit_review > out_of_range > ok """ if not page_texts: norm_pages: list[str] = [] else: norm_pages = [_normalize_text(t) for t in page_texts] for material in extraction.materials: for prop in material.properties: # always compute canonicalization so rows carry unit_canonical/value_si unit_canonical, value_si, unit_problem = canonicalize(prop) prop.unit_canonical = unit_canonical prop.value_si = value_si reasons: list[str] = [] status = "ok" if _empty_value(prop): status = "empty_value" reasons.append("empty_value") else: # grounding if norm_pages: page_idx = (prop.page - 1) if prop.page else None page_in_range = page_idx is not None and 0 <= page_idx < len(norm_pages) cited = [norm_pages[page_idx]] if page_in_range else [] if page_idx is not None and not page_in_range: # A cited page that doesn't exist is a wrong page too: # ground anywhere, but say so. found = _grounded(prop.value_raw, norm_pages) if found: reasons.append("grounded_off_page") else: found = _grounded(prop.value_raw, cited or norm_pages) if not found and cited: # Fallback: any page. Still grounded (the number IS in # the PDF) but the cited page was wrong — record it as # a soft reason so reviewers can see the weaker chain. found = _grounded(prop.value_raw, norm_pages) if found: reasons.append("grounded_off_page") if not found and prop.source_quote: chunk = prop.source_quote[:40] found = _grounded(chunk, norm_pages) if found: reasons.append("grounded_via_quote") if not found: status = "unverified" reasons.append("value_not_in_pdf_text") # unit problem if status == "ok" and unit_problem: status = "unit_review" reasons.append(unit_problem) # plausibility (only meaningful once unit is sane) if status == "ok": rng = plausibility_problem(prop) if rng: status = "out_of_range" reasons.append(rng) prop.status = status prop.flag_reason = "; ".join(reasons) return extraction # --------------------------------------------------------------------------- # Classification (Task 5) # --------------------------------------------------------------------------- COMPOSITE_KEYWORDS = ( "composite", "laminate", "reinforced", "prepreg", "cf/", "gf/", "fiber-reinforced", "fibre-reinforced", "ud ", "uni-directional", "unidirectional", "woven", "glass-filled", "glass filled", "carbon-filled", ) FIBER_KEYWORDS = ("fiber", "fibre", "yarn", "tow", "roving", "filament") def classify_material(material: Material) -> str: """Return 'Polymer' | 'Fiber' | 'Composite' (Task 5). Primary signal is the model's `material_class` enum; a deterministic keyword fallback (using sections + Vf presence) only runs when the field is missing or invalid. The old haystack bug (repeating material_name N times, never reading section) is gone. """ field = (material.material_class or "").strip().capitalize() if field in MATERIAL_CLASS_ENUM: return field # --- deterministic fallback --- # A reported fiber volume fraction is a strong composite signal. if material.fiber_volume_fraction.strip(): return "Composite" haystack = " ".join( [ material.material_name or "", material.material_abbreviation or "", material.trade_grade or "", material.matrix or "", material.fiber or "", ] + [p.section or "" for p in material.properties] + [p.property_name or "" for p in material.properties] ).lower() if any(k in haystack for k in COMPOSITE_KEYWORDS): return "Composite" if any(k in haystack for k in FIBER_KEYWORDS): return "Fiber" return "Polymer" def _autoabbr(name: str) -> str: if not name: return "UNKNOWN" parts = [p[0] for p in name.split() if p and p[0].isalpha()] return "".join(parts).upper() or name[:6].upper() def material_key(material: Material) -> str: """Stable material identity (Task 6): normalized name, plus the trade grade. ```` when no trade grade is known, else ``|``. The grade is part of the identity: a datasheet describing PEEK 150G and PEEK 450G yields two materials named "PEEK", and without the grade in the key any property both grades share (density, Tg, ...) dedup-collapsed into one row and the other was silently dropped. Falls back to the abbreviation when the name is empty. """ name = (material.material_name or "").strip().lower() name = re.sub(r"\s+", " ", name) if not name: name = (material.material_abbreviation or "unknown").strip().lower() grade = re.sub(r"\s+", " ", (material.trade_grade or "").strip().lower()) if grade and grade != name and grade not in name: return f"{name}|{grade}" return name # --------------------------------------------------------------------------- # Flatten to DB rows (Tasks 2/5/6) # --------------------------------------------------------------------------- def to_rows(extraction: Extraction, source_pdf: str, source_sha1: str) -> list[PropertyRow]: """Flatten an Extraction into PropertyRows, iterating materials x properties.""" rows: list[PropertyRow] = [] for material in extraction.materials: mclass = classify_material(material) mkey = material_key(material) abbr = material.material_abbreviation or _autoabbr(material.material_name) for prop in material.properties: rows.append( PropertyRow( material_name=material.material_name, material_abbreviation=abbr, material_key=mkey, material_class=mclass, section=prop.section, property_name=prop.property_name, value=prop.value_raw, # legacy column == value_raw (back-compat) unit=prop.unit, english="", # legacy alt-units column; model no longer emits it test_condition=prop.test_condition, comments=prop.comments, trade_grade=material.trade_grade, manufacturer=material.manufacturer, matrix=material.matrix, fiber=material.fiber, fiber_volume_fraction=material.fiber_volume_fraction, value_raw=prop.value_raw, value_num=prop.value_num, value_min=prop.value_min, value_max=prop.value_max, qualifier=prop.qualifier, unit_canonical=prop.unit_canonical, value_si=prop.value_si, source_pdf=source_pdf, source_sha1=source_sha1, page=prop.page, source_quote=prop.source_quote, status=prop.status, flag_reason=prop.flag_reason, model=extraction.model, prompt_version=extraction.prompt_version, ) ) return rows