# ==========================================
# 【App A】 統合GPV解析エンジン (MSM/GSM/ANAL/MEPS対応版)
# VERSION: MERGED_6.0 (真のハイブリッド完成版)
# ------------------------------------------
# 【処理方針】
# - 安定版(Ver5.3)の「GPVキャンバス結合処理(パッチワーク)」「気温・風速等の取得」「ANAL最優先処理」を完全維持。
# - 最新版(Ver5.6)の「ガイダンス(GUID)からの降水・降雪・最大降水・確率の期間別完全分離」を融合。
# - P-all / P-rrsf の両ファイルが揃うまで待機する「同期抽出ロジック」を搭載。
# ==========================================
import sys, os, glob, re, subprocess, time
from datetime import datetime
os.environ['QT_API'] = 'pyqt6'
from PyQt6.QtWidgets import (QApplication, QWidget, QVBoxLayout, QHBoxLayout,
QPushButton, QLabel, QListWidget, QFileDialog, QMessageBox,
QSystemTrayIcon, QMenu, QGridLayout)
from PyQt6.QtCore import Qt, QTimer, QThread, pyqtSignal, QSettings
from PyQt6.QtGui import QIcon, QPixmap, QPainter, QColor, QAction
try:
import cfgrib
except ImportError as e:
print("\n" + "="*60)
print("【起動エラー】cfgrib または依存ライブラリの読み込みに失敗しました。")
print(f"エラー詳細: {e}")
print("="*60)
input("\nEnterキーを押すとウィンドウを閉じます...")
sys.exit(1)
APP_DIR = os.getcwd()
DEFAULT_OUTPUT_DIR = os.path.join(APP_DIR, "gpv_cache_npz")
CURRENT_OUTPUT_DIR = DEFAULT_OUTPUT_DIR
WGRIB2_EXE = os.path.join(APP_DIR, "wgrib2_data", "wgrib2.exe")
os.makedirs(CURRENT_OUTPUT_DIR, exist_ok=True)
def write_syslog(msg):
log_file = os.path.join(CURRENT_OUTPUT_DIR, "system_log.txt")
try:
with open(log_file, "a", encoding="utf-8") as f:
f.write(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}\n")
except Exception: pass
WORKER_SCRIPT = os.path.join(APP_DIR, "_gpv_worker.py")
def create_lightning_icon():
pixmap = QPixmap(64, 64)
pixmap.fill(Qt.GlobalColor.transparent)
painter = QPainter(pixmap)
font = painter.font()
font.setPixelSize(50)
painter.setFont(font)
painter.drawText(pixmap.rect(), Qt.AlignmentFlag.AlignCenter, "⚡")
painter.end()
return QIcon(pixmap)
# ==========================================
# 統合ワーカーコード
# ==========================================
WORKER_CODE = """
import sys, traceback
def main():
try:
import os, gc, json, subprocess, time, re
import numpy as np
import warnings; warnings.filterwarnings("ignore")
import cfgrib
os.environ["ECCODES_MAX_VALUES"] = "5000000"
if sys.platform == "win32":
conda_dir = os.path.dirname(sys.executable)
dll_paths = [os.path.join(conda_dir, "Library", "bin"), os.path.join(conda_dir, "bin")]
for p in dll_paths:
if os.path.exists(p):
os.environ["PATH"] = f"{p};{os.environ.get('PATH', '')}"
try: os.add_dll_directory(p)
except Exception: pass
def calculate_vorticity(u, v, lon, lat):
try:
R = 6371000.0; LON, LAT = np.meshgrid(lon, lat) if lon.ndim == 1 else (lon, lat)
rad_lat = np.deg2rad(LAT); rad_lon = np.deg2rad(LON)
dy = R * np.gradient(rad_lat, axis=0); dx = R * np.cos(rad_lat) * np.gradient(rad_lon, axis=1)
dx[dx == 0] = 1e-10; dy[dy == 0] = 1e-10
return ((np.gradient(v, axis=1) / dx) - (np.gradient(u, axis=0) / dy)) * 1e5
except Exception: return np.zeros_like(u)
model = sys.argv[1]; mode = sys.argv[2]; cache = sys.argv[3]
init = sys.argv[4]; target_fts_str = sys.argv[5]
f1 = sys.argv[6]; f2 = sys.argv[7]; wgrib2_path = sys.argv[8]
target_fts = [int(x) for x in target_fts_str.split(',')]
creationflags = 0x08000000 if sys.platform == "win32" else 0
# === 全モデル共通: wgrib2による高速スライス関数 ===
def slice_with_wgrib2(fin, ft_val, suffix):
if fin == "NONE" or not os.path.exists(fin): return "NONE"
fout = os.path.join(cache, f"temp_{model}_{init}_{ft_val}_{suffix}.bin")
if ft_val == 0:
match_str = r":(anl|0 hour [^:]+|[0-9]*[-]0 hour [^:]+):"
else:
patterns = [f"{ft_val} hour", f"[0-9]*[-]{ft_val} hour"]
if mode == "MEPS_GUID" and ft_val >= 3:
offset_start = max(0, ft_val - 3)
offset_end = offset_start + 1
patterns.append(f"{offset_start}-{offset_end} hour")
if ft_val % 24 == 0:
days = ft_val // 24
patterns.extend([f"{days} day", f"[0-9]*[-]{days} day", f"{days} d", f"[0-9]*[-]{days} d"])
pat_joined = "|".join(patterns)
match_str = f":({pat_joined})[^:]*:"
cmd = [wgrib2_path, fin, "-match", match_str, "-grib", fout]
try:
subprocess.run(cmd, creationflags=creationflags, capture_output=True, text=True, timeout=120)
if os.path.exists(fout) and os.path.getsize(fout) > 0: return fout
except subprocess.TimeoutExpired:
pass
except Exception: pass
return "NONE"
# ============================================================
# 【MEPS_GUID専用】確率処理・パーセンタイル (Ver 5.6 準拠)
# ============================================================
if mode == "MEPS_GUID":
f_pall = f1; f_prrst = f2
d_p90 = {ft: {} for ft in target_fts}; d_p10 = {ft: {} for ft in target_fts}
d_prob1 = {ft: {} for ft in target_fts}; d_prob10 = {ft: {} for ft in target_fts}
d_prob20 = {ft: {} for ft in target_fts}; d_prob30 = {ft: {} for ft in target_fts}
for t_ft in target_fts:
f1_mini = slice_with_wgrib2(f_pall, t_ft, "pall")
f2_mini = slice_with_wgrib2(f_prrst, t_ft, "prrsf")
def extract_and_process(fin, src_name):
if fin == "NONE" or not os.path.exists(fin): return
res = subprocess.run([wgrib2_path, fin, "-s"], capture_output=True, text=True, creationflags=creationflags)
if not res.stdout.strip(): return
parsed_lines = []
for line in res.stdout.strip().split('\\n'):
if not line: continue
parts = line.split(':')
if len(parts) < 6: continue
param = parts[3]; time_str = parts[5].lower()
m = re.search(r'([0-9]+)-([0-9]+)\\s+(hour|day)', time_str)
if not m:
m2 = re.search(r'([0-9]+)\\s+(hour|day)', time_str)
if m2:
end_val = int(m2.group(1)); unit = m2.group(2)
if unit == 'day': end_val *= 24
start_val = end_val
duration = 0
else: continue
else:
start_val = int(m.group(1)); end_val = int(m.group(2)); unit = m.group(3)
if unit == 'day': end_val *= 24; start_val *= 24
duration = end_val - start_val
if duration == 1: mapped_ft = start_val + 3
else: mapped_ft = end_val
if mapped_ft != t_ft: continue
var_name = param.lower()
if param in ["TPRATE", "APCP", "PRECIP"]: var_name = "tp"
elif param in ["TSRATE", "WEASD", "SNOW", "ASNOW"]: var_name = "snow"
elif param in ["TSTM", "THUND", "PRSTM"]: var_name = "thund"
elif "vis" in var_name: var_name = "vis"
elif "parm=192" in time_str or param in ["WEA"]: var_name = "wea"
is_prob = "prob" in time_str
step_type = "accum"
if " max " in time_str or time_str.endswith(" max"): step_type = "max"
fixed_key = None
if var_name == "tp":
if is_prob: fixed_key = f"prob_precip{duration}"
elif src_name == "prrsf" or step_type == "max": fixed_key = f"precip{duration}max"
else: fixed_key = "precip" if duration == 3 else f"precip{duration}"
elif var_name == "snow": fixed_key = f"snow{duration}"
elif var_name == "thund": fixed_key = "thund"
elif var_name == "vis": fixed_key = "vis"
elif var_name == "wea": fixed_key = "wea"
if fixed_key: parsed_lines.append( (fixed_key, line) )
grouped = {}
for k, line in parsed_lines: grouped.setdefault(k, []).append(line)
for fixed_key, group_lines in grouped.items():
inv_input = "\\n".join(group_lines) + "\\n"
temp_bin = os.path.join(cache, f"~tmp_ext_{model}_{init}_{t_ft}_{src_name}_{fixed_key}.bin")
try:
cmd_extract = [wgrib2_path, fin, "-i", "-grib", temp_bin]
process = subprocess.Popen(cmd_extract, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, creationflags=creationflags)
process.communicate(input=inv_input.encode('utf-8'), timeout=60)
except: pass
if os.path.exists(temp_bin) and os.path.getsize(temp_bin) > 0:
try:
dss = cfgrib.open_datasets(temp_bin, backend_kwargs={'indexpath': ''})
all_fields = []; lon_1d, lat_1d = None, None
for ds in dss:
if lon_1d is None:
lon = ds.coords['longitude'].values if 'longitude' in ds.coords else (ds.longitude.values if hasattr(ds, 'longitude') else None)
lat = ds.coords['latitude'].values if 'latitude' in ds.coords else (ds.latitude.values if hasattr(ds, 'latitude') else None)
if lon is not None and lat is not None:
lon_1d = lon[0, :] if lon.ndim == 2 else lon
lat_1d = lat[:, 0] if lat.ndim == 2 else lat
for v in ds.data_vars:
da = ds[v]
da_step = da.isel(step=0) if 'step' in da.dims else da
val = da_step.values.copy()
if 'number' in da_step.dims:
num_axis = da_step.dims.index('number')
val = np.moveaxis(val, num_axis, 0)
for i in range(val.shape[0]): all_fields.append(val[i])
else: all_fields.append(val)
for ds in dss: ds.close()
if all_fields and lon_1d is not None and lat_1d is not None:
stacked_val = np.stack(all_fields, axis=0)
def set_coords(d_dict):
d_dict[f'lon_{fixed_key}'] = lon_1d.copy()
d_dict[f'lat_{fixed_key}'] = lat_1d.copy()
if 'lon' not in d_dict:
d_dict['lon'] = lon_1d.copy(); d_dict['lat'] = lat_1d.copy()
for d_dict in [d_p90[t_ft], d_p10[t_ft], d_prob1[t_ft], d_prob10[t_ft], d_prob20[t_ft], d_prob30[t_ft]]:
set_coords(d_dict)
with warnings.catch_warnings():
warnings.simplefilter("ignore", category=RuntimeWarning)
d_p90[t_ft][fixed_key] = np.nanpercentile(stacked_val, 90, axis=0)
d_p10[t_ft][fixed_key] = np.nanpercentile(stacked_val, 10, axis=0)
valid_counts = np.sum(~np.isnan(stacked_val), axis=0)
with np.errstate(divide='ignore', invalid='ignore'):
d_prob1[t_ft][fixed_key] = np.where(valid_counts > 0, np.sum(stacked_val >= 1.0, axis=0) / valid_counts * 100.0, np.nan)
d_prob10[t_ft][fixed_key] = np.where(valid_counts > 0, np.sum(stacked_val >= 10.0, axis=0) / valid_counts * 100.0, np.nan)
d_prob20[t_ft][fixed_key] = np.where(valid_counts > 0, np.sum(stacked_val >= 20.0, axis=0) / valid_counts * 100.0, np.nan)
d_prob30[t_ft][fixed_key] = np.where(valid_counts > 0, np.sum(stacked_val >= 30.0, axis=0) / valid_counts * 100.0, np.nan)
except Exception: pass
finally:
gc.collect()
try: os.remove(temp_bin)
except: pass
extract_and_process(f1_mini, "pall")
extract_and_process(f2_mini, "prrsf")
try:
if f1_mini != "NONE" and os.path.exists(f1_mini): os.remove(f1_mini)
if f2_mini != "NONE" and os.path.exists(f2_mini): os.remove(f2_mini)
except: pass
if d_p90[t_ft]:
stat_dicts = {"MAX": d_p90[t_ft], "MIN": d_p10[t_ft], "PROB1": d_prob1[t_ft], "PROB10": d_prob10[t_ft], "PROB20": d_prob20[t_ft], "PROB30": d_prob30[t_ft]}
for stat_name, data_dict in stat_dicts.items():
if not data_dict: continue
final_filepath = os.path.join(cache, f"{model}_GUID_{stat_name}_{init}_FT{t_ft:02d}.npz")
temp_filepath = os.path.join(cache, f"~tmp_{model}_GUID_{stat_name}_{init}_FT{t_ft:02d}.npz")
try:
np.savez_compressed(temp_filepath, **data_dict)
success = False
for _ in range(10):
try:
os.replace(temp_filepath, final_filepath); success = True; break
except PermissionError: time.sleep(0.5)
if not success:
try: os.remove(final_filepath)
except: pass
np.savez_compressed(final_filepath, **data_dict)
try:
if os.path.exists(temp_filepath): os.remove(temp_filepath)
except: pass
except Exception: pass
print(f"SUCCESS:{t_ft}", flush=True)
return
# ============================================================
# 【ANAL (毎時大気解析) 専用処理】 (Ver 5.3 準拠)
# ============================================================
elif mode == "ANAL":
d_all = {ft: {} for ft in target_fts}
for t_ft in target_fts:
d = d_all[t_ft]
f1_mini = slice_with_wgrib2(f1, t_ft, "1")
if f1_mini == "NONE": continue
try:
dss = cfgrib.open_datasets(f1_mini, backend_kwargs={'indexpath': ''})
for ds in dss:
lon = ds.longitude.values if hasattr(ds, 'longitude') else None
lat = ds.latitude.values if hasattr(ds, 'latitude') else None
if lon is None or lat is None: continue
lon_1d = lon[0, :] if lon.ndim == 2 else lon
lat_1d = lat[:, 0] if lat.ndim == 2 else lat
for v in ds.data_vars:
da = ds[v]
da_step = da.isel(step=0) if 'step' in da.dims else da
if 'number' in da_step.coords: da_step = da_step.isel(number=0)
sName = str(da.attrs.get('GRIB_shortName', v)).lower()
is_upper = 'isobaricInhPa' in da_step.coords or 'level' in da_step.coords
lon_key = 'lon_pall' if is_upper else 'lon_surf'
lat_key = 'lat_pall' if is_upper else 'lat_surf'
if not is_upper:
val = da_step.values.copy()
while val.ndim > 2: val = val[0]
try:
if sName in ['sp', 'pres'] and 'psfc' not in d:
_psfc = val / 100.0 if np.nanmax(val) > 2000 else val
if 300.0 < np.nanmax(_psfc) < 1200.0: d['psfc'] = _psfc.copy()
except Exception: pass
assign_key = None
if sName in ['prmsl', 'msl', 'mslet']: assign_key = 'slp'
elif sName in ['10u', 'u', 'u10']: assign_key = 'u10'
elif sName in ['10v', 'v', 'v10']: assign_key = 'v10'
elif sName in ['2t', 't', 't2m', 'temp']: assign_key = 't2m'
elif sName in ['2r', 'r', 'rh2m', 'rh']: assign_key = 'rh2m'
if assign_key:
if assign_key == 'slp' and np.nanmax(val) > 2000: val = val / 100.0
if assign_key == 't2m' and np.nanmax(val) > 150: val = val - 273.15
d[assign_key] = val
d[lon_key] = lon_1d; d[lat_key] = lat_1d
if 'lon' not in d: d['lon'] = lon_1d; d['lat'] = lat_1d
else:
levels = []
if 'isobaricInhPa' in da_step.coords: levels = np.atleast_1d(da_step.isobaricInhPa.values)
elif 'level' in da_step.coords: levels = np.atleast_1d(da_step.level.values)
for l_idx, lvl in enumerate(levels):
lvl = int(lvl)
if lvl in [300, 500, 600, 700, 850, 925, 950, 975]:
dim_n = 'isobaricInhPa' if 'isobaricInhPa' in da_step.coords else 'level'
val_l = da_step.isel(**{dim_n: l_idx}).values.copy() if len(levels) > 1 else da_step.values.copy()
while val_l.ndim > 2: val_l = val_l[0]
assign_key = None
if sName in ['t', 'temp']: assign_key = f't{lvl}'
elif sName in ['u', 'u-component']: assign_key = f'u{lvl}'
elif sName in ['v', 'v-component']: assign_key = f'v{lvl}'
elif sName in ['r', 'rh', 'humidity']: assign_key = f'r{lvl}'
elif sName in ['w', 'v-velocity', 'dz']: assign_key = f'w{lvl}'
elif sName in ['gh', 'z', 'geopotential']: assign_key = f'gh{lvl}'
if assign_key:
if assign_key.startswith('t') and np.nanmax(val_l) > 150: val_l = val_l - 273.15
d[assign_key] = val_l
d[lon_key] = lon_1d; d[lat_key] = lat_1d
for ds in dss: ds.close()
except Exception: pass
finally: gc.collect()
try:
if os.path.exists(f1_mini): os.remove(f1_mini)
except: pass
for t_ft in target_fts:
d = d_all[t_ft]
if not d: continue
for lvl in [300, 500, 600, 700, 850, 925, 950, 975]:
tc = d.get(f't{lvl}'); rh = d.get(f'r{lvl}')
if tc is not None and rh is not None:
rh_c = np.clip(rh, 0.1, 100)
e = 6.112 * np.exp((17.67*tc)/(tc+243.5)) * (rh_c/100.0)
td = (243.5*np.log(e/6.112))/(17.67-np.log(e/6.112))
d[f'tddep{lvl}'] = tc - td
tk = tc + 273.15; theta = tk*(1000.0/lvl)**0.2854; w = 0.622*e/(lvl-e)
d[f'ep{lvl}'] = theta * np.exp((2.5e6*w)/(1004.0*tk))
u500 = d.get('u500'); v500 = d.get('v500')
if u500 is not None and v500 is not None and 'lon_pall' in d and 'lat_pall' in d:
lon_2d, lat_2d = np.meshgrid(d['lon_pall'], d['lat_pall'])
rad_lat = np.deg2rad(lat_2d); rad_lon = np.deg2rad(lon_2d)
dy = 6371000.0 * np.gradient(rad_lat, axis=0)
dx = 6371000.0 * np.cos(rad_lat) * np.gradient(rad_lon, axis=1)
dx[dx == 0] = 1e-10; dy[dy == 0] = 1e-10
d['vort500'] = ((np.gradient(v500, axis=1) / dx) - (np.gradient(u500, axis=0) / dy)) * 1e5
final_filepath = os.path.join(cache, f"{model}_{init}_FT{t_ft:02d}.npz")
temp_filepath = os.path.join(cache, f"~tmp_{model}_{init}_FT{t_ft:02d}.npz")
try:
np.savez_compressed(temp_filepath, **d)
success = False
for _ in range(10):
try:
os.replace(temp_filepath, final_filepath); success = True; break
except PermissionError: time.sleep(0.5)
if not success:
try: os.remove(final_filepath)
except Exception: pass
np.savez_compressed(final_filepath, **d)
try:
if os.path.exists(temp_filepath): os.remove(temp_filepath)
except Exception: pass
print(f"SUCCESS:{t_ft}", flush=True)
except Exception as e:
print(f"DEBUG_EXCEPTION: Error saving FT={t_ft} - {e}", flush=True)
# ============================================================
# 【GPV (MSM/GSM/GSM_JP)】 キャンバス結合(パッチワーク)対応版 (Ver 5.3 準拠)
# ============================================================
elif mode == "GPV":
d_all = {ft: {} for ft in target_fts}
for t_ft in target_fts:
d = d_all[t_ft]
f1_mini = slice_with_wgrib2(f1, t_ft, "1")
f2_mini = slice_with_wgrib2(f2, t_ft, "2")
def process_file_gpv(filepath):
if filepath == "NONE": return
try:
dss = cfgrib.open_datasets(filepath, backend_kwargs={'indexpath': ''})
for ds in dss:
lon = ds.longitude.values if hasattr(ds, 'longitude') else None
lat = ds.latitude.values if hasattr(ds, 'latitude') else None
for v in ds.data_vars:
da = ds[v]
da_step = da.isel(step=0) if 'step' in da.dims else da
if 'number' in da_step.coords: da_step = da_step.isel(number=0)
sName = str(da.attrs.get('GRIB_shortName', v)).lower()
attrs_str = str(da.attrs).lower()
disc = da.attrs.get('GRIB_discipline', -1)
cat = da.attrs.get('GRIB_parameterCategory', -1)
num = da.attrs.get('GRIB_parameterNumber', -1)
is_upper = 'isobaricInhPa' in da_step.coords or 'level' in da_step.coords
def assign(k, val_to_assign, is_upper_flag):
if lon is None or lat is None:
d[k] = val_to_assign; return
lon_1d = lon[0, :] if lon.ndim == 2 else lon
lat_1d = lat[:, 0] if lat.ndim == 2 else lat
lon_key = 'lon_pall' if is_upper_flag else 'lon_surf'
lat_key = 'lat_pall' if is_upper_flag else 'lat_surf'
if k not in d:
d[k] = val_to_assign.copy()
d[lon_key] = lon_1d.copy(); d[lat_key] = lat_1d.copy()
if not is_upper_flag and 'lon' not in d:
d['lon'] = lon_1d.copy(); d['lat'] = lat_1d.copy()
else:
old_lon = d[lon_key]; old_lat = d[lat_key]; old_val = d[k]
if old_val.shape == val_to_assign.shape and np.array_equal(old_lon, lon_1d):
d[k] = val_to_assign.copy(); return
# 高度なパッチワーク結合 (全球データ + 局地データの融合)
new_lon = np.unique(np.concatenate([np.round(old_lon, 4), np.round(lon_1d, 4)]))
new_lat = np.unique(np.concatenate([np.round(old_lat, 4), np.round(lat_1d, 4)]))
new_lat = np.sort(new_lat)[::-1] if old_lat[0] > old_lat[-1] else np.sort(new_lat)
new_lon = np.sort(new_lon)
canvas = np.full((len(new_lat), len(new_lon)), np.nan)
lat_idx_old = np.where(np.isin(np.round(new_lat, 4), np.round(old_lat, 4)))[0]
lon_idx_old = np.where(np.isin(np.round(new_lon, 4), np.round(old_lon, 4)))[0]
if canvas[np.ix_(lat_idx_old, lon_idx_old)].shape == old_val.shape:
canvas[np.ix_(lat_idx_old, lon_idx_old)] = old_val
lat_idx_new = np.where(np.isin(np.round(new_lat, 4), np.round(lat_1d, 4)))[0]
lon_idx_new = np.where(np.isin(np.round(new_lon, 4), np.round(lon_1d, 4)))[0]
if canvas[np.ix_(lat_idx_new, lon_idx_new)].shape == val_to_assign.shape:
existing = canvas[np.ix_(lat_idx_new, lon_idx_new)]
mask = np.isnan(existing)
existing[mask] = val_to_assign[mask]
canvas[np.ix_(lat_idx_new, lon_idx_new)] = existing
d[k] = canvas; d[lon_key] = new_lon; d[lat_key] = new_lat
if 'lon' not in d:
d['lon'] = new_lon; d['lat'] = new_lat
if not is_upper:
val = da_step.values.copy()
while val.ndim > 2: val = val[0]
try:
if sName in ['sp', 'pres'] and 'psfc' not in d:
_psfc = val / 100.0 if np.nanmax(val) > 2000 else val
if 300.0 < np.nanmax(_psfc) < 1200.0: d['psfc'] = _psfc.copy()
except Exception: pass
if sName in ['prmsl', 'msl', 'mslet']: assign('slp', val / 100.0 if np.nanmax(val) > 2000 else val, is_upper)
elif sName in ['pres', 'sp'] or 'pressure' in attrs_str:
if 'slp' not in d and 'pres' not in d: assign('pres', val / 100.0 if np.nanmax(val) > 2000 else val, is_upper)
elif sName in ['10u', 'u', 'u10'] or ('u-component' in attrs_str and '10' in attrs_str): assign('u10', val, is_upper)
elif sName in ['10v', 'v', 'v10'] or ('v-component' in attrs_str and '10' in attrs_str): assign('v10', val, is_upper)
elif sName in ['2t', 't', 't2m', 'temp'] or ('temperature' in attrs_str and '2' in attrs_str): assign('t2m', val - 273.15 if np.nanmax(val) > 150 else val, is_upper)
elif sName in ['2r', 'r', 'rh2m', 'rh'] or ('humidity' in attrs_str): assign('rh2m', val, is_upper)
elif sName in ['tcc', 'hcc', 'mcc', 'lcc'] or (disc == 0 and cat == 6): assign(sName if sName != 'unknown' else f"var_{disc}_{cat}_{num}", val, is_upper)
elif sName in ['tp', 'apcp', 'pr', 'precip'] or 'precip' in attrs_str or 'accum' in attrs_str:
if 'precip' not in d: assign('precip', np.nan_to_num(val, nan=0.0), is_upper)
else:
levels = []
if 'isobaricInhPa' in da_step.coords: levels = np.atleast_1d(da_step.isobaricInhPa.values)
elif 'level' in da_step.coords: levels = np.atleast_1d(da_step.level.values)
for l_idx, lvl in enumerate(levels):
lvl = int(lvl)
if lvl in [300, 500, 600, 700, 850, 925, 950, 975]:
if len(levels) > 1:
dim_n = 'isobaricInhPa' if 'isobaricInhPa' in da_step.coords else 'level'
val_l = da_step.isel(**{dim_n: l_idx}).values.copy()
else: val_l = da_step.values.copy()
while val_l.ndim > 2: val_l = val_l[0]
if sName in ['t', 'temp'] or 'temperature' in attrs_str: assign(f't{lvl}', val_l - 273.15 if np.nanmax(val_l) > 150 else val_l, is_upper)
elif sName in ['u', 'u-component']: assign(f'u{lvl}', val_l, is_upper)
elif sName in ['v', 'v-component']: assign(f'v{lvl}', val_l, is_upper)
elif sName in ['r', 'rh', 'humidity']: assign(f'r{lvl}', val_l, is_upper)
elif sName in ['w', 'v-velocity', 'dz']: assign(f'w{lvl}', val_l, is_upper)
elif sName in ['gh', 'z', 'geopotential']: assign(f'gh{lvl}', val_l, is_upper)
for ds in dss: ds.close()
except Exception: pass
finally: gc.collect()
process_file_gpv(f1_mini)
process_file_gpv(f2_mini)
try:
if f1_mini != "NONE" and os.path.exists(f1_mini): os.remove(f1_mini)
if f2_mini != "NONE" and os.path.exists(f2_mini): os.remove(f2_mini)
except Exception: pass
for t_ft in target_fts:
d = d_all[t_ft]
if not d: continue
for lvl in [300, 500, 600, 700, 850, 925, 950, 975]:
tc = d.get(f't{lvl}'); rh = d.get(f'r{lvl}')
if tc is not None and rh is not None:
rh_c = np.clip(rh, 0.1, 100)
e = 6.112 * np.exp((17.67*tc)/(tc+243.5)) * (rh_c/100.0)
td = (243.5*np.log(e/6.112))/(17.67-np.log(e/6.112))
d[f'tddep{lvl}'] = tc - td
tk = tc + 273.15; theta = tk*(1000.0/lvl)**0.2854; w = 0.622*e/(lvl-e)
d[f'ep{lvl}'] = theta * np.exp((2.5e6*w)/(1004.0*tk))
u500 = d.get('u500'); v500 = d.get('v500')
if u500 is not None and v500 is not None:
vort_lon = d.get('lon_pall') if 'lon_pall' in d else d.get('lon_surf')
vort_lat = d.get('lat_pall') if 'lat_pall' in d else d.get('lat_surf')
if vort_lon is not None and vort_lat is not None:
d['vort500'] = calculate_vorticity(u500, v500, vort_lon, vort_lat)
final_filepath = os.path.join(cache, f"{model}_{init}_FT{t_ft:02d}.npz")
temp_filepath = os.path.join(cache, f"~tmp_{model}_{init}_FT{t_ft:02d}.npz")
try:
np.savez_compressed(temp_filepath, **d)
success = False
for _ in range(10):
try:
os.replace(temp_filepath, final_filepath); success = True; break
except PermissionError:
time.sleep(0.5)
if not success:
try: os.remove(final_filepath)
except Exception: pass
np.savez_compressed(final_filepath, **d)
try:
if os.path.exists(temp_filepath): os.remove(temp_filepath)
except Exception: pass
print(f"SUCCESS:{t_ft}", flush=True)
except Exception as e:
print(f"DEBUG_EXCEPTION: Error saving FT={t_ft} - {e}", flush=True)
# ============================================================
# 【GUID 専用処理 (MSM/GSMガイダンス)】 完全分離対応 ＋ 基礎気象要素復旧
# ============================================================
elif mode == "GUID":
d_all = {ft: {} for ft in target_fts}
def extract_guid_unified(fin, src_type, d_target, t_ft):
if fin == "NONE" or not os.path.exists(fin): return
res = subprocess.run([wgrib2_path, fin, "-s"], capture_output=True, text=True, creationflags=creationflags)
if not res.stdout.strip(): return
parsed_lines = []
for line in res.stdout.strip().split('\\n'):
if not line: continue
parts = line.split(':')
if len(parts) < 6: continue
param = parts[3]; time_str = parts[5].lower()
m = re.search(r'([0-9]+)-([0-9]+)\\s+(hour|day)', time_str)
duration = 0
if not m:
m2 = re.search(r'([0-9]+)\\s+(hour|day)', time_str)
if m2:
end_val = int(m2.group(1)); unit = m2.group(2)
if unit == 'day': end_val *= 24
duration = 0
else:
start_val = int(m.group(1)); end_val = int(m.group(2)); unit = m.group(3)
if unit == 'day': end_val *= 24; start_val *= 24
duration = end_val - start_val
var_name = "unknown"
if param in ["TPRATE", "APCP", "PRECIP"]: var_name = "tp"
elif param in ["TSRATE", "WEASD", "SNOW", "ASNOW"]: var_name = "snow"
elif param in ["TSTM", "THUND", "PRSTM"]: var_name = "thund"
elif param in ["VIS"] or "vis" in time_str: var_name = "vis"
elif "parm=192" in time_str or param in ["WEA", "NSWRS", "NSWRV", "WEATHER"]: var_name = "wea"
# --- Ver6.0 復旧: 気温、湿度、風速の抽出 ---
elif param in ["TMP", "T", "T2M", "TEMP"]: var_name = "t2m"
elif param in ["RH", "2R", "RH2M"]: var_name = "rh2m"
elif param in ["UGRD", "U10", "10U"]: var_name = "u10"
elif param in ["VGRD", "V10", "10V"]: var_name = "v10"
is_prob = "prob" in time_str
fixed_key = None
if var_name == "tp":
if is_prob: fixed_key = f"prob_precip{duration}"
elif src_type == "prrsf": fixed_key = f"precip{duration}max"
else: fixed_key = "precip" if duration == 3 else f"precip{duration}"
elif var_name == "snow":
fixed_key = f"snow{duration}"
elif var_name == "thund":
fixed_key = "thund" if duration == 3 else f"thund{duration}"
elif var_name == "vis":
fixed_key = "vis"
elif var_name == "wea":
fixed_key = "wea"
elif var_name in ["t2m", "rh2m", "u10", "v10"]:
fixed_key = var_name
if fixed_key:
parsed_lines.append( (fixed_key, line) )
grouped = {}
for k, line in parsed_lines: grouped.setdefault(k, []).append(line)
for fixed_key, group_lines in grouped.items():
inv_input = "\\n".join(group_lines) + "\\n"
temp_bin = os.path.join(cache, f"~tmp_ext_guid_{init}_{t_ft}_{src_type}_{fixed_key}.bin")
try:
cmd_extract = [wgrib2_path, fin, "-i", "-grib", temp_bin]
process = subprocess.Popen(cmd_extract, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, creationflags=creationflags)
process.communicate(input=inv_input.encode('utf-8'), timeout=60)
except: pass
if os.path.exists(temp_bin) and os.path.getsize(temp_bin) > 0:
try:
dss = cfgrib.open_datasets(temp_bin, backend_kwargs={'indexpath': ''})
for ds in dss:
lon = ds.coords['longitude'].values if 'longitude' in ds.coords else (ds.longitude.values if hasattr(ds, 'longitude') else None)
lat = ds.coords['latitude'].values if 'latitude' in ds.coords else (ds.latitude.values if hasattr(ds, 'latitude') else None)
lon_1d = None
if lon is not None and lat is not None:
lon_1d = lon[0, :] if lon.ndim == 2 else lon
lat_1d = lat[:, 0] if lat.ndim == 2 else lat
for v in ds.data_vars:
da = ds[v]
da_step = da.isel(step=0) if 'step' in da.dims else da
val = da_step.values.copy()
while val.ndim > 2: val = val[0]
# --- Ver6.0 復旧: 単位変換 ---
if fixed_key == "t2m" and np.nanmax(val) > 150: val = val - 273.15
if "snow" in fixed_key: val = val * 100.0 # m -> cm
d_target[fixed_key] = val
if lon_1d is not None:
d_target[f'lon_{fixed_key}'] = lon_1d
d_target[f'lat_{fixed_key}'] = lat_1d
if 'lon' not in d_target:
d_target['lon'] = lon_1d
d_target['lat'] = lat_1d
for ds in dss: ds.close()
except: pass
finally:
try: os.remove(temp_bin)
except: pass
for t_ft in target_fts:
f1_mini = slice_with_wgrib2(f1, t_ft, "pall")
f2_mini = slice_with_wgrib2(f2, t_ft, "prrsf")
extract_guid_unified(f1_mini, "pall", d_all[t_ft], t_ft)
extract_guid_unified(f2_mini, "prrsf", d_all[t_ft], t_ft)
try:
if f1_mini != "NONE" and os.path.exists(f1_mini): os.remove(f1_mini)
if f2_mini != "NONE" and os.path.exists(f2_mini): os.remove(f2_mini)
except: pass
d = d_all[t_ft]
if not d: continue
final_filepath = os.path.join(cache, f"{model}_GUID_{init}_FT{t_ft:02d}.npz")
temp_filepath = os.path.join(cache, f"~tmp_{model}_GUID_{init}_FT{t_ft:02d}.npz")
try:
np.savez_compressed(temp_filepath, **d)
success = False
for _ in range(10):
try:
os.replace(temp_filepath, final_filepath); success = True; break
except PermissionError:
time.sleep(0.5)
if not success:
try: os.remove(final_filepath)
except Exception: pass
np.savez_compressed(final_filepath, **d)
try:
if os.path.exists(temp_filepath): os.remove(temp_filepath)
except Exception: pass
print(f"SUCCESS:{t_ft}", flush=True)
except Exception as e:
print(f"DEBUG_EXCEPTION: Error saving FT={t_ft} - {e}", flush=True)
except BaseException as e:
print(f"CRITICAL_ERROR: {traceback.format_exc()}", flush=True)
sys.exit(1)
if __name__ == '__main__': main()
"""
try:
with open(WORKER_SCRIPT, "w", encoding="utf-8") as f: f.write(WORKER_CODE)
except Exception: pass
# ==========================================
# スキャン・スケジューリング担当スレッド
# ==========================================
class DataParserThread(QThread):
log_signal = pyqtSignal(str); progress_signal = pyqtSignal(int, str); finished_signal = pyqtSignal(bool)
MODEL_CYCLE = ["MSM_GPV", "MSM_GUID", "MEPS_GUID", "GSM_GUID", "GSM_GPV"]
MSM_INTERRUPT_KEYS = ["MSM_GPV", "MSM_GUID"]
def __init__(self, folder_paths, output_dir, last_reported, scan_mode="12h"):
super().__init__()
self.folder_paths = [path for path in folder_paths if path and os.path.exists(path)]
self.output_dir = output_dir
self.last_reported = last_reported
self.scan_mode = scan_mode
self.abort = False
def _scan_sources(self):
now_utc = datetime.utcnow()
sources = {m: {} for m in self.MODEL_CYCLE}
anal_groups_all = {}
meps_gpv_groups_all = {}
def get_all_init_groups(file_list, allow_all_time=True):
groups = {}
for f in file_list:
m = re.search(r'_(\d{14})_', os.path.basename(f))
if not m: continue
init = m.group(1)
if not allow_all_time:
try:
file_time = datetime.strptime(init, '%Y%m%d%H%M%S')
if (now_utc - file_time).total_seconds() > 12 * 3600:
continue
except Exception: pass
if init not in groups: groups[init] = []
groups[init].append(f)
return groups
gsm_ft_target = list(range(0, 135, 3)) + list(range(138, 265, 6))
for folder_path in self.folder_paths:
msm_files_all = glob.glob(os.path.join(folder_path, '**', '*MSM*GPV*.bin'), recursive=True)
gsm_files_raw = glob.glob(os.path.join(folder_path, '**', '*GSM*GPV*.bin'), recursive=True)
gsm_jp_files = [f for f in gsm_files_raw if 'Rgl' not in f and ('Rjp' in f or 'Japan' in f or 'jp' in f.lower())]
gsm_files_all = gsm_jp_files if gsm_jp_files else gsm_files_raw
# --- GUID (P-all と P-rrsf/prr を分離・Ver 5.6) ---
msm_guid_p = glob.glob(os.path.join(folder_path, '**', '*MSM_GUID*P-all*Toorg*.bin'), recursive=True) + glob.glob(os.path.join(folder_path, '**', '*MSM_GUID*Pall*Toorg*.bin'), recursive=True)
msm_guid_pr = glob.glob(os.path.join(folder_path, '**', '*MSM_GUID*P-prr*Toorg*.bin'), recursive=True) + glob.glob(os.path.join(folder_path, '**', '*MSM_GUID*prr*Toorg*.bin'), recursive=True)
gsm_guid_p = glob.glob(os.path.join(folder_path, '**', '*GSM_GUID*P-all*Toorg*.bin'), recursive=True) + glob.glob(os.path.join(folder_path, '**', '*GSM_GUID*Pall*Toorg*.bin'), recursive=True) + glob.glob(os.path.join(folder_path, '**', '*GSM_GUID*gsm*Toorg*.bin'), recursive=True)
gsm_guid_pr = glob.glob(os.path.join(folder_path, '**', '*GSM_GUID*P-rrsf*Toorg*.bin'), recursive=True) + glob.glob(os.path.join(folder_path, '**', '*GSM_GUID*rrsf*Toorg*.bin'), recursive=True)
msm_guid_p_groups = get_all_init_groups(msm_guid_p)
msm_guid_pr_groups = get_all_init_groups(msm_guid_pr)
if not msm_guid_p_groups:
fallback_msm = glob.glob(os.path.join(folder_path, '**', '*MSM_GUID*Toorg*.bin'), recursive=True)
msm_guid_p_groups = get_all_init_groups([f for f in fallback_msm if 'prr' not in f.lower()])
gsm_guid_p_groups = get_all_init_groups(gsm_guid_p)
gsm_guid_pr_groups = get_all_init_groups(gsm_guid_pr)
if not gsm_guid_p_groups:
fallback_gsm = glob.glob(os.path.join(folder_path, '**', '*GSM_GUID*Toorg*.bin'), recursive=True)
gsm_guid_p_groups = get_all_init_groups([f for f in fallback_gsm if 'rrsf' not in f.lower()])
meps_gpv = glob.glob(os.path.join(folder_path, '**', '*MEPS_GPV*.bin'), recursive=True)
anal_files_all = glob.glob(os.path.join(folder_path, '**', '*ANAL_grib2*.bin'), recursive=True)
meps_guid_p = glob.glob(os.path.join(folder_path, '**', '*MEPS_GUID*P-all*grib2*'), recursive=True) + glob.glob(os.path.join(folder_path, '**', '*MEPS*Pall*grib2*'), recursive=True)
meps_guid_pr = glob.glob(os.path.join(folder_path, '**', '*MEPS_GUID*Prrsf*grib2*'), recursive=True) + glob.glob(os.path.join(folder_path, '**', '*MEPS*Prrst*grib2*'), recursive=True)
for init_str, files in get_all_init_groups(msm_files_all).items():
sources["MSM_GPV"].setdefault(init_str, {"model_name": "MSM", "mode": "GPV", "ft_list": list(range(0, 79, 1)), "files": []})["files"].extend(files)
for init_str, files in get_all_init_groups(gsm_files_all).items():
sources["GSM_GPV"].setdefault(init_str, {"model_name": "GSM_JP", "mode": "GPV", "ft_list": gsm_ft_target, "files": []})["files"].extend(files)
# --- GUID ペア登録 ---
for init_str in set(msm_guid_p_groups.keys()) | set(msm_guid_pr_groups.keys()):
entry = sources["MSM_GUID"].setdefault(init_str, {
"model_name": "MSM", "mode": "GUID", "ft_list": list(range(0, 85, 1)),
"f_pall": "NONE", "f_prrst": "NONE"
})
if init_str in msm_guid_p_groups: entry["f_pall"] = msm_guid_p_groups[init_str][0]
if init_str in msm_guid_pr_groups: entry["f_prrst"] = msm_guid_pr_groups[init_str][0]
for init_str in set(gsm_guid_p_groups.keys()) | set(gsm_guid_pr_groups.keys()):
entry = sources["GSM_GUID"].setdefault(init_str, {
"model_name": "GSM", "mode": "GUID", "ft_list": gsm_ft_target,
"f_pall": "NONE", "f_prrst": "NONE"
})
if init_str in gsm_guid_p_groups: entry["f_pall"] = gsm_guid_p_groups[init_str][0]
if init_str in gsm_guid_pr_groups: entry["f_prrst"] = gsm_guid_pr_groups[init_str][0]
# --- MEPS ペア登録 ---
pall_groups = get_all_init_groups(meps_guid_p)
prrst_groups = get_all_init_groups(meps_guid_pr)
for init_str in set(pall_groups.keys()) | set(prrst_groups.keys()):
entry = sources["MEPS_GUID"].setdefault(init_str, {
"model_name": "MEPS", "mode": "MEPS_GUID", "ft_list": list(range(3, 40, 3)),
"f_pall": "NONE", "f_prrst": "NONE"})
if init_str in pall_groups: entry["f_pall"] = pall_groups[init_str][0]
if init_str in prrst_groups: entry["f_prrst"] = prrst_groups[init_str][0]
for init_str, files in get_all_init_groups(anal_files_all).items():
anal_groups_all.setdefault(init_str, {"model_name": "ANAL", "mode": "ANAL", "ft_list": [0], "files": []})["files"].extend(files)
for init_str, files in get_all_init_groups(meps_gpv).items():
meps_gpv_groups_all.setdefault(init_str, {"model_name": "MEPS", "mode": "GPV", "ft_list": list(range(0, 40, 3)), "files": []})["files"].extend(files)
return sources, anal_groups_all, meps_gpv_groups_all
def _sorted_inits_desc(self, sources, model_key):
return sorted(sources.get(model_key, {}).keys(), reverse=True)
def _within_window(self, init_str):
if self.scan_mode == "all": return True
try:
file_time = datetime.strptime(init_str, '%Y%m%d%H%M%S')
return (datetime.utcnow() - file_time).total_seconds() <= 12 * 3600
except Exception:
return True
def _target_npz_names(self, model_key, model_name, mode, init_str, ft_list):
if mode == "MEPS_GUID":
names = []
for ft in ft_list:
for stat in ["MAX", "MIN", "PROB1", "PROB10", "PROB20", "PROB30"]:
names.append(f"{model_name}_GUID_{stat}_{init_str}_FT{ft:02d}.npz")
return names
pfx = "GUID_" if mode == "GUID" else ""
return [f"{model_name}_{pfx}{init_str}_FT{ft:02d}.npz" for ft in ft_list]
def _is_fully_done(self, model_key, entry, init_str):
names = self._target_npz_names(model_key, entry["model_name"], entry["mode"], init_str, entry["ft_list"])
return all(os.path.exists(os.path.join(self.output_dir, n)) for n in names)
def _missing_fts(self, model_key, entry, init_str):
if entry["mode"] == "MEPS_GUID":
missing = []
for ft in entry["ft_list"]:
names = [f"MEPS_GUID_{stat}_{init_str}_FT{ft:02d}.npz" for stat in ["MAX", "MIN", "PROB1", "PROB10", "PROB20", "PROB30"]]
if not all(os.path.exists(os.path.join(self.output_dir, n)) for n in names):
missing.append(ft)
return missing
pfx = "GUID_" if entry["mode"] == "GUID" else ""
missing = []
for ft in entry["ft_list"]:
name = f"{entry['model_name']}_{pfx}{init_str}_FT{ft:02d}.npz"
if not os.path.exists(os.path.join(self.output_dir, name)):
missing.append(ft)
return missing
def _get_file_for_ft(self, files_list, keyword, ft):
for f in files_list:
if keyword not in f: continue
m_fd = re.search(r'FD(\d{2})(\d{2})-(\d{2})(\d{2})', os.path.basename(f))
if m_fd:
start_h = int(m_fd.group(1))*24 + int(m_fd.group(2))
end_h = int(m_fd.group(3))*24 + int(m_fd.group(4))
if start_h <= ft <= end_h: return f
continue
m_fh = re.search(r'FH(\d{2,3})-(\d{2,3})', os.path.basename(f))
if m_fh:
start_h = int(m_fh.group(1)); end_h = int(m_fh.group(2))
if start_h <= ft <= end_h: return f
continue
return f
return None
def _process_one(self, model_key, entry, init_str, creationflags, env):
missing = self._missing_fts(model_key, entry, init_str)
if not missing: return False
model_name = entry["model_name"]; mode = entry["mode"]
# 【MEPS / GUID の同期抽出チェック】 P-all と Prrsf の両方が揃うまで処理保留
if mode in ["MEPS_GUID", "GUID"]:
f_pall = entry.get("f_pall", "NONE"); f_prrst = entry.get("f_prrst", "NONE")
if f_pall == "NONE" or f_prrst == "NONE": return False
proc_label = "MEPS(アンサンブル)" if mode == "MEPS_GUID" else "ガイダンス"
self.log_signal.emit(f"🚀 [{model_key}] {model_name} {proc_label} ({init_str}) 抽出開始 (対象: {len(missing)}件)...")
total_extracted = 0
for i in range(0, len(missing), 5):
if self.abort: break
chunk_fts = missing[i:i + 5]; ft_str = ",".join(map(str, chunk_fts))
cmd = [sys.executable, WORKER_SCRIPT, model_name, mode, self.output_dir, init_str, ft_str, f_pall, f_prrst, WGRIB2_EXE]
total_extracted = self._run_worker_and_report(cmd, model_name, init_str, proc_label, len(missing), total_extracted)
return True
files = entry.get("files", [])
file_groups = {}
for ft in missing:
if mode == "GPV":
f1 = self._get_file_for_ft(files, "Lsurf", ft) or "NONE"
f2 = self._get_file_for_ft(files, "L-pall", ft) or "NONE"
if f1 == "NONE": continue
# L-pallの隙間処理
if f2 == "NONE" and ft not in [16, 17, 34, 35, 40, 41, 52, 53, 87]: continue
elif mode == "ANAL":
f1 = files[0] if files else "NONE"; f2 = "NONE"
if f1 == "NONE": continue
else:
f1 = files[0] if files else "NONE"; f2 = "NONE"
if f1 == "NONE": continue
key = (f1, f2)
file_groups.setdefault(key, []).append(ft)
if not file_groups: return False
proc_label = "GPV" if mode in ["GPV", "ANAL"] else "ガイダンス"
total_target = sum(len(fts) for fts in file_groups.values())
self.log_signal.emit(f"🚀 [{model_key}] {model_name} {proc_label} ({init_str}) 抽出開始 (対象: {total_target}件)...")
total_extracted = 0
for (f1, f2), fts in file_groups.items():
if self.abort: break
for i in range(0, len(fts), 5):
if self.abort: break
chunk_fts = fts[i:i + 5]; ft_str = ",".join(map(str, chunk_fts))
cmd = [sys.executable, WORKER_SCRIPT, model_name, mode, self.output_dir, init_str, ft_str, f1, f2, WGRIB2_EXE]
total_extracted = self._run_worker_and_report(cmd, model_name, init_str, proc_label, total_target, total_extracted)
return True
def _run_worker_and_report(self, cmd, model_name, init_str, proc_label, total_target, total_extracted=0):
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
conda_dir = os.path.dirname(sys.executable); lib_bin = os.path.join(conda_dir, "Library", "bin")
env = os.environ.copy(); env["PATH"] = f"{lib_bin};{env.get('PATH', '')}"; env["PYTHONIOENCODING"] = "utf-8"
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding='utf-8', errors='replace', creationflags=creationflags, env=env)
for line in iter(process.stdout.readline, ''):
if self.abort:
try: process.kill()
except Exception: pass
break
line = line.strip()
if not line: continue
if line.startswith("SUCCESS:"):
ft_success = line.split(":")[1]; total_extracted += 1
self.progress_signal.emit(int((total_extracted / max(total_target, 1)) * 100), f"⚡ {model_name} {init_str} FT={ft_success}h...")
self.log_signal.emit(f"⚡ 抽出完了: {model_name} {proc_label} FT={ft_success}h")
elif line.startswith("WARN:") or "CRITICAL_ERROR:" in line or "Traceback" in line or "Error:" in line or "File " in line:
self.log_signal.emit(f"⚠️ {line}")
try:
process.stdout.close(); process.wait()
except Exception: pass
return total_extracted
def run(self):
if not self.folder_paths: self.finished_signal.emit(False); return
if not os.path.exists(WGRIB2_EXE):
self.log_signal.emit("⚠️ wgrib2が見つかりません。"); self.finished_signal.emit(False); return
has_new_data = False
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
env = os.environ.copy()
try:
sources, anal_groups_all, meps_gpv_groups_all = self._scan_sources()
round_idx = {m: 0 for m in self.MODEL_CYCLE}
known_latest_anal = None
known_latest_msm = {}
for k in self.MSM_INTERRUPT_KEYS:
inits = self._sorted_inits_desc(sources, k)
known_latest_msm[k] = inits[0] if inits else None
def try_process(model_key, init_str):
nonlocal has_new_data
entry = sources[model_key][init_str]
if self._is_fully_done(model_key, entry, init_str): return False
did = self._process_one(model_key, entry, init_str, creationflags, env)
if did: has_new_data = True
return did
def try_process_anal(init_str):
nonlocal has_new_data
entry = anal_groups_all[init_str]
if self._is_fully_done("ANAL", entry, init_str): return False
did = self._process_one("ANAL", entry, init_str, creationflags, env)
if did: has_new_data = True
return did
while not self.abort:
progressed = False
# ★ 0. 【ANAL 最優先割り込み処理】
if anal_groups_all:
inits_anal = sorted(anal_groups_all.keys(), reverse=True)
latest_anal = inits_anal[0] if inits_anal else None
if latest_anal and latest_anal != known_latest_anal:
if self._within_window(latest_anal):
self.log_signal.emit(f"⚡ 割り込み: 毎時大気解析(ANAL)に新しい初期時刻({latest_anal})を検出、最優先処理します")
if try_process_anal(latest_anal): progressed = True
known_latest_anal = latest_anal
# ★ 1. MSM 割り込み
for msm_key in self.MSM_INTERRUPT_KEYS:
if self.abort: break
inits = self._sorted_inits_desc(sources, msm_key)
latest = inits[0] if inits else None
if latest and latest != known_latest_msm.get(msm_key):
self.log_signal.emit(f"⚡ 割り込み: {msm_key} に新しい初期時刻({latest})を検出、優先処理します")
if try_process(msm_key, latest): progressed = True
known_latest_msm[msm_key] = latest
# ★ 2. メインサイクル
for model_key in self.MODEL_CYCLE:
if self.abort: break
if anal_groups_all:
inits_anal = sorted(anal_groups_all.keys(), reverse=True)
latest_anal = inits_anal[0] if inits_anal else None
if latest_anal and latest_anal != known_latest_anal:
if self._within_window(latest_anal):
self.log_signal.emit(f"⚡ 割り込み: 毎時大気解析(ANAL)に新しい初期時刻({latest_anal})を検出、最優先処理します")
if try_process_anal(latest_anal): progressed = True
known_latest_anal = latest_anal
for msm_key in self.MSM_INTERRUPT_KEYS:
inits = self._sorted_inits_desc(sources, msm_key)
latest = inits[0] if inits else None
if latest and latest != known_latest_msm.get(msm_key):
self.log_signal.emit(f"⚡ 割り込み: {msm_key} に新しい初期時刻({latest})を検出、優先処理します")
if try_process(msm_key, latest): progressed = True
known_latest_msm[msm_key] = latest
inits = self._sorted_inits_desc(sources, model_key)
idx = round_idx[model_key]
if idx >= len(inits): continue
init_str = inits[idx]
if idx > 0 and not self._within_window(init_str): continue
if self._is_fully_done(model_key, sources[model_key][init_str], init_str):
round_idx[model_key] += 1; continue
if try_process(model_key, init_str): progressed = True
round_idx[model_key] += 1
if model_key in self.MSM_INTERRUPT_KEYS:
inits2 = self._sorted_inits_desc(sources, model_key)
if inits2: known_latest_msm[model_key] = inits2[0]
# ★ 3. 通常サイクル後の処理 (MEPS_GPVなど)
for label, groups in [("MEPS_GPV", meps_gpv_groups_all)]:
if self.abort: break
inits = sorted(groups.keys(), reverse=True)
if not inits: continue
init_str = inits[0]
entry = groups[init_str]
if self._within_window(init_str) and not self._is_fully_done(label, entry, init_str):
if self._process_one(label, entry, init_str, creationflags, env):
progressed = True; has_new_data = True
if not progressed: break
if not self.abort and has_new_data:
self.log_signal.emit("✅ 全てのスキャン・抽出が完了しました")
except Exception as e:
self.log_signal.emit(f"⚠️ スキャン処理で予期しないエラーが発生しました: {e}")
finally:
self.finished_signal.emit(has_new_data)
class EngineStatusWindow(QWidget):
def __init__(self):
super().__init__()
self.setWindowTitle("GPV 統合解析エンジン (Ver MERGED_6.0 ハイブリッド完成版)")
self.setFixedSize(900, 660)
self.setStyleSheet("""
QWidget { background-color: #0A192F; color: #E0E0E0; font-family: 'MS Gothic'; font-size: 11pt; }
QLabel { color: #64FFDA; font-weight: bold; }
QPushButton { background-color: #1D3557; color: white; padding: 6px; border-radius: 4px; font-weight: bold; border: 1px solid #457B9D; }
QPushButton:hover { background-color: #457B9D; }
QListWidget { background-color: #112240; border: 1px solid #457B9D; color: #64FFDA; padding: 5px; font-family: Consolas, monospace; font-size: 10pt; }
""")
self.is_quitting = False
layout = QVBoxLayout(self)
self.status_label = QLabel("状態: 起動中..."); self.status_label.setStyleSheet("font-size: 14pt;")
layout.addWidget(self.status_label)
control_layout = QHBoxLayout()
self.pause_btn = QPushButton("⏹ 抽出を即時停止")
self.pause_btn.setStyleSheet("background-color: #C0392B;")
self.resume_btn = QPushButton("▶ 手動強制スキャン")
self.mode_btn = QPushButton("🔄 モード: 最新12時間 (推奨)")
self.mode_btn.setStyleSheet("background-color: #D35400;")
for btn in [self.pause_btn, self.resume_btn, self.mode_btn]: control_layout.addWidget(btn)
self.clear_btn = QPushButton("🧹 キャッシュ全削除"); self.clear_btn.setStyleSheet("background-color: #8E44AD;")
control_layout.addWidget(self.clear_btn)
layout.addLayout(control_layout)
out_layout = QHBoxLayout()
self.out_folder_btn = QPushButton("💾 出力先フォルダを設定"); self.out_folder_btn.setStyleSheet("background-color: #2980B9;")
out_layout.addWidget(self.out_folder_btn)
self.out_folder_label = QLabel("未設定"); self.out_folder_label.setStyleSheet("background: #112240; padding: 4px; border: 1px solid #457B9D; color: #8892B0;")
out_layout.addWidget(self.out_folder_label, stretch=1)
layout.addLayout(out_layout)
self.log_list = QListWidget(); layout.addWidget(self.log_list)
self.copy_btn = QPushButton("📋 ログをコピー"); self.copy_btn.setStyleSheet("background-color: #16A085;"); layout.addWidget(self.copy_btn)
folder_grid = QGridLayout()
self.folder_btns = [QPushButton(f"📁 読込元フォルダ {i+1}") for i in range(6)]
for i, b in enumerate(self.folder_btns):
folder_grid.addWidget(b, i // 3, i % 3)
layout.addLayout(folder_grid)
self.exit_btn = QPushButton("システムを完全に終了"); self.exit_btn.setStyleSheet("background-color: #8b0000;"); layout.addWidget(self.exit_btn)
def log(self, message):
self.log_list.addItem(f"[{datetime.now().strftime('%H:%M:%S')}] {message}")
self.log_list.scrollToBottom(); write_syslog(message)
def closeEvent(self, event):
if getattr(self, 'is_quitting', False):
event.accept()
else:
event.ignore()
self.hide()
if hasattr(self, 'tray_msg_callback'):
self.tray_msg_callback()
class TrayApp(QApplication):
def __init__(self, sys_argv):
super().__init__(sys_argv)
self.setQuitOnLastWindowClosed(False)
self.settings = QSettings("SapporoWeatherApp", "GPVEngine")
self.monitor_folders = [self.settings.value(f"watch_dir_{i+1}", "") for i in range(6)]
global CURRENT_OUTPUT_DIR
self.output_dir = self.settings.value("output_dir", DEFAULT_OUTPUT_DIR)
CURRENT_OUTPUT_DIR = self.output_dir
os.makedirs(self.output_dir, exist_ok=True)
self.last_reported = {}
self.scan_mode = "12h"
self.window = EngineStatusWindow()
self.window.tray_msg_callback = self.show_tray_message
self.tray_icon = QSystemTrayIcon(self)
self.tray_icon.setIcon(create_lightning_icon())
self.tray_icon.setToolTip("GPV統合抽出エンジン (稼働中)")
tray_menu = QMenu(); show_action = QAction("ウィンドウを表示"); show_action.triggered.connect(self.restore_window)
quit_action = QAction("完全に終了する"); quit_action.triggered.connect(self.quit_app)
tray_menu.addAction(show_action); tray_menu.addAction(quit_action)
self.tray_icon.setContextMenu(tray_menu); self.tray_icon.activated.connect(self.tray_icon_activated); self.tray_icon.show()
with open(os.path.join(self.output_dir, "system_log.txt"), "w", encoding="utf-8") as f: pass
self.window.log("===============================")
self.window.log("システム起動: Ver MERGED_6.0 (真のハイブリッド完成版)")
self.window.log("===============================")
self.window.out_folder_label.setText(self.output_dir)
self.window.out_folder_btn.clicked.connect(self.select_output_folder)
folders_set = False
for i in range(6):
self.window.folder_btns[i].clicked.connect(lambda checked, idx=i: self.select_folder(idx))
self.update_folder_btn_text(i)
if self.monitor_folders[i]: folders_set = True
if not folders_set: self.window.status_label.setText("状態: 読込元フォルダ未設定")
self.window.copy_btn.clicked.connect(self.copy_logs)
self.window.pause_btn.clicked.connect(self.stop_and_pause)
self.window.resume_btn.clicked.connect(self.force_scan)
self.window.mode_btn.clicked.connect(self.toggle_mode)
self.window.clear_btn.clicked.connect(self.clear_cache)
self.window.exit_btn.clicked.connect(self.quit_app)
self.timer = QTimer(); self.timer.timeout.connect(self.run_parser); self.timer.start(10000)
self.is_processing = False; self.run_parser(); self.window.show()
def kill_current_worker(self):
if hasattr(self, 'worker') and self.worker.isRunning():
self.window.log("⚠️ 設定変更のため、現在の抽出処理を即座に中断しています...")
self.worker.abort = True
self.is_processing = False
def tray_icon_activated(self, reason):
if reason == QSystemTrayIcon.ActivationReason.DoubleClick: self.restore_window()
def restore_window(self):
self.window.showNormal(); self.window.activateWindow()
def show_tray_message(self):
self.tray_icon.showMessage("エンジン待機中", "ウィンドウを閉じましたが、裏で抽出を続けています。終了する場合は右クリックから「完全に終了する」を選んでください。", QIcon(self.tray_icon.icon()), 3000)
def quit_app(self):
self.kill_current_worker()
self.window.is_quitting = True
self.tray_icon.showMessage("システム終了", "エンジンを完全に停止しました。", QIcon(self.tray_icon.icon()), 2000)
self.quit()
def stop_and_pause(self):
self.timer.stop()
self.kill_current_worker()
self.window.log("⏸ 処理を即時停止しました。再開するには強制スキャンを押してください。")
self.window.status_label.setText("状態: 停止中")
def toggle_mode(self):
self.kill_current_worker()
if self.scan_mode == "12h":
self.scan_mode = "all"
self.window.mode_btn.setText("🔄 モード: すべてのデータ")
self.window.mode_btn.setStyleSheet("background-color: #27AE60;")
self.window.log("🔄 モード変更 [すべてのデータ] -> 再スキャンします")
else:
self.scan_mode = "12h"
self.window.mode_btn.setText("🔄 モード: 最新12時間 (推奨)")
self.window.mode_btn.setStyleSheet("background-color: #D35400;")
self.window.log("🔄 モード変更 [最新12時間] -> 再スキャンします")
self.run_parser()
def select_output_folder(self):
global CURRENT_OUTPUT_DIR
folder = QFileDialog.getExistingDirectory(self.window, "出力先フォルダ(共有)を選択", self.output_dir)
if folder:
self.kill_current_worker()
self.output_dir = folder; CURRENT_OUTPUT_DIR = folder; self.settings.setValue("output_dir", folder)
self.window.out_folder_label.setText(folder); self.window.log(f"💾 出力先フォルダを更新しました: {folder}")
self.run_parser()
def clear_cache(self):
reply = QMessageBox.question(self.window, "確認", "現在の出力先のデータをすべて削除し再抽出しますか？", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No)
if reply == QMessageBox.StandardButton.Yes:
self.timer.stop()
self.kill_current_worker()
self.window.log("🗑️ キャッシュデータのお掃除を開始します...")
for f in glob.glob(os.path.join(self.output_dir, "*.npz")) + glob.glob(os.path.join(self.output_dir, "*.json")) + glob.glob(os.path.join(self.output_dir, "*.bin")):
try: os.remove(f)
except Exception: pass
for d in self.monitor_folders:
if d:
for idx_f in glob.glob(os.path.join(d, '**', '*.idx'), recursive=True):
try: os.remove(idx_f)
except Exception: pass
self.last_reported = {}; self.window.log("✨ データを削除しました。再抽出を開始します！")
self.run_parser(); self.timer.start(10000)
def select_folder(self, index):
folder = QFileDialog.getExistingDirectory(self.window, "読込元フォルダ選択")
if folder:
self.kill_current_worker()
self.monitor_folders[index] = folder; self.settings.setValue(f"watch_dir_{index+1}", folder)
self.update_folder_btn_text(index); self.window.log(f"フォルダ{index+1}を更新: {folder}")
self.run_parser()
def update_folder_btn_text(self, i):
f = self.monitor_folders[i]; self.window.folder_btns[i].setText(f"📁 {os.path.basename(f)}" if f else f"📁 未設定 {i+1}")
def copy_logs(self): QApplication.clipboard().setText("\n".join([self.window.log_list.item(i).text() for i in range(self.window.log_list.count())]))
def force_scan(self):
self.window.log("▶ 手動強制スキャンを開始します...")
self.timer.start(10000)
self.kill_current_worker()
self.run_parser()
def run_parser(self):
if self.is_processing: return
if not any(self.monitor_folders): self.window.status_label.setText("状態: 読込元フォルダ未設定"); return
self.is_processing = True
self.worker = DataParserThread(self.monitor_folders, self.output_dir, self.last_reported, self.scan_mode)
self.worker.log_signal.connect(self.window.log); self.worker.progress_signal.connect(lambda v, t: self.window.status_label.setText(f"状態: {t}"))
self.worker.finished_signal.connect(self.on_parse_finished)
self.worker.start()
def on_parse_finished(self, has_new_data):
self.is_processing = False
if not hasattr(self.worker, 'abort') or not self.worker.abort:
self.window.status_label.setText("状態: 待機中 (自動スキャン継続中)")
if __name__ == '__main__':
import multiprocessing
multiprocessing.freeze_support()
if len(sys.argv) > 1 and '_gpv_worker.py' in sys.argv[1]:
sys.argv = sys.argv[1:]
try:
with open(sys.argv[0], "r", encoding="utf-8") as f:
worker_code = f.read()
exec(worker_code, {'__name__': '__main__'})
except Exception:
pass
sys.exit(0)
app = TrayApp(sys.argv); sys.exit(app.exec())
未分類