Unknown user at 08:52, 13 September 2025

2025-09-13T08:52:07Z

Unknown user: Created page with "Mantap. Aku siapkan script siap-pakai untuk mendeteksi bottleneck dari file XES terlampir (`/mnt/data/BPIC_2012_A.xes`) memakai PM4Py. Script menghitung: * **Serv..."

2025-09-13T00:00:01Z

Created page with "Mantap. Aku siapkan **script siap-pakai** untuk mendeteksi bottleneck dari **file XES terlampir** (`/mnt/data/BPIC_2012_A.xes`) memakai **PM4Py**. Script menghitung: * **Serv..."

New page

Mantap. Aku siapkan **script siap-pakai** untuk mendeteksi bottleneck dari **file XES terlampir** (`/mnt/data/BPIC_2012_A.xes`) memakai **PM4Py**.
Script menghitung:

* **Service time** per aktivitas dari pasangan `lifecycle:start`–`complete`
* **Waiting time** antar-aktivitas di dalam tiap case
* Ringkasan metrik (mean/median/p95/count) + **bottleneck score** (gabungan z-score service & waiting)
* Ekspor ke **CSV** dan tampilkan **TOP-N bottleneck**

> **Install (sekali saja)**
>
> ```bash
> pip install pm4py pandas numpy
> ```

---

### 1) Script utama — `bottleneck_pm4py.py`

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import sys
from pathlib import Path
import numpy as np
import pandas as pd

from pm4py.objects.log.importer.xes import importer as xes_importer
from pm4py import convert_to_dataframe

def pair_start_complete(df: pd.DataFrame) -> pd.DataFrame:
"""Pasangkan start/complete per (case, activity) -> satu baris per eksekusi aktivitas."""
if "lifecycle:transition" not in df.columns:
return pd.DataFrame(columns=["case", "activity", "start_time", "complete_time", "service_sec"])

dfl = df.dropna(subset=["lifecycle:transition"]).copy()
dfl["transition"] = dfl["lifecycle:transition"].str.lower()
dfl = dfl[dfl["transition"].isin(["start", "complete"])]
if dfl.empty:
return pd.DataFrame(columns=["case", "activity", "start_time", "complete_time", "service_sec"])

dfl = dfl.sort_values(["case:concept:name", "concept:name", "time:timestamp", "transition"]).copy()
dfl["start_rank"] = dfl["transition"].eq("start").groupby(
[dfl["case:concept:name"], dfl["concept:name"]]
).cumsum()
dfl["complete_rank"] = dfl["transition"].eq("complete").groupby(
[dfl["case:concept:name"], dfl["concept:name"]]
).cumsum()

starts = dfl[dfl["transition"] == "start"].rename(columns={"time:timestamp": "start_time"})
comps = dfl[dfl["transition"] == "complete"].rename(columns={"time:timestamp": "complete_time"})

merged = pd.merge(
starts[["case:concept:name", "concept:name", "start_rank", "start_time"]],
comps[["case:concept:name", "concept:name", "complete_rank", "complete_time"]],
left_on=["case:concept:name", "concept:name", "start_rank"],
right_on=["case:concept:name", "concept:name", "complete_rank"],
how="inner",
).rename(columns={"case:concept:name": "case", "concept:name": "activity"})

merged["service_sec"] = (merged["complete_time"] - merged["start_time"]).dt.total_seconds()
merged = merged[(merged["service_sec"] >= 0) & np.isfinite(merged["service_sec"])]
return merged[["case", "activity", "start_time", "complete_time", "service_sec"]]

def compute_waiting_times(df: pd.DataFrame, exec_df: pd.DataFrame) -> pd.DataFrame:
"""
Hitung waiting time antar aktivitas di tiap case.
- Jika ada start/complete: tunggu = start(curr) - complete(prev)
- Jika tidak ada lifecycle: tunggu = time(curr) - time(prev)
"""
has_lifecycle = "lifecycle:transition" in df.columns and \
df["lifecycle:transition"].str.lower().isin(["start", "complete"]).any()

rows = []
if has_lifecycle and not exec_df.empty:
per_case = exec_df.sort_values(["case", "start_time"])
for case, g in per_case.groupby("case"):
g = g.sort_values("start_time")
prev_complete, prev_act = None, None
for _, r in g.iterrows():
if prev_complete is not None:
wt = (r["start_time"] - prev_complete).total_seconds()
if wt >= 0:
rows.append({"case": case, "from_activity": prev_act,
"to_activity": r["activity"], "waiting_sec": wt})
prev_complete, prev_act = r["complete_time"], r["activity"]
else:
# fallback tanpa lifecycle
df2 = df.sort_values(["case:concept:name", "time:timestamp"])
for case, g in df2.groupby("case:concept:name"):
g = g.sort_values("time:timestamp")
prev_time, prev_act = None, None
for _, r in g.iterrows():
if prev_time is not None:
wt = (r["time:timestamp"] - prev_time).total_seconds()
if wt >= 0:
rows.append({"case": case, "from_activity": prev_act,
"to_activity": r["concept:name"], "waiting_sec": wt})
prev_time, prev_act = r["time:timestamp"], r["concept:name"]

if not rows:
return pd.DataFrame(columns=["case", "from_activity", "to_activity", "waiting_sec"])
w = pd.DataFrame(rows)
return w[["case", "from_activity", "to_activity", "waiting_sec"]]

def zscore(series: pd.Series) -> pd.Series:
mu = np.nanmean(series)
sd = np.nanstd(series, ddof=0)
if sd == 0 or np.isnan(sd):
return pd.Series(np.zeros(len(series)), index=series.index)
return (series - mu) / sd

def main():
ap = argparse.ArgumentParser(description="Bottleneck Detection from XES using PM4Py")
ap.add_argument("xes_path", type=str, help="Path ke file .xes")
ap.add_argument("--top", type=int, default=10, help="Top-N bottleneck yang ditampilkan (default 10)")
ap.add_argument("--out", type=str, default="bottlenecks_summary.csv", help="Output CSV ringkasan")
args = ap.parse_args()

xes_path = Path(args.xes_path)
if not xes_path.exists():
print(f"[ERROR] File tidak ditemukan: {xes_path}", file=sys.stderr)
sys.exit(1)

# 1) Load XES -> DataFrame
log = xes_importer.apply(str(xes_path))
df = convert_to_dataframe(log)

for c in ["case:concept:name", "concept:name", "time:timestamp"]:
if c not in df.columns:
print(f"[ERROR] Kolom wajib hilang di event log: {c}", file=sys.stderr)
sys.exit(1)

df["time:timestamp"] = pd.to_datetime(df["time:timestamp"], errors="coerce")
df = df.dropna(subset=["time:timestamp"]).copy()

# 2) Service time per aktivitas
exec_df = pair_start_complete(df)

# 3) Waiting time antar aktivitas
wait_df = compute_waiting_times(df, exec_df)

# 4) Agregasi per aktivitas
if not exec_df.empty:
service_stats = exec_df.groupby("activity")["service_sec"].agg(
service_mean_sec="mean",
service_median_sec="median",
service_p95_sec=lambda x: np.nanpercentile(x, 95),
service_count="count",
).reset_index()
else:
service_stats = pd.DataFrame(columns=["activity","service_mean_sec","service_median_sec","service_p95_sec","service_count"])

if not wait_df.empty:
wait_stats = wait_df.groupby("to_activity")["waiting_sec"].agg(
wait_mean_sec="mean",
wait_median_sec="median",
wait_p95_sec=lambda x: np.nanpercentile(x, 95),
wait_count="count",
).reset_index().rename(columns={"to_activity": "activity"})
else:
wait_stats = pd.DataFrame(columns=["activity","wait_mean_sec","wait_median_sec","wait_p95_sec","wait_count"])

summary = pd.merge(service_stats, wait_stats, on="activity", how="outer").fillna(0)

# 5) Skor bottleneck (gabungan z-score)
summary["z_service"] = zscore(summary["service_mean_sec"])
summary["z_wait"] = zscore(summary["wait_mean_sec"])
summary["z_service_p95"] = zscore(summary["service_p95_sec"])
summary["z_wait_p95"] = zscore(summary["wait_p95_sec"])

summary["bottleneck_score"] = (
0.4 * summary["z_service"] +
0.4 * summary["z_wait"] +
0.1 * summary["z_service_p95"] +
0.1 * summary["z_wait_p95"]
)

summary_sorted = summary.sort_values("bottleneck_score", ascending=False)
summary_sorted.to_csv(args.out, index=False)

# 6) Tampilkan Top-N ringkas
n = min(args.top, len(summary_sorted))
print(f"\n=== TOP {n} BOTTLENECK ACTIVITIES ===")
if n == 0:
print("Tidak ada aktivitas terdeteksi. Cek struktur log Anda.")
sys.exit(0)

for _, r in summary_sorted.head(n).iterrows():
print(
f"- {r['activity']}: score={r['bottleneck_score']:.3f} | "
f"svc_mean={r['service_mean_sec']:.1f}s (p95={r['service_p95_sec']:.1f}s, n={int(r['service_count'])}) | "
f"wait_mean={r['wait_mean_sec']:.1f}s (p95={r['wait_p95_sec']:.1f}s, n={int(r['wait_count'])})"
)

print(f"\nCSV ringkasan disimpan ke: {args.out}")
print("Prioritaskan aktivitas dengan score & p95 tinggi, terutama jika wait_mean besar (antrian).")

if __name__ == "__main__":
main()
```

**Jalankan ke file terlampir** (path sesuai unggahan Anda):

```bash
python bottleneck_pm4py.py /mnt/data/BPIC_2012_A.xes --top 15 --out bottlenecks_BPICA.csv
```

---

### 2) (Opsional) Cross-check cepat pakai **Performance DFG** PM4Py

Ini alternatif ringkas untuk melihat **mean waiting time** antar-aktivitas (edge) dan **sojourn time** (di node). Cocok untuk sanity check bottleneck transisi.

```python
#!/usr/bin/env python3
# perf_dfg_quickcheck.py
from pm4py.objects.log.importer.xes import importer as xes_importer
from pm4py.algo.discovery.dfg import algorithm as dfg_discovery
from pm4py.statistics.sojourn_time.log import get as soj_get
import pandas as pd
import sys

xes = sys.argv[1]
log = xes_importer.apply(xes)

# Mean performance DFG (edge durations)
perf_dfg = dfg_discovery.apply(log, variant=dfg_discovery.Variants.PERFORMANCE)
df_edges = pd.DataFrame(
[{"from": a, "to": b, "mean_sec": v} for (a, b), v in perf_dfg.items()]
).sort_values("mean_sec", ascending=False)

# Sojourn time per activity (durasi berada di node)
soj = soj_get.apply(log) # returns dict {activity: mean_seconds}
df_nodes = pd.DataFrame(
[{"activity": k, "sojourn_mean_sec": v} for k, v in soj.items()]
).sort_values("sojourn_mean_sec", ascending=False)

df_edges.to_csv("perf_dfg_edges.csv", index=False)
df_nodes.to_csv("sojourn_nodes.csv", index=False)

print("Top 10 edges by mean_sec:")
print(df_edges.head(10))
print("\nTop 10 activities by sojourn_mean_sec:")
print(df_nodes.head(10))
```

**Jalankan:**

```bash
python perf_dfg_quickcheck.py /mnt/data/BPIC_2012_A.xes
```

---

### Catatan penting

* **Akurasi service time** bergantung pada hadirnya pasangan `lifecycle:start/complete`. Jika dataset hanya punya `complete`, fokuskan interpretasi pada **waiting antar event** (edge) dan **sojourn** (node).
* **p95** membantu mengungkap **ekor panjang** (spikes jarang tapi berat) yang sering jadi bottleneck meski mean tidak terlalu tinggi.
* Untuk investigasi mendalam, gabungkan hasil `bottlenecks_summary.csv` dengan **variasi per resource**, **per channel**, atau **per case attribute** (mis. `org:resource`, `org:role`, `application type`, dll.) lalu lakukan **groupby** tambahan.

Kalau mau, saya bisa lanjutkan dengan **visualisasi** (bar chart Top-N) atau **filter per resource**—tinggal sebutkan atribut yang ingin dianalisis.

PM: Bottleneck Detection - Revision history

Unknown user at 08:52, 13 September 2025

Unknown user: Created page with "Mantap. Aku siapkan **script siap-pakai** untuk mendeteksi bottleneck dari **file XES terlampir** (`/mnt/data/BPIC_2012_A.xes`) memakai **PM4Py**. Script menghitung: * **Serv..."

Unknown user: Created page with "Mantap. Aku siapkan script siap-pakai untuk mendeteksi bottleneck dari file XES terlampir (`/mnt/data/BPIC_2012_A.xes`) memakai PM4Py. Script menghitung: * **Serv..."