Neqsim Technical Document Reading

Technical Document Reading Skill

Extract structured engineering data from technical documents and convert it into formats usable by process simulation, mechanical design, and engineering analysis.

Use the public community skill neqsim-document-intelligence-extraction before this skill when the task receives files. It owns source inventory, native/OCR/vision routing, provenance, confidence, review gates, and cross-document conflicts. This skill consumes that evidence package and adds document-type schemas, engineering normalization, physical validation, and NeqSim handoffs. Keeping this boundary explicit prevents OCR or multimodal output from being mistaken for verified engineering data.

Core Principle

Classify → Extract → Normalize → Validate → Output

1. Classify the document type to select the right extraction strategy
2. Extract raw data using format-specific tools (PDF, Word, Excel)
3. Normalize units, component names, and field names to standard conventions
4. Validate extracted data against physical bounds and completeness checks
5. Output structured JSON/dict for downstream consumption

For P&ID-driven operational studies, also load neqsim-pid-process-operations. That skill converts symbols into a process graph, classifies valve functions, maps instrument bubbles to historian tags, and defines steady-state or dynamic NeqSim scenario actions.

For water-hammer or liquid-hammer screening, also load neqsim-water-hammer. Extract route geometry, wall thickness, roughness/piping class, fittings, valve closure timing, design pressure, and tagreader event references so the output can feed WaterHammerStudy or MCP runWaterHammer.

For trapped-liquid fire rupture studies, also load neqsim-trapped-liquid-fire-rupture. Extract segment boundaries, line numbers, pipe geometry, material grade/certificate data, flange/gasket/bolt ratings, fire/PFP basis, relief availability, acceptance criteria, and explicit evidence gaps before handing data to the solver.

1. Document Type Classification

Classification Decision Tree

When a document is provided, classify it first:

Auto-Detection Heuristics

When document type isn't stated, look for these patterns:

DOCUMENT_SIGNATURES = {
    "equipment_data_sheet": [
        r"(?i)data\s*sheet", r"(?i)tag\s*no", r"(?i)design\s+press",
        r"(?i)operating\s+press", r"(?i)nozzle\s+schedule"
    ],
    "technical_requirement": [
        r"(?i)technical\s+requirement", r"(?i)\bTR[\s-]?\d+",
        r"(?i)\bshall\b.*\bdesign", r"(?i)design\s+factor"
    ],
    "design_basis": [
        r"(?i)design\s+basis", r"(?i)basis\s+of\s+design",
        r"(?i)operating\s+envelope", r"(?i)ambient\s+conditions"
    ],
    "heat_mass_balance": [
        r"(?i)heat\s+.*mass\s+balance", r"(?i)stream\s+(table|summary)",
        r"(?i)mol\s*%|mole\s+frac", r"(?i)vapour\s+fraction"
    ],
    "well_test": [
        r"(?i)well\s+test", r"(?i)choke\s+size", r"(?i)GOR",
        r"(?i)productivity\s+index", r"(?i)IPR"
    ],
    "inspection_report": [
        r"(?i)inspection\s+report", r"(?i)wall\s+thickness",
        r"(?i)corrosion\s+rate", r"(?i)remaining\s+life"
    ],
    "material_certificate": [
        r"(?i)material\s+cert", r"(?i)heat\s+number",
        r"(?i)SMYS|SMTS", r"(?i)charpy|impact\s+test"
    ],
    "piping_spec": [
        r"(?i)piping\s+class", r"(?i)material\s+class",
        r"(?i)line\s+class", r"(?i)pressure\s+rating"
    ],
    "line_list_route_table": [
        r"(?i)line\s*list", r"(?i)route\s*table",
        r"(?i)stress\s*iso(metric)?", r"(?i)from\s+node.*to\s+node",
        r"(?i)nominal\s+(size|diameter)", r"(?i)straight\s+length"
    ],
    "trapped_liquid_fire_rupture": [
        r"(?i)trapped\s+liquid", r"(?i)blocked[-\s]?in\s+liquid",
        r"(?i)thermal\s+expansion", r"(?i)fire\s+exposure",
        r"(?i)no\s+pressure\s+relief", r"(?i)passive\s+fire\s+protection|PFP",
        r"(?i)flange\s+(failure|leak|rupture)", r"(?i)pipe\s+rupture"
    ],
    "engineering_drawing_pid": [
        r"(?i)piping\s+.*instrument.*diagram", r"(?i)P&ID",
        r"(?i)P\s*&\s*I\s*D", r"(?i)instrument\s+diagram"
    ],
    "mechanical_arrangement": [
        r"(?i)general\s+arrangement", r"(?i)GA\s+drawing",
        r"(?i)section\s+view", r"(?i)plan\s+view",
        r"(?i)mechanical\s+arrangement", r"(?i)elevation\s+view"
    ],
    "vendor_api_datasheet_image": [
        r"(?i)API\s+6[19][0247]", r"(?i)dry\s+gas\s+seal",
        r"(?i)seal\s+gas", r"(?i)vendor\s+data\s*sheet",
        r"(?i)mechanical\s+seal", r"(?i)compressor\s+data\s*sheet"
    ],
    "performance_map": [
        r"(?i)performance\s+(map|curve)", r"(?i)compressor\s+map",
        r"(?i)pump\s+curve", r"(?i)phase\s+envelope",
        r"(?i)head\s+vs\s+flow", r"(?i)surge\s+line",
        r"(?i)operating\s+point", r"(?i)cricondentherm"
    ],
}

Trapped-Liquid Fire Rupture Extraction Schema

When the downstream task is trapped liquid, blocked-in liquid, fire rupture, PFP demand, or no pressure relief on liquid-filled piping, build this study input block:

{
    "study_type": "trapped_liquid_fire_rupture",
    "segments": [
        {
            "segment_id": "...",
            "line_numbers": ["..."],
            "isolation_boundary": {
                "upstream": {"tag": "...", "source": "..."},
                "downstream": {"tag": "...", "source": "..."},
                "vents_drains_relief_paths": []
            },
            "pipe_geometry": {
                "internal_diameter_m": {"value": null, "source": "...", "confidence": 0.0},
                "wall_thickness_m": {"value": null, "source": "...", "confidence": 0.0},
                "exposed_length_m": {"value": null, "source": "...", "confidence": 0.0}
            },
            "material": {
                "grade": {"value": null, "source": "..."},
                "smys_MPa": {"value": null, "source": "..."},
                "smts_MPa": {"value": null, "source": "..."}
            },
            "flange_gasket_bolt": {
                "class": {"value": null, "source": "..."},
                "gasket_type": {"value": null, "source": "..."},
                "temperature_rating_source": "..."
            },
            "fluid_and_conditions": {
                "fluid_description": "...",
                "composition": [],
                "operating_pressure_bara": {"value": null, "source": "..."},
                "operating_temperature_C": {"value": null, "source": "..."}
            },
            "fire_and_pfp": {
                "fire_type": "api521_pool_fire | fixed_heat_flux | radiative_fire | unknown",
                "heat_flux_W_m2": {"value": null, "source": "..."},
                "pfp_required_endurance_s": {"value": null, "source": "..."}
            },
            "relief_and_acceptance": {
                "relief_available": {"value": null, "source": "..."},
                "relief_set_pressure_bara": {"value": null, "source": "..."},
                "acceptance_criteria": []
            },
            "evidence_gaps": []
        }
    ]
}

Extraction priority for conflicting values: material certificate > piping spec > line list > P&ID annotation > narrative report > inferred/default. Always report conflicts rather than silently choosing one value.

2. File Format Handlers

2.1 PDF Extraction

Recommended library: pdfplumber (table-aware), fallback to pymupdf (fitz)

import pdfplumber

def extract_pdf(filepath):
    """Extract text and tables from a PDF document."""
    pages = []
    tables = []
    with pdfplumber.open(filepath) as pdf:
        for i, page in enumerate(pdf.pages):
            # Extract text
            text = page.extract_text() or ""
            pages.append({"page": i + 1, "text": text})

            # Extract tables (pdfplumber detects table boundaries)
            for table in page.extract_tables():
                if table and len(table) > 1:
                    headers = [str(h).strip() if h else "" for h in table[0]]
                    rows = []
                    for row in table[1:]:
                        rows.append([str(c).strip() if c else "" for c in row])
                    tables.append({
                        "page": i + 1,
                        "headers": headers,
                        "rows": rows
                    })
    return {"pages": pages, "tables": tables}

Extracting figures and diagrams from PDFs (PREFERRED):

Use devtools/pdf_to_figures.py to convert PDF pages to PNG images for visual analysis by AI tools (view_image). This is the fastest way to inspect engineering drawings, P&IDs, charts, tables, and compressor maps:

# CLI — convert all PDFs in the references folder
python devtools/pdf_to_figures.py step1_scope_and_research/references/ --outdir figures/

# CLI — specific pages only (1-indexed)
python devtools/pdf_to_figures.py path/to/document.pdf --pages 1 3 5 --outdir figures/

# Python / Notebook — programmatic use
from devtools.pdf_to_figures import pdf_to_pngs, pdf_folder_to_pngs

pngs = pdf_to_pngs("references/compressor_sketch.pdf", outdir="figures/")
all_pngs = pdf_folder_to_pngs("step1_scope_and_research/references/", outdir="figures/")

Then use view_image on the extracted PNGs to read diagrams, extract data from charts, identify equipment layouts, and analyze engineering drawings. Requires pymupdf.

Handling scanned PDFs:

• If extract_text() returns empty, the PDF is likely scanned/image-based
• First try pdf_to_figures.py to render pages as images, then use view_image for AI reading
• Use OCR as fallback: pytesseract + pdf2image, or run devtools/pdf_ocr.py (auto-detects scanned PDFs and falls back to OCR — see the neqsim-pdf-ocr skill for P&ID tag extraction and OCRmyPDF usage)
• Flag to user: "This appears to be a scanned document. OCR extraction may have errors."

2.2 Word Document Extraction

Library: python-docx

from docx import Document

def extract_docx(filepath):
    """Extract paragraphs and tables from a Word document."""
    doc = Document(filepath)
    paragraphs = []
    tables = []

    for para in doc.paragraphs:
        if para.text.strip():
            paragraphs.append({
                "text": para.text.strip(),
                "style": para.style.name,
                "level": _heading_level(para.style.name)
            })

    for i, table in enumerate(doc.tables):
        headers = [cell.text.strip() for cell in table.rows[0].cells]
        rows = []
        for row in table.rows[1:]:
            rows.append([cell.text.strip() for cell in row.cells])
        tables.append({"index": i, "headers": headers, "rows": rows})

    return {"paragraphs": paragraphs, "tables": tables}


def _heading_level(style_name):
    """Return heading level (1-9) or 0 for body text."""
    if style_name.startswith("Heading"):
        try:
            return int(style_name.split()[-1])
        except ValueError:
            return 0
    return 0

2.3 Excel / CSV Extraction

Libraries: openpyxl for .xlsx, pandas for both

import pandas as pd

def extract_excel(filepath, sheet_name=None):
    """Extract tables from Excel workbook."""
    sheets = {}
    xls = pd.ExcelFile(filepath)
    target_sheets = [sheet_name] if sheet_name else xls.sheet_names

    for name in target_sheets:
        df = pd.read_excel(xls, sheet_name=name)
        # Drop fully empty rows/columns
        df = df.dropna(how="all").dropna(axis=1, how="all")
        sheets[name] = {
            "headers": list(df.columns),
            "rows": df.values.tolist(),
            "shape": list(df.shape)
        }
    return sheets


def extract_csv(filepath):
    """Extract table from CSV file."""
    df = pd.read_csv(filepath)
    df = df.dropna(how="all").dropna(axis=1, how="all")
    return {
        "headers": list(df.columns),
        "rows": df.values.tolist(),
        "shape": list(df.shape)
    }

3. Extraction Patterns by Document Type

3.1 Equipment Data Sheet

Equipment data sheets follow standard API/ISO layouts. Key sections:

DATASHEET_SECTIONS = {
    "general": {
        "patterns": [
            r"(?i)tag\s*(?:no|number|#)[\s:]+(\S+)",
            r"(?i)service[\s:]+(.+?)(?:\n|$)",
            r"(?i)quantity[\s:]+(\d+)",
            r"(?i)type[\s:]+(.+?)(?:\n|$)",
        ],
        "fields": ["tag_number", "service", "quantity", "equipment_type"]
    },
    "design_conditions": {
        "patterns": [
            r"(?i)design\s+press(?:ure)?[\s:]+([0-9.]+)\s*(bar|bara|barg|psi|psig|MPa|kPa)",
            r"(?i)design\s+temp(?:erature)?[\s:]+([0-9.+-]+)\s*(°?[CFK]|degC|degF)",
            r"(?i)oper(?:ating)?\s+press(?:ure)?[\s:]+([0-9.]+)\s*(bar|bara|barg|psi|psig|MPa|kPa)",
            r"(?i)oper(?:ating)?\s+temp(?:erature)?[\s:]+([0-9.+-]+)\s*(°?[CFK]|degC|degF)",
        ],
        "fields": ["design_pressure", "design_temperature",
                    "operating_pressure", "operating_temperature"]
    },
    "dimensions": {
        "patterns": [
            r"(?i)(?:ID|inner\s+diameter|internal\s+diameter)[\s:]+([0-9.]+)\s*(mm|m|in|inch|ft)",
            r"(?i)(?:OD|outer\s+diameter|outside\s+diameter)[\s:]+([0-9.]+)\s*(mm|m|in|inch|ft)",
            r"(?i)length[\s:]+([0-9.]+)\s*(mm|m|ft)",
            r"(?i)(?:t-t|tan.*tan|seam.*seam|height)[\s:]+([0-9.]+)\s*(mm|m|ft)",
            r"(?i)wall\s+thick(?:ness)?[\s:]+([0-9.]+)\s*(mm|in|inch)",
        ],
        "fields": ["inner_diameter", "outer_diameter", "length",
                    "height", "wall_thickness"]
    },
    "material": {
        "patterns": [
            r"(?i)(?:shell|body)\s+material[\s:]+(\S+(?:\s+\S+)?)",
            r"(?i)material\s+grade[\s:]+(\S+)",
            r"(?i)(?:SA|ASTM|A)\s*-?\s*(\d+)(?:\s*-?\s*(?:Gr\.?\s*)?(\S+))?",
        ],
        "fields": ["shell_material", "material_grade"]
    }
}

Table extraction for data sheets:

def extract_datasheet_table(table):
    """Parse an equipment data sheet table into structured fields.

    Handles common layouts where col 0 is field name, col 1+ are values.
    Also handles multi-case tables (normal/upset/design columns).
    """
    result = {}
    headers = table["headers"]
    for row in table["rows"]:
        if len(row) >= 2 and row[0]:
            field_name = row[0].strip().lower()
            # Multi-case: create dict with case names as keys
            if len(headers) > 2 and len(row) > 2:
                for i, header in enumerate(headers[1:], start=1):
                    if i < len(row) and row[i]:
                        case_key = header.strip() if header else f"case_{i}"
                        if field_name not in result:
                            result[field_name] = {}
                        result[field_name][case_key] = _parse_value(row[i])
            else:
                result[field_name] = _parse_value(row[1])
    return result


def _parse_value(text):
    """Parse a text value, attempting numeric conversion."""
    text = str(text).strip()
    if not text or text == "-" or text.lower() == "n/a":
        return None
    try:
        return float(text.replace(",", ""))
    except ValueError:
        return text

3.2 Heat & Mass Balance (Stream Table)

Stream tables are the most common source of process data. They vary in layout:

Layout A — Streams as columns:

Layout B — Streams as rows:

def extract_stream_table(table, layout="auto"):
    """Extract stream data from a heat & mass balance table.

    Args:
        table: dict with "headers" and "rows" keys
        layout: "columns" (streams as columns), "rows" (streams as rows), or "auto"

    Returns:
        dict mapping stream names to their properties
    """
    headers = table["headers"]
    rows = table["rows"]

    if layout == "auto":
        layout = _detect_stream_layout(headers, rows)

    streams = {}

    if layout == "columns":
        # First column is property name, remaining columns are streams
        stream_names = [h.strip() for h in headers[1:] if h and h.strip()]
        for row in rows:
            if not row[0]:
                continue
            prop_name, prop_unit = _parse_property_header(row[0])
            for i, stream_name in enumerate(stream_names):
                col_idx = i + 1
                if col_idx < len(row) and row[col_idx]:
                    if stream_name not in streams:
                        streams[stream_name] = {}
                    streams[stream_name][prop_name] = {
                        "value": _parse_value(row[col_idx]),
                        "unit": prop_unit
                    }
    else:
        # First column (or first few) is stream name, rest are properties
        for row in rows:
            if not row[0]:
                continue
            stream_name = str(row[0]).strip()
            streams[stream_name] = {}
            for j, header in enumerate(headers[1:], start=1):
                if j < len(row) and row[j]:
                    prop_name, prop_unit = _parse_property_header(header)
                    streams[stream_name][prop_name] = {
                        "value": _parse_value(row[j]),
                        "unit": prop_unit
                    }

    return streams


def _detect_stream_layout(headers, rows):
    """Heuristic: if first header looks like a property label, it's column layout."""
    first_header = str(headers[0]).strip().lower()
    property_keywords = ["property", "parameter", "description", "item",
                         "stream", "temperature", "pressure", "flow"]
    if any(kw in first_header for kw in property_keywords[:4]):
        return "columns"
    return "rows"


def _parse_property_header(text):
    """Split 'Temperature (°C)' into ('temperature', '°C')."""
    import re
    text = str(text).strip()
    match = re.match(r"(.+?)\s*[\(\[]([^\)\]]+)[\)\]]", text)
    if match:
        return match.group(1).strip().lower(), match.group(2).strip()
    return text.lower(), ""

3.3 Fluid Composition Tables

# Common header variations for composition columns
COMPOSITION_HEADERS = {
    "component": [
        r"(?i)component", r"(?i)species", r"(?i)compound", r"(?i)name"
    ],
    "mole_fraction": [
        r"(?i)mol\s*%", r"(?i)mole\s+frac", r"(?i)mol\s+frac",
        r"(?i)y\s*[\(\[]", r"(?i)x\s*[\(\[]", r"(?i)z\s*[\(\[]",
        r"(?i)composition\s*\(mol", r"(?i)molar"
    ],
    "mass_fraction": [
        r"(?i)wt\s*%", r"(?i)mass\s+frac", r"(?i)weight\s+frac",
        r"(?i)mass\s*%"
    ],
    "volume_fraction": [
        r"(?i)vol\s*%", r"(?i)volume\s+frac", r"(?i)liq\s+vol\s*%"
    ]
}

# Component name mapping (common variations → NeqSim names)
COMPONENT_NAME_MAP = {
    # Hydrocarbons
    "c1": "methane", "ch4": "methane", "methane": "methane",
    "c2": "ethane", "c2h6": "ethane", "ethane": "ethane",
    "c3": "propane", "c3h8": "propane", "propane": "propane",
    "ic4": "i-butane", "i-c4": "i-butane", "isobutane": "i-butane",
    "nc4": "n-butane", "n-c4": "n-butane",
    "ic5": "i-pentane", "i-c5": "i-pentane", "isopentane": "i-pentane",
    "nc5": "n-pentane", "n-c5": "n-pentane",
    "nc6": "n-hexane", "n-c6": "n-hexane", "hexane": "n-hexane",
    "nc7": "n-heptane", "n-c7": "n-heptane", "heptane": "n-heptane",
    "nc8": "n-octane", "n-c8": "n-octane", "octane": "n-octane",
    "nc9": "n-nonane", "n-c9": "n-nonane",
    "nc10": "nC10", "n-c10": "nC10",
    "cyclohexane": "cyclohexane", "c-c6": "cyclohexane",
    "benzene": "benzene", "toluene": "toluene",
    # Non-hydrocarbons
    "co2": "CO2", "carbon dioxide": "CO2",
    "h2s": "H2S", "hydrogen sulfide": "H2S", "hydrogen sulphide": "H2S",
    "n2": "nitrogen", "nitrogen": "nitrogen",
    "h2": "hydrogen", "hydrogen": "hydrogen",
    "h2o": "water", "water": "water",
    "o2": "oxygen", "oxygen": "oxygen",
    "ar": "argon", "argon": "argon",
    "he": "helium", "helium": "helium",
    "co": "CO",
    # Glycols
    "meg": "MEG", "monoethylene glycol": "MEG",
    "deg": "DEG", "diethylene glycol": "DEG",
    "teg": "TEG", "triethylene glycol": "TEG",
    # Mercaptans
    "methyl mercaptan": "methyl-mercaptan", "ch3sh": "methyl-mercaptan",
    "ethyl mercaptan": "ethyl-mercaptan", "c2h5sh": "ethyl-mercaptan",
}


def normalize_component_name(raw_name):
    """Map a raw component name to the NeqSim standard name."""
    key = raw_name.strip().lower().replace("_", " ").replace("-", " ")
    # Direct lookup
    if key in COMPONENT_NAME_MAP:
        return COMPONENT_NAME_MAP[key]
    # Try removing spaces
    key_nospace = key.replace(" ", "")
    if key_nospace in COMPONENT_NAME_MAP:
        return COMPONENT_NAME_MAP[key_nospace]
    # Return original (user may need to map manually)
    return raw_name.strip()


def extract_composition(table):
    """Extract fluid composition from a table.

    Returns dict of {neqsim_name: mole_fraction} and metadata about
    the composition basis (mole%, mass%, etc.).
    """
    import re
    headers = table["headers"]

    # Find component column and value column
    comp_col = None
    value_col = None
    basis = "mole_fraction"

    for i, h in enumerate(headers):
        h_str = str(h).strip()
        for pattern in COMPOSITION_HEADERS["component"]:
            if re.search(pattern, h_str):
                comp_col = i
                break
        for basis_type in ["mole_fraction", "mass_fraction", "volume_fraction"]:
            for pattern in COMPOSITION_HEADERS[basis_type]:
                if re.search(pattern, h_str):
                    value_col = i
                    basis = basis_type
                    break

    # Fallback: assume col 0 = component, col 1 = value
    if comp_col is None:
        comp_col = 0
    if value_col is None:
        value_col = 1

    composition = {}
    for row in table["rows"]:
        if comp_col < len(row) and value_col < len(row):
            name = str(row[comp_col]).strip()
            value = _parse_value(row[value_col])
            if name and value is not None:
                neqsim_name = normalize_component_name(name)
                composition[neqsim_name] = float(value)

    # Normalize: if values sum to ~100, convert to fractions
    total = sum(composition.values())
    if total > 1.5:  # Likely percentages
        composition = {k: v / 100.0 for k, v in composition.items()}
        total = sum(composition.values())

    return {
        "components": composition,
        "basis": basis,
        "total": total,
        "normalized": abs(total - 1.0) < 0.02
    }

3.4 Technical Requirements

TR documents contain design rules expressed as requirements. Extract:

def extract_requirements(text):
    """Extract numbered requirements from a technical requirements document.

    Looks for 'shall' and 'should' statements with associated values.
    """
    import re
    requirements = []

    # Pattern: numbered clause + requirement text
    clause_pattern = re.compile(
        r"(\d+(?:\.\d+)*)\s+(.*?(?:shall|should|must|may)\s+.*?)(?=\n\d+(?:\.\d+)*\s|\Z)",
        re.IGNORECASE | re.DOTALL
    )

    for match in clause_pattern.finditer(text):
        clause_num = match.group(1)
        req_text = match.group(2).strip()

        # Extract numeric values with units
        values = re.findall(
            r"(\d+(?:\.\d+)?)\s*(bar[ag]?|°?[CFK]|mm|m|kg|psi|MPa|kPa|%|hr|min)",
            req_text
        )

        # Classify requirement type
        if re.search(r"(?i)shall\b", req_text):
            level = "mandatory"
        elif re.search(r"(?i)should\b", req_text):
            level = "recommended"
        else:
            level = "informative"

        requirements.append({
            "clause": clause_num,
            "text": req_text,
            "level": level,
            "values": [{"value": float(v), "unit": u} for v, u in values]
        })

    return requirements

3.5 Well Test Reports

WELL_TEST_FIELDS = {
    "flow_rate": [
        r"(?i)(?:oil|gas|liquid|water)\s+(?:flow\s+)?rate[\s:]+([0-9.,]+)\s*(Sm3/d|bbl/d|MMSCFD|kg/hr|m3/d)",
    ],
    "gor": [
        r"(?i)GOR[\s:]+([0-9.,]+)\s*(Sm3/Sm3|scf/bbl|m3/m3)",
    ],
    "water_cut": [
        r"(?i)water\s*cut[\s:]+([0-9.,]+)\s*(%)?",
        r"(?i)BSW[\s:]+([0-9.,]+)\s*(%)?",
    ],
    "reservoir_pressure": [
        r"(?i)reservoir\s+press(?:ure)?[\s:]+([0-9.,]+)\s*(bar[ag]?|psi[ag]?|MPa|kPa)",
    ],
    "reservoir_temperature": [
        r"(?i)reservoir\s+temp(?:erature)?[\s:]+([0-9.,]+)\s*(°?[CFK])",
    ],
    "wellhead_pressure": [
        r"(?i)(?:WHP|wellhead\s+press(?:ure)?)[\s:]+([0-9.,]+)\s*(bar[ag]?|psi[ag]?|MPa)",
    ],
    "wellhead_temperature": [
        r"(?i)(?:WHT|wellhead\s+temp(?:erature)?)[\s:]+([0-9.,]+)\s*(°?[CFK])",
    ],
    "choke_size": [
        r"(?i)choke[\s:]+([0-9.,]+)(?:/64)?[\s]*(mm|inch|in|/64)",
    ],
    "productivity_index": [
        r"(?i)(?:PI|productivity\s+index)[\s:]+([0-9.,]+)\s*(Sm3/d/bar|bbl/d/psi)",
    ],
}

3.6 Inspection Reports

def extract_thickness_data(table):
    """Extract wall thickness measurements from an inspection table.

    Common format: Location | Nominal (mm) | Measured (mm) | Min (mm) | Corrosion Rate
    """
    result = []
    for row in table["rows"]:
        entry = {}
        for i, header in enumerate(table["headers"]):
            h = str(header).strip().lower()
            if i < len(row) and row[i]:
                val = _parse_value(row[i])
                if "location" in h or "position" in h or "point" in h:
                    entry["location"] = str(row[i]).strip()
                elif "nominal" in h or "original" in h:
                    entry["nominal_mm"] = val
                elif "measured" in h or "actual" in h or "remaining" in h:
                    entry["measured_mm"] = val
                elif "min" in h:
                    entry["minimum_mm"] = val
                elif "corrosion" in h and "rate" in h:
                    entry["corrosion_rate_mm_yr"] = val
        if entry:
            result.append(entry)
    return result

3.7 Image and Figure Analysis (Engineering Drawings, P&IDs, Performance Maps)

Many engineering documents contain critical data embedded in images rather than text — P&IDs, mechanical arrangement drawings, vendor datasheets rendered as scanned PDFs, compressor performance maps, and phase envelopes. Use the view_image tool for multimodal analysis of these visual documents.

3.7.1 Workflow: Image-Based Document Analysis

PDF/Image Document → pdf_to_figures.py → PNG pages → view_image → Structured Data

Step 1: Convert PDF pages to images

# Single file
python devtools/pdf_to_figures.py path/to/document.pdf --outdir figures/ --dpi 200

# Specific pages
python devtools/pdf_to_figures.py document.pdf --outdir figures/ --pages 1,3,5

# Batch (all PDFs in a folder)
python devtools/pdf_to_figures.py references/ --outdir figures/

Step 2: View and analyze each page with view_image

Use view_image on each extracted PNG. For each image, systematically extract:

1. Title block — document number, revision, title, date, originator
2. Equipment tags — vessel tags (V-xxx), pump tags (P-xxx), compressor tags (BCL-xxx)
3. Instrument tags — transmitters (PT-, TT-, LT-, FT-), valves (XV-, PV-, LV-)
4. Piping information — line sizes, piping class, connection types (flanged, welded)
5. Operating conditions — T, P, flow annotations on streams
6. Notes and legends — design notes, material callouts, hold/revision clouds
7. Dimensional data — lengths, diameters, elevations (for GA drawings)

Step 3: Structure the extracted data

Convert visual observations into the standard extraction result JSON format (Section 6.1), adding an image_analysis block.

3.7.2 Engineering Drawing Types and Extraction Patterns

3.7.3 P&ID Data Extraction Pattern

When reading a P&ID image with view_image, extract data in this structured format:

PID_EXTRACTION = {
    "document": {
        "drawing_number": "SOK-xxxxxx",
        "revision": "C2",
        "title": "Oil/Seal Gas Piping — Compressor BCL-304/D",
        "area": "26",
        "unit": "GIC Compressor"
    },
    "equipment": [
        {"tag": "BCL-304/D", "type": "Centrifugal Compressor", "service": "Gas Injection"},
        {"tag": "FLT-26110", "type": "Coalescing Filter", "service": "Seal Gas Supply"},
    ],
    "valves": [
        {"tag": "VB26-0130", "type": "Ball Valve", "size_inch": 0.75,
         "class": "2500 RTJ", "service": "Seal Gas Supply Isolation"},
        {"tag": "XV-S26102.05", "type": "Shutdown Valve",
         "service": "Primary Seal Gas", "normally": "open"},
    ],
    "instruments": [
        {"tag": "PT-S26102.03", "type": "Pressure Transmitter",
         "service": "Seal Gas Supply Pressure", "range": "0-200 barg"},
        {"tag": "TP-S26102.11", "type": "Temperature Point",
         "service": "Primary Vent Temperature"},
        {"tag": "FT-S26102.01", "type": "Flow Transmitter",
         "service": "Seal Gas Flow"},
    ],
    "piping": [
        {"line_number": "26-G1H-001", "size_inch": 0.75,
         "piping_class": "G1H", "rating": "2500#",
         "from": "FLT-26110", "to": "BCL-304/D NDE Seal"},
    ],
    "connections": [
        {"from": "FLT-26110.outlet", "to": "BCL-304/D.seal_gas_inlet",
         "type": "seal_gas_supply"},
        {"from": "BCL-304/D.primary_vent", "to": "VB26-0180",
         "type": "seal_vent", "destination": "Flare/Vent"},
    ],
    "route_segments": [
        {"segment_id": "26-G1H-001-S1", "from_node": "FLT-26110", "to_node": "XV-S26102.05",
         "line_number": "26-G1H-001", "size_inch": 0.75,
         "internal_diameter_m": 0.019, "length_m": 4.5,
         "elevation_change_m": 0.0,
         "minor_losses": [{"type": "ball valve", "k_value": 0.05}],
         "source_ref": "P&ID page 1, line 26-G1H-001"}
    ]
}

For route pressure-drop work, pass route_segments to neqsim.process.equipment.pipeline.routing.PipingRouteBuilder and save route.toJson() with the task results.

For operational studies, also create a pid_operational_model block as defined in neqsim-pid-process-operations. Include symbol semantics, valve normal and fail positions when stated, control links, drains, vents, bypasses, check valves, and logical tag names for plant data binding. Flag uncertain symbol readings; do not infer live valve state from drawing normal position alone.

3.7.4 Vendor API Datasheet Image Extraction

For vendor datasheets rendered as images (common for API 692 seal datasheets, API 617 compressor datasheets):

VENDOR_DATASHEET_EXTRACTION = {
    "equipment_tag": "BCL-304/D",
    "vendor": "John Crane / Flowserve / EagleBurgmann",
    "api_standard": "API 692",
    "data_sections": {
        "operating_conditions": {
            "suction_pressure_barg": 125.0,
            "discharge_pressure_barg": 190.0,
            "seal_gas_supply_pressure_barg": 155.0,
            "operating_temperature_C": 45.0,
            "shaft_speed_rpm": 11000,
        },
        "seal_data": {
            "seal_type": "Tandem dry gas seal",
            "seal_arrangement": "API Plan 74 (primary) + API Plan 76 (secondary)",
            "primary_leakage_nm3hr": 12.0,
            "secondary_leakage_nm3hr": 2.0,
            "buffer_gas": "Nitrogen",
        },
        "gas_supply": {
            "required_flow_nm3hr": 25.0,
            "required_pressure_above_ref_barg": 3.5,
            "filtration_micron": 3,
            "gas_temperature_range_C": [-10, 60],
        },
        "materials": {
            "face_material": "Silicon Carbide",
            "seat_material": "Carbon",
            "o_ring_material": "Viton / FFKM",
            "spring_material": "Inconel 718",
        }
    }
}

3.7.5 Performance Map / Phase Envelope Extraction

When analyzing compressor maps or phase envelopes from images:

PERFORMANCE_MAP_EXTRACTION = {
    "chart_type": "compressor_map",  # or "phase_envelope", "pump_curve"
    "axes": {
        "x": {"label": "Inlet Volume Flow", "unit": "m3/hr"},
        "y": {"label": "Polytropic Head", "unit": "kJ/kg"}
    },
    "rated_point": {
        "flow": 5000, "head": 85.0, "efficiency_pct": 82.0, "speed_rpm": 11000
    },
    "surge_point": {
        "flow": 3200, "head": 92.0
    },
    "curves": [
        {"speed_rpm": 11000, "points": [
            {"flow": 3200, "head": 92.0},
            {"flow": 4000, "head": 89.0},
            {"flow": 5000, "head": 85.0},
            {"flow": 6000, "head": 78.0},
        ]},
    ],
    "operating_window": {
        "min_flow": 3500, "max_flow": 5800,
        "min_head": 70.0, "max_head": 95.0
    }
}

PHASE_ENVELOPE_EXTRACTION = {
    "chart_type": "phase_envelope",
    "axes": {
        "x": {"label": "Temperature", "unit": "C"},
        "y": {"label": "Pressure", "unit": "bar"}
    },
    "critical_point": {"temperature_C": -82.0, "pressure_bar": 46.0},
    "cricondentherm": {"temperature_C": -5.4, "pressure_bar": 38.0},
    "cricondenbar": {"temperature_C": -25.0, "pressure_bar": 55.0},
    "key_points": [
        {"label": "Operating Point", "temperature_C": 25.0, "pressure_bar": 45.0,
         "phase_region": "single_phase_gas"},
        {"label": "JT Outlet", "temperature_C": -20.0, "pressure_bar": 3.0,
         "phase_region": "single_phase_gas"},
    ],
    "retrograde_region": {
        "temperature_range_C": [-40, -5],
        "pressure_range_bar": [30, 55],
        "max_liquid_fraction_pct": 8.0
    }
}

3.7.6 Mechanical Arrangement Drawing Extraction

For GA drawings showing physical layout and dimensions:

…(truncated)