Primavera P6 Export Parsing

Primavera P6 is where the schedule of record lives, but the schedule leaves P6 as a file, not an API call — either the tab-delimited XER export the scheduler emails around, or a P6 XML export dumped from the client. The precise sub-problem this page solves is how a pipeline turns that P6 export into typed, WBS-bound ScheduleActivity records — each carrying an activity id, a resolved WBS node, timezone-aware start and finish dates, an integer total-float value, and a critical-path flag — so a change-order pipeline can reason about schedule impact deterministically instead of re-keying dates off a PDF Gantt chart. When the schedule stays trapped in a proprietary export, an approved change order never posts its delay against the right activity, a slipped critical path goes unnoticed until the look-ahead meeting, and float that has quietly evaporated shows up first as a liquidated-damages claim. This page details the export-to-record pipeline: how the XER file is structured as delimited tables (TASK, TASKPRED, PROJWBS, CALENDAR), how to read it without silent data loss, how to parse the P6 XML variant with the standard-library XML parser, and how each parsed activity binds to the same Work Breakdown Structure that governs cost. It belongs to schedule sync integration and feeds the downstream comparison stages, and it targets Python automation builders, schedulers, and project controls engineers who need predictable activity data under real-world export variance.

Prerequisites

This subsystem sits downstream of the raw export drop and upstream of the schedule-comparison stages. Before implementing the patterns below, you need:

• Python 3.11+ with pydantic v2 for typed validation, plus the standard-library xml.etree.ElementTree, csv, decimal, datetime, re, enum, and logging modules. Dates become timezone-aware datetime objects and any monetary field is a Decimal; no float ever touches a cost or a variance total.
• A canonical WBS taxonomy to resolve each activity against. A parsed activity carries P6’s own internal WBS id, which is meaningless outside the export; binding it to a project-stable node reuses the same WBS mapping contract that drives cost allocation, so an activity and its budget line agree on where the work sits.
• The confidence bands used everywhere a match or classification happens on this site. When P6’s WBS short name has to be fuzzy-matched to the canonical node, a score of 0.92 or above auto-binds, 0.75–0.92 flags the activity for a scheduler’s review, and below 0.75 quarantines it rather than binding to a guessed node.
• A task queue with a dead-letter path — Celery on Redis or RabbitMQ — so a malformed table row or an unbindable activity is parked and replayed instead of dropped. Escalation for parked records is owned by fallback alert routing.
• A structural gate at the ingestion boundary. The export has already cleared basic schema validation rules — file type, size, non-empty — so the work here is P6-specific decoding, typing, and binding, not raw upload handling.

The pipeline assumes the export is a single project snapshot. Once activities are typed and bound, they feed critical path delta detection and schedule baseline variance, both of which depend on the float and date fields being correctly typed here first.

Architecture: decode, type, and bind

A P6 export is not a document you read top to bottom; it is a set of relational tables serialized into one file. The XER format prepends a version header, then emits each table as a %T table-name line, a %F field-header line, and a run of %R data rows, terminated by %E. The P6 XML format carries the same tables as nested elements under a namespaced root. Whichever variant arrives, the pipeline decodes it into rows, lifts the four tables that matter for scheduling, builds a typed ScheduleActivity per TASK row, and binds each activity to a canonical WBS node — parking anything it cannot decode or bind. Keeping decode, type, and bind as distinct stages is what lets a malformed calendar reference fail one activity without aborting the whole import.

Step-by-step implementation

Step 1 — Define the typed activity contract

The ScheduleActivity is the whole point of the exercise: a strict, versioned record that downstream comparison logic can trust. Identity is the pair (project_id, activity_id) because P6’s user-facing activity code is unique only within a project — the same A1010 appears in every schedule ever exported. Dates are timezone-aware per the ISO 8601 date and time standard; a naive datetime is rejected because float and variance math across project sites in different zones would silently drift. Total float is stored as an integer day count, not hours, so that comparison and reporting speak the project’s language, and is_critical is derived, not trusted from a flag that P6 sets under its own scheduling options.

from __future__ import annotations

import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Literal, Optional

from pydantic import BaseModel, Field, field_validator, model_validator

logger = logging.getLogger("p6.parse")

Discipline = Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]
RelationType = Literal["FS", "SS", "FF", "SF"]
BindState = Literal["auto_bind", "human_review", "quarantine"]

WBS_PATTERN = r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$"  # e.g. PROJ-014-STR-02


class ScheduleActivity(BaseModel):
    """A typed, WBS-bound activity lifted from a P6 export."""

    schema_version: Literal["1.0"] = "1.0"
    project_id: str = Field(min_length=1, max_length=40)
    activity_id: str = Field(min_length=1, max_length=40)   # P6 task_code, e.g. A1010
    activity_name: str = Field(min_length=1, max_length=240)
    wbs_node: str = Field(pattern=WBS_PATTERN)              # canonical, resolved at bind
    discipline: Discipline
    start_date: datetime
    finish_date: datetime
    total_float_days: int                                   # negative on a driven critical path
    is_critical: bool
    bind_confidence: float = Field(ge=0.0, le=1.0)

    @field_validator("start_date", "finish_date")
    @classmethod
    def require_tz_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("schedule dates must be timezone-aware (ISO 8601)")
        return v.astimezone(timezone.utc)

    @model_validator(mode="after")
    def finish_after_start(self) -> "ScheduleActivity":
        if self.finish_date < self.start_date:
            raise ValueError("finish_date precedes start_date")
        # is_critical is derived from float, never copied blindly from P6's flag.
        if self.is_critical != (self.total_float_days <= 0):
            raise ValueError("is_critical must equal total_float_days <= 0")
        return self

    @property
    def uid(self) -> str:
        """Cross-project-safe key: activity codes repeat between schedules."""
        return f"{self.project_id}:{self.activity_id}"

Step 2 — Read the XER as delimited tables

The XER file is a Windows-1252 (cp1252) text file, not UTF-8 — the single most common cause of a failed import is opening it with the default codec and hitting a stray 0x92 smart-quote. Each line begins with a marker: %T names a table, %F lists that table’s field headers, %R is a data row, and %E ends the file. Everything after the marker is tab-delimited. The reader below streams the file once, tracks the current table and its headers, and yields (table_name, row_dict) pairs so a caller can pull only the four tables scheduling cares about without holding the whole export in memory.

from __future__ import annotations

import csv
from collections.abc import Iterator
from pathlib import Path

WANTED_TABLES = frozenset({"PROJWBS", "TASK", "TASKPRED", "CALENDAR"})


def read_xer_tables(path: Path) -> Iterator[tuple[str, dict[str, str]]]:
    """Yield (table_name, row) for the scheduling tables in an XER export.

    XER is CP1252, tab-delimited, with %T/%F/%R/%E record markers. Using the
    csv module (not str.split) keeps embedded tabs and quoting intact.
    """
    current_table: str | None = None
    headers: list[str] = []

    # newline="" lets csv handle embedded CR/LF inside a field correctly.
    with path.open("r", encoding="cp1252", newline="") as fh:
        reader = csv.reader(fh, delimiter="\t")
        for cells in reader:
            if not cells:
                continue
            marker = cells[0]
            if marker == "%T":              # table declaration: %T <TABLE_NAME>
                current_table = cells[1]
                headers = []
            elif marker == "%F":            # field header row for the current table
                headers = cells[1:]
            elif marker == "%R":            # data row
                if current_table in WANTED_TABLES and headers:
                    yield current_table, dict(zip(headers, cells[1:]))
            elif marker == "%E":            # explicit end-of-file marker
                break
            # ERMHDR and any other leading lines are ignored by design.

The csv reader matters here: P6 activity names routinely contain tabs and newlines that a naive line.split("\t") would shred into phantom columns. Reading the header row per table means a field added in a newer P6 version lands in the row dict by name rather than shifting every positional index.

Step 3 — Parse the P6 XML variant with ElementTree

When the export arrives as P6 XML instead of XER, the same tables are nested elements under a namespaced root. The standard-library xml.etree.ElementTree parser handles it without a third-party dependency, but every find/findall must carry the P6 namespace or it silently returns nothing. Register the namespace once and pass it through. The function below yields the same shape of row dict as the XER reader, so the downstream builder is format-agnostic.

from __future__ import annotations

import xml.etree.ElementTree as ET
from collections.abc import Iterator
from pathlib import Path

# P6 XML declares a default namespace on the root; findall must be qualified.
P6_NS = {"p6": "http://xmlns.oracle.com/Primavera/P6/V19.12/API/BusinessObjects"}


def read_p6_xml_activities(path: Path) -> Iterator[dict[str, str]]:
    """Yield one row dict per <Activity> element in a P6 XML export."""
    tree = ET.parse(path)                      # raises ParseError on malformed XML
    root = tree.getroot()

    for activity in root.iterfind(".//p6:Activity", P6_NS):
        row: dict[str, str] = {}
        for child in activity:
            # Strip the namespace URI from the tag: '{...}Id' -> 'Id'.
            tag = child.tag.rsplit("}", 1)[-1]
            row[tag] = (child.text or "").strip()
        yield row


def _first_ns(root: ET.Element) -> dict[str, str]:
    """Recover the actual namespace from a document whose P6 version differs."""
    if root.tag.startswith("{"):
        uri = root.tag[1:].split("}", 1)[0]
        return {"p6": uri}
    return {}

Because the namespace URI carries the P6 version (V19.12 here) and a client on a different release ships a different URI, _first_ns recovers it straight from the root tag rather than hard-coding a version that would make iterfind come back empty on the next export.

Step 4 — Build the activity, bind the WBS, and route

The final stage joins the pieces: it reads the PROJWBS and CALENDAR tables into lookups, converts P6’s total_float_hr_cnt (hours) into whole days using the activity’s calendar, resolves P6’s internal WBS id to a canonical node by fuzzy-matching the WBS short name, and constructs a validated ScheduleActivity. The site-canonical bands govern the bind: a WBS short-name match of 0.92 or above auto-binds, 0.75–0.92 builds the record but flags it for a scheduler, and below 0.75 the activity is quarantined rather than bound to a guessed node.

from __future__ import annotations

from datetime import datetime
from difflib import SequenceMatcher

AUTO_BIND = 0.92        # >= 0.92 -> auto_bind
REVIEW_FLOOR = 0.75     # 0.75-0.92 -> human_review; < 0.75 -> quarantine


def classify_bind(score: float) -> BindState:
    if score >= AUTO_BIND:
        return "auto_bind"
    if score >= REVIEW_FLOOR:
        return "human_review"
    return "quarantine"


def _hours_to_days(total_float_hr: str, day_hr_cnt: float) -> int:
    # P6 stores float in hours against a calendar; report it in whole days.
    hours = float(total_float_hr or 0.0)
    return int(round(hours / day_hr_cnt)) if day_hr_cnt else 0


def _parse_p6_datetime(raw: str) -> datetime:
    # XER date form: 'YYYY-MM-DD HH:MM'; treat P6 clock time as project-local UTC.
    return datetime.strptime(raw, "%Y-%m-%d %H:%M").replace(tzinfo=timezone.utc)


def build_activity(
    task: dict[str, str],
    wbs_short_by_id: dict[str, str],         # wbs_id -> P6 short name
    canonical_wbs: dict[str, str],           # canonical short name -> PROJ-NNN-...
    day_hr_by_calendar: dict[str, float],    # clndr_id -> hours per day
    project_id: str,
    discipline: Discipline,
) -> ScheduleActivity:
    short_name = wbs_short_by_id.get(task["wbs_id"], "")
    node, score = _best_wbs(short_name, canonical_wbs)
    day_hours = day_hr_by_calendar.get(task.get("clndr_id", ""), 8.0)
    float_days = _hours_to_days(task.get("total_float_hr_cnt", "0"), day_hours)

    activity = ScheduleActivity(
        project_id=project_id,
        activity_id=task["task_code"],
        activity_name=task["task_name"],
        wbs_node=node,
        discipline=discipline,
        start_date=_parse_p6_datetime(task["target_start_date"]),
        finish_date=_parse_p6_datetime(task["target_end_date"]),
        total_float_days=float_days,
        is_critical=float_days <= 0,          # driven activities carry zero/negative float
        bind_confidence=round(score, 4),
    )
    if classify_bind(score) == "quarantine":
        logger.warning("p6.quarantine", extra={"uid": activity.uid, "score": score})
    return activity


def _best_wbs(short_name: str, canonical: dict[str, str]) -> tuple[str, float]:
    key = short_name.strip().lower()
    best_node, best_score = "PROJ-000-MEP-00", 0.0   # quarantine sentinel node
    for name, node in canonical.items():
        score = SequenceMatcher(None, key, name.strip().lower()).ratio()
        if score > best_score:
            best_node, best_score = node, score
    return best_node, best_score

The builder returns a fully typed record and never writes anything itself, so the outer queue task can retry it safely; the quarantine sentinel node satisfies the wbs_node regex while a bind_confidence below 0.75 keeps the record out of the diff stages until a scheduler resolves it.

Schema and configuration reference

Binding constants are site-canonical and must appear identically wherever P6 activities are bound: AUTO_BIND = 0.92 and REVIEW_FLOOR = 0.75. The default calendar day of 8.0 hours is a per-project configuration value, not a constant to hard-code across clients.

Verification and testing

Treat decoding, float conversion, and binding as pure functions and assert against known rows. The serialized contract is round-tripped with model_dump_json so the wire format is part of the test surface.

from datetime import datetime, timezone


def _task(**overrides) -> dict[str, str]:
    base = {
        "task_code": "A1010",
        "task_name": "Erect structural steel — Level 3",
        "wbs_id": "9001",
        "clndr_id": "C1",
        "target_start_date": "2026-07-20 08:00",
        "target_end_date": "2026-07-31 17:00",
        "total_float_hr_cnt": "0",
    }
    base.update(overrides)
    return base


def _build(**overrides):
    return build_activity(
        _task(**overrides),
        wbs_short_by_id={"9001": "Structural Steel"},
        canonical_wbs={"Structural Steel": "PROJ-014-STR-02"},
        day_hr_by_calendar={"C1": 8.0},
        project_id="TWR-014",
        discipline="STR",
    )


def test_zero_float_is_critical():
    act = _build(total_float_hr_cnt="0")
    assert act.is_critical is True
    assert act.total_float_days == 0


def test_float_hours_convert_to_days():
    act = _build(total_float_hr_cnt="40")   # 40h / 8h-day = 5 days
    assert act.total_float_days == 5
    assert act.is_critical is False


def test_exact_wbs_name_auto_binds():
    act = _build()
    assert act.wbs_node == "PROJ-014-STR-02"
    assert classify_bind(act.bind_confidence) == "auto_bind"


def test_unknown_wbs_quarantines():
    act = build_activity(
        _task(wbs_id="9099"),
        wbs_short_by_id={"9099": "Unmapped Temporary Works"},
        canonical_wbs={"Structural Steel": "PROJ-014-STR-02"},
        day_hr_by_calendar={"C1": 8.0},
        project_id="TWR-014", discipline="STR",
    )
    assert classify_bind(act.bind_confidence) == "quarantine"


def test_activity_round_trips():
    act = _build()
    restored = ScheduleActivity.model_validate_json(act.model_dump_json())
    assert restored.uid == act.uid

Run the suite with pytest -q. For a quick smoke test on a real export, pipe the first TASK row through the builder and confirm the wbs_node echoes back in canonical PROJ-NNN-DIV-NN form and total_float_days is a plausible whole number.

Troubleshooting

• UnicodeDecodeError opening the XER file. P6 writes XER as Windows-1252, so a smart quote or an em dash in an activity name (0x92, 0x97) blows up a UTF-8 read. Open the file with encoding="cp1252" as shown; never errors="ignore", which silently deletes characters and corrupts activity names used for WBS matching.
• The same activity id appears in two projects and collides. P6’s task_code (A1010) is unique only within its proj_id, so keying a lookup on the code alone merges two schedules. Always key on the uid — project_id plus activity_id — so a multi-project import can never overwrite one activity with another’s dates.
• Every activity reports the wrong float in days. total_float_hr_cnt is stored in hours against the activity’s calendar, not a fixed eight-hour day. If the calendar id in TASK.clndr_id is not resolved against the CALENDAR table’s day_hr_cnt, a six-day-week or ten-hour-shift calendar converts float wrongly. Build the calendar lookup first and fall back to the project default only when the id is genuinely absent.
• findall on the P6 XML returns an empty list. The export declares a default namespace on the root, so an unqualified .//Activity matches nothing. Pass the namespace map to iterfind, and recover the URI from the root tag (_first_ns) rather than hard-coding a P6 version, because a client on a different release ships a different namespace URI.
• Activities land on is_critical=True that the scheduler says are not critical. P6 can mark a longest-path activity through its own scheduling options that your derivation does not reproduce. This page derives criticality strictly from total_float_days <= 0; if the project uses a non-zero float threshold for “near-critical,” make that threshold an explicit configuration value and document it rather than copying P6’s driving_path_flag blindly.

Frequently Asked Questions

Related

• Parsing Primavera P6 XER Files with Python
• Extracting Activity Relationships from P6 XML Exports
• Critical Path Delta Detection
• Schedule Baseline Variance
• WBS Mapping Strategies

← Back to Schedule Sync & Critical Path Integration