neqsim-plant-data

# Plant Data Integration with Tagreader Patterns for connecting NeqSim process simulations to real plant data from OSIsoft PI and Aspen IP.21 historians using the tagreader package. For P&ID-driven operational studies, also load `neqsim-pid-process-operations` to map instrument bubbles and valve symbols to logical tag names, infer active state, and drive steady-state or dynamic NeqSim scenarios. For water-hammer or liquid-hammer screening, also load `neqsim-water-hammer` and extract short event windows with inlet pressure, temperature, flow rate, valve position, pump speed/status, and observed closure/trip timing for MCP `runWaterHammer`. ## Installation ```bash pip install tagreader neqsim ``` ## Tagreader API Reference ### Listing Available Data Sources ```python import tagreader # Discover what historian servers are accessible pi_sources = tagreader.list_sources("piwebapi") # OSIsoft PI Web API aspen_sources = tagreader.list_sources("aspenone") # Aspen IP.21 REST API ``` For custom historian deployments, supply `url`, `auth`, and optionally `verify_ssl`: ```python from requests_ntlm import HttpNtlmAuth auth = HttpNtlmAuth("domain\\user", "password") sources = tagreader.list_sources("aspenone", url="https://api.mycompany.com/aspenone", auth=auth, verify_ssl=False) ``` ### Creating and Connecting a Client ```python # PI Web API c = tagreader.IMSClient("MY_PI_SOURCE", "piwebapi") c.connect() # Aspen IP.21 c = tagreader.IMSClient("MY_ASPEN_SOURCE", "aspenone") c.connect() # Custom deployment with explicit auth c = tagreader.IMSClient( datasource="myplant", imstype="aspenone", url="https://api.mycompany.com/aspenone", auth=auth, verify_ssl=False, tz="US/Central" # default is "Europe/Oslo" ) c.connect() ``` **Constructor parameters:** | Parameter | Required | Description | |-----------|----------|-------------| | `datasource` | Yes | Name of the data source (from `list_sources()`) | | `imstype` | Yes | `"piwebapi"` or `"aspenone"` | | `tz` | No | Timezone for timestamps. Default: `"Europe/Oslo"` | | `url` | No | Server URL for custom deployments | | `verify_ssl` | No | SSL verification. Default: `True` | | `auth` | No | Auth object. Default is determined by the configured source | | `cache` | No | Local disk cache for avoiding re-reads | ### Searching for Tags ```python # Search by tag name (wildcards supported) results = c.search("*COMP*232*") # Search by description results = c.search(desc="*temperature*") # Both must match results = c.search(tag="BA:*", desc="*Temperature*") # With timeout (avoids long-running searches) results = c.search("*COMP*", timeout=30) ``` > **Gotcha — historian point names rarely equal the P&ID/STID tag number.** A PI > point is often prefixed and suffixed (e.g. STID `20PIT1111` → historian > `KRI.20PIT1111/Y/PRIM`, and a valve position as `.../OutPosition/PRIM`). Search > with **both-side wildcards** `*<tag>*` (a trailing-only `<tag>*` will miss the > prefix and resolve nothing), then map STID tag → resolved point and **verify the > resolved count** before `read()`. Also check the per-tag `::status`/quality > column — a value with bad quality (e.g. status 2) must not be trusted as a > current-operation input. > **Block suffixes select what a point returns.** On PI sources the same > instrument commonly exposes a family of blocks under > `<PREFIX>.<TAG>/<BLOCK>/PRIM`: `Meas1`/`Meas` = measured value, > `SPntOut` = setpoint, `ConOut` = controller output, `Manual` = mode, > `OutControl` = valve output/position demand, `Position` = valve position > feedback, and `Running`/`Start`/`Stop`/`PowerDemand` = motor status. Pick the > block deliberately — a "flow" or "power" point may be a status block that > holds a near-constant code rather than a measurement, so check > `series.nunique()` before trusting it. Prefix casing can differ between tags on > the same source, so search case-insensitively instead of hand-building names. > Site-specific source names, prefixes and suffix families are recorded in the > enterprise `enterprise-plant-data` skill. Returns a list of tuples: `[(tag_name, description), ...]` ### Reading Data ```python tags = ["TAG1.PV", "TAG2.PV"] start = "01.06.2025 06:00:00" # dd.mm.yyyy HH:MM:SS end = "01.06.2025 18:00:00" interval = 300 # seconds (5 minutes) # Interpolated (default) — evenly spaced data points df = c.read(tags, start, end, interval) # Time-weighted average per interval df = c.read(tags, start, end, interval, read_type=tagreader.ReaderType.AVG) # Minimum value per interval df = c.read(tags, start, end, interval, read_type=tagreader.ReaderType.MIN) # Maximum value per interval df = c.read(tags, start, end, interval, read_type=tagreader.ReaderType.MAX) # Variance per interval df = c.read(tags, start, end, interval, read_type=tagreader.ReaderType.VAR) # Standard deviation per interval df = c.read(tags, start, end, interval, read_type=tagreader.ReaderType.STD) # Range (max - min) per interval df = c.read(tags, start, end, interval, read_type=tagreader.ReaderType.RNG) # Raw stored values (no interpolation) df = c.read(tags, start, end, read_type=tagreader.ReaderType.RAW) # Snapshot (current/latest value, one tag at a time) df = c.read(["TAG1.PV"], read_type=tagreader.ReaderType.SNAPSHOT) # Snapshot at specific time df = c.read(["TAG1.PV"], end_time="01.06.2025 12:00:00", read_type=tagreader.ReaderType.SNAPSHOT) ``` **Read parameters:** | Parameter | Required | Description | |-----------|----------|-------------| | `tags` | Yes | List of tag names (no wildcards) | | `start_time` | Yes* | Start of period (*not needed for SNAPSHOT) | | `end_time` | Yes* | End of period (*optional for SNAPSHOT) | | `ts` | No | Interval in seconds. Default: 60 | | `read_type` | No | `ReaderType.INT` (default), `AVG`, `MIN`, `MAX`, `VAR`, `STD`, `RNG`, `RAW`, `SNAPSHOT` | | `get_status` | No | Include data quality column. Default: `False` | **Returns:** `pandas.DataFrame` with `DatetimeIndex` and one column per tag. ### Saving Data to CSV (MANDATORY for Task Workflows) **All plant data read via tagreader MUST be saved as CSV inside the task folder.** This creates a reproducible data snapshot so analysis can be rerun without live historian access, and keeps the task self-contained. ```python import os, pathlib # ── Resolve task directory (same pattern as notebooks) ── NOTEBOOK_DIR = pathlib.Path(globals().get( "__vsc_ipynb_file__", os.path.abspath("step2_analysis/notebook.ipynb") )).resolve().parent TASK_DIR = NOTEBOOK_DIR.parent REFS_DIR = TASK_DIR / "step1_scope_and_research" / "references" REFS_DIR.mkdir(parents=True, exist_ok=True) # ── Read from historian ── tags = list(TAG_MAP.values()) start = "01.06.2025 06:00:00" end = "01.06.2025 18:00:00" df = c.read(tags, start, end, 300) # 5-min interpolated # ── Save raw plant data as CSV ── csv_path = REFS_DIR / "plant_data_raw.csv" df.to_csv(str(csv_path)) print(f"Saved {len(df)} rows × {len(df.columns)} tags → {csv_path}") # ── Save tag mapping alongside the data ── import json tag_map_path = REFS_DIR / "tag_mapping.json" with open(str(tag_map_path), "w") as f: json.dump(TAG_MAP, f, indent=2) print(f"Tag mapping saved → {tag_map_path}") ``` **Loading saved CSV for offline analysis** (no historian needed): ```python import pandas as pd, json, pathlib REFS_DIR = TASK_DIR / "step1_scope_and_research" / "references" # Load data + tag map df_plant = pd.read_csv(str(REFS_DIR / "plant_data_raw.csv"), index_col=0, parse_dates=True) with open(str(REFS_DIR / "tag_mapping.json")) as f: TAG_MAP = json.load(f) print(f"Loaded {len(df_plant)} rows, tags: {list(TAG_MAP.keys())}") ``` **CSV naming conventions** — use descriptive filenames with date range: | Filename | Content | |----------|---------| | `plant_data_raw.csv` | Raw historian data (all tags, full period) | | `plant_data_20250601_20250615.csv` | Date-scoped raw data | | `plant_data_hourly_avg.csv` | Aggregated hourly averages | | `plant_data_cleaned.csv` | After quality filtering (NaN removed, outliers clipped) | | `tag_mapping.json` | Logical name → historian tag mapping | | `digital_twin_results.csv` | Simulated vs plant comparison output | ### Data Quality ```python df = c.read(tags, start, end, 60, get_status=True) # Adds extra columns: TAG1.PV_status, TAG2.PV_status ``` | Status Code | Meaning | |-------------|---------| | 0 | Good | | 1 | Suspect | | 2 | Bad | | 4 | Good/Modified | | 5 | Suspect/Modified | | 6 | Bad/Modified | ### Aspen IP.21 Tag Maps For IP.21 tags with maps (controller outputs, setpoints, etc.): ```python # Tag with map: "tag;map" df = c.read(["109-HIC005;CS A_AUTO"], start, end, 60) ``` ### Time Format Tagreader accepts multiple timestamp formats via `pandas.Timestamp`: ``` "02.09.2025 23:00:00" # dd.mm.yyyy HH:MM:SS (European) "05-Jan-2025 08:00:00" # dd-Mon-yyyy HH:MM:SS "05/01/25 11:30am" # US short format "2025-06-01T08:00:00" # ISO format datetime(2025, 6, 1, 8) # Python datetime objects ``` ## Tag Mapping Pattern Map logical process parameters to historian tag names: ```python TAG_MAP = { # Inlet separator 'inlet_pressure': 'PLANT-20PT0201.PV', 'inlet_temperature': 'PLANT-20TT0201.PV', # Compressor A 'compA_suction_P': 'PLANT-35PT0101A.PV', 'compA_discharge_P': 'PLANT-35PT0101B.PV', 'compA_suction_T': 'PLANT-35TT0101A.PV', 'compA_discharge_T': 'PLANT-35TT0101B.PV', 'compA_flow': 'PLANT-35FT0101.PV', 'compA_power': 'PLANT-35JI0101.PV', 'compA_speed': 'PLANT-35SI0101.PV', # Export 'export_pressure': 'PLANT-27PT0301.PV', 'export_flow': 'PLANT-27FT0301.PV', } ``` ### Helper to Extract Values by Parameter Name ```python import math def get_plant_value(df, tag_map, param_name, aggregation='mean'): """Get a plant value by logical parameter name. Args: df: DataFrame from tagreader tag_map: Dictionary mapping logical names to tag names param_name: Logical parameter name (key in tag_map) aggregation: 'mean', 'last', 'median', 'min', 'max' Returns: float value or float('nan') if no data """ tag = tag_map[param_name] if tag not in df.columns: return float('nan') series = df[tag].dropna() if len(series) == 0: return float('nan') if aggregation == 'mean': return float(series.mean()) elif aggregation == 'last': return float(series.iloc[-1]) elif aggregation == 'median': return float(series.median()) elif aggregation == 'min': return float(series.min()) elif aggregation == 'max': return float(series.max()) return float(series.mean())