haniwers.config

haniwers.config#

Configuration Management for Haniwers.

This module handles loading, parsing, and storing configuration settings needed to run the haniwers tools. It converts TOML configuration files into Python objects that are easy to work with.

What is Configuration?

Configuration describes how your equipment and experiments are set up:

  • Which detector port/device to connect to

  • How often to sample data (sampling interval)

  • Which detector channels are active

  • Where to store output files

  • Threshold values and other parameters

Key Classes

RunData : Experiment metadata for a single measurement run Stores run ID, description, data location, and processing parameters

Daq : Data acquisition configuration Specifies detector connection, sampling rate, and channel settings

UserSettings : User-specific preferences Stores paths to config directories and personal preferences

Configuration File Format

Configuration uses TOML files (.toml), which are human-readable text files:

[daq]
port = "/dev/ttyUSB0"
baudrate = 115200
interval = 60
channels = [1, 2, 3]

Example Usage

from haniwers.config import UserSettings, Daq

settings = UserSettings.load()  # Load from config directory
daq = Daq.from_toml("config.toml")  # Load DAQ settings
print(f"Connecting to {daq.port} at {daq.baudrate} baud")

For Beginners

If you’re new to configuration:

  1. Look at example configs in examples/ directory

  2. Copy one to your working directory

  3. Edit values to match your setup

  4. Pass the config file path to haniwers commands

Module Contents#

Classes#

RunData

ランごとの設定

Config

（削除予定）ラン設定ファイル用のクラス

Daq

DAQ設定クラス

UserSettings

ユーザー設定用のクラス

RunManager

ランの設定をスプレッドシートから読み込む

API#

class haniwers.config.RunData#

ランごとの設定

それぞれのランの情報／設定を整理するためのクラスです。 ランを走らせた時の設定は、Googleスプレッドシートで管理しています （更新・共有の手軽さを考慮してスプレッドシートを採用しました）。

TODO:

  • スプレッドシートをCSV形式でダウンロードして、RunDataオブジェクトに変換するツールを作りたい

run_id: int#

None

ラン番号

description: str = <Multiline-String>#

ランの簡単な説明。詳しい説明はGoogleスプレッドシートで説明

read_from: str = <Multiline-String>#

測定データが保存されているディレクトリ名

srcf: str#

‘*.csv’

測定データの拡張子。デフォルトはCSV形式

interval: int#

600

サンプリング間隔（秒）。デフォルトは600秒（10分）

datetime_offset: int#

0

測定器の時刻と実時刻の時間差（秒）。デフォルトは0秒

skip: bool#

False

データ処理スキップのフラグ。デフォルトはFalse

nfiles: int#

0

測定データのファイル数。デフォルトは0

raw2gz: str = <Multiline-String>#

出力ファイル名。リサンプルなし。指定がない場合は保存しない

raw2csv: str = <Multiline-String>#

出力ファイル名。リサンプルあり。指定がない場合は保存しない

query: str = <Multiline-String>#

データの抽出条件。有効なデータを指定する

__post_init__() → None#

RunDataクラスの初期化

get_fnames() → list[pathlib.Path]#

該当するランのファイル名の一覧を取得

read_fromとsrcfの値から、該当するファイルの一覧を、 pathlib.Pathオブジェクトのリストとして取得します。

Exception:

  • read_fromディレクトリが存在しない場合は終了

Returns:

  • fnames (list[Path]): ファイル名のリスト

_load_gzip() → Optional[pandas.DataFrame]#

gzipで保存したデータを読み込む

raw2gzで指定したファイル名をpd.DataFrameに変換します。

Exception:

  • raw2gzファイルが存在しない場合は終了

Returns:

  • data (pd.DataFrame | None): データフレーム

_load_csv() → Optional[pandas.DataFrame]#

csvで保存したデータを読み込む

raw2csvで指定したファイルを``pd.DataFrame`に変換します。

Exception:

  • raw2csvファイルが存在しない場合は終了

Returns:

  • data (pd.DataFrame | None) : データフレーム

load_data(kind: str) → Optional[pandas.DataFrame]#

形式を指定してデータを読み込む

kindで測定データの読み取り形式を指定します。 指定できる形式はcsvもしくはgzipです。

Args:

  • kind (str): 保存したデータの形式。[“csv”, “gzip”]

Returns:

  • data (pd.DataFrame): データフレーム

class haniwers.config.Config#

（削除予定）ラン設定ファイル用のクラス

TODO:

  • RunManagerクラスを別途作成した

  • 重複する箇所は置き換える

fname: str#

‘config.toml’

設定ファイル。デフォルトはconfig.toml

__post_init__() → None#

（削除予定）

  • self.config

  • self.rules

  • self.runs

load_config() → dict#

（削除予定）設定ファイルを読み込む

Returns:

  • config(dict) : 設定

Notes:

  • デフォルトの設定ファイル名は config.toml

  • 設定ファイルの名前は変更することができる

  • 設定ファイルが見つからない場合は、エラーを表示してスキップ（早期リターン）する

  • 設定ファイルは辞書型で読み込む

get_rules() → dict#

（削除予定）イベント条件に関する設定を取得する

Returns:

  • rules(dict): {イベント名 : イベント条件}

Notes:

  • 設定ファイルの [rules] セクションの内容を取得する

  • 条件名 = 条件式 の辞書型（map型）で定義されている

get_labels() → Optional[dict]#

（削除予定）カラム名を取得する

[labels] のセクションに、カラム名に対応した日本語を記述する

get_run(run_id: int) → haniwers.config.RunData#

（削除予定）ラン番号を指定してラン情報を取得する

get_runs() → list[haniwers.config.RunData]#

（削除予定）ランに関係する設定を取得する

Returns:

  • runs(list[RunData]): ランごとに設定をまとめたリスト

Notes:

  • 設定ファイルの [[rawdata]] セクションの内容を RunData クラスにまとめる

  • 除外するランがある場合は skip = true を設定する

class haniwers.config.Daq#

DAQ設定クラス

Example:

daq = Daq()
daq.load_toml(fname="TOML形式のファイル")

# 現在日時をベースにした保存先に変更
now = pendulum.now().format("YYYYMMDD")
daq.saved =  str(Path(daq.saved) / now)
saved: str#

‘.’

測定データを保存するディレクトリ。デフォルトは.（カレントディレクトリ）

prefix: str#

‘data’

測定データのファイル名につける接頭辞。デフォルトはdata

suffix: str#

‘.csv’

測定データの拡張子。デフォルトは.csv

skip: int#

10

max_rows: int#

10000

1ファイルあたりのデータ数。デフォルトは10000

max_files: int#

100

1ランあたりのファイル数。デフォルトは100

quiet: bool#

False

quietモード用フラグ。デフォルトはFalse

append: bool#

False

追記モード用フラグ。デフォルトはFalse

device: str#

‘/dev/ttyUSB0’

シリアル通信のポート名。デフォルトは/dev/ttyUSB0（Linux用）

baudrate: int#

115200

シリアル通信レート（ボーレート）。デフォルトは115200 [bps]

timeout: int#

1000

シリアル通信のタイムアウト設定（秒）。デフォルトは1000秒

fname_logs: str#

‘threshold_logs.csv’

ファイル名。スレッショルド設定のログ

fname_scan: str#

‘threshold_scan.csv’

ファイル名。スレッショルド測定の結果

load_toml(fname: str) → None#

Load DAQ configuration from TOML

class haniwers.config.UserSettings(/, **data: Any)#

Bases: pydantic.BaseModel

ユーザー設定用のクラス

read_from に指定したファイルからユーザー設定を読み込みます。 設定ファイルはTOML形式で作成してください。 その他の形式に対応する予定はいまのところありません。 また、読み込み時のファイル形式のチェックもしていません。

TODO:

  • daq.toml、scan.tomlからhnw.tomlに移行する

  • クラス名をDaqManagerに変更する

Versionadded:

0.12.0

Example:

us = UserSettings(load_from="../sandbox/hnw.toml")
us.settings
{
    'default': {'saved': '', 'suffix': '.csv', 'skip': 10, 'max_rows': 1000},
    'device': {'port': '/dev/ttyUSB0', 'baudrate': 115200, 'timeout': 100},
    'daq': {'prefix': 'osechi_data', 'max_files': 1000},
    'scan': {'prefix': 'scan_data', 'max_files': 10, 'timeout': 10},
    'threshold': {'logs': {'fname': 'threshold_logs.csv',
    'names': ['time', 'ch', 'vth', 'success']},
    'scan': {'fname': 'threshold_scan.csv',
    'names': ['time', 'duration', 'ch', 'vth', 'events']
}

us.sections
dict_keys(
    ['default', 'device', 'daq', 'scan', 'threshold', 'loguru']
)

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

load_from: str#

None

設定ファイルを指定する

settings: dict#

None

読み込んだ設定値の一覧

sections: list#

[]

読み込んだ設定セクションの一覧

model_post_init(__context: Any) → None#

UserSettingsクラスの初期化

load_toml(load_from: str) → dict#

TOML形式の設定を読み込む

UserSettingsクラスのオブジェクトを生成するときに __post_init__の中で実行している。 プロダクション環境では、このメソッドをわざわざ実行する必要はありません。

新しい設定ファイルを作成した場合の内容確認のために使えると思います。

Args:

  • load_from (str): ファイル名

Returns:

  • settings (dict): ユーザー設定

_get_settings(keys: list) → dict#

キーを指定してユーザー設定を取得

キーはTOMLファイルのセクションに対応しています。 キーは複数指定できます。 同じ設定項目がある場合、あとに設定した値が優先されます。

指定したキーが設定ファイルに存在しない場合はスキップします。

Args:

keys (list): 設定キーをリストで指定する

Returns:

settings (dict): 設定値の一覧

Example:

us = UserSettings(load_from="../sandbox/hnw.toml")
keys = ["default", "device", "scan", "threshold"]
settings = s._get_settings(keys)
get_daq_settings() → dict#

DAQの設定を取得する

Note:

keys = ["default", "device", "daq"]に相当

Returns:

  • settings (dict): DAQの設定に必要な項目

Example:

us = UserSettings(load_from="../sandbox/hnw.toml")
settings = us.get_daq_settings()
settings
{
    'saved': '',
    'suffix': '.csv',
    'skip': 10,
    'max_rows': 1000,
    'port': '/dev/ttyUSB0',
    'baudrate': 115200,
    'timeout': 100,
    'prefix': 'osechi_data',
    'max_files': 1000
}
get_scan_settings() → dict#

スレッショルド測定の設定を取得

Note:

keys = ["default", "device", "scan", "threshold"]に相当

Returns:

  • settings (dict): スレッショルド測定に必要な項目

Example:

us = UserSettings(load_from="../sandbox/hnw.toml")
settings = us.get_scan_settings()
settings
{
    'saved': '',
    'suffix': '.csv',
    'skip': 10,
    'max_rows': 1000,
    'port': '/dev/ttyUSB0',
    'baudrate': 115200,
    'timeout': 10,
    'prefix': 'scan_data',
    'max_files': 10,
    'logs': {'fname': 'threshold_logs.csv', 'names': ['time', 'ch', 'vth', 'success']},
    'scan': {'fname': 'threshold_scan.csv', 'names': ['time', 'duration', 'ch', 'vth', 'events']},
    'fit': {'fname': 'threshold_fit.csv', 'names': ['time', 'ch', '0sigma', '1sigma', '3sigma', '5sigma']}
}
get_loguru(level: str = 'DEBUG') → loguru.logger#

ロガーを初期化する

loguruパッケージを使ったロガーを初期化します。 level オプションで指定したログレベルで、標準エラー出力（sys.stderr）のハンドラーを作成します。

以下のような [loguru] セクションを作成することで、ファイル出力のハンドラーを追加できます。 このハンドラーのログレベルはDEBUG、出力形式はJSON形式にハードコードしています。

[loguru]
sink = "required"
format = "required"
retention = "optional"
rotation = "optional"
Args:

  • level (str): 標準エラー出力のログレベル。デフォルトは"DEBUG"。

Returns:

  • logger (loguru.logger): ロガー

Example:

us = UserSettings(load_from="../sandbox/hnw.toml")
logger = us.get_loguru()
logger.info("ロガーを初期化した")
class haniwers.config.RunManager#

ランの設定をスプレッドシートから読み込む

load_from: str#

None

ファイル名。データを記録したスプレッドシートを指定する

query: str#

‘run_id > 0’

クエリ。データを読み込む条件を指定する

drive: str#

‘data’

ディレクトリ名。上記ファイルまでの相対パスを指定する

is_valid: bool#

True

データ用フラグ。スレッショルド測定の解析時はFalseにする

is_test: bool#

False

テスト用フラグ。ユニットテストで実行するときはTrueにする

__post_init__()#
_load_data() → pandas.DataFrame#

ファイルから設定を読み込む

get_records(query: str, is_valid: bool) → pandas.DataFrame#

条件にマッチしたレコードを取得

queryにマッチしたレコードを取得します。

Args:

  • query (str): クエリ条件

  • is_valid (bool) : 有効データのフラグ。 Defaults to True

Returns:

  • matched (pd.DataFrame): クエリにマッチしたデータフレーム

Example:

rm = RunManager("./_data/run.csv")
rm.get_records("run_id==1")
rm.get_records("run_type=='古墳'")
rm.get_records("50 <= run_id <= 84")
rm.get_records("50 <= run_id <= 84", is_valid=False)
rm.get_records("50 <= run_id <= 84 and run_type='テスト'")
_to_rundata(row: pandas.DataFrame) → haniwers.config.RunData#
_get_runs(is_valid: bool) → list[haniwers.config.RunData]#
get_run(run_id: int) → haniwers.config.RunData#

Get RunData.

ラン番号（run_id）を指定して、RunData情報を取得する。

例外:

KeyError --

  • ラン番号が見つからない場合

目次