Source module: clarity:backend/src-tauri/src/processing_api/
Three Rust modules handle specialized data operations: elog (electronic logbook reports), EMS (energy management), and ingest (sensor data ingestion). Route registration in main.rs is selective — see per-section notes.
Sources:
clarity:backend/src-tauri/src/processing_api/elog.rsclarity:backend/src-tauri/src/processing_api/ems.rsclarity:backend/src-tauri/src/processing_api/ingest.rselog.rsRegistered at: clarity:backend/src-tauri/src/main.rs:3137
.or(crate::processing_api::elog::elog_routes(db.clone()))
The elog subsystem generates formatted reports from electronic shift logbooks. All handlers fetch elog configuration and time-series data by calling the local SQLite API at https://localhost:3030/exactapi via reqwest. Time-series data is fetched via POST /exactapi/fast_query. Excel files are generated with rust_xlsxwriter (v0.71.0, confirmed in Cargo.toml:68).
clarity:backend/src-tauri/src/processing_api/elog.rs:10
Authentication: None — elog routes have no JWT filter. Confirmed: clarity:backend/src-tauri/src/main.rs:3137 adds elog routes without an auth wrapper, and the handler functions contain no with_auth filter. See wiki/dev/architecture/api-server.md for the complete route listing.
| Method | Path | Description |
|---|---|---|
GET or POST |
/elog/group/data/health |
Health check — returns string "healthy" |
POST |
/elog/group/data |
Generate elog parameter report as Excel |
POST |
/elog/group/data/deviation-report |
Generate off-limit parameter report as Excel |
POST |
/elog/group/data/downloadcsv |
Export raw tag data as Excel (name is misleading — returns .xlsx, not CSV) |
POST |
/elog/summary/download |
Export logbook submission summary as Excel |
clarity:backend/src-tauri/src/processing_api/elog.rs:273-305
POST /elog/group/dataGenerates a full elog parameter report as a downloadable Excel workbook.
clarity:backend/src-tauri/src/processing_api/elog.rs:570-682
Request body: ElogDataRequest (camelCase JSON)
{
"elogId": "string",
"startTime": 1700000000000,
"endTime": 1700086400000,
"customerId": "string",
"agg": [{ "name": "avg", "samplingValue": 1, "samplingUnit": "hours" }]
}
agg is optional; defaults to first at the elog's own dataInterval (hours).
samplingUnit accepts: milliseconds, seconds, minutes, hours, days.
Response: application/vnd.openxmlformats-officedocument.spreadsheetml.sheet
{elog_name}_report.xlsxInternal flow:
GET /exactapi/activities?filter={"where":{"type":"elog","id":"{elogId}"}} elog.rs:82-93activities to split time range into per-shift chunks (create_epoch_chunks). Offset applied: +5:30 IST. elog.rs:96-166/exactapi/units/{id}, /exactapi/sites/{id}, /exactapi/orgs/{id}. elog.rs:168-186/exactapi/fast_query with pipeline ops per tag. elog.rs:227-241elog.rs:651-672elog.rs:311-428POST /elog/group/data/deviation-reportSame request shape as /elog/group/data. Returns only the columns where at least one value breaches the configured limit.lower or limit.upper threshold.
clarity:backend/src-tauri/src/processing_api/elog.rs:684-854
Response: Excel .xlsx file, filename {elog_name}_deviation_report.xlsx
{"message": "No columns with off-limit values found."}{"error": "No data found for the given parameters"}{"error": "elog configuration not found"}POST /elog/group/data/downloadcsvNOTE: Despite the path name, this endpoint returns an Excel
.xlsxfile, not CSV.
Exports raw tag data for an arbitrary tag list.
clarity:backend/src-tauri/src/processing_api/elog.rs:856-920
Request body: DownloadCsvRequest (camelCase JSON)
{
"tagList": ["TAG_A", "TAG_B"],
"startTime": 1700000000000,
"endTime": 1700086400000,
"agg": [{ "name": "gaps", "samplingValue": 1, "samplingUnit": "minutes" }],
"unitId": "optional-unit-id"
}
agg accepts string or array. Default: gaps at 1-minute sampling.
Response: Excel .xlsx file, filename raw_data.xlsx. Timestamps formatted as dd/mmm/yyyy hh:mm AM/PM in IST (+5:30).
POST /elog/summary/downloadExports the shift logbook submission history for a unit as an Excel summary. Groups rows by logbook name (one worksheet per logbook), shows latest submission per day+shift.
clarity:backend/src-tauri/src/processing_api/elog.rs:922-954
Request body: SummaryDownloadRequest (camelCase JSON)
{
"unitsId": "string",
"startTime": 1700000000000,
"endTime": 1700086400000
}
Queries useractivities entity with filter activity: {inq: ["Workflow Action", "Approved", "Approval", "Rejected"]} and source: {inq: ["elog"]}.
Response: Excel .xlsx, filename {unit_name}_logbookSummary.xlsx
ems.rsNOTE:
ems_routes()is not registered inmain.rs. The function is defined atems.rs:1455but never called. The/dataflow/ems/*paths are served by Python services viapython_proxy_routes()(the catch-all atmain.rs:3144). The Rust implementation inems.rsis currently dead code.
clarity:backend/src-tauri/src/main.rs:3132-3144
The 6 endpoint handlers in ems.rs document the intended Rust EMS API. Until ems_routes() is registered, these paths are handled by a Python service whose port is resolved at runtime from python_services_config.json via register_routes().
The EMS value path was widened
f32→f64inc9fceb2(calculate_period_differences(series: &[(u64, f64)])and all KPI / load-share / consumption-trend handlers), matching the f64 wire formats.
clarity:backend/src-tauri/src/api/python_proxy.rs:136-142
ems.rs — currently served by Python proxy)All endpoints accept both GET (query params) and POST (JSON body) via the ems_payload filter, except tags/summary/download which is POST-only.
clarity:backend/src-tauri/src/processing_api/ems.rs:1429-1453
| Method | Path | Request type | Description |
|---|---|---|---|
GET/POST |
/dataflow/ems/kpiparams |
KpiParamsRequest |
EMS KPI parameters: kWh, MD, PF, THD I/V, SEC, Carbon Footprint, Energy Losses, Load Factor, Load |
GET/POST |
/dataflow/ems/load-share-donut/linewise |
LoadShareRequest |
Load share breakdown by source/utilization system |
GET/POST |
/dataflow/ems/line-consumption-trend |
ConsumptionTrendRequest |
Line-level consumption trend; supports line and bar modes, DTD/WTD/MTD/YTD deltas |
GET/POST |
/dataflow/ems/tags/summary |
TagsSummaryRequest |
Statistical summary per tag: Live Value, Mean, Peak, SD |
POST |
/dataflow/ems/tags/summary/download |
TagsSummaryRequest |
Same summary exported as Excel .xlsx |
GET/POST |
/dataflow/ems/meters/group-config |
MetersGroupConfigRequest |
Meter group hierarchy from DCS graphics activity config |
clarity:backend/src-tauri/src/processing_api/ems.rs:1455-1489
KpiParamsRequest (ems.rs:121-134):
{
"startTime": 1700000000000,
"endTime": 1700086400000,
"unitsId": 42,
"tagType": ["kWh", "MD", "PF"],
"agg": [{"name": "avg"}]
}
LoadShareRequest (ems.rs:231-242):
{
"startTime": 1700000000000,
"endTime": 1700086400000,
"unitsId": 42,
"system": "Source"
}
system values: "Utilization", "Source", "Misc", "Misc Utilization", "Sys Utilization".
ConsumptionTrendRequest (ems.rs:244-263):
{
"startTime": 1700000000000,
"endTime": 1700086400000,
"unitsId": 42,
"mode": "line",
"agg": [{"name": "avg"}],
"tagType": ["kWh"],
"system": "Source",
"taglist": []
}
TagsSummaryRequest (ems.rs:265-276): startTime, endTime, unitsId, taglist: string[].
MetersGroupConfigRequest (ems.rs:278-285): unitsId, equipmentlist: string[].
ems.rs)kpiparams, load-share-donut, line-consumption-trend, tags/summary:
/exactapi/units/{id}, /exactapi/sites/{id}, /exactapi/orgs/{id}. ems.rs:289-307/exactapi/units/{unitsId}/tagmeta?filter=... selecting dataTagId, measureUnit, system, equipment, etc. ems.rs:14-36/exactapi/units/{unitsId}/equipment. ems.rs:38-60/exactapi/fast_query with pipeline ops per tag. ems.rs:309-387ems.rs:389-465meters/group-config:
dcsgraphics activity from /exactapi/activities. ems.rs:1295-1302nodes and containment.contains. ems.rs:1313-1344equipmentlist, return tree structure. ems.rs:1349-1423ems_payload<T>() accepts both GET (query string) and POST (JSON body) for the same endpoint. unitsId and list fields accept number, string, or JSON-encoded string (for GET compatibility).
clarity:backend/src-tauri/src/processing_api/ems.rs:1429-1453
ingest.rsRegistered via ingest_routes() (clarity:backend/src-tauri/src/processing_api/ingest.rs:1608), composed into the main route chain in main.rs.
The ingest API receives sensor data from external IoT clients, parses and normalizes it, writes to the historian first (in-process, selectable backend — see below), then hands the batch to a decoupled, lossy MQTT fan-out task for realtime display. Restructured in c41a03b (PRs #283–#288, "ingest-mqtt"): the historian is the lossless source of truth and can never be delayed by MQTT publishing.
A module-level shared pooled HTTP client, processing_api::internal_http_client() (processing_api/mod.rs), replaces the per-call reqwest::Client::builder() instances across ingest.rs, elog.rs (3 sites), ems.rs (6 sites), and the PI metadata connector.
| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/ingest/health |
None | Health check; also reports MQTT connection status |
POST |
/ingest/v2/{clientId}/{configId} |
Required | Ingest live sensor data — fast path (see below), ordered before v1 |
POST |
/ingest/{clientId}/{configId} |
Required | Ingest live sensor data (v1) |
POST |
/ingest/backfill/{clientId}/{configId} |
Required | Ingest historical (backfill) data |
clarity:backend/src-tauri/src/processing_api/ingest.rs:1191-1218; v2 route registered at clarity:backend/src-tauri/src/processing_api/ingest.rs:1656-1662 (ordered before v1 at :1680).
POST /ingest/v2/{clientId}/{configId} — fast path (ingest_v2.rs)A drop-in fast alternative to v1 with an identical wire format (same flat-JSON body {"tag": value, …, "timestamp": ms}, same auth, rate limiting, and response codes). clarity:backend/src-tauri/src/processing_api/ingest_v2.rs:1-29
What differs is the parse/write path:
Value tree (per-point String allocations, double regrouping); v2 does a streaming borrowed-key parse with no Value tree, via custom deserializers (NumSeed/KeySeed/SnapshotSeed). clarity:backend/src-tauri/src/processing_api/ingest_v2.rs:99-323(clientId, configId) cached IngestSession maps each key to a (scope, ring column) binding, revalidated per request via one atomic generation load (Storage::ingest_generation). clarity:backend/src-tauri/src/processing_api/ingest_v2.rs:57-90storage.ingest_indexed(...), "slow" groups use storage.ingest_grouped_writes(...), run in parallel on spawn_blocking. clarity:backend/src-tauri/src/processing_api/ingest_v2.rs:642-691handle_ingest_v1). MQTT fan-out is preserved and flag-gated (the body is only re-parsed when publishing is on). clarity:backend/src-tauri/src/processing_api/ingest_v2.rs:18-24, 514-553, 704-745Handler: handle_ingest_v2(param1, param2, raw_body: Bytes) at clarity:backend/src-tauri/src/processing_api/ingest_v2.rs:514-517. The v1 batch grouping and fast write were also widened f32 → f64 (group_batch_by_scope now yields Vec<(u64, Option<f64>)>). clarity:backend/src-tauri/src/processing_api/ingest.rs:1226-1250
/ingest/health and /ingest/{clientId}/{configId} are explicitly excluded from the Python proxy catch-all. clarity:backend/src-tauri/src/api/python_proxy.rs:163-165
GET /ingest/healthReturns:
{ "message": "Healthy", "mqtt_connected": true }
clarity:backend/src-tauri/src/processing_api/ingest.rs:1061-1067
POST /ingest/{clientId}/{configId}Auth: Bearer token required (via with_auth).
Payload limit: config-driven ingest_max_payload_bytes (default 10 MB); as of c41a03b the handler takes raw_body: bytes::Bytes and checks size on the raw bytes before parsing — oversize returns 413 PAYLOAD_TOO_LARGE. ingest.rs:1474-1499
Rate limit: 1,000 requests/minute per clientId (properties keys renamed to clarity.rate_limit.ingest.max_requests / .window_seconds).
Rate limit response: HTTP 429 with {"error": "Rate limit exceeded. Retry in N seconds."}
Request body: JSON object with tag names as keys and numeric values, plus an optional timestamp (epoch milliseconds). If timestamp is absent, server time is used.
Payload parsers applied in order (ingest.rs:1101-1130):
imei and dtm keys present, flattens gateway/meter group structure. ingest.rs:1107-1109BYTE are split into 8 individual bit fields. ingest.rs:1112DeviceID key present, strips DeviceID/DT and converts DT to epoch ms. ingest.rs:1115-1117Processing flow (restructured in c41a03b):
clientId and configId as safe identifiers (alphanumeric + _. -&#()/, 1–50 chars; or 24-char hex MongoDB ObjectId).clientId from INGEST_CONFIG_CACHE — refresh now happens on a background loop (ensure_cache_refresh_task, tick = ingest_config_cache_ttl_seconds, min 30 s, default 120 s), never on the request path; cold-start bootstrap gated on a CACHE_BOOTSTRAPPED AtomicBool. ingest.rs:778TAG_SCOPE_MAP under one read lock (producing Arc<Scope>); misses trigger maybe_refresh_tag_scope_map_on_miss (serialized by a REFRESH_IN_FLIGHT mutex + 200 ms debounce) and one retry before falling back to the config-ID-derived scope — see Scope resolution. ingest.rs:555,1309historian_write (ingest.rs:1287), backend selected by clarity.ingest.write.method.mqtt_fanout::enqueue, non-blocking, gated on clarity.ingest.mqtt.publish). ingest.rs:1451Responses: success = HTTP 200 {"status": "received"}; a historian rejection (write-buffer full / storage error) = 503 SERVICE_UNAVAILABLE (real backpressure — new in c41a03b); oversize = 413. ingest.rs:1537-1543, 1570-1605
POST /ingest/backfill/{clientId}/{configId}Same request format and processing as the live endpoint; same auth, rate limit, and payload parsers apply. Used for historical data imports. ingest.rs:1138-1188
Reworked in 9847dba to stop silently dropping data when TAG_SCOPE_MAP is stale or a tag maps to an empty unit, then restructured in c41a03b:
Historian write dispatch (new in c41a03b). historian_write (ingest.rs:1287) selects by clarity.ingest.write.method (buffered | fast | http, code default buffered, but the shipped clarity.properties pins fast as of 45a686e — so a default on-prem install takes the fast/mmap path; validated in config.rs:694-705):
"buffered" → WriteBuffer::enqueue_batch in-process (historian_write_buffered, ingest.rs:1215); a full buffer surfaces Err → the endpoint's 503."fast" → synchronous storage.store_data_mmap in one spawn_blocking per scope group (historian_write_fast, ingest.rs:1238)."http" → the legacy HTTPS loopback to /exactapi/write_buffered (kept as a rollback lever).init_ingest_backends (ingest.rs:1172, called from main.rs:3290). The old adaptive-batch HTTPS loopback (store_datapoints_write_fast, 10k–100k points) survives only as the "http" mode. As of 45a686e, when clarity.ha.shadow_targets is configured, store_datapoints_write_fast also fire-and-forgets a best-effort (1 s-timeout) mirror of each batch to POST {target}/exactapi/shadow_write on each peer, feeding the HA secondary shadow cache for zero-loss failover. clarity:backend/src-tauri/src/processing_api/ingest.rs:32-43, 1124-1148.In-process TAG_SCOPE_MAP refresh (replaces the /collections HTTP fetch). fetch_collections_and_build_scope_cache — and its HTTPS GET /collections — is removed; refresh_tag_scope_map (ingest.rs:475) walks local Storage via tag_resolver::handle_list_collections instead. The "never overwrite a populated map with empty data" guard is retained. Miss-triggered refresh is serialized through a REFRESH_IN_FLIGHT mutex with a 200 ms debounce (MISS_REFRESH_DEBOUNCE_MS, ingest.rs:555), replacing the old 5 s cooldown that suppressed legitimate new-collection misses.
Reverse ID→name caches. The ingest config now carries org_id / site_id / unit_id (read from orgsId / siteId / unitsId in the config metadata, tolerating both number and string). To turn those numeric IDs into the names the storage/MQTT paths expect, three reverse caches are built alongside the existing UNIT_ID_CACHE: ORG_NAME_BY_ID, SITE_NAME_BY_ID, and UNIT_NAME_BY_ID (populated by fetch_orgs_and_build_name_cache / fetch_sites_and_build_name_cache from GET /orgs and GET /sites, and the reverse half of fetch_units_and_build_unit_id_cache). clarity:backend/src-tauri/src/processing_api/ingest.rs:135-137, 264-292, 576-895
/collection → /collections endpoint fix. fetch_collections_and_build_scope_cache fetched {base}/collection (singular), which does not exist — create_collections_routes mounts at warp::path("collections"). The bad URL 404'd silently, leaving TAG_SCOPE_MAP empty, so every ingest payload skipped its MQTT publishes (skipped_empty_unit = N). The URL is now {base}/collections. clarity:backend/src-tauri/src/processing_api/ingest.rs:456-462
Never overwrite a populated map with empty data. The TAG_SCOPE_MAP rebuild inside fetch_collections_and_build_scope_cache now skips the clear-and-replace when the freshly-fetched scope set is empty but the map already has entries (e.g. /collections returned no tags because the walker hasn't indexed yet, or the server is still starting) — it logs a warning and preserves the boot-time population done by build_tag_map. clarity:backend/src-tauri/src/processing_api/ingest.rs:496-524
Per-tag scope with config fallback. In process_and_publish, each tag's scope comes from TAG_SCOPE_MAP unless the entry is missing or resolves to an empty unit; misses are set aside, the map refresh runs (debounced, see above), and a second lookup retries before falling back to a scope built from the ingest config's own org_id/site_id/unit_id (resolved to names via the reverse caches; grid = "default_grid"; a cache miss uses the raw ID as a last resort). The batch tuple type is now (String, f64, u64, Arc<Scope>). clarity:backend/src-tauri/src/processing_api/ingest.rs:1309
Tag-prefix double-apply guard. When the config has a tag_prefix, a payload key is prefixed only if it does not already start with the prefix (tag_prefix.is_empty() || key.starts_with(&tag_prefix)), preventing prefix_prefix_tag. clarity:backend/src-tauri/src/processing_api/ingest.rs:1137-1143, 1199-1204
mqtt_fanout.rsNew module in c41a03b (clarity:backend/src-tauri/src/processing_api/mqtt_fanout.rs, 179 lines). All in-handler MQTT machinery (per-tag topic building, 500-per-batch chunking, publish_wait_drain, the 100-error abort, the 30 s publish watchdog) is removed from ingest.rs — the handler just calls mqtt_fanout::enqueue(FanoutBatch{client_id, config_id, points}) (mqtt_fanout.rs:25,64; sole caller ingest.rs:1451).
Design: a decoupled, lossy, freshest-wins realtime display plane — the historian is the lossless source of truth (mqtt_fanout.rs:1-13).
mpsc::channel with QUEUE_CAPACITY = 8 batches, task lazily spawned on first enqueue; enqueue uses try_send and never blocks — on Full the whole batch is dropped and DROPPED_BATCHES/DROPPED_POINTS counters increment. mqtt_fanout.rs:34,64-81mqtt_fanout.rs:95-116try_publish_nonblocking): primary u/{uid}/{tag}/r|sd|v/e (only when the scope's unit resolves to a UID in UNIT_ID_CACHE; rate-limited cache refresh, ≥30 s apart, on misses) and secondary {clientId}/{ingestConfigId}/{tag} — always, and note the last segment is the full resolved tag name, not dataTagId. Payload: one-element array [{"t": <ms>, "v": <f64>}]. mqtt_fanout.rs:128-160mqtt_fanout.rs:162-176clarity.ingest.mqtt.publish (code default true; the shipped clarity.properties sets it OFF — the file wins when present).ingest.rs generates its own short-lived JWT to call internal APIs (/exactapi/ingestconfigs, /exactapi/collections, /exactapi/orgs, /exactapi/sites, /exactapi/units, /exactapi/write_fast). Token TTL: 3600 s, refreshed 300 s before expiry. Role: admin, sub: internal_ingest.
clarity:backend/src-tauri/src/processing_api/ingest.rs:27-98
The ingest-config cache is populated from GET /exactapi/ingestconfigs, served by the SQLite dynamic layer — which as of f3f16e6 stringifies top-level ID fields (id, clientsId, …) in responses. fetch_ingest_configs_from_api therefore reads IDs through a parse_id helper that accepts both Value::Number (legacy / internal callers) and Value::String, returning 0 for missing or unparseable values (preserving the prior unwrap_or(0) semantics). clientsId is read via parse_id; when it is 0/absent the reader falls back to a string clientId field. clarity:backend/src-tauri/src/processing_api/ingest.rs:324-371
The diagram below covers POST /ingest/{clientId}/{configId} — the live ingest path. Backfill uses the same path without the rate limit distinction.
Sources: clarity:backend/src-tauri/src/processing_api/ingest.rs, clarity:backend/src-tauri/src/api/tag_resolver.rs
Diagram generated from source. If implementation has changed, update the Mermaid source in this file directly.
Last updated: 2026-07-17 from commit 6800acc