ディレクトリ構成の再検討#
ツールの構成メンバー#
モジュール名 |
役割 |
|---|---|
|
設定ファイル管理の担当者 |
|
データ取得実行の担当者 |
|
データ取得モニタリングの担当者 |
|
スレショルドスキャン実行の担当者 |
|
スレッショルド解析の担当者 |
|
生データから解析データ変換の担当者 |
|
データ描画の担当者 |
|
ロギングの担当者 |
|
データ保存の担当者 |
|
モックデータ作成の担当者 |
|
ヘルプ表示の担当者 |
|
コマンド統合の担当者 |
|
イベント管理の担当者 |
ディレクトリ構成案#
haniwers/
├── __init__.py
├── cli.py # CLI統括（commander）
├── config/ # 設定管理（configer）
│ ├── __init__.py
│ ├── loader.py
│ └── schema.py
├── daq/ # データ取得まわり
│ ├── __init__.py
│ ├── device.py # 物理接続とIO制御
│ │ ├── connect(config: DeviceConfig) -> serial.Serial # 標準の接続入口
│ │ ├── is_available(config: DeviceConfig) -> bool # ポートの確認
│ │ ├── read_line(device: serial.Serial) -> str # デコード、strip処理
│ │ ├── write_line(device: serial.Serial, data) -> bool # 閾値設定などで書き込み
│ │ ├── flush(device: serial.Serial) -> None # 測定前のバッファクリア
│ │ ├── with_timeout(device: serial.Serial, sec: float) -> Generator[serial.Serial, None, None] # 一時的なタイムアウト変更 @contextmanager として使う
│ │ ├── read_n_lines(device: serial.Serial, n: int) -> list[str] # option: 初期化時のガベージ除去など
│ │ ├── disconnect(device: serial.Serial) -> bool # option: 明示的に .close()
│ │ ├── get_port_description(device: serial.Serial) -> str # option: ログやGUI表示に便利
│ │ ├── get_status(device: serial.Serial) -> dict # option: 通信状況や設定確認用（高度）
│ ├── sampler.py # イベント収集
│ │ ├── sample(device: serial.Serial) -> RawEvent
│ │ ├── for_events(device: serial.Serial, n: int) -> list[RawEvent]
│ │ ├── for_duration(device: serial.Serial, t: float) -> list[RawEvent]
│ │ ├── to_csv(events: list[RawEvent], path: Path) -> bool
│ │ ├── create_writer(path: Path) -> csv.writer # option: to_csvの内部処理
│ │ ├── append_events(writer: csv.writer, events: list[RawEvent]) -> None # option: to_csvの内部処理
│ │ ├── tqdm_wrapper(iterator, total: int | None) -> Iterator[RawEvent] # option: 進捗バー付きのイテレータラッパー
│ │ ├── sanitize(event: RawEvent) -> RawEvent # option: RawEvent の整形・欠損チェック・補正など
│ │ ├── parse_line(line: str) -> RawEvent # option: 1行の生データをRawEventにパースする
│ │ ├── mock_sample(index: int = 0) -> RawEvent # option: モックしたRawEventを取得
│ │ ├── run_session(device: serial.Serial, config: DaqConfig | ScanConfig) -> None # option: 設定に従ってDAQランを実行・ファイル出力する
│ ├── mocker.py # 開発用のデバイス模擬
│ │ ├── MockSerial(csv_path: Path, shuffle: bool = False, speed: float = 1.0) # CSVベースの再現モック（旧FileBackedMockSerial）
│ │ ├── RandomMockSerial(rate: float = 1.0, jitter: float = 0.0) # ランダムに行を生成（旧MockSerial）
│ │ ├── create_mock_device(config: DeviceConfig) -> serial.Serial # config.port に応じて適切なモックを生成
│ │ ├── create_mock_from_file(path: Path, **kwargs) -> MockSerial # 明示的にCSVモックを作成
│ │ ├── load_csv_lines(path: Path) -> list[str] # CSVファイルから行データを抽出
│ │ ├── parse_timestamps(lines: list[str], column: int) -> list[datetime] # ISO8601時刻列をdatetime型に変換
│ │ ├── compute_time_deltas(times: list[datetime]) -> list[float] # イベント間の秒数差を計算
│ │ ├── is_mock_port(port: str) -> bool # ポート名がモック用かどうか判定
│ │ ├── inject_noise(line: str, col_indices: list[int], stddev: float = 0.5) -> str # 指定列にノイズを加える
│ │ ├── drop_columns(line: str, col_indices: list[int]) -> str # 特定列を除去して行を再構成
├── thresholds/ # スレッショルドスキャン関係
│ ├── __init__.py
│ ├── scanner.py # スレショルドスキャンを実行
│ │ ├── scan_channel(device: serial.Serial, channel: str, config: SensorConfig, duration: float) -> list[tuple[int, int]] # Returns [ (threshold, count), ...]
│ │ ├── scan_all_channels_serial(device: serial.Serial, sensors: dict[str, SensorConfig], duration: float, mask_value: int = 500) -> dict[str, list[tuple[int, int]]]
│ │ ├── scan_all_channels_parallel(device: serial.Serial, sensors: dict[str, SensorConfig], duration: float) -> dict[str, list[tuple[int, int]]]
│ │ ├── to_csv(scan_result: dict[str, list[tuple[int, int]]], path: Path) -> None # スキャン結果をCSV形式で保存
│ │ ├── from_csv(path: Path) -> dict[str, list[tuple[int, int]]] # スキャン結果をCSVから読み込む
│ │ ├── _set_threshold(device: serial.Serial, channel: str, value: int) -> None
│ │ ├── _count_hits(device: serial.Serial, channel: str, duration: float) -> int
│ ├── fitter.py # スキャン結果をフィットしてスレッショルド値を計算
│ │ ├── fit_scan_curve(x: list[float], y: list[int]) -> FittingResult
│ │ ├── estimate_thresholds(result: FittingResult, sigmas: list[float] = [0, 1, 3, 5]) -> dict[str, float]
│ │ ├── plot_fit_result(x: list[float], y: list[int], result: FittingResult, save_path: Optional[Path] = None) -> None
│ │ ├── FittingResult
│ ├── applier.py # チャンネルごとにスレッショルドを設定（apply）
│ │ ├── from_csv(path: Path) -> dict[str, int] # CSVからチャンネルごとの閾値を読み込む
│ │ ├── to_csv(thresholds: dict[str, int], path: Path) -> None # 設定した値をCSVに書き込む（読み込みとは別のログ用）
│ │ ├── apply_thresholds(device: serial.Serial, thresholds: dict[str, int]) -> bool # 複数チャンネルに順番に閾値を設定
│ │ ├── apply_threshold(device: serial.Serial, channel: str, value: int) -> bool # 単一チャンネルの閾値設定
│ │ ├── verify_threshold(device: serial.Serial, channel: str) -> int # 設定されている閾値を確認（デバイスが対応していないかも）
│ │ ├── interactive(device: serial.Serial, config: HaniwersConfig) -> None # 対話的に適用
│ └── writer.py # スレッショルド関係のファイル出力
├── preprocess/ # 前処理・変換（converter）
│ ├── __init__.py
│ └── converter.py
├── viz/ # 可視化まわり（plotter）
│ ├── __init__.py
│ └── plotter.py
├── log/ # ログ（logger）
│ ├── __init__.py
│ └── logger.py
├── monitor/
│ ├── watcher.py # watchdogによるファイル監視
│ │ ├── watch_csv_file
│ ├── parser.py # RawEvent / ProcessedEvent の生成
│ │ ├── parse_raw_event
│ │ ├── process_event
│ ├── dashboard.py # streamlit or panel による可視化
│ │ ├── hvplot + panel
│ │ ├── hvplot + streamlit
│ ├── state.py # 共有データ構造・キャッシュ管理
│ │ ├── Shared State
│ ├── remote.py # WebSocket通知でリアルタイム配信
│ ├── summary.py # 統計的処理や異常検知の自動化
│ ├── alerts.py # 閾値越えをSlack等に通知