haniwers.v0.daq

`haniwers.v0.daq`#

Module Contents#

Classes#

RealEvent

実イベント

Functions#

`init_saved`	（削除予定）
`mkdir_saved`	データを保存するディレクトリを作成
`get_savef`	（削除予定）データを保存するファイル名を生成する
`get_savef_with_timestamp`	データを保存するファイル名を生成する（時刻付き）
`open_serial_connection`	シリアル通信を開始する
`write_vth`	Write threshold to individual channel.
`set_vth`	Set threshold to individual channel.
`set_vth_retry`	Set threshold with retries
`_read_event`	Read single event from the opened port.
`_loop_events_for_rows`	イベント取得ループ（回数指定）
`_loop_events_for_duration`	イベント取得ループ（時間指定）
`loop_and_save`	イベント保存ループ
`events_to_dataframe`	測定データをデータフレームに変換する
`run_daq`	Run DAQ（回数指定）
`scan_daq`	Run DAQ（時間指定）
`run`	メインのDAQ
`time_daq`	（削除予定）測定時間を指定してDAQを走らせます。
`scan_ch_vth`	（削除予定）Run threshold scan.
`scan_ch_thresholds`	（削除予定）Run threshold scan.
`_read_serial_data_as_list`	（削除予定）Read serial data from port.
`read_serial_data`	（削除予定）
`_loop_and_save_events`	（削除予定）イベント保存ループ
`loop_and_save_events`	（削除予定）イベント保存ループ
`save_serial_data`	（削除予定）

API#

class haniwers.v0.daq.RealEvent(/, **data: typing.Any)#

Bases: pydantic.BaseModel

実イベント

OSECHIに接続したUSBポートから、シリアル通信で受け取った値を格納するためのデータクラス。ファイルに書き出したり、pandas.DataFrameに変換できるように自作メソッドを追加。

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.

timestamp: datetime.datetime#

‘now(…)’

測定時刻。宇宙線イベントが通過した日時。タイムゾーン付きの日付オブジェクト

top: int#

topレイヤーのヒット。0 - 10の値

mid: int#

midレイヤーのヒット。0 - 10の値

btm: int#

btmレイヤーのヒット。0 - 10の値

adc: int#

topレイヤーにヒットがあったときのADC値。0 - 1023の値

tmp: float#

0.0

BME280で測定した気温。[degC]

atm: float#

0.0

BME280で測定した気圧。[Pa]

hmd: float#

0.0

BME280で測定した湿度。[%]

to_list_string() → list[str]#

メンバー変数を文字列にしたリストに変換

Returns:: list (str): データ文字列のリスト

>>> real_data = RealEvent()
>>> read_data.to_list_string()
['2024-05-21 08:44:20.389786+09:00', '0', '0', '0', '0', '0', '0', '0']

to_csv_string() → str#

Comma Separated Values

メンバー変数をCSV形式（カンマ区切り）の文字列に変換します。 OSECHIから受け取ったデータを、ファイルに保存する際に使うことを想定したメソッドです。

Returns:: str: CSV形式の文字列

>>> real_data = RealEvent()
>>> real_data.to_csv_string()
'2024-05-21 08:44:20.389786+09:00,0,0,0,0,0,0,0'

to_tsv_string() → str#

Tab Separated Values

メンバー変数をTSV形式（タブ区切り）の文字列に変換します。

Returns:: str: TSV形式の文字列

>>> real_data = RealEvent()
>>> real_data.to_tsv_string()
'2024-05-21 08:44:20.389786+09:00       0       0       0       0       0       0       0'

to_ltsv_string() → str#

Labeled Tab-Separated Values

メンバー変数をLTSV形式（ラベルありのタブ区切り）の文字列に変換します。

Returns:: str: LTSV形式の文字列

>>> real_data = RealEvent()
>>> real_data.to_ltsv_string()
'timestamp:2024-05-21 08:44:20.389786+09:00     top:0   mid:0   btm:0   adc:0   tmp:0   atm:0   hmd:0'

haniwers.v0.daq.init_saved(daq: haniwers.v0.config.Daq) → None#: （削除予定）

haniwers.v0.daq.mkdir_saved(daq: haniwers.v0.config.Daq) → None#

データを保存するディレクトリを作成

データを保存するディレクトリを初期化します。ディレクトリが存在しない場合は、新しく作成します。ディレクトリが存在する場合は、そのままにします。

Args:

daq(Daq): DAQ設定オブジェクト

Returns:

None

haniwers.v0.daq.get_savef(args: haniwers.v0.config.Daq, fid: int | str) → pathlib.Path#

（削除予定）データを保存するファイル名を生成する

保存するファイル名を生成します。 DAQ設定の接頭辞（prefix）と拡張子（suffix）の値を使って {prefix}_{連番:06}.{suffix}の形式で生成します。

Args:

args(Daq): DAQ設定オブジェクト
- prefix: ファイルの接頭辞
- suffix: ファイルの拡張子
n(int): ファイル番号

Example:

osechi_data_000000.csv
osechi_data_000001.csv
osechi_data_000002.csv

haniwers.v0.daq.get_savef_with_timestamp(args: haniwers.v0.config.Daq, fid: int | str) → pathlib.Path#

データを保存するファイル名を生成する（時刻付き）

作成日を含んだファイル名を生成する。ファイル名は、DAQ設定の接頭辞（prefix）、ファイルを開いた時刻（pendulum.now）と拡張子（suffix）の値を使って生成する。

時刻のフォーマットは、ファイル名が分かりやすいように独自フォーマットにした。

Args:

args (Daq): Daqオブジェクト
fid (int|str): ファイル識別子

Returns:

savef (Path): ファイル名（Pathオブジェクト）

Examples:

ファイル数の上限を指定して、Pathオブジェクトを生成

max_files = 100
for nfile in range(max_files):
    savef = get_savef_with_timestamp(daq, nfile)
    # savefを使ったファイル処理

Examples:

DAQを走らせると生成されるファイル名のサンプル

20240520/osechi_data_2024-05-20T12h34m56s_000000.csv
20240520/osechi_data_2024-05-20T13h53m24s_000001.csv
20240520/osechi_data_2024-05-20T14h46m23s_000000.csv  // DAQを走らせ直すとリセット
20240520/osechi_data_2024-05-20T14h36m32s_000001.csv

haniwers.v0.daq.open_serial_connection(daq: haniwers.v0.config.Daq) → serial.Serial#

シリアル通信を開始する

シリアル通信（UART）に使うポートを準備します。 with構文で使う想定です。

通信に使うUSBポート名（device）、ボーレート（baudrate）、通信開始／書き込みのタイムアウト秒（timeout）は DAQ用の設定ファイルで変更できるようにしてあります。

Args:

daq (Daq): DAQ設定オブジェクト
- device: USBポート名
- baudrate: ボーレート（通信速度）[bps]
- timeout: タイムアウト秒 [sec]

Returns:

port (serial.Serial): 通信を開始したSerialオブジェクト

Example:

with open_serial_connection(daq) as port:
    # データ測定の処理
    success = write_vth(port, ch, vth)

haniwers.v0.daq.write_vth(port: serial.Serial, ch: int, vth: int) → bool#

Write threshold to individual channel.

シリアル通信（UART）を使って、チャンネルに閾値を書き込みます。 ESP32のバッファサイズの制限から、閾値は2回に分割して転送しています。

val1 は vth を右に6ビットシフト（=64で割る）して、head を足した値
val2 は vth を左に2ビットシフト（=4をかける）して、下位8ビットを取り出した値

詳細はこの関数のソースコードを確認してください。

閾値を書き込んだあとに、値が読み出せるかを確認します。レスポンスは以下の2つの形式に対応しています：

JSON形式: {"type":"response","status":"ok"/"error","channel":ch,...}
レガシー形式: dame （失敗）またはチャンネル番号 1 / 2 / 3 （成功）

Args:

port (serial.Serial): 接続ポート（Serialオブジェクト）
ch (int): 閾値を設定するチャンネル番号
vth (int): 閾値

Returns:

success (bool): 閾値を書き込んだ結果

Examples:

チャンネルごとに閾値を設定

success = write_vth(port, 1, 270)
success = write_vth(port, 2, 281)
success = write_vth(port, 3, 297)

TODO:: ユーザーが直接あつかわない関数でよいので、内部変数（の命名規則）に変更する（write_vth -> _write_vth）

haniwers.v0.daq.set_vth(daq: haniwers.v0.config.Daq, ch: int, vth: int) → bool#

Set threshold to individual channel.

シリアル通信を開始し、チャンネル番号を指定して閾値を設定します。チャンネル番号は1 - 3 の範囲で指定してください。閾値は1 - 1023 の範囲で指定してください。

書き込みに成功すると success=True を返します。書き込みに失敗した場合は、警告メッセージを表示します。このとき、設定済みの閾値はそのままになります。

Args:

daq (Daq): DAQ設定オブジェクト
ch (int): チャンネル番号。1 - 3の範囲で指定してください
vth (int): 閾値の値。1 - 1023 の範囲で指定してください

Returns:

success (bool): 閾値を書き込んだ結果

Exceptions:

チャンネル番号が範囲外の場合は終了
閾値が範囲外の場合は終了
serial.SerialException: シリアル通信ができなかった場合は終了
Exception: その他の予期せぬエラーの場合も終了

Examples:

success = set_vth(daq, 1, 280)
success = set_vth(daq, 2, 280)
success = set_vth(daq, 4, 280)  # チャンネル番号が範囲外
success = set_vth(daq, 1, 2000) # 閾値が範囲外

haniwers.v0.daq.set_vth_retry(daq: haniwers.v0.config.Daq, ch: int, vth: int, max_retry: int) → bool#

Set threshold with retries

閾値の書き込みに失敗した場合、成功するまで max_retry回繰り返します。

Args:

daq (Daq): Daqオブジェクト
ch (int): チャンネル番号
vth (int): スレッショルド値
max_retry (int): リトライする最大回数

Returns:

success (bool): 閾値を書き込んだ結果

Examples:

# 5回繰り返す
success = set_vth_retry(daq, 1, 280, 5)
# 10回繰り返す
success = set_vth_retry(daq, 2, 290, 10)

haniwers.v0.daq._read_event(port: serial.Serial) → haniwers.v0.daq.RealEvent#

Read single event from the opened port.

シリアル接続しているポートから1イベント分のデータを読み出します。引数に指定するポートは、接続済みのポートを渡してください。読み出したデータは適切な型に変換して、RealEventオブジェクトに代入します。

Note:

run_daqやtime_daqでデータを取得するために使います。

Args:

port (serial.Serial): 接続済みのSerialオブジェクト

Returns:

event (RealEvent): 1イベント分のデータ。読み出した時刻も自動で追加される。

Examples:

with open_serial_connection() as port:
    event = _read_event(port)
print(event)
# [pendulum.now(), top, mid, btm, adc, tmp, atm, hmd]

haniwers.v0.daq._loop_events_for_rows(port: serial.Serial, max_rows: int) → Generator[haniwers.v0.daq.RealEvent, None, None]#

イベント取得ループ（回数指定）

測定回数を指定してデータを読み出します。ジェネレーター関数になっています。

Args:

port (serial.Serial): 接続済みのSerialオブジェクト
max_rows (int): 測定回数

Yields:

event (RealEvent): 1イベント分の測定データ（RealEventオブジェクト）

Example:

# 100回測定する
for event in _loop_events_for_rows(port, 100):
    print(event.to_ltsv_string())

haniwers.v0.daq._loop_events_for_duration(port: serial.Serial, max_duration: int) → Generator[haniwers.v0.daq.RealEvent, None, None]#

イベント取得ループ（時間指定）

測定時間を指定してデータを読み出します。ジェネレーター関数になっています。

Args:

port (serial.Serial): 接続済みのSerialオブジェクト
max_duration (int): 測定時間（秒）

Yields:

event (RealEvent): 1イベント分の測定データ（RealEventオブジェクト）

Example:

# 10秒間測定する
for event in _loop_events_for_rows(port, 10):
    print(event.to_ltsv_string())

haniwers.v0.daq.loop_and_save(fname: pathlib.Path, generator: Generator) → list[haniwers.v0.daq.RealEvent]#

イベント保存ループ

指定したファイルに、1イベントずつファイルに書き足して保存します。

保存形式はファイル名の拡張子で判定します。有効な拡張子は[".csv", ".dat", ".tsv", ".json", ".jsonl"]です。

イベントの取得方法はgeneratorで指定します。有効なジェネレーターは_loop_events_for_rowsと_loop_events_for_durationです。

Args:

fname (Path): データを追記するファイル名
generator (Generator): データの取得方法。[_loop_events_for_rows, _loop_events_for_duration]から選択。

Returns:

events (list[RealEvent]): 測定したデータのリスト

haniwers.v0.daq.events_to_dataframe(events: list[haniwers.v0.daq.RealEvent]) → pandas.DataFrame#

測定データをデータフレームに変換する

測定データは、データを取得するたびにファイルに書き出してますが、同時にRealEventオブジェクトのリストとしてもストアしています。このままだと使いにくいので、データフレームに変換できるようにしました。そのときに、preprocess.add_hitとpreprocess.add_hit_typeの処理もしています。測定時刻の調整やリサンプル処理はしていません。

Args:

events (list[RealEvent]): 測定データ（RealEventオブジェクト）のリスト

Returns:

data (pd.DataFrame): 測定データのデータフレーム

Notes:

haniwers.v0.daq.run_daq(port: serial.Serial, daq: haniwers.v0.config.Daq) → None#

Run DAQ（回数指定）

OSECHIを接続したUSBポートとシリアル通信をして、データ取得する。指定したファイル数と行数をでループ処理する。

Args:

port (serial.Serial): 接続するポート（Serialオブジェクト）
daq (Daq): 設定（Daqオブジェクト）

Returns:

None: 測定時間が長くなると、メモリリークするかもしれないため、 1イベントのデータをファイルに書き出したあとは潔く破棄している

Example:

run_daq(port, daq)

haniwers.v0.daq.scan_daq(args: haniwers.v0.config.Daq, fname: str, duration: int) → pandas.DataFrame#

Run DAQ（時間指定）

1回のランあたりの測定時間を指定してデータを取得する。スレッショルド測定するために作ったDAQ関数です。

Args:

args (Daq): Daqオブジェクト
duration (int): 測定時間を秒で指定

Returns:

data (pd.DataFrame): 測定結果のデータフレーム。データフレームを次の処理に渡したいため。

Example:

data = scan_daq(daq, fname, duration)

haniwers.v0.daq.run(port: serial.Serial, args: haniwers.v0.config.Daq)#

メインのDAQ

run_daqのラッパー。例外処理などで囲んだもの。

Versionadded:: 0.6.0.

haniwers.v0.daq.time_daq(args: haniwers.v0.config.Daq, duration: int) → pandas.DataFrame#

（削除予定）測定時間を指定してDAQを走らせます。

Args:

args (Daq): Daqオブジェクト
duration (int): 測定時間を秒で指定

Returns:

data (pd.DataFrame): 測定結果のデータフレーム

haniwers.v0.daq.scan_ch_vth(daq: haniwers.v0.config.Daq, duration: int, ch: int, vth: int) → list#

（削除予定）Run threshold scan.

Args:

daq (Daq): Daqオブジェクト
duration (int): 測定時間（秒）
ch (int): 測定するチャンネル番号
vth (int): スレッショルド値

Returns:

data (list): [測定時刻、チャンネル番号、スレッショルド値、イベント数]のリストを返します。

haniwers.v0.daq.scan_ch_thresholds(daq: haniwers.v0.config.Daq, duration: int, ch: int, thresholds: list[int]) → list[list]#

（削除予定）Run threshold scan.

Args:

daq (Daq): Daqオブジェクト
duration (int): 測定時間（秒）
ch (int): チャンネル番号
thresholds (list[int]): スレッショルド値のリスト

Returns:

rows (list[list]): [測定時刻、チャンネル番号、スレッショルド値、イベント数]のリスト

haniwers.v0.daq._read_serial_data_as_list(port: serial.Serial) → list#

（削除予定）Read serial data from port.

OSECHIが接続されているポートからデータを読み出します。引数に指定するポートは、あらかじめ開いたものを渡してください。 run_daqやtime_daqでデータを取得するために使います。

Args:

port (serial.Serial): Serialオブジェクト

Returns:

row (list): 読み出した時刻を追加したデータのリスト

Examples:

>>> with open_serial_connection() as port:
>>>     row = read_serial_data(port)
>>>     row
[日付, top, mid, btm, adc, tmp, atm, hmd]

haniwers.v0.daq.read_serial_data(port: serial.Serial) → list#: （削除予定）

haniwers.v0.daq._loop_and_save_events(fname: pathlib.Path, port: serial.Serial, max_rows: int, suffix: str = '.csv') → list#

（削除予定）イベント保存ループ

Args:

f (typing.TextIO): 開いたファイルオブジェクト
port (serial.Serial) : 接続済みのSerialオブジェクト
max_rows (int): 1ファイルあたりの行数の最大値
suffix (str): ファイルの拡張子

Return:

events (list): 複数イベントの測定データ

haniwers.v0.daq.loop_and_save_events(f: TextIO, daq: haniwers.v0.config.Daq, port: serial.Serial) → list[str]#

（削除予定）イベント保存ループ

Args:

f (TextIO): データを書き込むファイルオブジェクト
daq (Daq): Daqオブジェクト
port (serial.Serial): 接続済みのSerialオブジェクト

haniwers.v0.daq.save_serial_data(f, daq: haniwers.v0.config.Daq, port: serial.Serial) → list#

（削除予定）

Args:

f: ファイルポインタ
daq (Daq): Daqオブジェクト
port (serial.Serial): Serialオブジェクト

Return:

rows (list[list]): 取得したデータのリスト

TODO:

Daqオブジェクトに依存しない関数にしたい（ジェネレーターにするのがいいのかな？）
pd.DataFrameを返した方がいいかもしれない？

haniwers.v0.daq

目次

haniwers.v0.daq#

Module Contents#

Classes#

Functions#

API#

`haniwers.v0.daq`#