Connects to OSIsoft PI System via the PI Web API (WebPI REST). Three modules cover metadata traversal, live data polling, and historic batch backfill. All three modules authenticate with PI using HTTP Basic Auth; SSL verification is caller-controlled via verify_ssl.
Source files:
clarity:backend/src-tauri/src/connectors/webpi/webpi_meta_connector.rsclarity:backend/src-tauri/src/connectors/webpi/webpi_live_data_connector.rsclarity:backend/src-tauri/src/connectors/webpi/webpi_historic_data_connector.rsAll Clarity-side endpoints are registered under the warp router as /pi/<name> and exposed at http://localhost:3030/exactapi/pi/<name>.
Source: clarity:backend/src-tauri/src/connectors/webpi/webpi_meta_connector.rs:26-81
All models deserialize PI Web API JSON responses. Serde rename is PascalCase unless noted.
| Struct | Serde | Key fields |
|---|---|---|
PiItem |
PascalCase | web_id, id, name, description, path, has_children — all Option<String> except has_children: Option<bool> |
PiItemSummary |
camelCase | web_id: String, element_count: usize, attribute_count: usize — used in list-endpoint responses |
PiAttribute |
PascalCase | web_id, id, name, description, path, attr_type (field named Type in PI), default_units_name, default_uom, default_units_name_abbreviation, config_string, data_tag_id, data_reference_plug_in, links: Option<HashMap<String, String>>, point: Option<PiPoint> |
PiPoint |
none | name, descriptor, engineering_units, id, web_id — nested inside PiAttribute |
PiElementNode |
PascalCase | name, web_id, description, attributes: Vec<PiAttribute>, children: Vec<PiElementNode> — recursive tree node |
PiDatabaseNode |
PascalCase | name, web_id, elements: Vec<PiElementNode> |
PiServerNode |
PascalCase | name, web_id, databases: Vec<PiDatabaseNode> |
PiHierarchy |
— | servers: Vec<PiServerNode> — internal full-tree type; not returned directly in API responses |
data_tag_id resolution on PiAttribute (lines 436–448): priority is (1) resolved PiPoint.name via Links.Point URL, (2) last path component of ConfigString for "PI Point" data reference plugin, stripping query parameters after ?.
webpi_meta_connector.rsPiConnector wraps all outbound requests to the PI Web API. HTTP Basic Auth, 30-second timeout, optional SSL cert bypass (danger_accept_invalid_certs(!verify_ssl)).
Source: clarity:backend/src-tauri/src/connectors/webpi/webpi_meta_connector.rs:161-168
| Method | PI endpoint called | Notes |
|---|---|---|
list_asset_servers() |
GET /assetservers |
Returns HashMap<String, String> (name → web_id) |
list_asset_servers_summary() |
GET /assetservers + parallel per-server GET /assetservers/{id}/assetdatabases |
Returns HashMap<String, PiItemSummary> |
get_databases(server_web_id) |
GET /assetservers/{id}/assetdatabases |
Returns HashMap<String, String> |
get_databases_summary(server_web_id) |
GET /assetservers/{id}/assetdatabases + parallel element count |
Returns HashMap<String, PiItemSummary> |
get_root_elements(db_web_id) |
GET /assetdatabases/{id}/elements |
Returns Vec<Value> |
get_child_elements(element_web_id) |
GET /elements/{id}/elements |
Returns Vec<Value> |
get_child_elements_summary(web_id, is_database) |
GET /assetdatabases/{id}/elements (is_database) or GET /elements/{id}/elements, then parallel attribute count |
Returns HashMap<String, PiItemSummary> |
get_attributes(element_web_id) |
GET /elements/{id}/attributes |
Returns Vec<Value> |
get_attribute_details(attr_web_id) |
GET /attributes/{id} |
Returns full attribute Value |
get_pipoint_info(link_or_web_id) |
GET /points/{id} or full URL from Links.Point |
Returns PiPoint |
Source: clarity:backend/src-tauri/src/connectors/webpi/webpi_meta_connector.rs:211-386
Source: clarity:backend/src-tauri/src/connectors/webpi/webpi_meta_connector.rs:1568-1807
Route function pi_meta_routes_full registers all 10 endpoints.
POST /exactapi/pi/list_asset_serversSource: lines 1580–1583, handler at lines 877–888
Request body: {api_url, username, password?, verify_ssl?}
Response: { "ServerName": {"web_id": "...", "element_count": N, "attribute_count": N}, ... }
NOTE: Old
PI_ONBOARDING_API.mddocuments response as{WebId, Name, Path}. Source returnsPiItemSummary{web_id, element_count, attribute_count}— old doc is stale.
POST /exactapi/pi/list_databasesSource: lines 1585–1588, handler at lines 890–902
Request body: {api_url, username, password?, verify_ssl?, web_id} — accepts web_id, webid, or server_web_id (all aliases).
Response: { "DatabaseName": {"web_id": "...", "element_count": N, "attribute_count": N}, ... }
POST /exactapi/pi/list_childrenSource: lines 1590–1593, handler at lines 904–917
Request body: {api_url, username, password?, verify_ssl?, web_id, is_database: bool} — accepts web_id, webid, or element_web_id. Set is_database: true for first call under a database; false for subsequent element traversal.
Response: { "ElementName": {"web_id": "...", "element_count": N, "attribute_count": N}, ... }
POST /exactapi/pi/list_attributesSource: lines 1595–1598, handler at lines 1554–1566
Request body: {api_url, username, password?, verify_ssl?, web_id}
Response: Raw attribute array from PI Web API — each entry is a full PI attribute JSON object including Name, WebId, Type, DefaultUnitsName, ConfigString.
POST /exactapi/pi/dump_hierarchySource: lines 1617–1621, handler at lines 1506–1552
Traverses the entire AF hierarchy starting from all asset servers; writes result to disk. Does not stream.
Request body: {api_url, username, password?, verify_ssl?, connectionId?}
Output file: pi_onboarding/{connectionId}_AF.json (or full_hierarchy.json if connectionId omitted).
Response: {"status":"success","path":"/absolute/path/to/file"}
POST /exactapi/pi/all_hierarchy_streamSource: lines 1795–1798, handler at lines 1718–1783
Streams the AF hierarchy as NDJSON without writing to disk. Recursion depth limited to 5.
Request body: {api_url, username, password?, verify_ssl?}
Response: Content-Type: application/x-ndjson, one JSON object per line:
| type | Fields |
|---|---|
"server" |
name, webId |
"database" |
name, webId |
"element" |
name, webId, depth, hasChildren, attributeCount |
POST /exactapi/pi/dump_and_streamSource: lines 1800–1804, handler at lines 1810–1987
Combined: streams NDJSON for real-time UI and writes the full hierarchy to {connectionId}_AF.json. Recommended for interactive UI.
Request body: {api_url, username, password?, verify_ssl?, connectionId?}
Response: Content-Type: application/x-ndjson. Each line adds tree-rendering fields:
| Field | Type | Description |
|---|---|---|
type |
string | "init", "server", "database", "element", "complete", "error" |
id |
string | Same as webId — unique node key |
parentId |
string/null | Parent's webId |
pathIds |
string[] | Ancestor chain from root to parent |
name |
string | Display label |
isExpandable |
bool | Has child elements |
isLastChild |
bool | Last sibling in parent's child list (for branch-line rendering) |
attributeCount |
number | PI attribute count |
depth |
number | Nesting level (element nodes only) |
Final line: {"type":"complete","message":"...","dumpPath":"/path/to/_AF.json","totalNodes":N}
POST /exactapi/pi/onboard_unitSource: lines 1600–1608, handler at lines 919–1275
Crawls an AF element and all its direct attributes; writes tagmap, metadata, and progress files; then launches the live data driver. Requires Authorization: Bearer <token> header for post-onboarding steps (collection metadata update, Qdrant trigger).
Request body: {api_url, username, password?, verify_ssl?, web_id, connectionId} — accepts web_id, webid, or unit_element_webid; connectionId may be string or number.
Response: Content-Type: application/x-ndjson streaming:
| Event | Shape |
|---|---|
| Start | {"status":"started","connectionId":"N"} |
| Per tag | {"status":"tag","tag":{"dataTagId":"...","description":"...","DefaultUnitsName":"...","DefaultUom":"...","DefaultUnitsNameAbbreviation":"..."}} |
| Success | {"status":"completed","total_tags":N} |
| Error | {"status":"error","error":"..."} |
NOTE: Old
PI_ONBOARDING_API.mddocuments this as a non-blocking call returning{status:"onboarding_started"}immediately. Current source implements a streaming NDJSON response — old doc is stale.
Output files written to pi_onboarding/ under app data dir:
| File | Contents |
|---|---|
{connectionId}_tagmap.ndjson |
One JSON object per line; fields: AttributeWebId, AttributeName, AttributeDescription, AttributeDefaultUom, AttributeDefaultUnitsName, AttributeDefaultUnitsNameAbbreviation, PointDescriptor, PointEngineeringUnits, TagName, PointWebId, ConfigString, DataTagId, Path |
{connectionId}_meta.json |
Array of PiElementNode trees for onboarded elements |
{connectionId}_progress.json |
Live progress; fields: status ("running" or "done"), message, elements_processed, attributes_processed, tags_found |
{connectionId}_connectionid |
Merged hierarchy tree (skeleton parents + direct attributes) for PI live driver |
Source for file names and fields: lines 961–963, 544–551, 815–821
NOTE: Old
PI_ONBOARDING_API.mddocuments progress fields as{progress, total_elements, processed_elements}. Source uses{elements_processed, attributes_processed, tags_found}— old doc is stale.
Post-onboarding steps (run after stream closes, lines 1029–1259):
connectionId's collectionId; update tags and descriptions fields in collection metadatapi-driver-{connectionId} managed Rust task (live data loop)POST https://localhost:3030/exactapi/create_qdrant_collectionPOST /exactapi/pi/start_backfillSource: lines 1610–1615, handler at lines 1639–1716
Registers and starts a pi-backfill-{timestamp} managed Rust task via ProcessManager. Backfill state is persisted; interrupted runs resume from last checkpoint.
Request body:
| Parameter | Type | Default | Description |
|---|---|---|---|
connection_id |
string | — | Connection ID |
start_time |
string | — | ISO8601 or "AUTO" (→ Unix epoch) |
end_time |
string | — | ISO8601 or "AUTO" (→ now) |
pi_url |
string | — | PI Web API URL |
username |
string | — | |
password |
string | — | |
resolution_ms |
number | 60000 |
Data grid resolution (ms) |
concurrency |
number | 4 |
Parallel worker count |
pacing_ms |
number | 0 |
Inter-request delay (ms) |
tag_filter |
string[] | — | WebIds to backfill; required — no all-tag fallback |
Source for defaults: webpi_meta_connector.rs:1650–1652
Response: {"status":"success","message":"Backfill task started","process_id":N} or {"status":"error","message":"..."}
POST /exactapi/pi/map_tagsSource: lines 1623–1628, handler at lines 1277–1503
Sends a batch of tags to the ML mapping service and orchestrates post-mapping entity creation for "Verified" tags.
Request body: {industry, equipment, unitsId (or unitId), siteId, orgsId, tags: [...]}
tags array element shape: {dataTagId, description, DefaultUnitsName, DefaultUom, DefaultUnitsNameAbbreviation}
ML service call: POST https://localhost:8200/map/tags with {industry, equipment, tags} — source: line 1314. Uses HTTPS with danger_accept_invalid_certs(true).
Response: NDJSON stream forwarded directly from ML service. For format see ML Tag Mapping section.
Post-mapping orchestration for "Verified" tags (lines 1358–1486):
POST /exactapi/equipment — create equipment entityPOST /exactapi/dashboards — create dashboardPOST /exactapi/boiler-assets — create boiler asset linking equipment and dashboardPOST /exactapi/tagmeta/bulk — bulk insert tagmeta records with ML-mapped fieldsAll calls use https://localhost:3030/exactapi with the auth token from the incoming request header.
webpi_live_data_connector.rsNot REST-accessible. PiDriver runs as a managed Rust task launched by the process manager after a successful onboard_unit call. It is not directly callable via HTTP.
Source: clarity:backend/src-tauri/src/connectors/webpi/webpi_live_data_connector.rs
AUTH_REFRESH_SECS = 55 * 60 (3300 s) — refreshes the local Clarity API auth token (not the PI token). PI credentials are sent with every request via HTTP Basic Auth.
Source: lines 15, 114–125
{local_api_url}/loginEach iteration (source: lines 279–330):
{connectionId}_tagmap.ndjson — maps AttributeWebId → DataTagId (falls back to TagName)GET /connections, resolve collectionId to {org}/{site}/{unit}/{grid} pathRECONNECT_INTERVAL_SECS = 60 secondsPI endpoint called: GET {pi_web_api_url}/streamsets/value?webId=...&webId=...
Adaptive batching (source: lines 338–342, 464–485):
Value quality filtering: skips tags where Value.Good == false.
Value extraction: handles numeric Value.Value, numeric top-level Value, or {Value: N} nested object.
Collected (tag_name → [f64]) map posted to local {local_api_url}/write as:
{"organization":"...","site":"...","unit":"...","grid":"...","data":[{"timestamps":[N],"tag_values":{...}}]}
Source: lines 511–552
webpi_historic_data_connector.rsNot REST-accessible. HistoricBackfiller runs as a managed Rust task (pi-backfill-{timestamp}) spawned by start_backfill.
Source: clarity:backend/src-tauri/src/connectors/webpi/webpi_historic_data_connector.rs
Source: lines 13–25
| Constant | Value | Meaning |
|---|---|---|
MAX_EVENTS |
1000 | Events per tag per window before splitting window |
MIN_WINDOW_MS |
5 000 ms | Minimum window size; won't split further |
AUTH_REFRESH_SECS |
3300 s | Local API token refresh interval |
INITIAL_BACKOFF_SECS |
2 s | First retry delay |
MAX_BACKOFF_SECS |
60 s | Retry backoff ceiling |
MAX_RETRIES |
5 | Retries before skipping a window segment |
DENSITY_HIGH |
3600 events/hr | >1 event/sec — high-frequency tags |
DENSITY_MEDIUM |
60 events/hr | >1 event/min — medium-frequency tags |
HistoricBackfillConfig fieldsSource: lines 41–60
pub struct HistoricBackfillConfig {
pub connection_id: String,
pub start_time: String, // ISO8601 or "AUTO" (→ Unix epoch)
pub end_time: String, // ISO8601 or "AUTO" (→ now)
pub pi_web_api_url: String,
pub username: String,
pub password: String,
pub resolution_ms: u64,
pub tag_filter: Option<Vec<String>>, // WebIds; required
pub concurrency: usize,
pub pacing_ms: u64,
pub local_api_url: String,
pub local_user: String,
pub local_pass: String,
pub tagmap_path: String,
pub direction: Option<String>, // "Forward" or "Backward" (default: "Backward")
}
Default direction is "Backward" (newest → oldest). Source: line 211.
Before backfill begins, the connector samples 1 hour of data (50 tags per chunk) starting from start_time to measure events per tag. Tags are grouped by density:
Source: lines 638–710
| Density | Threshold | Window size |
|---|---|---|
| High | ≥3600 events/hr | 10 minutes |
| Medium | ≥60 events/hr | 4 hours |
| Low | <60 events/hr | 24 hours |
Each group runs a separate concurrent fetch loop. Window size adapts automatically — if a tag returns ≥1000 events (MAX_EVENTS) in one window, the window is bisected recursively until either the data fits or the window drops below MIN_WINDOW_MS = 5000ms.
GET {pi_web_api_url}/streamsets/recorded?webId=...&startTime=...&endTime=...
Per-request timeout: 30 seconds.
Source: lines 713–757
Exponential: starts at 2s, doubles each retry, caps at 60s, max 5 retries. Applied on HTTP 5xx, 429, or timeout. HTTP 4xx is treated as fatal (no retry).
Source: lines 564–628
Backward (default): iterates from end_time toward start_time. Progress = oldest timestamp synced.
Forward: iterates from start_time toward end_time. Progress = newest timestamp synced.
State persisted in ProcessManager SQLite via pm.upsert_backfill_state / pm.get_backfill_state. On restart, each tag group resumes from its checkpoint — backward resumes from the maximum (most-recent) checkpoint across the group's tags; forward from the minimum (least-recent). This avoids gaps.
Source: lines 215–266, 335–376
Termination condition: 100 consecutive empty windows with no data. Source: line 456.
Raw PI events are interpolated to a resolution_ms grid using linear interpolation (interpolate(), lines 910–926). Grid-aligned data posted to {local_api_url}/write. Progress checkpointed after each successful write.
Source: lines 802–907
Called via POST /exactapi/pi/map_tags (Clarity endpoint) or directly at https://localhost:8200/map/tags.
Source for ML URL: clarity:backend/src-tauri/src/connectors/webpi/webpi_meta_connector.rs:1314
Response stream (NDJSON) from ML service — format documented in clarity:docs/_archive/developer/PI_ONBOARDING_API.md:
Per-tag row fields:
| Field | Description |
|---|---|
dataTagId |
Original tag ID |
measureUnit |
Matched unit (e.g. "Degree Celsius") |
system |
System classification (e.g. "Boiler") |
equipment |
Equipment type |
component |
Component (e.g. "Fan", "Motor") |
measureProperty |
What is measured (e.g. "Metal", "Power") |
measureType |
Type (e.g. "Temperature", "Current") |
Rerank_Score |
Cross-encoder confidence score |
Mapping_Status |
"Verified" (score ≥0.5), "Low Confidence" (score <0.5), "Skipped - empty query", "No Match Found" |
Qdrant equipment collections (6 types): Id Fan, Fd Fan, Pa Fan, Air Preheater, Economizer, Furnace — case-insensitive matching.
TODO-VERIFY: The ML service at port 8200 is described in old docs as a separate Python service. Verify whether it appears in
python_services_config.jsonor if it is a standalone service not managed by clarity's python_services system. Cannot be determined from source alone — requires inspection of the runtime deployment config file (python_services_config.json) which is not committed to the repository.
POST /pi/all_hierarchy_stream (large hierarchies) or POST /pi/dump_and_stream (with file dump for UI)POST /pi/list_databases, POST /pi/list_childrenPOST /pi/list_attributes on a candidate elementPOST /pi/onboard_unit; consume NDJSON stream; live driver starts automatically on completionPOST /pi/map_tags with equipment type; orchestration creates equipment/dashboard/tagmeta recordsPOST /pi/start_backfill with time range and resolutionAll endpoints require PI Web API credentials in the body:
| Parameter | Type | Notes |
|---|---|---|
api_url |
string | PI Web API URL, e.g. https://pi-server/piwebapi |
username |
string | Domain\username |
password |
string | Optional (Kerberos not tested) |
verify_ssl |
bool | Default true in source; old docs say default false — verify per deployment |
Source for verify_ssl default: clarity:backend/src-tauri/src/connectors/webpi/webpi_meta_connector.rs:881,894,908,1513 — all handler bodies use .unwrap_or(true).
NOTE: Old
OSI_PI_CONNECTORS.mdstatesverify_ssldefaults tofalse. Source code usesunwrap_or(true)in all handler bodies — old doc is wrong.
Last updated: 2026-07-11 from clarity@c41a03b
Range9847dba..c41a03b: the live-data connector changed logging only (println!→log::info!/warn!/trace!; polling, adaptive batching, and 429 handling identical). The metadata connector's onboarding/backfill/dump handlers now resolve the data dir viaconfig::resolved_data_dir_override()beforeapp_data_dir()(honoringclarity.data.dir/CLARITY_DATA_DIR), and its three ad-hoc TLS-permissive HTTP clients were replaced by the shared pooledprocessing_api::internal_http_client().