winファイルの振動データをPythonに取り込む必要に迫られました。
もともとは 既存の GUI プログラムでテキストに変換していたのですが、これだと CSV で保存してから Python で読みに行くといった無駄な時間が発生します。
C の共有メモリに Python でアクセスする方法が最も速いのですが、これは私には荷が重い。
ということで、刻々書き出される win ファイルを Python で読むことにしました。
といっても、今まで win format を読んだことがありません。確認すると、これがフクザツ。入れ子状態で説明が書かれています。
ということで、GPT-o1 の出番。
win format の解説と wdisk のソースをすべて読ませてから、読み込みスクリプトを作ってもらいました。
フムフム、ナルホド、コウイウコトダッタノカ。
- import pandas as pd
- from datetime import datetime, timedelta
-
- def bcd_byte_to_int(b: int) -> int:
- """1 バイトBCDを整数に変換 (0x23 -> 23など)"""
- hi = (b & 0xF0) >> 4
- lo = (b & 0x0F)
- return hi * 10 + lo
-
- def decode_bcd_time(bcd6: bytes) -> datetime:
- """
- 秒ヘッダー6バイト (BCD: YY, MM, DD, HH, mm, SS) を datetime に変換。
- ここでは 0~49 => 2000~2049, 50~99 => 1950~1999 の例。
- """
- if len(bcd6) < 6:
- return None
-
- year = bcd_byte_to_int(bcd6[0]) # 0 ~ 99
- month = bcd_byte_to_int(bcd6[1]) # 1 ~ 12
- day = bcd_byte_to_int(bcd6[2]) # 1 ~ 31
- hour = bcd_byte_to_int(bcd6[3]) # 0 ~ 23
- minu = bcd_byte_to_int(bcd6[4]) # 0 ~ 59
- sec = bcd_byte_to_int(bcd6[5]) # 0 ~ 59
-
- # 年の補正 (例: 年<50 => 2000年代, それ以外 => 1900年代)
- if year < 50:
- year_full = 2000 + year
- else:
- year_full = 1900 + year
-
- try:
- return datetime(year_full, month, day, hour, minu, sec)
- except ValueError:
- # 不正な日時など
- return None
-
- def read_4bit_2s_comp_no_cross(data: bytes, bit_offset: int) -> int:
- """
- data バッファの bit_offset ビット目から 4 ビットを取り出し、
- 2 の補数(4bit)として返す (バイト境界は跨がない前提)。
- - 0x0~0x7 => 0~7, 0x8~0xF => -8 ~ -1
- """
- byte_index = bit_offset // 8
- bit_in_byte = bit_offset % 8 # 0～7
-
- val = data[byte_index] if byte_index < len(data) else 0
- shift_amount = (8 - 4) - bit_in_byte
- nibble = (val >> shift_amount) & 0x0F
-
- if nibble >= 8:
- nibble -= 16 # => -8 ～ -1
- return nibble
-
-
- def parse_channel_block(data: bytes, offset: int):
- """
- 1つのチャネルブロックをパースして情報を返す。
- 戻り値: (消費バイト数, dict(channel=<ch_num>,
- sample_size_info=<info>,
- sampling_rate=<rate>,
- samples=[<int値>, ...]))
- """
- if offset + 4 > len(data):
- return (0, None)
- chan_header = data[offset : offset+4]
-
- # チャネル番号
- ch_num_raw = int.from_bytes(chan_header[0:2], 'big', signed=False)
- ch_num = f"{ch_num_raw:04X}" # 90EA のような16進大文字表記
- # サンプリングレート等
- last2 = int.from_bytes(chan_header[2:4], 'big', signed=False)
- sample_size_info = (last2 >> 12) & 0x0F
- sampling_rate = last2 & 0x0FFF
-
- pos = offset + 4
-
- # 先頭サンプル 4バイト
- if pos + 4 > len(data):
- return (4, {
- 'channel': ch_num,
- 'sample_size_info': sample_size_info,
- 'sampling_rate': sampling_rate,
- 'samples': [],
- })
- first_sample_bytes = data[pos : pos+4]
- pos += 4
-
- current_value = int.from_bytes(first_sample_bytes, 'big', signed=True)
- samples = [current_value]
-
- diff_count = sampling_rate - 1
- if sample_size_info == 0:
- # 4bit(0.5バイト) 差分
- bit_offset = pos * 8
- for _ in range(diff_count):
- #nibble = read_4bit_2s_comp(data, bit_offset)
- nibble = read_4bit_2s_comp_no_cross(data, bit_offset)
- bit_offset += 4
- current_value += nibble
- samples.append(current_value)
- pos = (bit_offset + 7)//8
-
- elif sample_size_info in [1, 2, 3, 4]:
- # 差分
- sample_len = sample_size_info
- for _ in range(diff_count):
- if pos + sample_len > len(data):
- break
- diff_val = int.from_bytes(data[pos : pos+sample_len], 'big', signed=True)
- pos += sample_len
- current_value += diff_val
- samples.append(current_value)
-
- elif sample_size_info == 5:
- # 4バイト絶対値
- sample_len = 4
- for _ in range(diff_count):
- if pos + sample_len > len(data):
- break
- abs_val = int.from_bytes(data[pos : pos+sample_len], 'big', signed=True)
- pos += sample_len
- current_value = abs_val
- samples.append(current_value)
-
- ch_info = {
- 'channel': ch_num,
- 'sample_size_info': sample_size_info,
- 'sampling_rate': sampling_rate,
- 'samples': samples
- }
- consumed = pos - offset
- return (consumed, ch_info)
-
- def read_win_file_as_dataframe(filename: str) -> pd.DataFrame:
- """
- ディスクファイル上の WINフォーマットを読み込み、
- ・行: 「各サンプル」の時刻 (秒ヘッダ + サンプルオフセット)
- ・列: チャネル番号
- ・値: 各サンプルの値
- として1つの DataFrame を返す
- """
- # "各サンプル"を行にするため、時刻単位(秒 + fraction)で紐づける。
- # 複数チャネルを同一時刻にマージするため、以下のような手順:
- # - data_dict: { pd.Timestamp: { ch_name: sampleVal, ...}, ... }
- # - 最後に data_dict -> DataFrame へ変換
-
- data_dict = {}
-
- with open(filename, 'rb') as f:
- while True:
- # 1) ブロックサイズ
- size_buf = f.read(4)
- if len(size_buf) < 4:
- break
- block_size = int.from_bytes(size_buf, 'big', signed=False)
- if block_size < 4:
- break
-
- # 2) 秒ブロック本体読み込み
- sec_block = f.read(block_size - 4)
- if len(sec_block) < (block_size - 4):
- break
-
- # a) BCD時刻 (先頭6バイト)
- if len(sec_block) < 6:
- continue
- bcd_part = sec_block[0:6]
- base_dt = decode_bcd_time(bcd_part) # datetime
- if base_dt is None:
- continue
-
- # b) チャネルブロックを読み込む
- offset = 6
- channel_results = []
- while offset < len(sec_block):
- consumed, ch_info = parse_channel_block(sec_block, offset)
- if not ch_info or consumed <= 0:
- break
- offset += consumed
- channel_results.append(ch_info)
-
- # c) チャネル結果 ⇒ サンプルの時刻と値を data_dict に格納
- for ch_info in channel_results:
- ch_num = ch_info['channel']
- sr = ch_info['sampling_rate']
- samples = ch_info['samples']
- # 1秒ブロック内で sr 個のサンプルがあると仮定（不足分は途中でbreakしてる場合もあり）
- # 各サンプル i 番は base_dt + i*(1/sr) 秒後とみなす
- # → Timedelta等で補正
- # カラム名は16進表記でも10進表記でもOK
- ch_name = ch_num
-
- for i, val in enumerate(samples):
- # サンプル時刻
- if sr > 1:
- frac_sec = i / sr
- else:
- frac_sec = 0.0
-
- sample_dt = base_dt + timedelta(seconds=frac_sec)
-
- # data_dict[sample_dt] に ch_name: val を入れる
- if sample_dt not in data_dict:
- data_dict[sample_dt] = {}
- data_dict[sample_dt][ch_name] = val
-
- # 辞書 -> DataFrame
- df = pd.DataFrame.from_dict(data_dict, orient='index')
- # 時刻順にソート
- df.sort_index(inplace=True)
- return df,sr
-
- fname = "input/25012123.36"
- df, sr = read_win_file_as_dataframe(fname)
-
-
これでA/D値を取り込めます。テストしたファイルは sample_size_info = 2 のみでしたが、これまでのプログラムと出力が一致していることを確認しました。他は追々。
data frame に格納したので、物理量への変換はもちろん指定時刻から前〇〇秒とか、欠測秒をNanで埋めるとか後で容易に加工できます。
コードを読むことでファイル構造を理解するという逆の手順ですが、GPTの性能向上によって可能となりました。ラクナジダイニナッタ。
