Initial commit: Dabit Time Manager project
Python-based time management application with UDP discovery, TCP protocol communication, time sync, and drift monitoring. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
103
services/export_service.py
Normal file
103
services/export_service.py
Normal file
@@ -0,0 +1,103 @@
|
||||
"""CSV export 서비스"""
|
||||
|
||||
import csv
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from models.reading import TimeReading
|
||||
|
||||
|
||||
def export_readings_csv(
|
||||
readings: list[TimeReading],
|
||||
filepath: str | Path,
|
||||
) -> str:
|
||||
"""시간 읽기 기록을 CSV로 내보내기 (컨트롤러별 정렬).
|
||||
|
||||
Returns: 저장된 파일 경로
|
||||
"""
|
||||
filepath = Path(filepath)
|
||||
# 컨트롤러별로 정렬
|
||||
sorted_readings = sorted(readings, key=lambda r: (r.controller_mac, r.pc_time))
|
||||
|
||||
with open(filepath, "w", newline="", encoding="utf-8-sig") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow([
|
||||
"컨트롤러",
|
||||
"MAC",
|
||||
"PC 시간",
|
||||
"컨트롤러 시간",
|
||||
"오차(초)",
|
||||
])
|
||||
for r in sorted_readings:
|
||||
writer.writerow([
|
||||
r.controller_label,
|
||||
r.controller_mac,
|
||||
r.pc_time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
r.controller_time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
f"{r.drift_seconds:.1f}",
|
||||
])
|
||||
return str(filepath)
|
||||
|
||||
|
||||
def export_per_controller_csv(
|
||||
readings: list[TimeReading],
|
||||
folder: str | Path,
|
||||
) -> list[str]:
|
||||
"""컨트롤러별 개별 CSV 파일 생성.
|
||||
|
||||
Returns: 생성된 파일 경로 목록
|
||||
"""
|
||||
folder = Path(folder)
|
||||
os.makedirs(folder, exist_ok=True)
|
||||
|
||||
by_mac: dict[str, list[TimeReading]] = {}
|
||||
for r in readings:
|
||||
by_mac.setdefault(r.controller_mac, []).append(r)
|
||||
|
||||
created = []
|
||||
for mac, rlist in by_mac.items():
|
||||
rlist.sort(key=lambda r: r.pc_time)
|
||||
# 파일명: 컨트롤러 이름 기반 (특수문자 제거)
|
||||
safe_name = rlist[0].controller_label
|
||||
for ch in r'<>:"/\\|?*':
|
||||
safe_name = safe_name.replace(ch, "_")
|
||||
filepath = folder / f"{safe_name}_drift.csv"
|
||||
export_readings_csv(rlist, filepath)
|
||||
created.append(str(filepath))
|
||||
|
||||
return created
|
||||
|
||||
|
||||
def export_summary_csv(
|
||||
summary: dict[str, dict],
|
||||
filepath: str | Path,
|
||||
) -> str:
|
||||
"""오차 요약 통계를 CSV로 내보내기.
|
||||
|
||||
Returns: 저장된 파일 경로
|
||||
"""
|
||||
filepath = Path(filepath)
|
||||
with open(filepath, "w", newline="", encoding="utf-8-sig") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow([
|
||||
"컨트롤러",
|
||||
"MAC",
|
||||
"읽기 횟수",
|
||||
"평균 오차(초)",
|
||||
"최대 오차(초)",
|
||||
"시간당 오차율(초/시간)",
|
||||
"첫 읽기",
|
||||
"마지막 읽기",
|
||||
])
|
||||
for mac, s in summary.items():
|
||||
writer.writerow([
|
||||
s["label"],
|
||||
mac,
|
||||
s["count"],
|
||||
f"{s['avg_drift']:.2f}",
|
||||
f"{s['max_drift']:.2f}",
|
||||
f"{s['drift_per_hour']:.3f}",
|
||||
s["first_time"].strftime("%Y-%m-%d %H:%M:%S"),
|
||||
s["last_time"].strftime("%Y-%m-%d %H:%M:%S"),
|
||||
])
|
||||
return str(filepath)
|
||||
Reference in New Issue
Block a user