From eb25c01c973ac16c3e8948fd22df4f30aaaba678 Mon Sep 17 00:00:00 2001 From: jjangddu Date: Mon, 13 Jul 2026 15:21:07 +0900 Subject: [PATCH] =?UTF-8?q?docs(tools):=20align=5Fanalyze.py=20=EC=9D=B4?= =?UTF-8?q?=EA=B4=80=20+=20CH3=20flicker=20fix=20Python=20replay=20?= =?UTF-8?q?=EA=B2=80=EC=A6=9D=20=EA=B2=B0=EA=B3=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tools/align_analyze.py 신규: AlignGuide4Stage 상태머신 Python 재현 - `--both` 모드로 NEW (majority + relaxed + soft-hint) vs LEGACY (연속 3 hit) 로직 병행 replay 지원 - Kotlin AlignmentConstants (V_WIN/V_MAJ/L_WIN/L_MAJ/STUCK/SOFT) 값 그대로 매핑 - tools/README.md: align_analyze.py 사용법 + 파라미터 매핑 표 + data123/ 4 세션 검증 결과 표 (Phantom / Human 0·1·3 CM) 정리 - tools/labdb_upload.py 신규 추가 (기존 문서만 있고 실체 미커밋 상태였음) - VesiScan_Android_Pipeline_Summary.md v7 patch (2026-07-13): align_analyze.py 도구 언급 + CH3 flicker fix 실측 검증 결과 요약 검증 결과 요약 (data123/ human 4 세션): - Human 0CM (CH3 31.8%): LEGACY 즉시 회귀 → NEW flick 1/4 흡수 유지 - Human 1CM (CH3 77.3%): LEGACY idx15 oscillation → NEW idx13 조기 진입 - Human 3CM (CH3 97.2%): 양쪽 정상 (회귀 없음) - Phantom: 양쪽 정상 (relaxed mode 오진입 없음) --- VesiScan_Android_Pipeline_Summary.md | 8 +- tools/README.md | 78 ++++++ tools/align_analyze.py | 385 +++++++++++++++++++++++++++ tools/labdb_upload.py | 270 +++++++++++++++++++ 4 files changed, 740 insertions(+), 1 deletion(-) create mode 100644 tools/align_analyze.py create mode 100644 tools/labdb_upload.py diff --git a/VesiScan_Android_Pipeline_Summary.md b/VesiScan_Android_Pipeline_Summary.md index 04e0c54..5e79fbc 100644 --- a/VesiScan_Android_Pipeline_Summary.md +++ b/VesiScan_Android_Pipeline_Summary.md @@ -2,7 +2,7 @@ ## 작성일: 2026-04-23 ## 작성자: dwjang -## 마지막 업데이트: 2026-07-10 (v7) — 1.2.0-demo (versionCode 26) +## 마지막 업데이트: 2026-07-13 (v7 patch) — 1.2.0-demo (versionCode 26) ## 권장 펌웨어: **VBTFW0120+** (mim FIFO 지원 필수. VBTFW0121 은 `mls mode 0` handler 취약) ### v7 주요 변경 요약 (2026-07-06 ~ 2026-07-10) @@ -62,6 +62,12 @@ - **`tools/labdb_upload.py` 신규** — LabdbUploader.kt 의 `buildPayload` 로직을 Python 으로 이식. 외부 저장 세션 폴더 일괄 업로드용. 자세한 사용법: [tools/README.md](tools/README.md). +- **`tools/align_analyze.py` 신규** (2026-07-13) — `AlignGuide4Stage` 상태머신을 + Python 으로 재현하여 alignment 세션 JSON 을 오프라인 replay. `--both` 모드로 + NEW (majority + relaxed + soft-hint) vs LEGACY (연속 3 hit) 로직 비교 지원. + `data123/` 인체 임상 4 세션 (Phantom / Human 0·1·3 CM) replay 결과 CH3 flicker + fix 회귀·조기 진입 개선 확인, relaxed mode false-positive 없음. + 파라미터 매핑 · 결과 표: [tools/README.md](tools/README.md). **참고** - 이번 세션에서 발견되었으나 아직 미조치: PiezoMonitoringView (2900L) / diff --git a/tools/README.md b/tools/README.md index 322b067..1c08901 100644 --- a/tools/README.md +++ b/tools/README.md @@ -170,3 +170,81 @@ summary: 19 uploaded, 0 skipped, 1 failed (total 20) 3. **다른 폰 → labdb 이관**: A 폰에서 측정 → `Downloads/VesiScan_Sessions/` 폴더 → PC 로 복사 → 스크립트로 서버 반영. +--- + +## align_analyze.py — Alignment session Python replay + +`AlignGuide4Stage` (Kotlin) 의 상태머신 로직을 Python 으로 그대로 재현하여 +alignment 세션 JSON 을 오프라인 replay 한다. Method D 파이프라인 +(`vesiscan_test.library`) 을 통해 실제 앱과 동일한 검출 결과 위에서 +phase 전이 · action 시퀀스 · CH3 flicker 통계를 산출한다. + +### Setup + +```bash +# piezophantomtest repo 경로 필요 (align_analyze.py 상단 sys.path.insert 참조) +# 기본값: c:/Projects/piezophantomtest/piezo-phantom-test +pip install numpy pandas +``` + +### Usage + +```bash +# 세션 요약 (detection matrix + per-channel 통계) +python tools/align_analyze.py session.json + +# NEW 로직 (majority + relaxed + soft-hint) 로 replay +python tools/align_analyze.py session.json --replay + +# LEGACY 로직 (연속 3회 hit) 로 replay — 비교용 +python tools/align_analyze.py session.json --replay --legacy + +# NEW + LEGACY 동시 실행 (fix 전후 비교) +python tools/align_analyze.py session.json --both + +# accum 조절 (기본 10 — BLE commit 주기 매칭) +python tools/align_analyze.py session.json --both --accum 10 --no-summary + +# 여러 세션 일괄 (files 생략 시 c:/Projects/medilightv2android/data123/*.json 전체) +python tools/align_analyze.py --both --no-summary +``` + +### 재현 파라미터 (Kotlin AlignmentConstants 매핑) + +| Python | Kotlin | 의미 | +|---|---|---| +| `V_WIN=6` | `VERTICAL_HIT_WINDOW` | VERTICAL_CLIMB 진입 sliding window | +| `V_MAJ=3` | `VERTICAL_HIT_MAJORITY` | window 내 최소 hit | +| `L_WIN=4` | `STABILIZE_LOST_WINDOW` | CH3 lost 판정 window | +| `L_MAJ=3` | `STABILIZE_LOST_MAJORITY` | window 내 최소 lost | +| `STUCK_TH=20` | `VERTICAL_STUCK_THRESHOLD` | relaxed mode 진입 threshold | +| `SOFT_AFTER=12` | `VERTICAL_SOFT_HINT_AFTER` | soft hint 문구 전환 시점 | + +### 활용 예 (CH3 flicker fix 검증 2026-07-10) + +`data123/` 인체 임상 4 세션에 `--both` 로 NEW / LEGACY 병행 replay 실행. +결과 요약: + +| 세션 | Records | LEGACY | NEW | +|---|---|---|---| +| Phantom 0CM | 62 | LR_BALANCE 도달 정상 | 동일 궤적, 회귀 없음 (false-positive 없음) | +| Human 0CM (CH3 31.8%) | 22 | CH3_STAB 진입 즉시 회귀 | flick 1/4 흡수하며 유지 | +| Human 1CM (CH3 77.3%) | 22 | idx 15 진입 + 2회 oscillation | idx 13 조기 진입 + 2 flick 흡수 | +| Human 3CM (CH3 97.2%) | 36 | CENTER_OPTIMIZE 도달 정상 | 동일 궤적, 회귀 없음 | + +결론: sliding-window majority + soft-hint 조합만으로 human 0CM/1CM 회귀·조기 +진입 개선 확인. relaxed mode 는 팬텀·정상 케이스에서 트리거되지 않아 false +positive 위험 없음. + +### 출력 (예) + +``` +--- RollingAligner replay NEW --- + window = 10 + idx phase relax chN CH3 ch0..3 chord CV action + 10 VERTICAL . 6 Y 67.3 104.5 103.8 103.5 0.17 MOVE_UP (hit 1/6) + 11 VERTICAL . 6 Y 65.3 104.5 103.8 103.5 0.18 MOVE_UP (hit 2/6) + 12 CH3_STAB . 6 Y 67.3 104.5 103.8 103.5 0.17 STOP → CH3_STAB (hit 3/6) + ... +``` + diff --git a/tools/align_analyze.py b/tools/align_analyze.py new file mode 100644 index 0000000..f88e06c --- /dev/null +++ b/tools/align_analyze.py @@ -0,0 +1,385 @@ +""" +Alignment session 상세 분석 — Python method_d 파이프라인으로 재현. + +목표: + 1. 각 record 별 채널별 검출 여부 (CH3 flicker 원인 pin-point) + 2. Cross-channel 보정 전/후 결과 비교 (base vs corrected) + 3. RollingAligner 실행 시 phase 전이와 action 시퀀스 + 4. CH3 raw 신호 특성 (low_echo span, peak amplitude) +""" +import os, sys, json, glob +from collections import Counter +os.environ.setdefault("PYTHONIOENCODING", "utf-8") +sys.path.insert(0, r"c:/Projects/piezophantomtest/piezo-phantom-test") + +import numpy as np +from vesiscan_test.library.method_d import detect_walls_multichannel_full +from vesiscan_test.library.method_d.detector import detect as method_d_detect +from vesiscan_test.library.method_d.preprocessing import preprocess_heavy, preprocess_light +from vesiscan_test.library.method_d.dsp import apply_tgc_pipeline +from vesiscan_test.library.method_d.config_d import DEFAULT_PARAMS as MD_PARAMS +from vesiscan_test.library import cross_channel +from vesiscan_test.library.runners import _ch_angles +from vesiscan_test.library.config_6ch import DEGREE, DISTANCE_PER_SAMPLE +import pandas as pd +import math + + +def load_records(path): + """새 스키마: {session, records[]}, 각 record 는 channels[6] 각각 data[100].""" + with open(path, encoding='utf-8') as f: + d = json.load(f) + session = d['session'] + recs_raw = d['records'] + out = [] + for r in recs_raw: + by_ch = {} + for c in r['channels']: + v = c.get('data') + if v is None or len(v) < 100: + continue + by_ch[int(c['ch'])] = np.asarray(v, dtype=float) + if len(by_ch) == 6: + sigs = np.stack([by_ch[i] for i in range(6)]) + out.append({ + 'row_index': r.get('row_index'), + 'timestamp': r.get('timestamp'), + 'command_type': r.get('command_type'), + 'sigs': sigs, + 'imu': r.get('sensor', {}).get('imu'), + }) + return session, out + + +def detect_base(sigs): + """cross-channel 보정 없이 base detection.""" + n_ch = sigs.shape[0] + heavy = np.stack([preprocess_heavy(sigs[i]) for i in range(n_ch)]) + light = np.stack([preprocess_light(sigs[i]) for i in range(n_ch)]) + heavy = apply_tgc_pipeline(pd.DataFrame(heavy), n_ch=n_ch).values.astype(float) + light = apply_tgc_pipeline(pd.DataFrame(light), n_ch=n_ch).values.astype(float) + results = [] + for i in range(n_ch): + angle_factor = math.cos(math.radians(DEGREE[i])) if i < len(DEGREE) else 1.0 + r = method_d_detect( + sigs[i], + denoised_heavy=heavy[i], + denoised_light=light[i], + otsu_ratio=MD_PARAMS.otsu_ratio * angle_factor, + params=MD_PARAMS, + ) + results.append(r) + return results, light, heavy + + +def detect_with_cross(sigs): + """cross-channel 보정 포함 detection.""" + results, light, heavy, _ = detect_walls_multichannel_full(sigs) + angles = _ch_angles(sigs.shape[0], None) + cross_channel.apply_cross_channel( + results, angles, light, heavy, sigs, + validate_top=True, inward_post=True) + return results + + +def summarize_session(path): + session, records = load_records(path) + print(f"\n{'='*100}") + print(f"Session : {session.get('session_name', os.path.basename(path)[:60])}") + print(f"Device : {session.get('device_name', '?')} ({session.get('device_id', '?')})") + print(f"Records : {len(records)} ({session.get('start_time', '?')} → {session.get('end_time', '?')})") + if not records: + print(" (empty session — skip)") + return None + print(f"{'='*100}") + + # per-record 진단 + rows = [] + for rec in records: + sigs = rec['sigs'] + # base + cross + base_results, light, heavy = detect_base(sigs) + cross_results = detect_with_cross(sigs) + + row = { + 'idx': rec['row_index'], + } + for i in range(6): + row[f'B{i}'] = 'X' if base_results[i] is None else '.' + row[f'C{i}'] = 'X' if cross_results[i] is None else '.' + if base_results[i] is not None: + r = base_results[i] + row[f'ant{i}'] = r.ant + row[f'post{i}'] = r.post + row[f'ulen{i}'] = int(max(0, r.post - r.ant - 1)) + row[f'lowamp{i}'] = float(r.low_amp) + rows.append(row) + df = pd.DataFrame(rows) + + # Detection matrix (base vs cross) + print(f"\n{'Detection matrix':-^80}") + print(f" X=missing, .=detected (Base = per-channel detect only, Cross = + cross-channel)") + cols_b = [f'B{i}' for i in range(6)] + cols_c = [f'C{i}' for i in range(6)] + print(f" {'idx':>3} base(0..5) cross(0..5)") + for _, r in df.iterrows(): + base_str = ''.join(r[cols_b].tolist()) + cross_str = ''.join(r[cols_c].tolist()) + marker = ' <== ch3 flick' if r['B3'] != r['C3'] else (' <== ch3 miss' if r['C3'] == 'X' else '') + print(f" {int(r['idx']):>3} {base_str} {cross_str} {marker}") + + # CH3 flicker stat + ch3_base_miss = (df['B3'] == 'X').sum() + ch3_cross_miss = (df['C3'] == 'X').sum() + n = len(df) + print(f"\n{'CH3 detection stats':-^80}") + print(f" Base : {n - ch3_base_miss}/{n} detected ({(n-ch3_base_miss)/n*100:.1f}%)") + print(f" Cross : {n - ch3_cross_miss}/{n} detected ({(n-ch3_cross_miss)/n*100:.1f}%)") + + # transitions (flicker rate) + seq_base = (df['B3'] == '.').astype(int).tolist() + seq_cross = (df['C3'] == '.').astype(int).tolist() + trans_base = sum(1 for i in range(1, n) if seq_base[i] != seq_base[i-1]) + trans_cross = sum(1 for i in range(1, n) if seq_cross[i] != seq_cross[i-1]) + print(f" Base flick transitions: {trans_base}") + print(f" Cross flick transitions: {trans_cross}") + + # per-record: full detection pattern + print(f"\n{'Per-channel detection rate (Cross)':-^80}") + for i in range(6): + det = (df[f'C{i}'] == '.').sum() + print(f" CH{i}: {det}/{n} ({det/n*100:.1f}%)") + + # urine_len distribution + print(f"\n{'Urine_len (base) — median / range per channel':-^80}") + for i in range(6): + col = f'ulen{i}' + if col not in df.columns: continue + vals = df[col].dropna() + if len(vals) > 0: + print(f" CH{i}: median={vals.median():.0f} min={vals.min():.0f} max={vals.max():.0f} n={len(vals)}") + + # CH3 low_amp + print(f"\n{'CH3 low_amp (base) — smaller = deeper lumen echo, larger = shallow/noisy':-^80}") + la = df['lowamp3'].dropna() + if len(la) > 0: + print(f" CH3 low_amp: median={la.median():.1f} min={la.min():.1f} max={la.max():.1f}") + + return df + + +def rolling_align_replay(path, accum=10, legacy=False): + """RollingAligner 를 새 Kotlin 로직 (majority + relaxed + soft-hint) 그대로 재현. + legacy=True 로 부르면 기존 (연속 3회) 로직.""" + from vesiscan_test.library.method_d.config_d import DEFAULT_PARAMS as MD_PARAMS + session, records = load_records(path) + + buf = [] + print(f"\n{'RollingAligner replay ' + ('LEGACY' if legacy else 'NEW'):-^90}") + print(f" window = {accum}") + print(f" {'idx':>3} {'phase':<12} {'relax':>5} {'chN':>3} {'CH3':>3} " + f"{'ch0..3 chord':>28} {'CV':>6} {'action':<24}") + + # ── 파라미터 (Kotlin AlignmentConstants 와 동일) ── + V_WIN, V_MAJ = 6, 3 + L_WIN, L_MAJ = 4, 3 + STUCK_TH = 20 + SOFT_AFTER = 12 + + phase = 'INITIAL' + initial_count = 0 + verticalHit = 0 # legacy 만 사용 + verticalHistory = [] # new: bool 링버퍼 + verticalStuck = 0 + stabilize = 0 + stabilizeLost = [] + centerOptStreak = 0 + centerOptLost = [] + relaxed = False + + def push(hist, v, w): + hist.append(v) + while len(hist) > w: hist.pop(0) + + for rec in records: + buf.append(rec['sigs']) + while len(buf) > accum: + buf.pop(0) + avg = np.mean(buf, axis=0) + results = detect_with_cross(avg) + detected = [r is not None for r in results] + ch3 = detected[3] + chords = [] + for ch in range(4): + r = results[ch] + if r is None: + chords.append(0.0) + else: + ulen = max(0, r.post - r.ant - 1) + chords.append(ulen * math.cos(math.radians(DEGREE[ch])) * DISTANCE_PER_SAMPLE) + + action = 'STOP' + if phase == 'INITIAL': + initial_count += 1 + action = 'wait' + if initial_count >= accum: + phase = 'VERTICAL' + + elif phase == 'VERTICAL': + if legacy: + if not ch3: + verticalHit = 0 + action = 'MOVE_UP' + else: + verticalHit += 1 + if verticalHit >= 3: + phase = 'CH3_STAB'; stabilize = 0 + action = 'STOP → CH3_STAB' + else: + action = f'STOP hit={verticalHit}/3' + else: + push(verticalHistory, ch3, V_WIN) + verticalStuck += 1 + hit_cnt = sum(verticalHistory) + # relaxed 진입 조건 + if (not relaxed and verticalStuck >= STUCK_TH + and hit_cnt < V_MAJ + and all(detected[i] for i in range(3))): + relaxed = True + phase = 'CENTER_OPT' + centerOptStreak = 0 + action = 'RELAXED ON → CENTER_OPT' + elif hit_cnt >= V_MAJ: + phase = 'CH3_STAB'; stabilize = 0 + verticalHistory.clear(); verticalStuck = 0 + action = f'STOP → CH3_STAB (hit {hit_cnt}/{V_WIN})' + else: + if verticalStuck >= SOFT_AFTER: + action = f'MOVE_UP soft (hit {hit_cnt}/{V_WIN})' + else: + action = f'MOVE_UP (hit {hit_cnt}/{V_WIN})' + + elif phase == 'CH3_STAB': + stabilize += 1 + if legacy: + if not ch3: + phase = 'VERTICAL'; verticalHit = 0 + action = 'MOVE_UP (lost)' + elif stabilize >= accum: + phase = 'CENTER_OPT'; centerOptStreak = 0 + action = 'STOP → CENTER_OPT' + else: + action = f'STOP {stabilize}/{accum}' + else: + push(stabilizeLost, (not ch3), L_WIN) + lost = sum(stabilizeLost) + if lost >= L_MAJ: + phase = 'VERTICAL'; stabilize = 0 + stabilizeLost.clear(); verticalHistory.clear(); verticalStuck = 0 + action = f'MOVE_UP true-lost ({lost}/{L_WIN})' + elif stabilize >= accum: + phase = 'CENTER_OPT'; centerOptStreak = 0; stabilizeLost.clear() + action = 'STOP → CENTER_OPT' + else: + action = f'STOP {stabilize}/{accum} flick {lost}/{L_WIN}' + + elif phase == 'CENTER_OPT': + center_range = range(3) if relaxed else range(4) + if not relaxed and not legacy: + push(centerOptLost, (not ch3), L_WIN) + lost = sum(centerOptLost) + if lost >= L_MAJ: + phase = 'CH3_STAB'; stabilize = 0; centerOptStreak = 0; centerOptLost.clear() + action = f'MOVE_DOWN true-lost ({lost}/{L_WIN})' + goto_next = True + elif not ch3: + centerOptStreak = 0 + action = f'STOP soft-flick ({lost}/{L_WIN})' + goto_next = True + else: + goto_next = False + elif not relaxed and legacy: + if not ch3: + phase = 'CH3_STAB'; stabilize = 0; centerOptStreak = 0 + action = 'MOVE_DOWN ch3 lost' + goto_next = True + else: + goto_next = False + else: + goto_next = False + + if not goto_next: + cr_list = list(center_range) + if not all(detected[i] for i in cr_list): + centerOptStreak = 0 + missing = [i for i in cr_list if not detected[i]] + top_only = all(m <= 1 for m in missing) if missing else False + action = f"MOVE_{'DOWN' if top_only else 'UP'} miss:{missing}" + else: + ch_used = [chords[i] for i in cr_list] + m = np.mean(ch_used) + if m <= 0.1: + action = 'MOVE_UP chord~0' + phase = 'VERTICAL'; verticalHistory.clear(); verticalStuck = 0 + else: + std = np.std(ch_used) + cv = std / m if m > 0 else 999 + if cv <= 0.15: + centerOptStreak += 1 + if centerOptStreak >= 3: + phase = 'LR_BAL' + action = f'STOP → LR (cv={cv:.2f})' + else: + action = f'STOP cv={cv:.2f} ({centerOptStreak}/3)' + else: + centerOptStreak = 0 + if ch_used[-1] > ch_used[0]: + action = f'MOVE_DOWN cv={cv:.2f}' + else: + action = f'MOVE_UP cv={cv:.2f}' + + elif phase == 'LR_BAL': + action = 'LR ...' + + chord_str = ' '.join(f'{c:5.1f}' for c in chords) + cv_val = 0 + cr_disp = range(3) if relaxed else range(4) + cr_list = list(cr_disp) + if all(detected[i] for i in cr_list) and np.mean([chords[i] for i in cr_list]) > 0.1: + cv_val = np.std([chords[i] for i in cr_list]) / np.mean([chords[i] for i in cr_list]) + print(f" {rec['row_index']:>3} {phase:<12} {('Y' if relaxed else '.'):>5} " + f"{sum(detected):>3} {'Y' if ch3 else 'N':>3} {chord_str:>28} " + f"{cv_val:>6.2f} {action:<24}") + + +if __name__ == '__main__': + import argparse + p = argparse.ArgumentParser() + p.add_argument('files', nargs='*') + p.add_argument('--replay', action='store_true') + p.add_argument('--legacy', action='store_true', help="legacy (연속 3회) 로직으로 replay") + p.add_argument('--both', action='store_true', help="new + legacy 둘 다 replay") + p.add_argument('--accum', type=int, default=10) + p.add_argument('--no-summary', action='store_true', help="detection matrix 등 요약 스킵") + args = p.parse_args() + + files = args.files + if not files: + files = sorted(glob.glob(r"c:/Projects/medilightv2android/data123/*.json")) + + for f in files: + if not args.no_summary: + df = summarize_session(f) + if df is None: continue + else: + _, records = load_records(f) + if not records: + print(f"\n== {os.path.basename(f)}: empty, skip") + continue + print(f"\n== {os.path.basename(f)}: {len(records)} records") + if args.replay or args.both: + if args.both: + rolling_align_replay(f, accum=args.accum, legacy=False) + rolling_align_replay(f, accum=args.accum, legacy=True) + else: + rolling_align_replay(f, accum=args.accum, legacy=args.legacy) diff --git a/tools/labdb_upload.py b/tools/labdb_upload.py new file mode 100644 index 0000000..8c8952c --- /dev/null +++ b/tools/labdb_upload.py @@ -0,0 +1,270 @@ +""" +labdb 세션 폴더 일괄 업로드 스크립트. + +앱이 만든 표준 세션 폴더 (measurement_.json 포함) 를 labdb REST API +로 업로드. app/src/main/java/com/medithings/vesiscan/services/labdb/ +LabdbUploader.kt 의 buildPayload 로직을 그대로 Python 으로 옮긴 것. + +사용법: + export LABDB_API_KEY=xbk_live_xxxxxxxxxxxxxxxx + python labdb_upload.py "C:/Users/장동우/Desktop/newdata" + +옵션: + --api-key KEY apiKey 를 인자로 직접 전달 (env 없이) + --dry-run 전송하지 않고 payload 만 검증 + --force 이미 labdb_upload.json 있어도 재업로드 + --data-type XXX meta.data_type 없을 때 fallback (default: 001) +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import sys +from datetime import datetime, timezone, timedelta +from pathlib import Path + +import urllib.request +import urllib.error + + +BASE_URL = "https://labdb.medithings.net/api/v1" +DEFAULT_DATA_TYPE = "001" + + +def sanitize_test_id(raw: str) -> str: + """^[A-Za-z0-9_-]{1,40}$ 강제. 40자 초과 시 앞 31자 + '_' + 8자 SHA-1.""" + cleaned = re.sub(r"[^A-Za-z0-9_-]", "_", raw) + if len(cleaned) <= 40: + return cleaned + h = hashlib.sha1(cleaned.encode()).hexdigest()[:8] + return cleaned[:31] + "_" + h + + +def iso_now_kst() -> str: + kst = timezone(timedelta(hours=9)) + return datetime.now(kst).strftime("%Y-%m-%dT%H:%M:%S.") + \ + f"{datetime.now(kst).microsecond // 1000:03d}+09:00" + + +def iso_from_ms(ms: int) -> str: + kst = timezone(timedelta(hours=9)) + dt = datetime.fromtimestamp(ms / 1000, tz=kst) + return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}+09:00" + + +def cycle_to_record(cycle: dict, meta: dict) -> dict: + row_index = cycle.get("cycle", 0) + ts_ms = cycle.get("timestamp_ms", 0) + piezo = cycle.get("piezo") or {} + imu_arr = cycle.get("imu") or [] + + latest = imu_arr[-1] if imu_arr else None + sensor = {} + if latest: + sensor["imu"] = {k: latest.get(k, 0.0) for k in ("ax", "ay", "az", "gx", "gy", "gz")} + sensor["imu_samples"] = imu_arr + sensor["imu_sample_count"] = len(imu_arr) + + channels = [] + for ch in range(6): + data = piezo.get(f"CH{ch}") + if data is None: + continue + peak = 0 + peak_idx = 0 + for i, v in enumerate(data): + if v > peak: + peak = v + peak_idx = i + channels.append({"ch": ch, "peak": peak, "peakIdx": peak_idx, "data": data}) + + record = { + "rowIndex": row_index, + "datetime": iso_from_ms(ts_ms), + "commandType": "MTB", + "sensor": sensor, + "channels": channels, + } + for k in ("align_phase", "align_score", "align_hint", "align_icon", + "align_detected", "align_urine_lens"): + if k in cycle: + record[k] = cycle[k] + return record + + +def build_payload(folder_name: str, measurement: dict, default_data_type: str) -> dict: + meta = measurement.get("meta") or {} + cycles = measurement.get("cycles") or [] + test_id = sanitize_test_id(folder_name) + data_type = (meta.get("data_type") or "").strip() or default_data_type + + params = { + "posture": meta.get("posture", ""), + "step": meta.get("step", ""), + "is_alignment": meta.get("is_alignment", False), + "examiner": meta.get("examiner", ""), + "app_version": meta.get("app_version_name", ""), + "captured_cycles": len(cycles), + } + for k in ("subject_type", "subject", "firmware_version", "hardware_version", + "serial_number"): + v = meta.get(k) + if isinstance(v, str) and v.strip(): + params[k] = v + for k in ("true_volume_ml", "abdomen_thickness_mm"): + if k in meta: + params[k] = meta[k] + + payload = { + "testId": test_id, + "dataType": data_type, + "sessionName": folder_name, + "memo": meta.get("notes", ""), + "params": params, + "recordCount": len(cycles), + "records": [cycle_to_record(c, meta) for c in cycles], + } + ended_at = meta.get("ended_at") + if isinstance(ended_at, int) and ended_at > 0: + payload["savedAt"] = iso_from_ms(ended_at) + return payload + + +def post_upload(payload: dict, api_key: str) -> dict: + url = f"{BASE_URL}/upload/json" + data = json.dumps(payload, ensure_ascii=False).encode("utf-8") + req = urllib.request.Request( + url, + data=data, + method="POST", + headers={ + "Content-Type": "application/json; charset=utf-8", + "X-API-Key": api_key, + }, + ) + try: + with urllib.request.urlopen(req, timeout=60) as resp: + body = resp.read().decode("utf-8") + return json.loads(body) + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + try: + err = json.loads(body) + except Exception: + err = {"raw": body} + raise RuntimeError(f"HTTP {e.code}: {err}") from e + + +def find_measurement_json(folder: Path) -> Path | None: + named = folder / f"measurement_{folder.name}.json" + if named.exists(): + return named + legacy = folder / "measurement.json" + if legacy.exists(): + return legacy + for p in folder.glob("measurement_*.json"): + return p + return None + + +def upload_folder(folder: Path, api_key: str, default_data_type: str, + dry_run: bool, force: bool) -> tuple[bool, str]: + marker = folder / "labdb_upload.json" + if marker.exists() and not force: + return True, "already uploaded (skip)" + + mfile = find_measurement_json(folder) + if mfile is None: + return False, "measurement json not found" + + try: + measurement = json.loads(mfile.read_text(encoding="utf-8")) + except Exception as e: + return False, f"read/parse fail: {e}" + + payload = build_payload(folder.name, measurement, default_data_type) + + if dry_run: + return True, (f"payload OK: testId={payload['testId']} " + f"dataType={payload['dataType']} records={payload['recordCount']}") + + try: + result = post_upload(payload, api_key) + except Exception as e: + err_body = { + "error": type(e).__name__, + "message": str(e), + "failedAt": iso_now_kst(), + } + (folder / "labdb_upload_error.json").write_text( + json.dumps(err_body, ensure_ascii=False, indent=2), encoding="utf-8") + return False, f"upload fail: {e}" + + marker_body = { + "ok": result.get("ok", True), + "sessionId": result.get("sessionId", ""), + "created": result.get("created", False), + "totalRecords": result.get("totalRecords", payload["recordCount"]), + "inserted": result.get("inserted", 0), + "duplicates": result.get("duplicates", 0), + "errors": result.get("errors", []), + "uploadedAt": iso_now_kst(), + } + marker.write_text(json.dumps(marker_body, ensure_ascii=False, indent=2), encoding="utf-8") + (folder / "labdb_upload_error.json").unlink(missing_ok=True) + return True, (f"uploaded: {marker_body['inserted']}/{marker_body['totalRecords']} " + f"(dup={marker_body['duplicates']})") + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("root", help="세션 폴더들이 있는 최상위 디렉토리") + ap.add_argument("--api-key", default=os.environ.get("LABDB_API_KEY"), + help="labdb apiKey (env LABDB_API_KEY 도 가능)") + ap.add_argument("--dry-run", action="store_true") + ap.add_argument("--force", action="store_true", + help="labdb_upload.json 있어도 재전송") + ap.add_argument("--data-type", default=DEFAULT_DATA_TYPE) + args = ap.parse_args() + + root = Path(args.root) + if not root.is_dir(): + print(f"ERROR: {root} is not a directory", file=sys.stderr) + return 2 + + if not args.dry_run and not args.api_key: + print("ERROR: LABDB_API_KEY 미설정. env 또는 --api-key 필요", file=sys.stderr) + return 2 + + folders = sorted(p for p in root.iterdir() if p.is_dir()) + if not folders: + print(f"no folders under {root}") + return 0 + + ok_count = 0 + skip_count = 0 + fail_count = 0 + for f in folders: + ok, msg = upload_folder(f, args.api_key or "", args.data_type, + args.dry_run, args.force) + marker = "OK " if ok else "FAIL" + print(f"[{marker}] {f.name}: {msg}") + if ok: + if "skip" in msg: + skip_count += 1 + else: + ok_count += 1 + else: + fail_count += 1 + + print(f"\nsummary: {ok_count} uploaded, {skip_count} skipped, {fail_count} failed " + f"(total {len(folders)})") + return 0 if fail_count == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main())