dw.jang/VesiscanClinicalAndroid
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(tools): align_analyze.py 이관 + CH3 flicker fix Python replay 검증 결과

Browse Source 
- 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 오진입 없음)
This commit is contained in:
jjangddu
2026-07-13 15:21:07 +09:00
parent 43ad32ba8a
commit eb25c01c97
4 changed files with 740 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
VesiScan_Android_Pipeline_Summary.md
+7 -1
View File
@@ -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) /
tools/README.md
+78
View File
@@ -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)
...
```
tools/align_analyze.py
+385
View File
@@ -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)
tools/labdb_upload.py
+270
View File
@@ -0,0 +1,270 @@
"""
labdb 세션 폴더 일괄 업로드 스크립트.
앱이 만든 표준 세션 폴더 (measurement_<folder>.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())