P3-4.4 哪些信号说明样本单位抓错了¶

# 这个例子检查样本单位选错时出现的行数、标签和特征信号。
import csv
from collections import defaultdict
from pathlib import Path

repeat_warning_threshold = 1
preview_row_count = 8

input_path = Path("docs/assets/part-03/chapter-04/p3_4_4_sample_unit_warning_log.csv")

with input_path.open(newline="", encoding="utf-8") as file:
    rows = list(csv.DictReader(file))

for row in rows:
    row["second"] = int(row["second"])
    row["flow"] = float(row["flow"])
    row["review_needed"] = int(row["review_needed"])

events = defaultdict(list)
for row in rows:
    events[row["event_id"]].append(row)

label_repetition = []
event_summary = []

for event_id, event_rows in sorted(events.items()):
    review_needed_values = [row["review_needed"] for row in event_rows]
    label_repetition.append(
        {
            "event_id": event_id,
            "row_count": len(event_rows),
            "review_needed_sum": sum(review_needed_values),
        }
    )
    event_summary.append(
        {
            "event_id": event_id,
            "duration_seconds": max(row["second"] for row in event_rows),
            "flow_mean": sum(row["flow"] for row in event_rows) / len(event_rows),
            "review_needed": max(review_needed_values),
        }
    )

max_row_count = max(item["row_count"] for item in label_repetition)
max_label_sum = max(item["review_needed_sum"] for item in label_repetition)

warning_check = [
    (
        "same event repeated across many rows",
        max_row_count > repeat_warning_threshold,
    ),
    (
        "same label repeated within one event",
        max_label_sum > repeat_warning_threshold,
    ),
    (
        "event-level features appear only after regrouping",
        bool(event_summary),
    ),
]

print("1) row-level table where labels repeat inside one event")
for row in rows[:preview_row_count]:
    print(
        f"{row['event_id']} at {row['second']}s: "
        f"flow={row['flow']:.1f}, review_needed={row['review_needed']}"
    )
print(f"... {len(rows) - preview_row_count} more time-point rows")
print()
print("2) repeated rows and repeated labels per event")
for item in label_repetition:
    print(
        f"{item['event_id']}: row_count={item['row_count']}, "
        f"review_needed_sum={item['review_needed_sum']}"
    )
print()
print("3) event-level summary that appears only after regrouping")
for item in event_summary:
    print(
        f"{item['event_id']}: duration={item['duration_seconds']}s, "
        f"flow_mean={item['flow_mean']:.2f}, "
        f"review_needed={item['review_needed']}"
    )
print()
print("4) warning signs that sample unit may be wrong")
for warning_sign, seen in warning_check:
    print(f"{warning_sign}: {'yes' if seen else 'no'}")

1) row-level table where labels repeat inside one event
A at 0s: flow=0.5, review_needed=1
A at 1s: flow=0.9, review_needed=1
A at 2s: flow=1.2, review_needed=1
A at 3s: flow=1.5, review_needed=1
A at 4s: flow=1.8, review_needed=1
A at 5s: flow=1.6, review_needed=1
A at 6s: flow=1.4, review_needed=1
A at 7s: flow=1.2, review_needed=1
... 28 more time-point rows

2) repeated rows and repeated labels per event
A: row_count=18, review_needed_sum=18
B: row_count=9, review_needed_sum=0
C: row_count=6, review_needed_sum=6
D: row_count=3, review_needed_sum=0

3) event-level summary that appears only after regrouping
A: duration=17s, flow_mean=0.94, review_needed=1
B: duration=8s, flow_mean=0.88, review_needed=0
C: duration=5s, flow_mean=1.07, review_needed=1
D: duration=2s, flow_mean=0.67, review_needed=0

4) warning signs that sample unit may be wrong
same event repeated across many rows: yes
same label repeated within one event: yes
event-level features appear only after regrouping: yes