import os
import tkinter as tk
from tkinter import filedialog, ttk, messagebox
from datetime import datetime, timedelta
# ドラッグ＆ドロップ機能のための外部ライブラリ
try:
from tkinterdnd2 import TkinterDnD, DND_FILES
except ImportError:
print("【エラー】tkinterdnd2 がインストールされていません。")
print("コマンドプロンプトで 'pip install tkinterdnd2' を実行してください。")
exit()
"""
=========================================================================================
【次世代AIへの引き継ぎ用・技術サマリー】
気象庁 AMeDAS BUFR (N1フォーマット) ピュアPython・ダイレクトデコーダー
■ 背景と開発経緯
- pybufrkit等の外部ライブラリでは、気象庁独自のローカル記述子のビット幅解決ができずエラーが多発。
- バイナリを直接透視し、Python標準機能のみでビットスライスを行う強固な専用デコーダーを構築した。
- アプリケーション内部の時刻ロジックは、データ同期エラーを防ぐため【すべてUTC（協定世界時）】で処理・保持し、
UI表示のタイミングでのみ JST (UTC+9) に変換する厳密な仕様としている。
■ 確定した N1フォーマットのバイナリ構造 (27 (1).pdf 準拠)
- 1地点あたりのデータブロックは、正確に【 205ビット 】の固定長である。
- 当該BUFR内には観測所の「ID」のみが格納されており、「緯度・経度・気圧」は含まれていない。
■ 205ビットの完全マッピングリスト
1. 府県表示番号 (7 bits)
2. 地点表示番号 (10 bits)
3. 観測日時 UTC [年(12), 月(4), 日(6), 時(5), 分(6)] (計 33 bits)
4. 障害状況の指示符 (8 bits)
5. 時間変位 (12 bits)
6. 降水量合計 (14 bits) [尺度1, 参照値 -1]
7. 降水メーター値 (14 bits) [尺度1, 参照値 0]
8. 降水AQC (4 bits)
9. 風向 (9 bits) [尺度0, 参照値 0]
10. 風速 (12 bits) [尺度1, 参照値 0]
11. 風向論理チェック (4 bits)
12. 風速AQC (4 bits)
13. 気温 (12 bits) [尺度1, 参照値 0] ※ K -> ℃ 変換 (-273.15)
14. 気温AQC (4 bits)
15. 時間変位 (12 bits)
16. 日照時間合計 (11 bits) [尺度0, 参照値 0]
17. 日照メーター (11 bits)
18. 日照論理チェック (4 bits)
19. 積雪深 (16 bits) [尺度2, 参照値 -2, 単位 m] -> cmへの変換
20. 積雪AQC (4 bits)
=========================================================================================
"""
# 主要な観測所名マッピング（GUI表示用）
KNOWN_STATIONS = {
"14163": "札幌 (Sapporo)",
"12411": "旭川 (Asahikawa)",
"18461": "函館 (Hakodate)",
"11211": "稚内 (Wakkanai)",
"11711": "網走 (Abashiri)",
"17321": "倶知安 (Kutchan)",
"44132": "東京 (Tokyo)"
}
class BitReader:
""" バイナリデータから任意のビット数を正確に読み取るための専用クラス """
def __init__(self, data, bit_offset=0):
self.data = data
self.bit_offset = bit_offset
def read(self, bits):
val = 0
for _ in range(bits):
byte_idx = self.bit_offset // 8
bit_idx = 7 - (self.bit_offset % 8)
bit = (self.data[byte_idx] >> bit_idx) & 1
val = (val << 1) | bit
self.bit_offset += 1
return val
class AmedasDragDropApp:
def __init__(self, root):
self.root = root
self.root.title("AMeDAS N1 BUFR Viewer (Pure Python)")
self.root.geometry("700x700")
# ドラッグ＆ドロップの登録
self.root.drop_target_register(DND_FILES)
self.root.dnd_bind('<<Drop>>', self.on_file_drop)
# データ保持用
self.current_stations = {}
self.current_file_path = ""
self.create_widgets()
def create_widgets(self):
# --- ドロップエリア＆ファイル選択 ---
frame_top = tk.LabelFrame(self.root, text="ファイル読み込み", pady=10, padx=10)
frame_top.pack(fill=tk.X, padx=10, pady=5)
self.lbl_drop = tk.Label(
frame_top,
text="ここに BUFRファイル (.bin) を\nドラッグ＆ドロップしてください",
bg="#f0f8ff",
relief="solid",
borderwidth=1,
width=50,
height=4,
font=("Meiryo", 10, "bold")
)
self.lbl_drop.pack(pady=5)
self.btn_file = tk.Button(frame_top, text="またはファイルを選択", command=self.select_file)
self.btn_file.pack(pady=5)
# --- コントロール領域 ---
frame_ctrl = tk.LabelFrame(self.root, text="抽出地点の選択", pady=10, padx=10)
frame_ctrl.pack(fill=tk.X, padx=10, pady=5)
tk.Label(frame_ctrl, text="地点選択:").pack(side=tk.LEFT)
self.cmb_station = ttk.Combobox(frame_ctrl, width=40, state="readonly")
self.cmb_station.pack(side=tk.LEFT, padx=10)
self.cmb_station.bind("<<ComboboxSelected>>", self.on_station_selected)
# --- データ表示領域 ---
frame_data = tk.LabelFrame(self.root, text="気象要素 (デコード結果)", pady=10, padx=10)
frame_data.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
scroll_y = tk.Scrollbar(frame_data)
scroll_y.pack(side=tk.RIGHT, fill=tk.Y)
self.txt_display = tk.Text(frame_data, wrap=tk.WORD, font=("Consolas", 11), yscrollcommand=scroll_y.set)
self.txt_display.pack(fill=tk.BOTH, expand=True)
scroll_y.config(command=self.txt_display.yview)
def on_file_drop(self, event):
file_path = event.data.strip('{}')
self.process_file(file_path)
def select_file(self):
file_path = filedialog.askopenfilename(filetypes=[("BUFR bin files", "*.bin"), ("All files", "*.*")])
if file_path:
self.process_file(file_path)
def process_file(self, file_path):
if not file_path.endswith('.bin'):
messagebox.showwarning("ファイル形式エラー", "拡張子が .bin のファイルを選択してください。")
return
self.current_file_path = file_path
self.current_stations.clear()
self.txt_display.delete(1.0, tk.END)
self.txt_display.insert(tk.END, f"ファイルを解析中...\n{os.path.basename(file_path)}\n")
self.root.update()
try:
time_jst_str = self.extract_time_from_header(file_path)
if not time_jst_str:
raise ValueError("ヘッダーから時刻情報を読み取れませんでした。")
self.decode_bufr_file(file_path, time_jst_str)
station_list = sorted(self.current_stations.keys())
self.cmb_station['values'] = station_list
self.lbl_drop.config(text=f"読み込み成功: {os.path.basename(file_path)}\n({len(station_list)} 地点抽出)", bg="#e6ffe6")
self.txt_display.insert(tk.END, f"展開完了: {len(station_list)} 地点のデータを抽出しました。\n上のリストから地点を選択してください。")
if station_list:
sapporo_key = next((k for k in station_list if "14163" in k), None)
if sapporo_key:
self.cmb_station.set(sapporo_key)
else:
self.cmb_station.current(0)
self.on_station_selected(None)
except Exception as e:
self.lbl_drop.config(text="エラーが発生しました", bg="#ffe6e6")
self.txt_display.insert(tk.END, f"\n\n処理エラー: {e}")
def extract_time_from_header(self, file_path):
""" セクション1の絶対位置からUTC時刻を読み取り、JSTに変換 """
try:
with open(file_path, 'rb') as f:
data = f.read(50)
if data[0:4] != b'BUFR':
return None
year = int.from_bytes(data[23:25], 'big')
month = data[25]
day = data[26]
hour = data[27]
minute = data[28]
utc_time = datetime(year, month, day, hour, minute)
jst_time = utc_time + timedelta(hours=9)
return jst_time.strftime("%Y/%m/%d %H:%M")
except Exception as e:
print(f"Header Parse Error: {e}")
return None
def decode_bufr_file(self, file_path, time_jst_str):
""" N1フォーマットの仕様書(27 (1).pdf)に完全準拠した205ビットスライス処理 """
with open(file_path, 'rb') as f:
raw_data = f.read()
reader = BitReader(raw_data, 241 * 8)
TOTAL_STATIONS = 1286
BITS_PER_STATION = 205 # 仕様書に基づく正確な1地点のビット長
for _ in range(TOTAL_STATIONS):
if (reader.bit_offset + BITS_PER_STATION) > len(raw_data) * 8:
break
# --- 1. 識別情報 ---
block_no = reader.read(7)
station_no = reader.read(10)
st_id = f"{block_no:02d}{station_no:03d}"
st_name = KNOWN_STATIONS.get(st_id, "名称不明")
display_key = f"{st_id} {st_name}"
# --- 2. 観測日時 (UTC) ---
year = reader.read(12)
month = reader.read(4)
day = reader.read(6)
hour = reader.read(5)
minute = reader.read(6)
# --- 3. 障害状況 ---
obs_status = reader.read(8)
# --- 4. 降水量 ---
time_diff_1 = reader.read(12)
precip_raw = reader.read(14)
precip_meter = reader.read(14)
precip_aqc = reader.read(4)
# 降水量の計算 (尺度1, 参照値 -1)
precip = (precip_raw - 1) / 10.0 if precip_raw != 0x3FFF else None
# --- 5. 風向・風速 ---
wind_dir_raw = reader.read(9)
wind_spd_raw = reader.read(12)
wind_dir_aqc = reader.read(4)
wind_spd_aqc = reader.read(4)
# 16方位（静穏は0）
wind_dir = wind_dir_raw if wind_dir_raw != 0x1FF else None
wind_spd = wind_spd_raw / 10.0 if wind_spd_raw != 0xFFF else None
# --- 6. 気温 ---
temp_raw = reader.read(12)
temp_aqc = reader.read(4)
# 気温の計算 (尺度1, ケルビンから摂氏への変換)
temp_c = (temp_raw / 10.0) - 273.15 if temp_raw != 0xFFF else None
# --- 7. 日照時間 ---
time_diff_2 = reader.read(12)
sun_raw = reader.read(11)
sun_meter = reader.read(11)
sun_aqc = reader.read(4)
sun = sun_raw if sun_raw != 0x7FF else None
# --- 8. 積雪深 ---
snow_raw = reader.read(16)
snow_aqc = reader.read(4)
# 積雪深の計算 (尺度2, 参照値 -2, 単位 m -> 1cm単位で格納されている値に変換)
snow_cm = (snow_raw - 2) if snow_raw != 0xFFFF else None
# 辞書に保存
self.current_stations[display_key] = {
"id": st_id,
"file_time_jst": time_jst_str,
"utc_time": f"{year:04d}-{month:02d}-{day:02d} {hour:02d}:{minute:02d}",
"temp_c": temp_c,
"precip": precip,
"wind_dir": wind_dir,
"wind_spd": wind_spd,
"sun": sun,
"snow_cm": snow_cm
}
def on_station_selected(self, event):
selected_key = self.cmb_station.get()
data = self.current_stations.get(selected_key)
if not data: return
self.txt_display.delete(1.0, tk.END)
out_text = f"==========================================\n"
out_text += f" 観測所ID : {data['id']}\n"
out_text += f" 観測所名 : {selected_key.split(' ')[1]}\n"
out_text += f" 観測時間 : {data['file_time_jst']} (JST)\n"
out_text += f"==========================================\n\n"
out_text += f" [観測データ]\n"
out_text += f" 気 温 : {data['temp_c']:.1f} ℃\n" if data['temp_c'] is not None else " 気 温 : 欠測\n"
out_text += f" 降水量 : {data['precip']:.1f} mm\n" if data['precip'] is not None else " 降水量 : 欠測\n"
if data['wind_spd'] is not None and data['wind_dir'] is not None:
out_text += f" 風 向 : {int(data['wind_dir'])} 度\n"
out_text += f" 風 速 : {data['wind_spd']:.1f} m/s\n"
else:
out_text += f" 風向・風速 : 欠測\n"
out_text += f" 日照時間 : {int(data['sun'])} 分\n" if data['sun'] is not None else " 日照時間 : 欠測\n"
out_text += f" 積雪深 : {int(data['snow_cm'])} cm\n" if data['snow_cm'] is not None else " 積雪深 : 欠測\n"
out_text += f"\n------------------------------------------\n"
out_text += f" ※N1仕様により、このBUFRには気圧・緯度経度の\n"
out_text += f" データは含まれていません。\n"
out_text += f" ※内部データ時刻記録: {data['utc_time']} (UTC)\n"
self.txt_display.insert(tk.END, out_text)
if __name__ == "__main__":
try:
root = TkinterDnD.Tk()
except NameError:
root = tk.Tk()
app = AmedasDragDropApp(root)
root.mainloop()
未分類
