Skip to content

petri_net_nn.xes

xes

IEEE XES (eXtensible Event Stream) log loader.

XES is the standard format for process execution logs and is mentioned explicitly in §10 Step 3 of the architecture spec as the data source for training. This scaffold implements a minimal reader covering the subset sufficient to drive train_on_traces in petri_net_nn.traces:

  • one <log> containing zero or more <trace> elements
  • each <trace> carrying zero or more <event> elements
  • typed attributes (<string>, <int>, <float>, <boolean>, <date>) at the log, trace, or event level — all surfaced as string-valued entries in the corresponding attributes dict

Each event's name is the value of its concept:name attribute, which is the XES convention for the activity (task) that fired. That string is what matches against the transition labels produced by parse_bpmn.

Out of scope: extensions, classifiers, globals, nested attributes (lists / containers). They parse without error but are ignored.

parse_xes

parse_xes(source)

Parse an IEEE XES log into a list of XESTrace objects.

The XES namespace is optional; both namespaced and non-namespaced documents parse identically.

Source code in petri_net_nn/xes.py
def parse_xes(source: str | Path | IO[str] | IO[bytes]) -> list[XESTrace]:
    """Parse an IEEE XES log into a list of XESTrace objects.

    The XES namespace is optional; both namespaced and non-namespaced
    documents parse identically.
    """
    root = _load_xml(source)
    if _local(root.tag) != "log":
        raise ValueError(
            f"expected XES root element <log>, got <{_local(root.tag)}>"
        )

    traces: list[XESTrace] = []
    for trace_el in root:
        if _local(trace_el.tag) != "trace":
            continue
        trace = XESTrace(attributes=_collect_attributes(trace_el))
        for event_el in trace_el:
            if _local(event_el.tag) != "event":
                continue
            attrs = _collect_attributes(event_el)
            event = XESEvent(name=attrs.get("concept:name", ""), attributes=attrs)
            trace.events.append(event)
        traces.append(trace)

    return traces