paper_example_log.py

"""Materialize the running example of Augusto et al. (2019) as an XES log.

The paper enumerates ten distinct traces, each recorded ten times. We write
them to ``outputs/paper_example.xes`` so both the Java and Python miners can
consume the same file, and we can verify whether either reproduces Fig. 3c.
"""

from __future__ import annotations

from pathlib import Path

import pm4py

from log_loader import PAPER_EXAMPLE_LOG


def write_xes(out_path: Path) -> Path:
    """Hand-rolled XES that the OpenXES Java parser used by Split Miner accepts.

    PM4Py's writer omits the ``<global>`` sections that the Java reference
    implementation expects, which makes ``LogParser.getComplexLog`` NPE.
    """
    out_path.parent.mkdir(parents=True, exist_ok=True)
    parts: list[str] = [
        '<?xml version="1.0" encoding="UTF-8" ?>',
        '<log xes.version="1.0" xes.features="nested-attributes" '
        'openxes.version="1.0RC7">',
        '\t<extension name="Lifecycle" prefix="lifecycle" '
        'uri="http://www.xes-standard.org/lifecycle.xesext"/>',
        '\t<extension name="Concept" prefix="concept" '
        'uri="http://www.xes-standard.org/concept.xesext"/>',
        '\t<extension name="Time" prefix="time" '
        'uri="http://www.xes-standard.org/time.xesext"/>',
        '\t<global scope="trace">',
        '\t\t<string key="concept:name" value="DEFAULT"/>',
        '\t</global>',
        '\t<global scope="event">',
        '\t\t<string key="lifecycle:transition" value="complete"/>',
        '\t\t<string key="concept:name" value="DEFAULT"/>',
        '\t\t<date key="time:timestamp" value="1970-01-01T00:00:00.000+00:00"/>',
        '\t</global>',
        '\t<classifier name="concept:name" keys="concept:name"/>',
    ]

    for i, trace in enumerate(PAPER_EXAMPLE_LOG):
        parts.append('\t<trace>')
        parts.append(f'\t\t<string key="concept:name" value="case_{i:03d}"/>')
        for j, label in enumerate(trace):
            parts.append('\t\t<event>')
            parts.append(f'\t\t\t<string key="concept:name" value="{label}"/>')
            parts.append('\t\t\t<string key="lifecycle:transition" value="complete"/>')
            ts = f"2026-01-01T00:{j // 60:02d}:{j % 60:02d}.000+00:00"
            parts.append(f'\t\t\t<date key="time:timestamp" value="{ts}"/>')
            parts.append('\t\t</event>')
        parts.append('\t</trace>')
    parts.append('</log>')

    with open(out_path, "w", encoding="utf-8") as f:
        f.write("\n".join(parts))
    return out_path


if __name__ == "__main__":
    out = Path(__file__).parent / "outputs" / "paper_example.xes"
    write_xes(out)
    print(f"wrote {out} ({len(PAPER_EXAMPLE_LOG)} distinct traces x 10 = "
          f"{sum(len(t) for t in PAPER_EXAMPLE_LOG)*10} events)")