PSI Knowledge Base

psi-data-pipeline

8 min readEdit this page

PSI Data Pipeline

System of scheduled jobs that move PSI manufacturing data from source systems (AFTEC, network shares, vendor APIs) into Azure SQL (PSI_Analytics), with SQLite cache layers and a live operational dashboard. Powers PSI Explorer analytics views, Power BI, and downstream consumers.

Overview

The PSI Data Pipeline is a Python-based ETL system running on PS-PROXY. Eight independent scheduled jobs read from AFTEC CSV exports, the UniData API, vendor APIs (Rockwell, GE, Emerson, Dell), and network-share spreadsheets, then land data in three storage tiers ending at Azure SQL.

PropertyValue
RepositoryProgressiveSurface/psi-data-pipeline
RunnerPS-PROXY (self-hosted GHE Actions, [self-hosted, ps-proxy])
OrchestrationPrefect server + work pool (psi-pipeline) running locally on PS-PROXY
Target databaseAzure SQL PSI_Analytics on procserv-proddata.database.windows.net
Live dashboardexplorer.progressivesurface.com/#pipeline
StackPython 3.12, SQLite, pyodbc, Prefect, MSAL service-principal auth

Architecture

Three storage tiers; data flows ingest → analytics → Azure SQL:

TierPathRoleLifecycle
psi_ingest.db\\ad.ptihome.com\DFS\Schedule\SS123\LEADTIME\psi_ingest.dbDurable cache, scraper-ownedPersists across runs; preserves enrichment via INSERT OR REPLACE
psi_analytics.db\\ad.ptihome.com\DFS\Schedule\SS123\LEADTIME\psi_analytics.dbRead-optimized analytics SQLiteRebuilt nightly from AFTEC CSVs + imports from psi_ingest.db. Atomic deploy via .tmp rename.
PSI_AnalyticsAzure SQL on procserv-proddata.database.windows.netProduction source of truthSynced incrementally from analytics tier; consumed by PSI Explorer, Power BI

Architecture doc in repo: docs/pipeline-architecture.md — authoritative for flow patterns, env vars, “how to add a new job”, and troubleshooting.

Data flow patterns

The pipeline has three distinct patterns:

Standard scraper:
    Source → load_*.py → psi_ingest.db → publish_ingest_tables.py
                       → psi_analytics.db → sync_table_to_azure.py
                       → Azure SQL → verify_passthrough_tables.py

Analytics build:
    AFTEC CSVs + psi_ingest.db imports → build_analytics_db.py
                       → psi_analytics.db
                       → (internal incremental Azure SQL sync, manifest-based)

Raw AFTEC:
    AFTEC CSV files → sync_aftec_raw.py → Azure SQL (no SQLite)

Orchestration

One scheduler: GHE Actions. Prefect is telemetry. (Decided 2026-07-20 — the earlier “migrate scheduling to Prefect” effort is retired; it kept cron in two racing places. Rationale: docs/pipeline-remediation-plan.md in the repo.)

LayerRole
GitHub ActionsThe scheduler and pass/fail source of truth. Each workflow’s cron: block is the only schedule for its job; workflow_dispatch: stays for manual triggers.
Prefect (psi-pipeline work pool)Telemetry only. @flow decorators report every run to the local Prefect server regardless of launcher; prefect_deployments.py registers deployments without schedules so the dashboard lists the jobs. Never add a cron there.
PSI Explorer dashboard (#pipeline)Operator-facing UI. Reads the Prefect API and renders jobs, recent runs, row counts.

Passthrough-table verification runs inside the nightly build flow (its Phase 6), and an Azure sync failure fails the run — so dashboard state = actual state.

Active pipeline jobs

WorkflowPrefect deploymentSchedule (UTC)Schedule (ET)DurationTables owned
sync-aftec-raw.ymlsync-aftec-raw0 6 * * *Daily 1 AM~2 hours31 raw AFTEC tables (tslabor2, openwo, partmaster, ecn, redbook, etc.)
nightly-data-build.yml (Build Analytics DB)build-analytics-db30 6 * * *Daily 1:30 AM~30 minprojects, labor_detail, redbooks, machine_dna, workforce_features, comprehensive_view, + all imported ingest tables (incl. Azure copies of the passthrough tables — single-writer rule)
sync-kg-data.yml (Sync KG Data)(none — see Gaps)0 7 * * *Daily 2 AM~1 minkg_* knowledge-graph tables (runs pipeline.cpq.ingest_to_graph)
sync-bom-data.ymlsync-bom-data0 8 * * 1-5Weekdays 3 AM~1hbom_parts (subset for active projects, via UniData API batch endpoint)
sync-prims.ymlsync-prims0 5 * * 1-5Weekdays midnight~2 minprims_installations
sync-dept-schedules.ymlsync-dept-schedules0 14 * * 1-5Weekdays 9 AM~2 mindept_schedules, floor_space
sync-rockwell.ymlrefresh-component-lifecycle0 10 * * 1Mondays 5 AM~5 mincomponent_lifecycle, project_components (Allen-Bradley + GE/Emerson) — writes psi_ingest.db
sync-product-taxonomy.ymlsync-product-taxonomy0 7 * * 1Mondays 2 AM~2 minproduct_classes, product_catalog — writes psi_ingest.db only (Azure via nightly build since 2026-07-20)
enrich-dell-warranty.ymlenrich-dell-warranty0 12 * * 6Saturday 7 AM~1 min(updates prims_installations dell_* columns via PSI TechDirect Warranty API key — GHE secrets DELL_API_CLIENT_ID/SECRET, key expires 2027-07-22; Dell Command CLI on PS-PROXY is the fallback, its Dell-side backend broken since 2026-04)
setup-env-vars.yml—manual——One-time: writes Azure secrets as machine env vars

Monitoring gaps

  • sync-kg-data (pipeline.cpq.ingest_to_graph) has no @flow — its logic runs directly under __main__ — so it doesn’t appear on the Explorer dashboard. GHE Actions is its only red/green. Tracked: repo issue #18.
  • (Fixed 2026-07-20: nightly-data-build and sync-rockwell are now Prefect flows, and verification runs inside the nightly build flow.)

Data sources

SourceRecordsWhat
AFTEC CSV exports9.25M+ rows / 31 tablesDaily exports from UniData/AFTEC (labor, work orders, BOM, parts, PO, redbook, ECN, etc.)
UniData API batch endpoint5.76M BOM rowsLive BOM exports for active projects
Redbook SQLite (redbook_coq.db)40K RFCsAI-classified quality issues
PRIMS Installations XLS650 PCsPSI machine PC fleet inventory
Department schedule .xlsm files7 filesWeekly capacity by department + floor space
Rockwell / GE / Emerson APIs488 componentsAllen-Bradley + GE + Emerson product lifecycle status
Dell CLIper-PC warrantyService tag → warranty end, model, ship date
Product taxonomy CSVs437 classes + 20,654 productsProduct class hierarchy and catalog
LDS Excel files~3,000 projectsPer-project Gantt schedules (extracted by extract_lds_gantt.py)
OneNote Lessons Learned872 projectsPost-project lessons (extracted from .one binary)
Handoff .docx notes~2,000 projectsRisk notes, comp-machine references (extract_handoff_notes)

Live dashboard

The PSI Explorer hosts a Data Pipeline dashboard at #pipeline that reads the Prefect API and renders:

  • 6 currently-deployed jobs as cards with status badges (Healthy / Partial / Failed)
  • Per-job row counts and table counts
  • Recent runs (50) with state, started, duration
  • Summary counts (completed / failed / running)
  • Manual “Run” button per job
  • Link to the Prefect UI on PS-PROXY

Backend: analytics-routes.js in the psi-explorer-web repo queries the local Prefect API.

Repository layout

psi-data-pipeline/
├── CLAUDE.md                          # Cold-start orientation for AI sessions
├── README.md                          # Quick start + doc index
├── BUILD_LOG.md                       # Dated change log (append on every change)
├── prefect_deployments.py             # Register flows with the psi-pipeline work pool
├── docs/
│   ├── pipeline-architecture.md       # Authoritative architecture, workflow table, "how to add a job"
│   ├── TODO.md                        # Active backlog
│   ├── data-flow-diagram.md           # Visual data flow
│   └── PSI_Engineering_Knowledge_Graph_v2.md
├── pipeline/
│   ├── build_analytics_db.py          # Main analytics builder (AFTEC CSV → psi_analytics.db → Azure SQL)
│   ├── build_comprehensive_dataset.py # AFTEC CSV loaders (imported by build_analytics_db.py)
│   ├── sync_aftec_raw.py              # Raw AFTEC CSV → Azure SQL (no SQLite)
│   ├── sync_table_to_azure.py         # Generic SQLite → Azure SQL sync (manifest-based incremental)
│   ├── sync_bom_data.py               # UniData API batch BOM export → Azure SQL
│   ├── publish_ingest_tables.py       # psi_ingest.db → psi_analytics.db copy
│   ├── verify_passthrough_tables.py   # Row-count check across all three DBs
│   ├── load_prims_standalone.py       # PRIMS PC fleet loader
│   ├── load_dept_schedules.py         # Department schedule + floor space loader
│   ├── load_product_taxonomy.py       # Product classes + catalog loader
│   ├── rockwell_lifecycle.py          # Rockwell + GE + Emerson lifecycle scraper
│   ├── enrich_dell_support.py         # Dell warranty CLI enrichment
│   ├── azure_sql.py                   # Azure SQL connection + bulk insert helper
│   └── cpq/                           # Knowledge graph extraction (CPQ subsystem)
└── .github/workflows/                 # 9 workflow files (see table above)

Environment

PS-PROXY (Windows Server):

  • Python 3.12 via actions/setup-python@v5
  • ODBC Driver 17/18 for SQL Server
  • Network access to \\ad.ptihome.com\DFS\* shares (always use this path, never \\fs1\ or \\PS-GR-FS01\)
  • Drive mappings: S: = \\ad.ptihome.com\DFS\DATA, I: = Quotes
  • Prefect server local at 127.0.0.1:4200 (no Prefect Cloud)
  • Azure credentials from GHE Actions secrets (AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET)

Common operations

Manually trigger a workflow

$env:GH_HOST = 'progressivesurface.ghe.com'
gh workflow run "Sync PRIMS" --repo ProgressiveSurface/psi-data-pipeline

Force a re-sync (bypass manifest-based skip)

The incremental sync compares row_count:col_count:content_hash signatures AND checks Azure’s live row count before skipping (since 2026-07-20 — drift self-heals nightly). To force a full re-sync anyway:

rm "\\ad.ptihome.com\DFS\Schedule\SS123\LEADTIME\.azure_sync_manifest.json"

Add a new pipeline job

See docs/pipeline-architecture.md → “How to Add a New Pipeline Job” for the full walkthrough (5 steps: create loader, register in publish list, register in analytics build, create workflow, test).

After registering, also add to prefect_deployments.py and re-run it to register the deployment with the work pool.

Check pipeline status

Known issues / class of bugs

The Azure SQL passthrough track had recurring data-fidelity bugs because each load_*.py defines its own table DDL and column widths. The 2026-07-20 remediation (repo PRs 17, docs/pipeline-remediation-plan.md) fixed the live instances at root cause:

  • String truncation — source values exceed guessed NVARCHAR(N) widths. Fixed instance: floor_space.project widened to NVARCHAR(100). If it recurs: widen the column or use NVARCHAR(MAX), and add column names to the long_cols set in build_analytics_db.py / sync_table_to_azure.py.
  • Unique-index conflicts — source data has duplicate keys that contradict declared uniqueness. Fixed instance: PRIMS now normalizes and dedupes service tags before creating its unique index.
  • Phantom rows / stale data in Azure SQL — fixed by design: the nightly sync signature now includes a content hash, and skipping additionally requires Azure’s live row count to match the source, so deletes, equal-count edits, and second-writer drift resync automatically.
  • Non-idempotent DDL against pre-existing tables — fixed instance: rockwell_lifecycle.py applies tables → column migration → indexes.

Long-term direction (still open): one shared additive-migration helper so jobs stop hand-rolling DDL — 13 files carry their own today.

Graph View

Backlinks