def run_grib_db_extraction(self, data_type_key, data_name, silent=False):
folder_path = self.grib_folders.get(data_type_key)
if not folder_path or not os.path.exists(folder_path):
if not silent: QMessageBox.warning(self, "警告", f"{data_name} のフォルダが設定されていません。")
return None
if data_type_key == "rain":
pattern = "**/*Prr60lv*.bin"
col_name = "Rainfall1h"
elif data_type_key == "snowfall":
pattern = "**/*Psflv*.bin"
col_name = "Snowfall1h"
else:
pattern = "**/*Psdlv*.bin"
col_name = "SnowDepth"
sql_upsert = f"INSERT INTO OBS_GP (ObsDay, ObsTime, MsCd, {col_name}) VALUES (?, ?, ?, ?) ON CONFLICT(ObsDay, ObsTime, MsCd) DO UPDATE SET {col_name} = excluded.{col_name};"
bin_files = glob.glob(os.path.join(folder_path, pattern), recursive=True)
if not bin_files:
if not silent: QMessageBox.warning(self, "ファイルなし", "対象ファイルが見つかりません。")
return None
def get_date_from_filename(f):
m = re.search(r'(20\d{10})', os.path.basename(f))
return m.group(1) if m else "000000000000"
bin_files.sort(key=get_date_from_filename)
target_files = bin_files[-1:] if self.grib_radio_latest.isChecked() else []
if not self.grib_radio_latest.isChecked():
sd_utc = (self.grib_date_start.date().toPyDate() - timedelta(days=1)).strftime("%Y%m%d")
ed_utc = self.grib_date_end.date().toPyDate().strftime("%Y%m%d")
for f in bin_files:
dt_str = get_date_from_filename(f)
if sd_utc <= dt_str[:8] <= ed_utc:
target_files.append(f)
if not target_files:
if not silent: QMessageBox.warning(self, "該当なし", "指定期間内にファイルがありません。")
return None
total_inserted = 0
cursor = self.conn.cursor()
try:
cursor.execute("UPDATE GPV_Points SET PointIDText = substr('00000' || PointID, -5, 5) WHERE PointIDText IS NULL")
self.conn.commit()
db_points = pd.read_sql_query("SELECT PointIDText, Lon, Lat FROM GPV_Points", self.conn)
# ★ 浮動小数点誤差対策: DB側の座標を丸める
db_points['Lon'] = db_points['Lon'].round(4)
db_points['Lat'] = db_points['Lat'].round(4)
total_pts = len(db_points)
temp_grib = "temp_subset.grib2" # 切り出し用一時ファイル
for target_file in target_files:
f_df = pd.DataFrame()
dt_str = get_date_from_filename(target_file)
if dt_str == "000000000000": continue
dt_utc = datetime.strptime(dt_str, "%Y%m%d%H%M")
obs_day = dt_utc.strftime("%Y-%m-%d")
obs_time = dt_utc.strftime("%H:%M")
if not obs_time.endswith("00"): continue
if not self.grib_radio_latest.isChecked() and total_pts > 0:
cursor.execute(f"SELECT COUNT({col_name}) FROM OBS_GP WHERE ObsDay=? AND ObsTime=? AND {col_name} IS NOT NULL", (obs_day, obs_time))
exist_count = cursor.fetchone()[0]
if exist_count >= total_pts:
continue
# ====================================================
# ★ 爆速化処理: 1kmメッシュ(860万点)を札幌圏だけに切り抜く
# ====================================================
lon_str = f"{self.lon_min - 0.05:.2f}:{self.lon_max + 0.05:.2f}"
lat_str = f"{self.lat_min - 0.05:.2f}:{self.lat_max + 0.05:.2f}"
if os.path.exists(temp_grib):
os.remove(temp_grib)
# wgrib2の機能で札幌圏だけを四角くくり抜いた小さなファイルを作成
subprocess.run([self.wgrib2_path, target_file, "-small_grib", lon_str, lat_str, temp_grib], capture_output=True)
# もし切り出しに失敗していれば、元の巨大ファイルをそのまま読み込む(安全策)
read_target = temp_grib if os.path.exists(temp_grib) and os.path.getsize(temp_grib) > 0 else target_file
# ====================================================
proc = subprocess.run([self.wgrib2_path, read_target, "-no_header", "-csv", "-"], capture_output=True, text=True)
if proc.returncode != 0 or not proc.stdout:
logging.warning(f"wgrib2 解析スキップ (破損または非対応): {os.path.basename(target_file)}")
continue
raw_df = pd.read_csv(io.StringIO(proc.stdout), header=None, engine='c')
val_1 = pd.to_numeric(raw_df.iloc[:, -3], errors='coerce')
val_2 = pd.to_numeric(raw_df.iloc[:, -2], errors='coerce')
raw_df["Value"] = pd.to_numeric(raw_df.iloc[:, -1], errors='coerce')
if val_1.mean() > 100:
raw_df["Lon"], raw_df["Lat"] = val_1, val_2
else:
raw_df["Lon"], raw_df["Lat"] = val_2, val_1
# ★ 浮動小数点誤差による「0件」バグ対策: 読み込み直後に丸める
raw_df["Lon"] = raw_df["Lon"].round(4)
raw_df["Lat"] = raw_df["Lat"].round(4)
raw_df = raw_df.dropna(subset=["Lon", "Lat", "Value"])
f_df = raw_df[(raw_df["Lat"] >= self.lat_min) & (raw_df["Lat"] <= self.lat_max) &
(raw_df["Lon"] >= self.lon_min) & (raw_df["Lon"] <= self.lon_max)].copy()
if f_df.empty: continue
unique_coords = f_df[['Lon', 'Lat']].drop_duplicates()
merged_check = pd.merge(unique_coords, db_points, on=['Lon', 'Lat'], how='left')
missing_coords = merged_check[merged_check['PointIDText'].isna()]
if not missing_coords.empty:
new_coords = list(missing_coords[['Lon', 'Lat']].itertuples(index=False, name=None))
cursor.executemany("INSERT OR IGNORE INTO GPV_Points (Lon, Lat) VALUES (?, ?)", new_coords)
cursor.execute("UPDATE GPV_Points SET PointIDText = substr('00000' || PointID, -5, 5) WHERE PointIDText IS NULL")
db_points = pd.read_sql_query("SELECT PointIDText, Lon, Lat FROM GPV_Points", self.conn)
# 再度丸める
db_points['Lon'] = db_points['Lon'].round(4)
db_points['Lat'] = db_points['Lat'].round(4)
f_df = pd.merge(f_df, db_points, on=['Lon', 'Lat'], how='left')
f_df = f_df.dropna(subset=['PointIDText', 'Value'])
f_df['ObsDay'] = obs_day
f_df['ObsTime'] = obs_time
f_df['MsCd'] = 'GPV_' + f_df['PointIDText'].astype(str)
bulk_data = list(f_df[['ObsDay', 'ObsTime', 'MsCd', 'Value']].itertuples(index=False, name=None))
cursor.executemany(sql_upsert, bulk_data)
total_inserted += len(bulk_data)
self.conn.commit()
if hasattr(self, 'update_t8_query_points'):
self.update_t8_query_points()
# 処理が終わったら一時ファイルを削除してディスクをクリーンに保つ
if os.path.exists(temp_grib):
os.remove(temp_grib)
if not silent:
QMessageBox.information(self, "書込完了", f"{data_name}の抽出・書込が完了しました。\n処理ファイル数: {len(target_files)}\n実更新データ数: {total_inserted} 件 (内部UTC軸・OBS_GPへ格納)")
return total_inserted
except Exception as e:
self.conn.rollback()
if not silent: QMessageBox.critical(self, "エラー", f"抽出処理エラー:\n{e}")
return None
未分類