import sys
import os
import gzip
import requests
import pandas as pd
import xml.etree.ElementTree as ET
from PyQt6.QtWidgets import (
QApplication,
QMainWindow,
QWidget,
QFileDialog,
QPushButton,
QLabel,
QListWidget,
QVBoxLayout,
QHBoxLayout,
QComboBox,
QLineEdit,
QMessageBox
)
from PyQt6.QtCore import Qt
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("GSM地点ガイダンスビューア")
self.resize(1400, 900)
self.station_dict = {}
self.station_names = {}
self.gsm_data = {}
self.msm_data = {}
self.humidity_codes = set()
self.init_ui()
self.load_station_master()
self.gsm_file_list.currentRowChanged.connect(
self.load_selected_gsm_file
)
self.msm_file_list.currentRowChanged.connect(
self.load_selected_msm_file
)
self.station_combo.currentTextChanged.connect(
self.station_changed
)
# --------------------------------------------------
# UI
# --------------------------------------------------
def init_ui(self):
from PyQt6.QtWidgets import QTableWidget
central = QWidget()
self.setCentralWidget(central)
layout = QVBoxLayout()
central.setLayout(layout)
# ==================================
# 上段
# 地点選択 + Excel出力
# ==================================
top_layout = QHBoxLayout()
top_layout.addWidget(
QLabel("地点")
)
self.station_search = QLineEdit()
self.station_search.setPlaceholderText(
"地点検索"
)
top_layout.addWidget(
self.station_search
)
self.station_combo = QComboBox()
top_layout.addWidget(
self.station_combo
)
self.btn_excel = QPushButton(
"Excel出力"
)
top_layout.addWidget(
self.btn_excel
)
layout.addLayout(
top_layout
)
# ==================================
# GSM / MSM フォルダ
# ==================================
folder_layout = QHBoxLayout()
# ----------------------------------
# GSM
# ----------------------------------
gsm_layout = QVBoxLayout()
gsm_layout.addWidget(
QLabel("GSM")
)
self.btn_gsm_folder = QPushButton(
"GSMフォルダ選択"
)
self.gsm_folder_label = QLabel(
"GSMフォルダ未選択"
)
self.gsm_file_list = QListWidget()
gsm_layout.addWidget(
self.btn_gsm_folder
)
gsm_layout.addWidget(
self.gsm_folder_label
)
gsm_layout.addWidget(
self.gsm_file_list
)
# ----------------------------------
# MSM
# ----------------------------------
msm_layout = QVBoxLayout()
msm_layout.addWidget(
QLabel("MSM")
)
self.btn_msm_folder = QPushButton(
"MSMフォルダ選択"
)
self.msm_folder_label = QLabel(
"MSMフォルダ未選択"
)
self.msm_file_list = QListWidget()
msm_layout.addWidget(
self.btn_msm_folder
)
msm_layout.addWidget(
self.msm_folder_label
)
msm_layout.addWidget(
self.msm_file_list
)
folder_layout.addLayout(
gsm_layout
)
folder_layout.addLayout(
msm_layout
)
layout.addLayout(
folder_layout
)
# ==================================
# GSMテーブル
# ==================================
layout.addWidget(
QLabel("GSM ガイダンス")
)
self.gsm_table = QTableWidget()
layout.addWidget(
self.gsm_table
)
# ==================================
# MSMテーブル
# ==================================
layout.addWidget(
QLabel("MSM ガイダンス")
)
self.msm_table = QTableWidget()
layout.addWidget(
self.msm_table
)
# ==================================
# シグナル
# ==================================
self.btn_gsm_folder.clicked.connect(
self.select_gsm_folder
)
self.btn_msm_folder.clicked.connect(
self.select_msm_folder
)
self.station_search.textChanged.connect(
self.filter_station
)
# --------------------------------------------------
# 地点マスタ
# --------------------------------------------------
def load_station_master(self):
try:
url = (
"https://www.jma.go.jp/"
"bosai/amedas/const/amedastable.json"
)
r = requests.get(url, timeout=20)
data = r.json()
df = pd.DataFrame.from_dict(
data,
orient="index"
)
self.station_dict = {
str(idx): row["kjName"]
for idx, row in df.iterrows()
}
self.station_names = {
row["kjName"]: str(idx)
for idx, row in df.iterrows()
}
self.station_combo.clear()
names = sorted(self.station_names.keys())
self.refresh_station_combo()
print(
f"地点数: {len(self.station_names)}"
)
except Exception as e:
QMessageBox.warning(
self,
"地点マスタ取得失敗",
str(e)
)
# --------------------------------------------------
# 検索
# --------------------------------------------------
def filter_station(self):
self.refresh_station_combo()
# --------------------------------------------------
# GSMフォルダ選択
# --------------------------------------------------
def select_gsm_folder(self):
folder = QFileDialog.getExistingDirectory(
self,
"GSMフォルダ選択"
)
if not folder:
return
self.gsm_folder_label.setText(
folder
)
self.load_gsm_files(folder)
# --------------------------------------------------
# MSMフォルダ選択
# --------------------------------------------------
def select_msm_folder(self):
folder = QFileDialog.getExistingDirectory(
self,
"MSMフォルダ選択"
)
if not folder:
return
self.msm_folder_label.setText(
folder
)
self.load_msm_files(folder)
# --------------------------------------------------
# GSMファイル検索
# --------------------------------------------------
def load_gsm_files(self, folder):
self.gsm_file_list.clear()
files = []
for file in os.listdir(folder):
if (
file.endswith(".xml")
or
file.endswith(".xml.gz")
):
files.append(file)
files.sort(reverse=True)
for file in files:
display = self.build_display_name(
file
)
self.gsm_file_list.addItem(
display
)
# --------------------------------------------------
# MSMファイル検索
# --------------------------------------------------
def load_msm_files(self, folder):
self.msm_file_list.clear()
files = []
for file in os.listdir(folder):
if (
file.endswith(".xml")
or
file.endswith(".xml.gz")
):
files.append(file)
files.sort(reverse=True)
for file in files:
display = self.build_display_name(
file
)
self.msm_file_list.addItem(
display
)
# --------------------------------------------------
# 表示名作成
# --------------------------------------------------
def build_display_name(self, filename):
try:
parts = filename.split("_")
dt = parts[3]
y = dt[0:4]
m = dt[4:6]
d = dt[6:8]
h = dt[8:10]
return (
f"{y}/{m}/{d} {h}UTC "
f"{filename}"
)
except:
return filename
def load_selected_gsm_file(self):
item = self.gsm_file_list.currentItem()
if item is None:
return
filename = item.text()
pos = filename.find("UTC")
if pos >= 0:
filename = filename[pos + 7:]
folder = self.gsm_folder_label.text()
filepath = os.path.join(
folder,
filename
)
self.parse_xml(
filepath,
"GSM"
)
def load_selected_msm_file(self):
item = self.msm_file_list.currentItem()
if item is None:
return
filename = item.text()
pos = filename.find("UTC")
if pos >= 0:
filename = filename[pos + 7:]
folder = self.msm_folder_label.text()
filepath = os.path.join(
folder,
filename
)
self.parse_xml(
filepath,
"MSM"
)
def refresh_station_combo(self):
current = self.station_search.text().strip()
self.station_combo.clear()
for name in sorted(self.station_names.keys()):
if current:
if current not in name:
continue
code = self.station_names[name]
display = f"{name} [{code}]"
self.station_combo.addItem(
display
)
def get_initial_hour(self, filepath):
try:
filename = os.path.basename(filepath)
parts = filename.split("_")
dtstr = parts[3]
return int(dtstr[8:10])
except:
return 0
def parse_xml(self, filepath,model):
try:
tree = ET.parse(filepath)
root = tree.getroot()
if model == "GSM":
self.gsm_data = {}
target = self.gsm_data
else:
self.msm_data = {}
target = self.msm_data
self.humidity_codes = set()
for tsi in root.findall(".//{*}TimeSeriesInfo"):
time_map = {}
for td in tsi.findall(".//{*}TimeDefine"):
tid = td.attrib.get("timeId")
dt = td.findtext(".//{*}DateTime")
if tid and dt:
time_map[tid] = dt
for item in tsi.findall(".//{*}Item"):
type_name = item.findtext(
".//{*}Type"
)
station = item.find(".//{*}Station")
if station is None:
continue
code = station.findtext(".//{*}Code")
if code is None:
continue
kind = item.find(".//{*}Kind")
if kind is None:
continue
prop = kind.find(".//{*}Property")
if prop is None:
continue
type_name = prop.findtext(
".//{*}Type"
)
if code not in target:
target[code] = {}
# --------------------------------
# 気温
# --------------------------------
if type_name == "気温":
target[code]["気温"] = []
for temp in prop.findall(".//{*}Temperature"):
refid = temp.attrib.get("refId")
dt = time_map.get(refid)
target[code]["気温"].append(
(dt, temp.text)
)
# --------------------------------
# 風
# --------------------------------
elif type_name == "風":
target[code]["風向"] = []
for elem in prop.findall(
".//{*}WindDirection"
):
refid = elem.attrib.get("refId")
dt = time_map.get(refid)
target[code]["風向"].append(
(dt, elem.text)
)
target[code]["風速"] = []
for elem in prop.findall(
".//{*}WindSpeed"
):
refid = elem.attrib.get("refId")
dt = time_map.get(refid)
target[code]["風速"].append(
(dt, elem.text)
)
# --------------------------------
# 最大風
# --------------------------------
elif type_name == "最大風":
target[code]["最大風向"] = []
for elem in prop.findall(
".//{*}WindDirection"
):
refid = elem.attrib.get("refId")
dt = time_map.get(refid)
target[code]["最大風向"].append(
(dt, elem.text)
)
target[code]["最大風速"] = []
for elem in prop.findall(
".//{*}WindSpeed"
):
refid = elem.attrib.get("refId")
dt = time_map.get(refid)
target[code]["最大風速"].append(
(dt, elem.text)
)
# --------------------------------
# 朝の最低気温
# --------------------------------
elif type_name == "朝の最低気温":
target[code]["朝の最低気温"] = []
for elem in prop.findall(
".//{*}Temperature"
):
refid = elem.attrib.get("refId")
dt = time_map.get(refid)
target[code]["朝の最低気温"].append(
(dt, elem.text)
)
# --------------------------------
# 日中の最高気温
# --------------------------------
elif type_name == "日中の最高気温":
target[code]["日中の最高気温"] = []
for elem in prop.findall(
".//{*}Temperature"
):
refid = elem.attrib.get("refId")
dt = time_map.get(refid)
target[code]["日中の最高気温"].append(
(dt, elem.text)
)
# --------------------------------
# 最小湿度
# --------------------------------
elif type_name == "最小湿度":
self.humidity_codes.add(code)
if code not in target:
target[code] = {}
target[code]["最小湿度"] = []
for elem in item.findall(".//{*}Humidity"):
refid = elem.attrib.get("refId")
dt = time_map.get(refid)
target[code]["最小湿度"].append(
(dt, elem.text)
)
if model == "GSM":
self.gsm_initial_hour = self.get_initial_hour(
filepath
)
else:
self.msm_initial_hour = self.get_initial_hour(
filepath
)
self.station_changed()
except Exception as e:
QMessageBox.warning(self, "読込失敗", str(e))
def station_changed(self):
name = self.station_combo.currentText()
if "[" in name:
name = name.split("[")[0].strip()
code = self.station_names.get(name)
if code is None:
return
if code in self.gsm_data:
self.fill_table(
self.gsm_table,
self.gsm_data,
code
)
if code in self.msm_data:
self.fill_table(
self.msm_table,
self.msm_data,
code
)
def fill_table(
self,
table,
data,
code
):
from PyQt6.QtWidgets import QTableWidgetItem
elements = [
"気温",
"風向",
"風速",
"最大風向",
"最大風速",
"朝の最低気温",
"日中の最高気温",
"最小湿度"
]
temps = data[code].get(
"気温",
[]
)
if len(temps) == 0:
return
table.clear()
table.setRowCount(
len(elements)
)
table.setColumnCount(
len(temps)
)
headers = []
jst_hours = []
for dt, value in temps:
jst = (
pd.to_datetime(
dt,
utc=True
).tz_convert(
"Asia/Tokyo"
)
)
headers.append(
jst.strftime("%m/%d\n%H")
)
jst_hours.append(
jst.hour
)
table.setHorizontalHeaderLabels(
headers
)
table.setVerticalHeaderLabels(
elements
)
for row, element in enumerate(elements):
values = data[code].get(
element,
[]
)
if len(values) == 0:
table.setItem(
row,
0,
QTableWidgetItem("-")
)
continue
# --------------------------------
# 風系
# --------------------------------
if element in [
"風向",
"風速",
"最大風向",
"最大風速"
]:
for dt, value in values:
hour = (
pd.to_datetime(
dt,
utc=True
)
.tz_convert(
"Asia/Tokyo"
)
.hour
)
try:
col = jst_hours.index(hour)
except:
continue
table.setItem(
row,
col,
QTableWidgetItem(
str(value)
)
)
continue
# --------------------------------
# 最高気温
# --------------------------------
if element == "日中の最高気温":
idx = 0
for col, hour in enumerate(jst_hours):
if hour == 15:
if idx < len(values):
table.setItem(
row,
col,
QTableWidgetItem(
str(values[idx][1])
)
)
idx += 1
continue
# --------------------------------
# 最低気温
# --------------------------------
if element == "朝の最低気温":
idx = 0
for col, hour in enumerate(jst_hours):
if hour == 3:
if idx < len(values):
table.setItem(
row,
col,
QTableWidgetItem(
str(values[idx][1])
)
)
idx += 1
continue
# --------------------------------
# 気温・湿度
# --------------------------------
for col, (dt, value) in enumerate(values):
table.setItem(
row,
col,
QTableWidgetItem(
str(value)
)
)
table.resizeColumnsToContents()
table.resizeRowsToContents()
if __name__ == "__main__":
app = QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec())
未分類