server_big.py

from __future__ import annotations
import calendar,gzip,io,os,random,re,sys,tarfile,tempfile,threading,time,requests,json,rasterio,rioxarray,math,uuid,hashlib, shutil, logging
from collections import deque,OrderedDict,defaultdict
from datetime import datetime,timedelta,timezone
from functools import lru_cache
from pathlib import Path
from typing import Optional,Callable,Any
import numpy as np
import xarray as xr
from fastapi import FastAPI,HTTPException,Query,Response,Body
from fastapi.staticfiles import StaticFiles
from rasterio.crs import CRS
from rasterio.enums import Resampling
from rasterio.transform import Affine,from_bounds
from rasterio.warp import reproject
from rio_cogeo.cogeo import cog_translate
from rio_cogeo.profiles import cog_profiles
from rio_tiler.errors import TileOutsideBounds
from rio_tiler.io import COGReader
from rio_tiler.utils import render
from starlette.responses import PlainTextResponse,Response
from huggingface_hub import HfApi,snapshot_download,hf_hub_download
from huggingface_hub._commit_api import CommitOperationAdd,CommitOperationDelete
from zoneinfo import ZoneInfo
import sqlite3
from rio_tiler.colormap import cmap
from fastapi.responses import Response as FastResponse
from pydantic import BaseModel
from starlette.middleware.gzip import GZipMiddleware
import huggingface_hub.utils._http as _hf_http
from requests.adapters import HTTPAdapter
import concurrent.futures
import requests.adapters as _req_adapters

os.environ["TZ"] = "America/Chicago"
time.tzset()
logging.Formatter.converter = time.localtime
os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "600"
os.environ["HF_HUB_ETAG_TIMEOUT"] = "600"

_orig_adapter_send = _req_adapters.HTTPAdapter.send
def _timeout_600_send(self, request, *args, **kwargs):
    if "timeout" not in kwargs or kwargs["timeout"] is None or (isinstance(kwargs.get("timeout"), (int, float)) and kwargs["timeout"] <= 30):
        kwargs["timeout"] = 600
    return _orig_adapter_send(self, request, *args, **kwargs)
_req_adapters.HTTPAdapter.send = _timeout_600_send

LOCAL_TZ=ZoneInfo(os.environ.get("LOCAL_TZ","America/Chicago"))
LOWRES_Z=int(os.environ.get("LOWRES_Z","6"))
LOWRES_MAX_REQUEST_Z=int(os.environ.get("LOWRES_MAX_REQUEST_Z","14"))
LOWRES_BOX_ZMIN=int(os.environ.get("LOWRES_BOX_ZMIN","4"))
LOWRES_BOX_ZMAX=int(os.environ.get("LOWRES_BOX_ZMAX","6"))
HF_MAX_OPS_PER_COMMIT=int(os.environ.get("HF_MAX_OPS_PER_COMMIT","300"))
HF_PUSH_SLEEP_SEC=float(os.environ.get("HF_PUSH_SLEEP_SEC","0.5"))
ROLL_WINDOW_DAYS=int(os.environ.get("ROLL_WINDOW_DAYS","6"))
SCHEDULE_HOUR_LOCAL=int(os.environ.get("SCHEDULE_HOUR_LOCAL","10")) #hour
SCHEDULE_MIN_LOCAL=int(os.environ.get("SCHEDULE_MIN_LOCAL","00")) #minute
WARM_ZMIN=int(os.environ.get("WARM_ZMIN","6"))
WARM_ZMAX=int(os.environ.get("WARM_ZMAX","11"))
WARM_RADIUS=int(os.environ.get("WARM_RADIUS","1"))
WARM_CAP_PER_ZOOM=int(os.environ.get("WARM_CAP_PER_ZOOM","7000"))
WARM_MAX_TILES_TOTAL=int(os.environ.get("WARM_MAX_TILES_TOTAL","50000"))
WARM_DOM=os.environ.get("WARM_DOM","zz").lower()
WARM_MILES=float(os.environ.get("WARM_MILES","8"))
BOOT_ROLL_DEFAULT=os.environ.get("RUN_BOOT_ROLL","0") #set to 1 to run the download sequence on boot, 0 to bypass
WARM_HOURS_LIST=[24,72]
CN_MBTILES_LOCAL=Path("cache")/"tracks"/"cn_tracks.mbtiles"
CN_MBTILES_REPO_PATH="tracks/cn_tracks.mbtiles"
CN_BOUNDS_REPO_PATH="tracks/cn_tracks.bounds.txt"
_VIEWGRID_LOCK=threading.Lock()
_VIEWGRID_CACHE:dict[tuple,tuple[float,Any]]={}
_VIEWGRID_TTL_S=45.0
_MELTCOGS_LOCK=threading.Lock()
_MELTCOGS_CACHE:dict[tuple,tuple[float,Any]]={}
_MELTCOGS_TTL_S=300.0
KEEP_TILE_DAYS=int(os.environ.get("KEEP_TILE_DAYS","6"))
KEEP_FORECAST_COG_DAYS=int(os.environ.get("KEEP_FORECAST_COG_DAYS","1"))
_COGREADER_CACHE_MAX=int(os.environ.get("COGREADER_CACHE_MAX","12"))
_COGREADER_CACHE:"OrderedDict[str,COGReader]"=OrderedDict()
_COGREADER_CACHE_LK=threading.Lock()
ROLL_SINGLETON_KEY="roll_job"
_ACTIVE_CLIENT: dict[str, dict[str, Any]] = {}
_ACTIVE_CLIENT_LK = threading.Lock()
_ACTIVE_CLIENT_TTL_S = 180.0
CACHE_CONTROL_BY_DATE=os.environ.get("CACHE_CONTROL_BY_DATE","public, max-age=31536000, immutable")
CACHE_CONTROL_LATEST=os.environ.get("CACHE_CONTROL_LATEST","public, max-age=300, stale-while-revalidate=600")
PNG_MAGIC=b"\x89PNG\r\n\x1a\n"
ZIP_MAGIC=b"PK\x03\x04"
RAW_TILE_EXT=".npz"
_COG_LRU_MAX=int(os.environ.get("COGREADER_LRU_MAX","36"))
_COG_LRU:"OrderedDict[str,COGReader]"=OrderedDict()
_COG_LRU_LK=threading.Lock()
_RATE_LIMIT_WINDOW=10.0
_RATE_LIMIT_MAX_REQUESTS=50

_PUSH_RESUME_TIMER = None
_PUSH_TIMER_LK = threading.Lock()

def _pause_push_for_user():
    """Pause background HF push for 30s to prioritize user requests."""
    global _PUSH_RESUME_TIMER
    _PUSH_PAUSED.set()
    with _PUSH_TIMER_LK:
        if _PUSH_RESUME_TIMER:
            _PUSH_RESUME_TIMER.cancel()
        _PUSH_RESUME_TIMER = threading.Timer(30.0, lambda: _PUSH_PAUSED.clear())
        _PUSH_RESUME_TIMER.daemon = True
        _PUSH_RESUME_TIMER.start()

def _pick_cache_root()->Path:
    env=(os.environ.get("SNODAS_CACHE")or"").strip()
    candidates=([env]if env else[])+["/data/snodas-cache","/home/user/snodas-cache","/tmp/snodas-cache","./cache"]
    for c in candidates:
        p=Path(c).expanduser()
        try:
            p.mkdir(parents=True,exist_ok=True)
            t=p/".wtest";t.write_text("ok");t.unlink(missing_ok=True)
            return p.resolve()
        except:pass
    return Path(tempfile.mkdtemp(prefix="snodas-cache-")).resolve()

CACHE=_pick_cache_root()
_HF_NEGATIVE_CACHE: dict[str, float] = {}
_HF_NEGATIVE_CACHE_LK = threading.Lock()
_HF_NEGATIVE_TTL = 300.0  # 5 minutes
_HF_NEGATIVE_MAX = 10000  # cap size

def _hf_negative_check(rel: str) -> bool:
    with _HF_NEGATIVE_CACHE_LK:
        ts = _HF_NEGATIVE_CACHE.get(rel)
        if ts is None:
            return False
        if (time.time() - ts) > _HF_NEGATIVE_TTL:
            del _HF_NEGATIVE_CACHE[rel]
            return False
        return True

def _hf_negative_mark(rel: str) -> None:
    with _HF_NEGATIVE_CACHE_LK:
        _HF_NEGATIVE_CACHE[rel] = time.time()
        # Prevent unbounded growth
        if len(_HF_NEGATIVE_CACHE) > _HF_NEGATIVE_MAX:
            # Remove oldest 20%
            sorted_keys = sorted(_HF_NEGATIVE_CACHE, key=_HF_NEGATIVE_CACHE.get)
            for k in sorted_keys[:_HF_NEGATIVE_MAX // 5]:
                del _HF_NEGATIVE_CACHE[k]



_NOAA_FILE_SIGS: dict[str, dict] = {}
_NOAA_FILE_SIGS_LK = threading.Lock()
_NOAA_SIGS_PATH = CACHE / "_noaa_file_sigs.json"

def _load_noaa_sigs() -> None:
    global _NOAA_FILE_SIGS
    try:
        if _NOAA_SIGS_PATH.exists():
            with _NOAA_FILE_SIGS_LK:
                _NOAA_FILE_SIGS = json.loads(_NOAA_SIGS_PATH.read_text("utf-8"))
    except:
        pass

def _save_noaa_sigs() -> None:
    try:
        with _NOAA_FILE_SIGS_LK:
            _NOAA_SIGS_PATH.write_text(json.dumps(_NOAA_FILE_SIGS, sort_keys=True), "utf-8")
    except:
        pass

def _get_noaa_file_sig(url: str) -> dict | None:
    """HEAD request to get Last-Modified and Content-Length from NOAA."""
    try:
        r = requests.head(url, timeout=15, allow_redirects=True, headers={"User-Agent": "snodas/1.3"})
        if r.status_code != 200:
            return None
        return {
            "last_modified": r.headers.get("Last-Modified", ""),
            "content_length": r.headers.get("Content-Length", ""),
            "etag": r.headers.get("ETag", ""),
        }
    except:
        return None

def _record_baked_sig(valid:str,dom:str,url:str,source_type:str|None=None)->None:
    """Record NOAA file sig. Key is now date|dom (not valid|dom) for consistency."""
    sig=_get_noaa_file_sig(url)
    if not sig:return
    date_ymd=valid[:8]if len(valid)>=8 else valid
    key=f"{date_ymd}|{dom}"
    sig["url"]=url
    sig["baked_at"]=time.time()
    sig["valid"]=valid
    if source_type is None:
        if"ttnats"in url.lower():
            source_type="ttna_11z"if url.lower().endswith("11dp000.nc")or"11dp000"in url.lower()else"ttna_05z"
        else:
            source_type="forecast"
    sig["source_type"]=source_type
    with _NOAA_FILE_SIGS_LK:
        _NOAA_FILE_SIGS[key]=sig
    _save_noaa_sigs()
    _log("INFO",f"[staleness] recorded sig for {key}: type={source_type} LM={sig['last_modified']} CL={sig['content_length']}")
    
def _record_baked_sig_v2(date_ymd:str,dom:str,url:str,source_type:str)->None:
    _record_baked_sig(date_ymd,dom,url,source_type=source_type)

# 1. NEW: Find 11Z TTNA melt files
def _find_melt_11044_ttna_11z(valid_date_ymd:str,dom_prefer:str="zz")->list[str]:
    """Find TTNA 11Z melt files for a given date (YYYYMMDD)."""
    dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid_date_ymd}11";out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low:continue
        if"__t0024ttnats"not in low:continue
        if want_ts not in low:continue
        if not low.endswith(".nc"):continue
        out.append(_as_url(base))
    return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))

# 2. NEW: Find 05Z TTNA melt files
def _find_melt_11044_ttna_05z(valid_date_ymd:str,dom_prefer:str="zz")->list[str]:
    """Find TTNA 05Z melt files for a given date (YYYYMMDD)."""
    dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid_date_ymd}05";out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low:continue
        if"__t0024ttnats"not in low:continue
        if want_ts not in low:continue
        if not low.endswith(".nc"):continue
        out.append(_as_url(base))
    return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))

def _pick_best_run_init_for_valid_11z(valid:str,dom_prefer:str="zz")->str|None:
    """Find the latest TT run init that has a TS...11 file."""
    valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";rows=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low or"__t0024tt"not in low or want_ts not in low or not low.endswith(".nc"):continue
        if"ttnats"in low:continue  # Skip TTNA files, only want forecast TT...06
        tt=_tt_from_name(base)
        if not tt:continue
        rows.append((tt,_score_dom(_as_url(base),dom_prefer)))
    if not rows:return None
    rows.sort(key=lambda t:(t[0],-t[1]));return rows[-1][0]

def _find_melt_11044_t0024_11z(run_init:str,valid:str,dom_prefer:str="zz")->list[str]:
    """Find forecast melt files matching a specific TT run init and 11Z valid time."""
    run_init=_norm_ts(run_init);valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_tt=f"tt{run_init}";want_ts=f"ts{valid}";out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low or"__t0024tt"not in low or want_tt not in low or want_ts not in low or not low.endswith(".nc"):continue
        if"ttnats"in low:continue  # Skip TTNA
        out.append(_as_url(base))
    return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))

def _find_best_melt_for_date(date_ymd:str,dom_prefer:str="zz")->dict:
    """Returns {ok, melt_urls, valid, run_init, source_type} preferring 11Z > 05Z."""
    dom_prefer=(dom_prefer or"zz").lower()
    # 1. TTNA 11Z
    urls_11z=_find_melt_11044_ttna_11z(date_ymd,dom_prefer=dom_prefer)
    if urls_11z:
        return{"ok":True,"melt_urls":urls_11z,"valid":f"{date_ymd}11","run_init":"TTNA","source_type":"ttna_11z"}
    # 2. Forecast 11Z (TT...06TS...11)
    valid_11=f"{date_ymd}11"
    run_init_11=_pick_best_run_init_for_valid_11z(valid_11,dom_prefer=dom_prefer)
    if run_init_11:
        melt_urls_11=_find_melt_11044_t0024_11z(run_init_11,valid_11,dom_prefer=dom_prefer)
        if melt_urls_11:
            return{"ok":True,"melt_urls":melt_urls_11,"valid":valid_11,"run_init":run_init_11,"source_type":"forecast_11z"}
    # 3. TTNA 05Z
    urls_05z=_find_melt_11044_ttna_05z(date_ymd,dom_prefer=dom_prefer)
    if urls_05z:
        return{"ok":True,"melt_urls":urls_05z,"valid":f"{date_ymd}05","run_init":"TTNA","source_type":"ttna_05z"}
    # 4. Forecast 05Z (TT...06TS...05)
    valid_05=_norm_ts(f"{date_ymd}05")
    run_init=_pick_best_run_init_for_valid_t0024(valid_05,dom_prefer=dom_prefer)
    if not run_init:
        return{"ok":False,"error":"no melt files found (11z, 05z, or forecast)","date":date_ymd}
    melt_urls=_find_melt_11044_t0024(run_init,valid_05,dom_prefer=dom_prefer)
    if not melt_urls:
        return{"ok":False,"error":"forecast run_init found but no melt files","run_init":run_init,"date":date_ymd}
    return{"ok":True,"melt_urls":melt_urls,"valid":valid_05,"run_init":run_init,"source_type":"forecast_05z"}

# 4. NEW: Find snowpack files matching the valid time we chose
# def _find_snowpack_for_best_melt(best_melt:dict,dom_prefer:str="zz")->list[str]:
#     """Find snowpack files that match the valid time from _find_best_melt_for_date."""
#     dom_prefer=(dom_prefer or"zz").lower()
#     valid=best_melt.get("valid","")
#     run_init=best_melt.get("run_init","")
#     source_type=best_melt.get("source_type","")
#     if source_type in("ttna_11z","ttna_05z"):
#         # For TTNA, search by valid time
#         html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";out=[]
#         for n in names:
#             base=n.split("?",1)[0].strip();low=base.lower()
#             if"ssmv11034"not in low and"ssmv11036"not in low:continue
#             if"hp001"not in low:continue
#             if want_ts not in low:continue
#             if not(low.endswith(".grib2")or low.endswith(".nc")):continue
#             out.append(_as_url(base))
#         if out:return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))
#         # Fallback: try 05Z snowpack if we're on 11Z melt
#         if source_type=="ttna_11z":
#             want_ts_05=f"ts{valid[:8]}05"
#             for n in names:
#                 base=n.split("?",1)[0].strip();low=base.lower()
#                 if"ssmv11034"not in low and"ssmv11036"not in low:continue
#                 if"hp001"not in low:continue
#                 if want_ts_05 not in low:continue
#                 if not(low.endswith(".grib2")or low.endswith(".nc")):continue
#                 out.append(_as_url(base))
#             if out:return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))
#         return[]
#     # Forecast source — use existing logic
#     if run_init and run_init!="TTNA":
#         return _find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom_prefer)
#     return[]
def _find_snowpack_for_best_melt(best_melt:dict,dom_prefer:str="zz")->list[str]:
    dom_prefer=(dom_prefer or"zz").lower()
    valid=best_melt.get("valid","")
    run_init=best_melt.get("run_init","")
    source_type=best_melt.get("source_type","")
    html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}"
    if source_type in("ttna_11z","ttna_05z"):
        # Prefer TTNA snowpack matching the valid time
        ttna_out=[]
        for n in names:
            base=n.split("?",1)[0].strip();low=base.lower()
            if("ssmv11034"not in low and"ssmv11036"not in low):continue
            if"hp001"not in low:continue
            if want_ts not in low:continue
            if"ttnats"not in low:continue  # Only TTNA snowpack
            if not(low.endswith(".grib2")or low.endswith(".nc")):continue
            ttna_out.append(_as_url(base))
        if ttna_out:
            return sorted(set(ttna_out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))
        # Fallback: TTNA snowpack at 05Z if we're on 11Z melt
        if source_type=="ttna_11z":
            want_ts_05=f"ts{valid[:8]}05"
            for n in names:
                base=n.split("?",1)[0].strip();low=base.lower()
                if("ssmv11034"not in low and"ssmv11036"not in low):continue
                if"hp001"not in low:continue
                if want_ts_05 not in low:continue
                if"ttnats"not in low:continue
                if not(low.endswith(".grib2")or low.endswith(".nc")):continue
                ttna_out.append(_as_url(base))
            if ttna_out:
                return sorted(set(ttna_out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))
    if source_type in("forecast_11z","forecast_05z","forecast"):
        # Use snowpack from the same run init only
        if run_init and run_init!="TTNA":
            return _find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom_prefer)
    # Last resort: any snowpack matching valid time but limit to preferred domain
    out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if("ssmv11034"not in low and"ssmv11036"not in low):continue
        if"hp001"not in low:continue
        if want_ts not in low:continue
        if not(low.endswith(".grib2")or low.endswith(".nc")):continue
        if _dom_from_url(_as_url(base))==dom_prefer:
            out.append(_as_url(base))
    if out:return sorted(set(out),key=lambda u:_dom_pri(_dom_from_url(u)))
    return[]


def _check_and_rebuild_stale_dates()->None:
    dom=(WARM_DOM or"zz").lower()
    window=_rolling_dates_local_window(lookback=2,lookahead=3)
    with _NOAA_FILE_SIGS_LK:
        known_sigs={k:dict(v)for k,v in _NOAA_FILE_SIGS.items()}
    stale_dates=[]
    for ymd in window:
        key=f"{ymd}|{dom}"
        old_sig=known_sigs.get(key)
        urls_11z=_find_melt_11044_ttna_11z(ymd,dom_prefer=dom)
        if urls_11z:
            url_11z=urls_11z[0]
            if old_sig and old_sig.get("source_type")=="ttna_11z":
                new_sig=_get_noaa_file_sig(url_11z)
                if new_sig:
                    old_lm=old_sig.get("last_modified","");new_lm=new_sig.get("last_modified","")
                    old_cl=old_sig.get("content_length","");new_cl=new_sig.get("content_length","")
                    if(old_lm and new_lm and old_lm!=new_lm)or(old_cl and new_cl and old_cl!=new_cl):
                        _log("INFO",f"[staleness] 11Z updated for {ymd}: LM {old_lm} -> {new_lm}")
                        stale_dates.append((ymd,url_11z))
                continue
            _log("INFO",f"[staleness] 11Z now available for {ymd} — was {old_sig.get('source_type','unknown') if old_sig else 'not baked'}, rebuilding")
            stale_dates.append((ymd,url_11z))
            continue
        if not old_sig:
            best=_find_best_melt_for_date(ymd,dom_prefer=dom)
            if best.get("ok")and best.get("melt_urls"):
                _record_baked_sig_v2(ymd,dom,best["melt_urls"][0],best.get("source_type","unknown"))
            continue
        if old_sig and old_sig.get("source_type")in("forecast","forecast_05z","ttna_05z"):
                    # Check if forecast 11Z is now available
                    valid_11=f"{ymd}11"
                    ri_11=_pick_best_run_init_for_valid_11z(valid_11,dom_prefer=dom)
                    if ri_11:
                        urls_11=_find_melt_11044_t0024_11z(ri_11,valid_11,dom_prefer=dom)
                        if urls_11:
                            _log("INFO",f"[staleness] forecast 11Z now available for {ymd}")
                            stale_dates.append((ymd,urls_11[0]))
                            continue
                    old_url=old_sig.get("url","")
                    best=_find_best_melt_for_date(ymd,dom_prefer=dom)
                    if best.get("ok")and best.get("melt_urls"):
                        new_url=best["melt_urls"][0]
                        if new_url!=old_url:
                            _log("INFO",f"[staleness] newer forecast run available for {ymd}: {old_url.split('/')[-1]} -> {new_url.split('/')[-1]}")
                            stale_dates.append((ymd,new_url))
                            continue
                        new_sig=_get_noaa_file_sig(new_url)
                        if new_sig:
                            old_lm=old_sig.get("last_modified","");new_lm=new_sig.get("last_modified","")
                            old_cl=old_sig.get("content_length","");new_cl=new_sig.get("content_length","")
                            if(old_lm and new_lm and old_lm!=new_lm)or(old_cl and new_cl and old_cl!=new_cl):
                                _log("INFO",f"[staleness] forecast file modified for {ymd}: LM {old_lm} -> {new_lm}")
                                stale_dates.append((ymd,new_url))
                                continue
    if not stale_dates:return
    # Priority: day 2 first, then day 1, then day 3, then others
    today=datetime.now(LOCAL_TZ).date()
    def _rebuild_priority(item):
        ymd=item[0]
        try:
            d=datetime.strptime(ymd,"%Y%m%d").date()
            offset=(d-today).days
            if offset==2:return 0   # day 2 first
            if offset==1:return 1   # day 1 second
            if offset==3:return 2   # day 3 third
            return 3+abs(offset)    # everything else after
        except:return 99
    stale_dates.sort(key=_rebuild_priority)
    _log("INFO",f"[staleness] rebuild order: {[d[0] for d in stale_dates]}")
    for ymd,url in stale_dates:
        _log("INFO",f"[staleness] hot-swapping {ymd}")
        try:_hot_swap_rebuild_date(ymd,dom)
        except Exception as e:_log("WARN",f"[staleness] rebuild failed for {ymd}: {e!r}")

def _staleness_monitor_loop()->None:
    _load_noaa_sigs()
    _log("INFO","[staleness] monitor started")
    time.sleep(300)
    while True:
        try:_check_and_rebuild_stale_dates()
        except Exception as e:_log("WARN",f"[staleness] monitor error: {e!r}")
        time.sleep(20*60)

def _nuke_and_rebuild_date(date:str,dom:str)->None:
    dom=(dom or"zz").lower();valid_05=f"{date}05";valid_11=f"{date}11"
    # 1. Delete local COGs for both 05Z and 11Z
    for valid in[valid_05,valid_11]:
        for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]:
            for p in _forecast_dir().glob(pattern):
                try:COG_CACHE.evict(p)
                except:pass
                try:p.unlink(missing_ok=True)
                except:pass
    # 2. Delete local tile cache
    for kind in["tilecache","rawtilecache"]:
        for h in WARM_HOURS_LIST:
            d=CACHE/kind/dom/str(h)/date
            if d.exists():shutil.rmtree(d,ignore_errors=True)
    # 3. Delete tar markers
    marker_dir=CACHE/"_tar_markers"
    if marker_dir.exists():
        for mk in marker_dir.glob(f"*_{date}.tar.gz.done"):
            try:mk.unlink(missing_ok=True)
            except:pass
    # 4. Delete tar archives from HF
    tok,repo=_hf_cfg();api=_hf_api()
    if tok and repo and api is not None:
        for kind in["tilecache","rawtilecache"]:
            for h in WARM_HOURS_LIST:
                archive_path=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz"
                try:
                    api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=archive_path)],commit_message=f"nuke: delete {archive_path}")
                    _log("INFO",f"[nuke] deleted HF archive: {archive_path}")
                except Exception as e:
                    err=repr(e)
                    if"404"not in err and"EntryNotFound"not in err and"does not exist"not in err:
                        _log("WARN",f"[nuke] HF delete failed: {archive_path}: {e!r}")
    # 5. Also delete 72h COGs and tiles for next 2 dates
    for offset in[1,2]:
        try:
            future_date=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
            for fv in[f"{future_date}05",f"{future_date}11"]:
                for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"):
                    try:COG_CACHE.evict(p)
                    except:pass
                    try:p.unlink(missing_ok=True);_log("INFO",f"[nuke] deleted 72h COG: {p.name}")
                    except:pass
            for kind in["tilecache","rawtilecache"]:
                d=CACHE/kind/dom/"72"/future_date
                if d.exists():shutil.rmtree(d,ignore_errors=True);_log("INFO",f"[nuke] deleted {kind}/72h tiles for {future_date}")
            if tok and repo and api is not None:
                for kind in["tilecache","rawtilecache"]:
                    ap=f"archives/{kind}_{dom}_h72_{future_date}.tar.gz"
                    try:
                        api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"nuke: delete stale 72h {ap}")
                        _log("INFO",f"[nuke] deleted HF 72h archive: {ap}")
                    except Exception as e:
                        err=repr(e)
                        if"404"not in err and"EntryNotFound"not in err and"does not exist"not in err:
                            _log("WARN",f"[nuke] HF 72h delete failed: {ap}: {e!r}")
        except:pass
    # 6. Clear negative cache
    with _HF_NEGATIVE_CACHE_LK:
        to_del=[k for k in _HF_NEGATIVE_CACHE if date in k]
        for k in to_del:del _HF_NEGATIVE_CACHE[k]
    # 7. Rebuild — _pick_forecast_for_valid now auto-selects 11Z > 05Z > forecast
    _log("INFO",f"[nuke] rebuilding COGs for {date} (will auto-select best available)")
    best=_find_best_melt_for_date(date,dom_prefer=dom)
    _log("INFO",f"[nuke] {date}: selected source_type={best.get('source_type','?')} valid={best.get('valid','?')}")
    # Use the valid time from best melt selection
    actual_valid=best.get("valid",f"{date}05")
    r=_ensure_forecast_for_valid(actual_valid,dom_prefer=dom)
    if not r.get("ok"):_log("WARN",f"[nuke] COG rebuild failed for {date}: {r}");return
    # 8. Re-bake tiles — sequential for CPU efficiency
    _log("INFO",f"[nuke] re-baking tiles for {date}")
    for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]:
        try:
            r=fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra)
            _log("INFO",f"[nuke] {date}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
        except Exception as e:_log("WARN",f"[nuke] {date}: {label} failed: {e!r}")
    for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
        try:fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra)
        except:pass
    # 9. Rebake 72h for next 2 dates — sequential, with tile invalidation
    for offset in[1,2]:
        try:
            future_date=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
            future_best=_find_best_melt_for_date(future_date,dom_prefer=dom)
            future_valid=future_best.get("valid",f"{future_date}05")if future_best.get("ok")else f"{future_date}05"
            try:_forecast_melt72h_end_cog(future_valid,dom_prefer=dom)
            except:pass
            _log("INFO",f"[nuke] rebaking 72h tiles for {future_date}")
            # Delete stale 72h tiles (nuke already deleted them in step 5,
            # but belt-and-suspenders in case the path changed)
            for kind in["tilecache","rawtilecache"]:
                d=CACHE/kind/dom/"72"/future_date
                if d.exists():shutil.rmtree(d,ignore_errors=True)
            for fn,extra in[(_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),(_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]:
                try:fn(date_yyyymmdd=future_date,dom=dom,hours_list=[72],zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra)
                except:pass
            for fn,extra in[(_bake_lowres_box_tiles_for_date,dict(conc=1)),(_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
                try:fn(date_yyyymmdd=future_date,dom=dom,hours_list=[72],zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra)
                except:pass
        except Exception as e:_log("WARN",f"[nuke] 72h rebake for +{offset} failed: {e!r}")
 
    # 10. Push corrected tiles
    push_dates=[date]
    for offset in[1,2]:
        try:push_dates.append((datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d"))
        except:pass
    _log("INFO",f"[nuke] pushing corrected tiles for {push_dates}")
    _hf_push_baked_as_tar(push_dates,dom,WARM_HOURS_LIST)
    # 11. Record signature
    if best.get("ok")and best.get("melt_urls"):
        _record_baked_sig(date,dom,best["melt_urls"][0],source_type=best.get("source_type"))
    _log("INFO",f"[nuke] {date}: rebuild complete (source={best.get('source_type','?')})")

        
def _client_mark_active(client_id: str, active_date: str, gen: int | None = None) -> None:
    with _ACTIVE_CLIENT_LK:
        _ACTIVE_CLIENT[client_id] = {
            "date": active_date,
            "gen": int(gen) if gen is not None else 0,
            "ts": time.time(),
        }

def _client_is_active_date(client_id: str, date_yyyymmdd: str) -> bool:
    now = time.time()
    with _ACTIVE_CLIENT_LK:
        st = _ACTIVE_CLIENT.get(client_id)
        if not st:
            return True
        if (now - float(st.get("ts", 0.0))) > _ACTIVE_CLIENT_TTL_S:
            return True
        return str(st.get("date", "")) == str(date_yyyymmdd)

def _client_is_active_gen(client_id: str, gen: int | None) -> bool:
    if not client_id or gen is None:
        return True  # no generation tracking → allow
    now = time.time()
    with _ACTIVE_CLIENT_LK:
        st = _ACTIVE_CLIENT.get(client_id)
        if not st:
            return True
        if (now - float(st.get("ts", 0.0))) > _ACTIVE_CLIENT_TTL_S:
            return True
        server_gen = int(st.get("gen", 0))
        return int(gen) >= server_gen

def _make_cancel_check(client_id: str | None, gen: int | None):
    if not client_id or gen is None:
        return None
    def _is_cancelled() -> bool:
        return not _client_is_active_gen(client_id, gen)
    return _is_cancelled
class CogCache:
    def __init__(self):
        self._datasets: dict[str, COGReader] = {}
        self._lock = threading.Lock()

    def get(self, cog: Path) -> COGReader:
        key = cog.as_posix()
        with self._lock:
            r = self._datasets.get(key)
            if r is None:
                r = COGReader(key)
                self._datasets[key] = r
            return r

    def evict(self, path: Path) -> None:
        with self._lock:
            ds = self._datasets.pop(str(path), None)
        if ds:
            try: ds.close()
            except: pass

    def close_all(self):
        with self._lock:
            for r in self._datasets.values():
                try: r.close()
                except: pass
            self._datasets.clear()

    def __enter__(self): return self
    def __exit__(self, *args): self.close_all()

COG_CACHE=CogCache()

def _cogreader_get(path:Path)->COGReader:
    key=path.as_posix()
    with _COGREADER_CACHE_LK:
        r=_COGREADER_CACHE.get(key)
        if r is not None:_COGREADER_CACHE.move_to_end(key,last=True);return r
        r=COGReader(key);_COGREADER_CACHE[key]=r;_COGREADER_CACHE.move_to_end(key,last=True)
        while len(_COGREADER_CACHE)>_COGREADER_CACHE_MAX:
            _,old_r=_COGREADER_CACHE.popitem(last=False)
            try:old_r.close()
            except:pass
        return r

class _RateLimitBuckets:
    def __init__(self):
        self._buckets:dict[str,list[float]]=defaultdict(list)
        self._locks:dict[str,threading.Lock]=defaultdict(threading.Lock)
        self._cleanup_lock=threading.Lock();self._last_cleanup=0.0
    def check(self,client_id:str)->bool:
        if not client_id:return True
        now=time.time();cutoff=now-_RATE_LIMIT_WINDOW
        with self._locks[client_id]:
            bucket=self._buckets[client_id];bucket[:]=[t for t in bucket if t>cutoff]
            if len(bucket)>=_RATE_LIMIT_MAX_REQUESTS:return False
            bucket.append(now)
        if now-self._last_cleanup>60.0:self._cleanup_old_clients(cutoff)
        return True
    def _cleanup_old_clients(self,cutoff:float)->None:
        if not self._cleanup_lock.acquire(blocking=False):return
        try:
            self._last_cleanup=time.time()
            empty=[cid for cid,bucket in self._buckets.items()if not bucket or all(t<=cutoff for t in bucket)]
            for cid in empty:self._buckets.pop(cid,None);self._locks.pop(cid,None)
        finally:self._cleanup_lock.release()

_rate_limiter=_RateLimitBuckets()
def _check_rate_limit(client_id:str)->bool:return _rate_limiter.check(client_id)

def _is_valid_png_file(p:Path)->bool:
    try:
        if not p.exists():return False
        if p.stat().st_size<64:return False
        with p.open("rb")as f:head=f.read(8)
        return head==PNG_MAGIC
    except:return False

def _ymd_from_cache_relpath(rel:str)->str|None:
    try:
        p=Path(rel).as_posix();parts=p.split("/")
        if len(parts)>=4 and parts[0]in("tilecache","rawtilecache"):
            ymd=parts[3];return ymd if re.fullmatch(r"[0-9]{8}",ymd)else None
        if parts and parts[0]=="forecast":
            m=re.search(r"([0-9]{8})",p);return m.group(1)if m and re.fullmatch(r"[0-9]{8}",m.group(1))else None
        base=parts[-1]if parts else p
        m=re.match(r"bydate_[a-z0-9]+_h\d+_([0-9]{8})_z\d+_\d+_\d+\.(png|npz)$",base)
        return m.group(1)if m else None
    except:return None

def _validate_npz_bytes(data:bytes)->bool:
    if not data or len(data)<16:return False
    if data[0:4]!=b'PK\x03\x04':return False
    return True

def _mem_cache_get(cache:dict,key:Any,ttl_s:int)->Any|None:
    try:
        v=cache.get(key)
        if not v:return None
        ts,payload=v
        if(time.time()-ts)>ttl_s:
            try:del cache[key]
            except:pass
            return None
        return payload
    except:return None

def _mem_cache_put(cache:dict,key:Any,payload:Any)->None:
    try:cache[key]=(time.time(),payload)
    except:pass

def _safe_unlink(p:Path)->None:
    try:p.unlink(missing_ok=True)
    except:pass

def _tile_cache_dir()->Path:
    d=CACHE/"tilecache";d.mkdir(parents=True,exist_ok=True);return d

def _tile_cache_path(dom:str,hours:int,date_yyyymmdd:str,z:int,x:int,y:int)->Path:
    dom=(dom or"zz").lower()
    return CACHE/"tilecache"/dom/str(int(hours))/date_yyyymmdd/str(int(z))/str(int(x))/f"{int(y)}.png"

def _raw_tile_cache_path(dom:str,hours:int,date_yyyymmdd:str,z:int,x:int,y:int)->Path:
    dom=(dom or"zz").lower()
    return CACHE/"rawtilecache"/dom/str(int(hours))/date_yyyymmdd/str(int(z))/str(int(x))/f"{int(y)}.npz"

def _raw_tile_cache_dir()->Path:
    d=CACHE/"rawtilecache";d.mkdir(parents=True,exist_ok=True);return d

def _is_valid_png_bytes(b:bytes)->bool:return bool(b)and len(b)>=64 and b[:8]==PNG_MAGIC
def _is_valid_npz_bytes(b:bytes)->bool:return bool(b)and len(b)>=16 and b[:4]==ZIP_MAGIC

def _atomic_write_bytes(path:Path,data:bytes)->None:
    path.parent.mkdir(parents=True,exist_ok=True)
    tmp=path.with_name(path.name+f".tmp.{uuid.uuid4().hex}")
    with tmp.open("wb")as f:f.write(data);f.flush();os.fsync(f.fileno())
    os.replace(tmp,path)

def _is_valid_npz_file(p:Path)->bool:
    try:
        if not p.exists():return False
        if p.stat().st_size<60:return False
        with p.open("rb")as f:head=f.read(2)
        return head==b"PK"
    except:return False

# def _raw_tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int,
#                         cancel_check=None) -> bytes | None:
#     p = _raw_tile_cache_path(dom, hours, ymd, z, x, y)
#     try:
#         if _is_valid_npz_file(p):
#             return p.read_bytes()
#         elif p.exists():
#             _safe_unlink(p)
#     except:
#         pass

#     if not _hf_pull_on_miss_enabled():
#         return None
#     if cancel_check and cancel_check():
#         return None

#     rel = p.relative_to(CACHE).as_posix()
#     if _hf_negative_check(rel):
#         return None

#     try:
#         if _hf_try_pull_file(rel, cancel_check=cancel_check):
#             if cancel_check and cancel_check():
#                 return None
#             if _is_valid_npz_file(p):
#                 return p.read_bytes()
#             elif p.exists():
#                 _safe_unlink(p)
#         _hf_negative_mark(rel)
#     except Exception as e:
#         if "404" not in str(e).lower():
#             _log("WARN", f"[hf_pull_raw] unexpected error for {rel}: {e!r}")
#         else:
#             _hf_negative_mark(rel)
#     return None

def _raw_tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int,
                        cancel_check=None) -> bytes | None:
    """Check local raw tile cache. No HF pull — same reasoning as _tile_cache_get."""
    p = _raw_tile_cache_path(dom, hours, ymd, z, x, y)
    try:
        if _is_valid_npz_file(p):
            return p.read_bytes()
        elif p.exists():
            _safe_unlink(p)
    except:
        pass
    # Per-tile HF pull removed: see _tile_cache_get comment.
    return None
    
def _tile_cache_put(dom,hours,ymd,z,x,y,png:bytes)->Path:
    p=_tile_cache_path(dom,hours,ymd,z,x,y)
    if not _is_valid_png_bytes(png):
        if p.exists():_safe_unlink(p)
        return p
    _atomic_write_bytes(p,png);return p

def _raw_tile_cache_put(dom,hours,ymd,z,x,y,npz_bytes:bytes)->Path:
    p=_raw_tile_cache_path(dom,hours,ymd,z,x,y)
    if not _is_valid_npz_bytes(npz_bytes):
        if p.exists():_safe_unlink(p)
        return p
    _atomic_write_bytes(p,npz_bytes);return p

_UPLOAD_STATE_DIR=CACHE/"_upload_state"
_UPLOAD_STATE_DIR.mkdir(parents=True,exist_ok=True)

def _upload_manifest_path(date_yyyymmdd:str)->Path:return _UPLOAD_STATE_DIR/f"uploaded_{date_yyyymmdd}.json"

def _load_upload_manifest(date_yyyymmdd:str)->dict:
    p=_upload_manifest_path(date_yyyymmdd)
    if not p.exists():return{"tilecache":{},"rawtilecache":{}}
    try:return json.loads(p.read_text())
    except:return{"tilecache":{},"rawtilecache":{}}

def _save_upload_manifest(date_yyyymmdd:str,manifest:dict)->None:
    try:_upload_manifest_path(date_yyyymmdd).write_text(json.dumps(manifest,sort_keys=True))
    except:pass

def _load_upload_manifest_global()->dict:
    try:
        p=CACHE/"_hf_upload_manifest.json"
        if p.exists():return json.loads(p.read_text("utf-8"))
    except:pass
    return{}

def _save_upload_manifest_global(m:dict)->None:
    try:
        p=CACHE/"_hf_upload_manifest.json";tmp=p.with_suffix(".tmp")
        tmp.write_text(json.dumps(m,separators=(",",":"),sort_keys=True),"utf-8");tmp.replace(p)
    except:pass

def _folder_signature(local_folder:Path,stable_age_sec:float=8.0)->tuple[int,int,float]:
    now=time.time();file_count=0;total_bytes=0;newest_mtime=0.0
    for p in local_folder.rglob("*"):
        try:
            if not p.is_file():continue
            st=p.stat()
            if st.st_size<=0:continue
            if(now-st.st_mtime)<stable_age_sec:continue
            file_count+=1;total_bytes+=int(st.st_size);newest_mtime=max(newest_mtime,float(st.st_mtime))
        except:continue
    return file_count,total_bytes,newest_mtime

def _hf_upload_folder_if_changed(*,local_folder:Path,repo_path:str,date_yyyymmdd:str,kind:str,dom:str,hours:int)->dict:
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"}
    if not local_folder.exists():return{"ok":True,"skipped":True,"reason":"folder_not_exists"}
    file_count,total_bytes,newest_mtime=_folder_signature(local_folder,stable_age_sec=8.0)
    if file_count<=0:return{"ok":True,"skipped":True,"reason":"empty_or_unstable_folder"}
    max_files=int(os.environ.get("HF_UPLOAD_MAX_FILES","8000"))
    if file_count>max_files:
        _log("WARN",f"[hf_upload_folder] skip too many files kind={kind} {repo_path} files={file_count} max={max_files}")
        return{"ok":True,"skipped":True,"reason":"too_many_files","files":file_count,"max":max_files}
    manifest=_load_upload_manifest_global();key=f"{kind}|{dom}|{int(hours)}|{repo_path}";prev=manifest.get(key)
    if prev:
        if prev.get("file_count")==file_count and prev.get("total_bytes")==total_bytes and prev.get("newest_mtime",0.0)>=newest_mtime:
            return{"ok":True,"skipped":True,"reason":"no_change"}
    try:
        api.upload_folder(folder_path=str(local_folder),repo_id=repo,repo_type="dataset",path_in_repo=repo_path,token=tok,
            commit_message=f"{kind}: upload {date_yyyymmdd} dom={dom} h={hours} ({file_count} files)",allow_patterns=["**/*"])
        manifest[key]={"file_count":file_count,"total_bytes":total_bytes,"newest_mtime":newest_mtime,"ts":time.time()}
        _save_upload_manifest_global(manifest)
        return{"ok":True,"skipped":False,"files":file_count,"bytes":total_bytes}
    except Exception as e:
        _log("WARN",f"[hf_upload_folder] failed kind={kind} date={date_yyyymmdd} err={e!r}")
        return{"ok":False,"error":repr(e)}
        
def _iter_zoom_dirs(date_folder: Path):
    if not date_folder.exists():
        return []
    out = []
    for p in sorted(date_folder.iterdir()):
        if p.is_dir() and p.name.isdigit():
            out.append(p)
    return out

def _hf_tiles_uploader_loop() -> None:
    _log("INFO", "[hf_uploader] disabled — roll job handles HF push directly")
    return
    
def _pack_raw_tile_npz(*,melt_mm:np.ndarray,valid_mask_u8:np.ndarray)->bytes:
    try:
        mm=melt_mm.astype("float32",copy=False);m=valid_mask_u8.astype("uint8",copy=False);valid=(m>0)
        inches=(mm/25.4).astype("float32",copy=False);q=np.zeros(inches.shape,dtype="uint16")
        qv=np.rint(np.clip(inches*100.0,0.0,65535.0)).astype("uint16");q[valid]=qv[valid]
        bio=io.BytesIO();np.savez_compressed(bio,q=q,m=m,s=np.uint8(100));return bio.getvalue()
    except:
        bio=io.BytesIO();np.savez_compressed(bio,q=np.zeros((256,256),dtype="uint16"),m=np.zeros((256,256),dtype="uint8"),s=np.uint8(100))
        return bio.getvalue()

_LOCK_ROOT=Path(os.environ.get("SNODAS_LOCK_DIR","/tmp/snodas-locks")).expanduser()
_LOCK_ROOT.mkdir(parents=True,exist_ok=True)
_LOCK_STALE_SECONDS=int(os.environ.get("LOCK_STALE_SECONDS",str(48*3600)))

def _acquire_singleton_lock(name:str)->bool:
    p=_LOCK_ROOT/f"{name}.lock";fd=None
    try:
        fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR)
        os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()}\n".encode("utf-8"));return True
    except FileExistsError:
        try:
            age=time.time()-p.stat().st_mtime
            if age>float(_LOCK_STALE_SECONDS):
                try:p.unlink(missing_ok=True)
                except:pass
                fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR)
                os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()} stale_recovered=1\n".encode("utf-8"));return True
        except:pass
        return False
    except:return False
    finally:
        if fd is not None:
            try:os.close(fd)
            except:pass

def _hf_cfg_norm():
    cfg=_hf_cfg()
    if isinstance(cfg,dict):
        ok=bool(cfg.get("ok"));repo=cfg.get("repo")or cfg.get("dataset_repo")or cfg.get("repo_id");token=cfg.get("token")or cfg.get("hf_token")
        return ok,repo,token,cfg
    if isinstance(cfg,tuple):
        if len(cfg)==0:return False,None,None,{"raw":cfg}
        if len(cfg)==1:return bool(cfg[0]),None,None,{"raw":cfg}
        if len(cfg)==2:return bool(cfg[0]),cfg[1],None,{"raw":cfg}
        return bool(cfg[0]),cfg[1],(cfg[2]if len(cfg)>=3 else None),{"raw":cfg}
    return False,None,None,{"raw":cfg}

_ROLL_LOCK_NAME="roll_job"
_ROLL_LOCK_STALE_SECONDS=int(os.environ.get("ROLL_LOCK_STALE_SECONDS",str(16*3600)))

def _run_lock_path(name:str)->Path:return _LOCK_ROOT/f"{name}.runlock"

def _try_acquire_run_lock(name:str,*,stale_seconds:int)->bool:
    p=_run_lock_path(name);fd=None
    try:
        fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR)
        os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()}\n".encode("utf-8"));return True
    except FileExistsError:
        try:
            age=time.time()-p.stat().st_mtime
            if age>float(stale_seconds):
                try:p.unlink(missing_ok=True)
                except:pass
                fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR)
                os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()} stale_recovered=1\n".encode("utf-8"));return True
        except:pass
        return False
    except:return False
    finally:
        if fd is not None:
            try:os.close(fd)
            except:pass

def _release_run_lock(name:str)->None:
    try:_run_lock_path(name).unlink(missing_ok=True)
    except:pass

CFGRIB_DIR=CACHE/"cfgrib"
CFGRIB_DIR.mkdir(parents=True,exist_ok=True)
os.environ.setdefault("CFGRIB_INDEXPATH",str(CFGRIB_DIR))
COLLAB_BASE="https://www.nohrsc.noaa.gov/pub/products/collaborators"
NSIDC_BASE="https://noaadata.apps.nsidc.org/NOAA/G02158"
_DIRLIST={"ts":0.0,"txt":""}
_COLLAB_LAST={"ts":0.0,"status":None,"error":None,"len":0}
_BUILD_LOCKS:dict[str,threading.Lock]={}
_BUILD_LOCKS_LOCK=threading.Lock()
_LOG=deque(maxlen=2000)
TILE_CACHE_DIR=(CACHE/"tile_png_cache"/"forecast_by_date")
TILE_CACHE_DIR.mkdir(parents=True,exist_ok=True)

def _q(v:float,step:float)->float:return round(v/step)*step

def _lock(key:str)->threading.Lock:
    with _BUILD_LOCKS_LOCK:
        l=_BUILD_LOCKS.get(key)
        if l is None:l=_BUILD_LOCKS[key]=threading.Lock()
        return l

# def _tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int,
#                     cancel_check=None) -> bytes | None:
#     p = _tile_cache_path(dom, hours, ymd, z, x, y)
#     try:
#         if _is_valid_png_file(p):
#             return p.read_bytes()
#         elif p.exists():
#             _safe_unlink(p)
#     except:
#         pass
#     if not _hf_pull_on_miss_enabled():
#         return None
#     if cancel_check and cancel_check():
#         return None
#     rel = p.relative_to(CACHE).as_posix()
#     if _hf_negative_check(rel):
#         return None
#     try:
#         if _hf_try_pull_file(rel, cancel_check=cancel_check):
#             if cancel_check and cancel_check():
#                 return None
#             if _is_valid_png_file(p):
#                 return p.read_bytes()
#             elif p.exists():
#                 _safe_unlink(p)
#         _hf_negative_mark(rel)
#     except Exception as e:
#         if "404" not in str(e).lower():
#             _log("WARN", f"[hf_pull] unexpected error for {rel}: {e!r}")
#         else:
#             _hf_negative_mark(rel)
#     return None

def _tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int,
                    cancel_check=None) -> bytes | None:
    """Check local tile cache. No HF pull — tiles only exist as tar archives
    in HF, not as individual files, so per-tile pulls always 404."""
    p = _tile_cache_path(dom, hours, ymd, z, x, y)
    try:
        if _is_valid_png_file(p):
            return p.read_bytes()
        elif p.exists():
            _safe_unlink(p)
    except:
        pass
    # Per-tile HF pull removed: individual tiles don't exist in the HF repo.
    # Tiles get populated via boot-time tar extraction or local baking.
    # Cache misses fall through to on-demand generation (fast w/ preloaded COGs).
    return None
    
def _hf_date_exists_remote(dom: str, hours: int, ymd: str) -> bool:
    """Check if HF repo has any tiles for this date/hours combo. Fast — only lists one dir."""
    tok, repo = _hf_cfg()
    if not tok or not repo: return False
    api = _hf_api()
    if api is None: return False
    prefix = f"tilecache/{dom}/{hours}/{ymd}"
    try:
        items = api.list_repo_tree(repo_id=repo, repo_type="dataset", path_in_repo=prefix, recursive=False)
        for _ in items:
            return True
    except Exception:
        pass
    return False

def _date_is_baked(dom: str, ymd: str, min_tiles: int = 50) -> bool:
    if _date_has_local_tiles_both(dom, ymd, min_tiles=min_tiles):
        return True
    # Temporarily check HF remote for testing
    for h in WARM_HOURS_LIST:
        if not _hf_date_exists_remote(dom, h, ymd):
            return False
    return True
def _incremental_hf_push(tile_paths:list[Path],repo_prefix:str,batch_name:str)->dict:
    if not tile_paths:return{"ok":True,"pushed":0,"skipped":0}
    if repo_prefix in("tilecache","rawtilecache"):return{"ok":True,"pushed":0,"skipped":len(tile_paths),"disabled":True}
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured","pushed":0,"skipped":0}
    ops:list[CommitOperationAdd]=[];skipped=0
    for p in tile_paths:
        try:
            if(not p.exists())or(p.stat().st_size<=0):skipped+=1;continue
            rel=p.relative_to(CACHE).as_posix()
            if not(rel.startswith("forecast/")or rel.startswith("tracks/")):skipped+=1;continue
            ops.append(CommitOperationAdd(path_in_repo=rel,path_or_fileobj=p.as_posix()))
        except:skipped+=1
    if not ops:return{"ok":True,"pushed":0,"skipped":len(tile_paths)}
    dedup:dict[str,CommitOperationAdd]={}
    for op in ops:dedup[op.path_in_repo]=op
    ops=list(dedup.values());attempts=3;last_err:str|None=None
    for i in range(attempts):
        try:
            api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"{repo_prefix}: {batch_name} ({len(ops)} files)")
            return{"ok":True,"pushed":len(ops),"skipped":(len(tile_paths)-len(ops))+skipped}
        except Exception as e:
            msg=repr(e);last_err=msg
            is_timeout=("ReadTimeout"in msg)or("Timeout"in msg)or("timed out"in msg.lower())
            _log("WARN",f"[incremental_push] failed attempt={i+1}/{attempts}: {msg}")
            if not is_timeout:return{"ok":False,"error":msg,"pushed":0,"skipped":0}
            time.sleep(2.0*(2**i)+random.random())
    return{"ok":False,"error":last_err or"unknown_error","pushed":0,"skipped":0}

def _ensure_cn_mbtiles()->Path:
    CN_MBTILES_LOCAL.parent.mkdir(parents=True,exist_ok=True)
    if CN_MBTILES_LOCAL.exists()and CN_MBTILES_LOCAL.stat().st_size>1024*1024:return CN_MBTILES_LOCAL
    tok,repo=_hf_cfg()
    if not tok or not repo:raise RuntimeError("HF not configured; cannot download cn_tracks.mbtiles")
    p=hf_hub_download(repo_id=repo,repo_type="dataset",filename=CN_MBTILES_REPO_PATH,token=tok,local_dir=str(CN_MBTILES_LOCAL.parent))#,local_dir_use_symlinks=False)
    downloaded=Path(p)
    if downloaded!=CN_MBTILES_LOCAL:
        try:downloaded.replace(CN_MBTILES_LOCAL)
        except:pass
    return CN_MBTILES_LOCAL

def _mbtiles_overzoom(z:int,x:int,y:int,zmax:int)->tuple[int,int,int]:
    z=int(z);x=int(x);y=int(y);zmax=int(zmax)
    if z<=zmax:return z,x,y
    dz=z-zmax;return zmax,(x>>dz),(y>>dz)

def _cogreader_lru_invalidate(cog:Path)->None:
    key=cog.as_posix()
    with _COG_LRU_LK:r=_COG_LRU.pop(key,None)
    if r is not None:
        try:r.close()
        except:pass

def _atomic_replace(src:Path,dst:Path)->None:
    dst.parent.mkdir(parents=True,exist_ok=True)
    try:src.replace(dst)
    except:
        if dst.exists():
            try:dst.unlink(missing_ok=True)
            except:pass
        src.replace(dst)

def _atomic_cog_translate(src_tif:Path,out_cog:Path)->None:
    out_cog.parent.mkdir(parents=True,exist_ok=True)
    tmp_cog=out_cog.with_suffix(out_cog.suffix+".tmp")
    try:tmp_cog.unlink(missing_ok=True)
    except:pass
    prof=cog_profiles.get("deflate")
    if prof is None:raise HTTPException(status_code=500,detail="rio_cogeo deflate profile missing")
    cog_translate(src_tif.as_posix(),tmp_cog.as_posix(),prof,quiet=True,overview_resampling="nearest")
    # Verify the COG is readable before replacing
    try:
        with rasterio.open(tmp_cog.as_posix())as verify:
            verify.read(1,window=rasterio.windows.Window(0,0,min(256,verify.width),min(256,verify.height)))
    except Exception as e:
        _log("WARN",f"[cog_translate] verification failed for {out_cog.name}: {e!r}")
        try:tmp_cog.unlink(missing_ok=True)
        except:pass
        raise HTTPException(status_code=500,detail=f"COG verification failed: {e!r}")
    _atomic_replace(tmp_cog,out_cog)
    try:src_tif.unlink(missing_ok=True)
    except:pass

def _mbtiles_xyz_to_tms(z:int,y:int)->int:return(1<<z)-1-y

# def _cn_mbtiles_conn()->sqlite3.Connection:
#     mb=_ensure_cn_mbtiles();return sqlite3.connect(f"file:{mb}?mode=ro",uri=True,check_same_thread=False)
_CN_CONN: sqlite3.Connection | None = None
_CN_CONN_LK = threading.Lock()

def _cn_mbtiles_conn() -> sqlite3.Connection:
    global _CN_CONN
    if _CN_CONN is not None:
        return _CN_CONN
    with _CN_CONN_LK:
        if _CN_CONN is not None:
            return _CN_CONN
        mb = _ensure_cn_mbtiles()
        _CN_CONN = sqlite3.connect(f"file:{mb}?mode=ro", uri=True, check_same_thread=False)
        return _CN_CONN
        
def _mbtiles_get_tile(conn:sqlite3.Connection,z:int,x:int,y:int)->bytes|None:
    ty=_mbtiles_xyz_to_tms(z,y)
    try:
        cur=conn.execute("SELECT tile_data FROM tiles WHERE zoom_level=? AND tile_column=? AND tile_row=?",(z,x,ty))
        row=cur.fetchone();return row[0]if row else None
    except:return None

@lru_cache(maxsize=1)
def _cn_mbtiles_maxzoom() -> int:
    try:
        conn=_cn_mbtiles_conn()
        cur=conn.execute("SELECT value FROM metadata WHERE name='maxzoom'")
        row=cur.fetchone()
        if row:return int(row[0])
        cur2=conn.execute("SELECT MAX(zoom_level) FROM tiles")
        row2=cur2.fetchone();return int(row2[0])if row2 and row2[0]else 10
    except:return 10

def _cn_overzoom_parent(z:int,x:int,y:int,maxz:int)->tuple[int,int,int]:
    if z<=maxz:return z,x,y
    dz=z-maxz;return maxz,x>>dz,y>>dz

def _tiles_for_boxes_zoom_range(boxes:list[tuple[float,float,float,float]],zmin:int,zmax:int,cap_per_zoom:int=5000)->list[tuple[int,int,int]]:
    out=[]
    for z in range(zmin,zmax+1):
        n=1<<z;zset=set()
        for w,s,e,n_ in boxes:
            x0=int((w+180)/360*n);x1=int((e+180)/360*n)
            y0=int((1-np.log(np.tan(np.radians(n_))+1/np.cos(np.radians(n_)))/np.pi)/2*n)
            y1=int((1-np.log(np.tan(np.radians(s))+1/np.cos(np.radians(s)))/np.pi)/2*n)
            x0,x1=max(0,min(x0,n-1)),max(0,min(x1,n-1));y0,y1=max(0,min(y0,n-1)),max(0,min(y1,n-1))
            if x0>x1:x0,x1=x1,x0
            if y0>y1:y0,y1=y1,y0
            for xx in range(x0,x1+1):
                for yy in range(y0,y1+1):zset.add((z,xx,yy))
        for t in list(zset)[:cap_per_zoom]:out.append(t)
    return out

def _generate_forecast_by_date_png_impl(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,max_in:float|None)->tuple[bytes,dict]:
    dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05")
    if int(hours)==24:
        sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
        if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="forecast data may not be available yet for this date";raise HTTPException(status_code=503,detail=detail)
        try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom)
        except Exception as e:raise HTTPException(status_code=503,detail={"error":"melt24_union_cog_unavailable","date":date_yyyymmdd,"hours":24,"upstream_error":repr(e)})
        try:melt_mm, melt_mask, oob = _tile_arrays_from_cog(melt24_union, z, x, y, cog_cache=None)
        except Exception as e:
            _log("WARN",f"[tile_read] melt24 failed date={date_yyyymmdd} dom={dom} z={z} x={x} y={y} err={e!r}")
            return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":"snowpack_unavailable","X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24","X-Read-Error":"melt24"}
        if oob or melt_mm is None or melt_mask is None:return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":"snowpack_unavailable","X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24"}
        melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255)
        snow=snow_valid=None;snow_info="snowpack_unavailable"
        if sel.get("snowpack_urls"):
            try:
                snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom)
                if snow_union:
                    try:
                        snow_arr,snow_mask,oob2=_tile_arrays_from_cog(snow_union,z,x,y,cog_cache=COG_CACHE)
                        if(not oob2)and(snow_arr is not None)and(snow_mask is not None):snow_valid=(snow_mask>0)&np.isfinite(snow_arr)&(snow_arr>-9990.0);snow=snow_arr;snow_info=f"snowpack_union:{snow_union.name}"
                    except Exception as e2:_log("WARN",f"[tile_read] snowpack24 failed date={date_yyyymmdd} dom={dom} z={z} x={x} y={y} err={e2!r}");snow_info=f"snowpack_unavailable:{e2!r}"
            except Exception as e:snow_info=f"snowpack_unavailable:{e!r}"
        png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_IN)
        return png,{"X-Allowed":"1","X-OOB":"0","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24"}
    try:
        melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom)
    except HTTPException as e:
        raise HTTPException(status_code=503,detail={"error":"melt72_cog_unavailable","date":date_yyyymmdd,"hours":72,"upstream_error":str(e.detail)if hasattr(e,"detail")else str(e)})
    except Exception as e:
        raise HTTPException(status_code=503,detail={"error":"melt72_cog_unavailable","date":date_yyyymmdd,"hours":72,"upstream_error":repr(e)})
    try:
        melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt72_cog,z,x,y,cog_cache=None)
    except Exception as e:
        _log("WARN",f"[tile_read] melt72 failed date={date_yyyymmdd} dom={dom} z={z} x={x} y={y} err={e!r}")
        return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-72h-COG":melt72_cog.name,"X-Read-Error":"melt72"}
    if oob or melt_mm is None or melt_mask is None:
        return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-72h-COG":melt72_cog.name}
    # Snowpack overlay is optional — try run_init but don't block on it
    run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom)
    snow=snow_valid=None;snow_info="snowpack_unavailable"
    if run_init:
        snow_urls=_find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom)
        if snow_urls:
            try:
                snow_cogs=_build_forecast_snowpack_cogs(valid,snow_urls)
                snow,snow_valid,snow_info=_union_tile(snow_cogs,z,x,y,label="snowpack")
            except Exception as e:snow_info=f"snowpack_unavailable:{e!r}"
    png=_melt_to_png(melt_mm,melt_mask,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_72H_IN)
    return png,{"X-Allowed":"1","X-OOB":"0","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":run_init or"none","X-Forecast-72h-COG":melt72_cog.name,"X-SnowPack-INFO":snow_info}

def _generate_forecast_by_date_png(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,max_in:float|None)->tuple[bytes,dict]:
    return _generate_forecast_by_date_png_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,max_in=max_in)

def _union_tile_cached(cogs:list[Path],z:int,x:int,y:int,*,label:str,cog_cache:CogCache,valid_pred:Optional[Callable[[np.ndarray,np.ndarray],np.ndarray]]=None)->tuple[Optional[np.ndarray],Optional[np.ndarray],str]:
    if not cogs:return None,None,f"{label}_no_cogs"
    out_arr:Optional[np.ndarray]=None;out_valid:Optional[np.ndarray]=None;used:list[str]=[]
    for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))):
        try:
            r=cog_cache.get(cog);data,mask=r.tile(x,y,z,tilesize=256,resampling_method="nearest")
            arr=data[0].astype("float32");m=mask[0];valid=(m>0)&np.isfinite(arr)&(arr>-9990.0)
            if valid_pred is not None:
                try:valid=valid&valid_pred(arr,m)
                except:pass
            if out_arr is None:out_arr=arr.copy();out_valid=valid.copy()
            else:
                fill=valid&(~out_valid)
                if fill.any():out_arr[fill]=arr[fill];out_valid[fill]=True
            used.append(cog.name)
        except TileOutsideBounds:continue
        except:continue
    if out_arr is None or out_valid is None:return None,None,f"{label}_unavailable_all_domains"
    info=f"{label}_union:"+",".join(used[:4])+("..."if len(used)>4 else"")
    return out_arr,out_valid,info

def _log(level:str,msg:str)->None:
    _LOG.append({"ts":int(datetime.utcnow().timestamp()*1000),"level":level,"msg":msg})
    print(f"{datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} | {level} | {msg}",flush=True)

_MODE=threading.local()
def _mode_get():return getattr(_MODE,"m","tiles")
class _Mode:
    def __init__(self,m):self.m=m
    def __enter__(self):_MODE.m=self.m
    def __exit__(self,*a):_MODE.m="tiles"

_SESSION=requests.Session()
POOL=int(os.environ.get("HTTP_POOL_SIZE","64"))
_ADAPTER=requests.adapters.HTTPAdapter(pool_connections=POOL,pool_maxsize=POOL,max_retries=0,pool_block=True)
_SESSION.mount("http://",_ADAPTER);_SESSION.mount("https://",_ADAPTER)
HF_PULL_ON_MISS_DEFAULT=(os.environ.get("HF_PULL_ON_MISS","1").strip()!="0")
_HF_PULL_ON_MISS_LOCAL=threading.local()

def _hf_pull_on_miss_enabled()->bool:
    v=getattr(_HF_PULL_ON_MISS_LOCAL,"enabled",None)
    return bool(HF_PULL_ON_MISS_DEFAULT)if v is None else bool(v)

class _HfPullOnMiss:
    def __init__(self,enabled:bool):self.enabled=bool(enabled);self.prev=None
    def __enter__(self):
        self.prev=getattr(_HF_PULL_ON_MISS_LOCAL,"enabled",None);_HF_PULL_ON_MISS_LOCAL.enabled=self.enabled;return self
    def __exit__(self,*a):
        if self.prev is None and hasattr(_HF_PULL_ON_MISS_LOCAL,"enabled"):
            try:delattr(_HF_PULL_ON_MISS_LOCAL,"enabled")
            except:_HF_PULL_ON_MISS_LOCAL.enabled=None
        else:_HF_PULL_ON_MISS_LOCAL.enabled=self.prev

_CPU_SEM_LOCK=threading.Lock()
def _cpu_sem_limits()->tuple[int,int]:
    t=int(os.environ.get("CPU_TILES_MAX_INFLIGHT","12"));v=int(os.environ.get("CPU_VALUE_MAX_INFLIGHT","16"));return t,v

def _cpu_set_limits(*,tiles:int|None=None,value:int|None=None)->None:
    global CPU_SEM_TILES,CPU_SEM_VALUE
    tiles0,value0=_cpu_sem_limits();t=int(tiles if tiles is not None else tiles0);v=int(value if value is not None else value0)
    t=max(1,t);v=max(1,v)
    with _CPU_SEM_LOCK:CPU_SEM_TILES=threading.BoundedSemaphore(t);CPU_SEM_VALUE=threading.BoundedSemaphore(v)

class _CpuLimits:
    def __init__(self,*,tiles:int|None=None,value:int|None=None):pass
    def __enter__(self):return self
    def __exit__(self,*a):pass

def _retry_get(url:str,*,timeout=(10,400),tries:int=6,backoff:float=1.6,headers=None,stream:bool=False,allow_redirects:bool=True)->requests.Response:
    hdr={"User-Agent":"snodas/1.3"}
    if headers:hdr.update(headers)
    last_exc=None
    for i in range(int(tries)):
        sem=HTTP_SEM_VALUE if _mode_get()=="value"else HTTP_SEM_TILES
        with sem:
            try:r=_SESSION.get(url,timeout=timeout,allow_redirects=allow_redirects,headers=hdr,stream=stream)
            except(requests.exceptions.Timeout,requests.exceptions.ConnectionError)as e:last_exc=e;r=None
        if r is None:
            if i<tries-1:time.sleep((backoff**i)+random.uniform(0,0.6));continue
            raise last_exc
        if r.status_code in(429,502,503,504):
            ra=r.headers.get("Retry-After")
            try:wait=float(ra)if ra else None
            except:wait=None
            if i<tries-1:
                if stream:
                    try:r.close()
                    except:pass
                time.sleep((wait if wait is not None else(backoff**i))+random.uniform(0,0.8));continue
            return r
        return r
    raise last_exc

def _forecast_melt24h_union_cog(valid:str,dom_prefer:str="zz")->Path:
    dom_prefer=(dom_prefer or"zz").lower();valid=_norm_ts(valid)
    out=_forecast_dir()/f"{valid}_{dom_prefer}_melt24_union_cog.tif"
    if out.exists()and out.stat().st_size>0:return out
    sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer)
    if not sel.get("ok"):raise HTTPException(status_code=503,detail={"error":"pick_failed","valid":valid,"dom":dom_prefer,"detail":sel})
    melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"])
    if not melt_cogs:raise HTTPException(status_code=503,detail={"error":"no_melt_cogs","valid":valid,"dom":dom_prefer})
    cogs=sorted(melt_cogs,key=lambda p:_dom_pri(_cog_dom(p)))
    with rasterio.open(cogs[0])as base:
        acc=base.read(1).astype("float32");base_transform=base.transform;base_crs=base.crs;base_w,base_h=base.width,base.height
        acc_valid=np.isfinite(acc)&(acc>-9990.0)
        for p in cogs[1:]:
            with rasterio.open(p)as src:
                arr=src.read(1).astype("float32")
                if(src.transform!=base_transform)or(src.crs!=base_crs)or(src.width!=base_w)or(src.height!=base_h):
                    tmp=np.full((base_h,base_w),np.nan,dtype="float32")
                    reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base_transform,dst_crs=base_crs,resampling=Resampling.nearest)
                    arr=tmp
                v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid)
                if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True
        tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True})
        with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1)
    _atomic_cog_translate(tmp_tif,out)
    try:_hf_enqueue_files([out])
    except:pass
    tmp_tif.unlink(missing_ok=True)
    _log("INFO",f"[build_forecast] melt24_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}")
    return out

def _forecast_snowpack_union_cog(valid:str,dom_prefer:str="zz")->Path|None:
    dom_prefer=(dom_prefer or"zz").lower();valid=_norm_ts(valid)
    out=_forecast_dir()/f"{valid}_{dom_prefer}_snow_union_cog.tif"
    if out.exists()and out.stat().st_size>0:return out
    sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer)
    if not sel.get("ok")or not sel.get("snowpack_urls"):return None
    try:snow_cogs=_build_forecast_snowpack_cogs(sel.get("snowpack_ts")or sel["valid"],sel["snowpack_urls"])
    except:return None
    if not snow_cogs:return None
    cogs=sorted(snow_cogs,key=lambda p:_dom_pri(_cog_dom(p)))
    with rasterio.open(cogs[0])as base:
        acc=base.read(1).astype("float32");base_transform=base.transform;base_crs=base.crs;base_w,base_h=base.width,base.height
        acc_valid=np.isfinite(acc)&(acc>-9990.0)
        for p in cogs[1:]:
            with rasterio.open(p)as src:
                arr=src.read(1).astype("float32")
                if(src.transform!=base_transform)or(src.crs!=base_crs)or(src.width!=base_w)or(src.height!=base_h):
                    tmp=np.full((base_h,base_w),np.nan,dtype="float32")
                    reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base_transform,dst_crs=base_crs,resampling=Resampling.nearest)
                    arr=tmp
                v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid)
                if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True
        tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True})
        with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1)
    _atomic_cog_translate(tmp_tif,out)
    try:_hf_enqueue_files([out])
    except:pass
    tmp_tif.unlink(missing_ok=True)
    _log("INFO",f"[build_forecast] snow_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}")
    return out

def _forecast_melt24h_union_end_cog(valid:str,dom_prefer:str="zz")->Path:
    valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    out=_forecast_dir()/f"fcst_melt24h_union_end_{valid}_{dom_prefer}_cog.tif"
    if out.exists()and out.stat().st_size>0:return out
    _hf_try_pull_file(out.relative_to(CACHE).as_posix())
    if out.exists()and out.stat().st_size>0:return out
    with _lock(f"fcst_melt24h_union_{valid}_{dom_prefer}"):
        if out.exists()and out.stat().st_size>0:return out
        sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer)
        if not sel.get("ok"):raise HTTPException(status_code=503,detail={"error":"forecast_selection_unavailable_for_24h_union","valid":valid,"dom":dom_prefer,"sel":sel})
        _log("INFO",f"[debug_union] valid={valid} run_init={sel['run_init']} melt_urls={sel['melt_urls']}")
        melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"])
        _log("INFO",f"[debug_union] melt_cogs built: {[c.name for c in melt_cogs]}")
        if not melt_cogs:raise HTTPException(status_code=503,detail={"error":"no_melt_cogs_for_24h_union","valid":valid,"dom":dom_prefer})
        cogs=sorted(melt_cogs,key=lambda p:_dom_pri(_cog_dom(p)))
        _log("INFO",f"[debug_union] sorted cogs for union: {[c.name for c in cogs]}")
        with rasterio.open(cogs[0])as base:
            acc=base.read(1).astype("float32");acc_valid=np.isfinite(acc)&(acc>-9990.0)
            _log("INFO",f"[debug_union] base cog={cogs[0].name} valid_pixels={acc_valid.sum()} mean_where_valid={float(np.nanmean(acc[acc_valid])):.2f}")
            for c in cogs[1:]:
                with rasterio.open(c)as src:
                    arr=src.read(1).astype("float32")
                    if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height):
                        tmp=np.full_like(acc,np.nan,dtype="float32")
                        reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest)
                        arr=tmp
                    v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid)
                    _log("INFO",f"[debug_union] overlay cog={c.name} valid_pixels={v.sum()} fill_pixels={fill.sum()} mean_where_valid={float(np.nanmean(arr[v])):.2f}")
                    if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True
            _log("INFO",f"[debug_union] final: total_valid={acc_valid.sum()} mean={float(np.nanmean(acc[acc_valid])):.2f} max={float(np.nanmax(acc[acc_valid])):.2f}")
            tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True})
            with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1)
        _atomic_cog_translate(tmp_tif,out)
        try:_hf_enqueue_files([out])
        except:pass
        tmp_tif.unlink(missing_ok=True)
        _log("INFO",f"[build_forecast] melt24h_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}")
        return out
        
def _forecast_snowpack_union_end_cog(valid:str,dom_prefer:str="zz")->Path|None:
    valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    out=_forecast_dir()/f"fcst_snowpack_union_end_{valid}_{dom_prefer}_cog.tif"
    if out.exists()and out.stat().st_size>0:return out
    _hf_try_pull_file(out.relative_to(CACHE).as_posix())
    if out.exists()and out.stat().st_size>0:return out
    with _lock(f"fcst_snowpack_union_{valid}_{dom_prefer}"):
        if out.exists()and out.stat().st_size>0:return out
        sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer)
        if not sel.get("ok")or not sel.get("snowpack_urls"):return None
        try:snow_cogs=_build_forecast_snowpack_cogs(sel.get("snowpack_ts")or sel["valid"],sel["snowpack_urls"])
        except:return None
        if not snow_cogs:return None
        cogs=sorted(snow_cogs,key=lambda p:_dom_pri(_cog_dom(p)))
        with rasterio.open(cogs[0])as base:
            acc=base.read(1).astype("float32");acc_valid=np.isfinite(acc)&(acc>-9990.0)
            for c in cogs[1:]:
                with rasterio.open(c)as src:
                    arr=src.read(1).astype("float32")
                    if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height):
                        tmp=np.full_like(acc,np.nan,dtype="float32")
                        reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest)
                        arr=tmp
                    v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid)
                    if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True
            tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True})
            with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1)
        _atomic_cog_translate(tmp_tif,out)
        try:_hf_enqueue_files([out])
        except:pass
        tmp_tif.unlink(missing_ok=True)
        _log("INFO",f"[build_forecast] snowpack_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}")
        return out

app=FastAPI(title="SNODAS Snowmelt Tiles",version="1.3.0")
app.add_middleware(GZipMiddleware, minimum_size=500)
app.mount("/web",StaticFiles(directory="web",html=True),name="web")

@app.get("/favicon.ico")
def favicon():return Response(status_code=204)

_TS10=re.compile(r"^\d{10}$")
_TS8U2=re.compile(r"^(\d{8})[_-](\d{2})$")
_HREF_RE=re.compile(r'href\s*=\s*(?P<q>[\'"])(?P<u>.*?)(?P=q)',re.I)
_TS_IN_NAME=re.compile(r"(?:TS|NATS)(\d{10})",re.I)
_TT_IN_NAME=re.compile(r"TT(\d{10})",re.I)

def _norm_ts(ts:str)->str:
    ts=(ts or"").strip()
    if _TS10.fullmatch(ts):return ts
    m=_TS8U2.fullmatch(ts)
    if m:return f"{m.group(1)}{m.group(2)}"
    raise HTTPException(status_code=404,detail="unknown forecast key")

US_BORDER_LAT=49.0

def _tile2lon(x:int,z:int)->float:return(float(x)/(1<<int(z)))*360.0-180.0
def _tile2lat(y:int,z:int)->float:
    n=np.pi-(2.0*np.pi*float(y)/(1<<int(z)));return(180.0/np.pi)*np.arctan(0.5*(np.exp(n)-np.exp(-n)))

def _tile_bounds(z:int,x:int,y:int)->tuple[float,float,float,float]:
    z=int(z);x=int(x);y=int(y);return _tile2lon(x,z),_tile2lat(y+1,z),_tile2lon(x+1,z),_tile2lat(y,z)

def _box_intersects(a,b)->bool:
    aw,as_,ae,an=a;bw,bs,be,bn=b;return not(ae<bw or be<aw or an<bs or bn<as_)

def _tile_allowed(z:int,x:int,y:int)->bool:
    z=int(z);x=int(x);y=int(y);n=1<<z
    if y<0 or y>=n or x<0 or x>=n:return True
    tb=_tile_bounds(z,x,y);box_a=(-170.0,49.0,-40.0,90.0);box_b=(-97.0,37.0,-63.0,49.0)
    return _box_intersects(tb,box_a)or _box_intersects(tb,box_b)

HTTP_SEM_TILES=threading.BoundedSemaphore(int(os.environ.get("HTTP_TILES_MAX_INFLIGHT","24")))
HTTP_SEM_VALUE=threading.BoundedSemaphore(int(os.environ.get("HTTP_VALUE_MAX_INFLIGHT","12")))
_cpu_set_limits()

@lru_cache(maxsize=1)
def _transparent_png_256()->bytes:
    rgb=np.zeros((3,256,256),dtype="uint8");a=np.zeros((256,256),dtype="uint8");return render(rgb,mask=a,img_format="PNG")

@lru_cache(maxsize=1)
def _zero_raw_npz_256()->bytes:
    return _pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8"))

_REMOTE_KEEP_DAYS=7

def _hf_cfg()->tuple[Optional[str],Optional[str]]:
    tok=(os.environ.get("HF_TOKEN")or"").strip();repo=(os.environ.get("HF_DATASET_REPO")or"").strip()
    return(tok,repo)if tok and repo else(None,None)

_HF_API_SINGLETON:Optional[HfApi]=None

def _hf_api()->Optional[HfApi]:
    global _HF_API_SINGLETON
    tok,repo=_hf_cfg()
    if not tok or not repo:return None
    if _HF_API_SINGLETON is None:_HF_API_SINGLETON=HfApi()
    return _HF_API_SINGLETON
    
def _date_ymd_utc(d:datetime.date|None=None)->str:
    if d is None:d=datetime.utcnow().date()
    return d.strftime("%Y%m%d")

# def _keep_cutoff_ymd(keep_days:int=_REMOTE_KEEP_DAYS)->str:
#     return(datetime.utcnow().date()-timedelta(days=int(keep_days)-1)).strftime("%Y%m%d")

def _file_ymd_from_name(name:str)->Optional[str]:
    m=re.search(r"(?:^|_)(\d{8})(?:_|\.|$)",name);return m.group(1)if m else None

def _is_cache_file_we_manage(rel:str)->bool:
    n=rel.replace("\\","/")
    if n.startswith("forecast/")and n.endswith("_cog.tif"):return True
    if n.startswith("tilecache/")and n.endswith(".png"):return True
    if n.startswith("rawtilecache/")and n.endswith(".npz"):return True
    base=n.rsplit("/",1)[-1]
    if re.fullmatch(r"bydate_[a-z0-9]+_h\d+_\d{8}_z\d+_\d+_\d+\.png",base):return True
    if re.fullmatch(r"bydate_[a-z0-9]+_h\d+_\d{8}_z\d+_\d+_\d+\.npz",base):return True
    return False

def _local_cache_paths_to_consider()->list[Path]:
    out:list[Path]=[]
    for p in CACHE.glob("melt24h_*_cog.tif"):out.append(p)
    for p in CACHE.glob("melt72h_end_*_cog.tif"):out.append(p)
    fc=_forecast_dir()
    if fc.exists():
        for p in fc.glob("*_cog.tif"):out.append(p)
    return out

def _is_lfs_pointer_bytes(b:bytes)->bool:
    if not b:return False
    if b.startswith(b"version https://git-lfs.github.com/spec/v1"):return True
    if b"git-lfs.github.com/spec/v1"in b[:200]:return True
    return False


def _hf_push_baked_as_tar(baked_ymds:list[str],dom:str,hours_list:list[int])->dict:
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"}
    dom=(dom or"zz").lower();pushed=0;errors=0;archives_pushed=0
    for ymd in baked_ymds:
        for h in hours_list:
            for kind in["tilecache","rawtilecache"]:
                base=CACHE/kind/dom/str(int(h))/ymd
                if not base.exists():continue
                while _PUSH_PAUSED.is_set():time.sleep(2.0)
                ext=".png"if kind=="tilecache"else".npz";min_sz=64 if kind=="tilecache"else 60
                files=[]
                for p in base.rglob(f"*{ext}"):
                    try:
                        if p.stat().st_size<min_sz:continue
                        files.append((p,p.relative_to(CACHE).as_posix()))
                    except:pass
                if not files:continue
                _log("INFO",f"[hf_push_tar] {kind}/{dom}/h{h}/{ymd}: archiving {len(files)} files")
                tar_buf=io.BytesIO()
                try:
                    with tarfile.open(fileobj=tar_buf,mode="w:gz",compresslevel=1)as tar:
                        for lp,rp in files:
                            try:tar.add(str(lp),arcname=rp)
                            except:pass
                except Exception as e:_log("WARN",f"[hf_push_tar] tar failed {kind}/{ymd}/h{h}: {e!r}");errors+=len(files);continue
                tar_bytes=tar_buf.getvalue();tar_mb=len(tar_bytes)/(1024*1024)
                _log("INFO",f"[hf_push_tar] {kind}/{dom}/h{h}/{ymd}: {tar_mb:.1f}MB ({len(files)} files)")
                op=CommitOperationAdd(path_in_repo=f"archives/{kind}_{dom}_h{h}_{ymd}.tar.gz",path_or_fileobj=tar_bytes)
                ok=False
                for attempt in range(3):
                    while _PUSH_PAUSED.is_set():time.sleep(2.0)
                    t0=time.time()
                    try:
                        api.create_commit(repo_id=repo,repo_type="dataset",operations=[op],commit_message=f"{kind}: {ymd} h{h} ({len(files)} files as tar.gz)")
                        pushed+=len(files);archives_pushed+=1;ok=True
                        _log("INFO",f"[hf_push_tar] {kind}/{dom}/h{h}/{ymd}: pushed {tar_mb:.1f}MB in {time.time()-t0:.1f}s");break
                    except Exception as e:
                        elapsed=time.time()-t0;err_str=repr(e)
                        if"ReadTimeout"in err_str and elapsed>=9.0:
                            _log("INFO",f"[hf_push_tar] {kind}/{dom}/h{h}/{ymd}: timeout after {elapsed:.1f}s — likely succeeded");pushed+=len(files);archives_pushed+=1;ok=True;break
                        _log("WARN",f"[hf_push_tar] {kind}/{dom}/h{h}/{ymd} attempt {attempt+1}: {type(e).__name__} after {elapsed:.1f}s")
                        if attempt<2:time.sleep(10.0*(attempt+1))
                if not ok:errors+=len(files)
                del tar_bytes,tar_buf
    _log("INFO",f"[hf_push_tar] done: pushed={pushed} errors={errors} archives={archives_pushed}")
    return{"ok":errors==0,"pushed":pushed,"errors":errors,"archives":archives_pushed}


def _hf_pull_tar_archives()->int:
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return 0
    window=set(_rolling_dates_local_window(lookback=2,lookahead=3))
    _log("INFO",f"[hf_pull_tar] window={sorted(window)}")
    try:
        items=list(api.list_repo_tree(repo_id=repo,repo_type="dataset",path_in_repo="archives/",recursive=False))
        archive_paths=[getattr(it,"path","")for it in items if getattr(it,"path","").endswith(".tar.gz")]
    except Exception as e:_log("WARN",f"[hf_pull_tar] list failed: {e!r}");return 0
    if not archive_paths:_log("INFO","[hf_pull_tar] no archives found");return 0
    _log("INFO",f"[hf_pull_tar] found {len(archive_paths)} archives")
    pulled=0;staging=CACHE/"_tar_staging";staging.mkdir(parents=True,exist_ok=True)
    marker_dir=CACHE/"_tar_markers";marker_dir.mkdir(parents=True,exist_ok=True)
    for ap in archive_paths:
        name=ap.rsplit("/",1)[-1];m=re.search(r"_(\d{8})\.tar\.gz$",name)
        if not m:continue
        ymd=m.group(1)
        if ymd not in window:continue
        marker=marker_dir/f"{name}.done"
        if marker.exists():continue
        parts=name.replace(".tar.gz","").split("_")
        if len(parts)>=4:
            edir=CACHE/parts[0]/parts[1]/parts[2].replace("h","")/parts[3]
            if edir.exists():
                ext="*.png"if parts[0]=="tilecache"else"*.npz"
                try:
                    cnt=sum(1 for _ in edir.rglob(ext))
                    if cnt>100:_log("INFO",f"[hf_pull_tar] skip {name}: {cnt} local files");marker.write_text(f"skipped:{cnt}");continue
                except:pass
        _log("INFO",f"[hf_pull_tar] downloading {ap}");t0=time.time()
        try:
            local=hf_hub_download(repo_id=repo,repo_type="dataset",filename=ap,token=tok,local_dir=str(staging))
            lp=Path(local)
            if not lp.exists():continue
            with lp.open("rb")as f:magic=f.read(2)
            if magic!=b"\x1f\x8b":_log("WARN",f"[hf_pull_tar] {name}: not gzip");lp.unlink(missing_ok=True);continue
            with tarfile.open(str(lp),mode="r:gz")as tar:
                safe=[m for m in tar.getmembers()if not m.name.startswith("/")and".."not in m.name]
                tar.extractall(path=str(CACHE),members=safe);pulled+=len(safe)
            _log("INFO",f"[hf_pull_tar] unpacked {name}: {len(safe)} files in {time.time()-t0:.1f}s")
            marker.write_text(f"unpacked:{len(safe)}:ts:{time.time()}")
            try:lp.unlink(missing_ok=True)
            except:pass
        except Exception as e:_log("WARN",f"[hf_pull_tar] failed {ap}: {e!r}")
    try:shutil.rmtree(staging,ignore_errors=True)
    except:pass
    _log("INFO",f"[hf_pull_tar] complete: {pulled} files");return pulled


# def _hf_pull_cache()->None:
#     tok,repo=_hf_cfg()
#     if not tok or not repo:_log("INFO","[hf] pull cache disabled");return
#     window=_rolling_dates_local_window(lookback=1,lookahead=3);dom=(WARM_DOM or"zz").lower()
#     _log("INFO",f"[hf] pull COGs for window={window}");pulled=0
#     for ymd in window:
#         for name in[f"forecast/fcst_melt24h_union_end_{ymd}05_{dom}_cog.tif",f"forecast/fcst_melt72h_end_{ymd}05_{dom}_cog.tif",f"forecast/fcst_snowpack_union_end_{ymd}05_{dom}_cog.tif"]:
#             dest=CACHE/name
#             if dest.exists()and dest.stat().st_size>1000:continue
#             try:hf_hub_download(repo_id=repo,repo_type="dataset",token=tok,filename=name,local_dir=str(CACHE));pulled+=1
#             except:pass
#     _log("INFO",f"[hf] pull COGs done: {pulled}")
#     _log("INFO","[hf] pull tar archives")
#     try:pt=_hf_pull_tar_archives();_log("INFO",f"[hf] pull tar archives done: {pt} files")
#     except Exception as e:_log("WARN",f"[hf] pull tar archives failed: {e!r}")

def _hf_pull_cache()->None:
    tok,repo=_hf_cfg()
    if not tok or not repo:_log("INFO","[hf] pull cache disabled");return
    window=_rolling_dates_local_window(lookback=1,lookahead=3);dom=(WARM_DOM or"zz").lower()
    _log("INFO",f"[hf] pull COGs for window={window}");pulled=0
    for ymd in window:
        # Try both 11z and 05z COGs — staleness monitor may have upgraded to 11z
        for valid_suffix in ["11", "05"]:
            for prefix in ["fcst_melt24h_union_end", "fcst_melt72h_end", "fcst_snowpack_union_end"]:
                name = f"forecast/{prefix}_{ymd}{valid_suffix}_{dom}_cog.tif"
                dest = CACHE / name
                if dest.exists() and dest.stat().st_size > 1000:
                    continue
                try:
                    hf_hub_download(repo_id=repo, repo_type="dataset", token=tok, filename=name, local_dir=str(CACHE))
                    pulled += 1
                except:
                    pass
    _log("INFO",f"[hf] pull COGs done: {pulled}")
    _log("INFO","[hf] pull tar archives")
    try:pt=_hf_pull_tar_archives();_log("INFO",f"[hf] pull tar archives done: {pt} files")
    except Exception as e:_log("WARN",f"[hf] pull tar archives failed: {e!r}")
 

def _hf_commit(ops:list,message:str)->None:
    api=_hf_api();tok,repo=_hf_cfg()
    if api is None or not repo or not ops:return
    api.create_commit(repo_id=repo,repo_type="dataset",operations=ops,commit_message=message)

def _hf_commit_with_backoff(ops:list,message:str)->None:
    api=_hf_api();tok,repo=_hf_cfg()
    if api is None or not repo or not ops:return
    for i in range(6):
        try:
            api.create_commit(repo_id=repo,repo_type="dataset",operations=ops,commit_message=message)
            if HF_PUSH_SLEEP_SEC>0:time.sleep(HF_PUSH_SLEEP_SEC)
            return
        except Exception as e:
            s=repr(e).lower();is_rate=("429"in s)or("rate"in s and"limit"in s)or("too many requests"in s)
            if(i<5)and is_rate:time.sleep(2.0*(2**i)+random.uniform(0,0.25));continue
            raise

_TILE_Z_RE=re.compile(r"_z(\d+)_")

def _hf_push_tilecache_for_window(window:list[str],dom:str,hours_list:list[int],zmin:int|None=None,zmax:int|None=None)->dict:
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"}
    dom=(dom or"zz").lower();root=_tile_cache_dir()
    if not root.exists():return{"ok":True,"pushed":0,"skipped":0}
    want_dates=set(window);want_hours=set(int(h)for h in hours_list)
    ops:list[CommitOperationAdd]=[];pushed=0;skipped=0
    for p in root.rglob("bydate_*.png"):
        name=p.name;m=re.match(r"^bydate_([a-z0-9]+)_h(\d+)_(\d{8})_z\d+_\d+_\d+\.png$",name)
        if not m:continue
        f_dom,f_hours,f_ymd=m.group(1),int(m.group(2)),m.group(3)
        if f_dom!=dom or f_hours not in want_hours or f_ymd not in want_dates:continue
        mz=_TILE_Z_RE.search(name);z=int(mz.group(1))if mz else None
        if z is not None and zmin is not None and z<int(zmin):continue
        if z is not None and zmax is not None and z>int(zmax):continue
        if not _is_valid_png_file(p):skipped+=1;continue
        ops.append(CommitOperationAdd(path_in_repo=f"tilecache/{name}",path_or_fileobj=p.as_posix()))
        if len(ops)>=int(HF_MAX_OPS_PER_COMMIT):
            api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"tilecache: update dom={dom} files={len(ops)}")
            pushed+=len(ops);ops=[]
            if HF_PUSH_SLEEP_SEC>0:time.sleep(float(HF_PUSH_SLEEP_SEC))
    if ops:
        api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"tilecache: update dom={dom} files={len(ops)}")
        pushed+=len(ops)
    return{"ok":True,"pushed":pushed,"skipped":skipped,"dom":dom}

def _hf_push_files(paths:list[Path])->None:
    tok,repo=_hf_cfg()
    if not tok or not repo:return
    ops=[]
    for p in paths:
        try:
            if not p.exists()or p.stat().st_size<=0:continue
            rel=p.relative_to(CACHE).as_posix()
            if not _is_cache_file_we_manage(rel):continue
            ops.append(CommitOperationAdd(path_in_repo=rel,path_or_fileobj=p.as_posix()))
        except:continue
    try:
        _hf_commit(ops,message=f"cache: add/update {len(ops)} files")
        if ops:_log("INFO",f"[hf_cache] push OK files={len(ops)}")
    except Exception as e:_log("INFO",f"[hf_cache] push FAIL err={e!r}")


def _hf_prune_remote_and_local(keep_days:int=_REMOTE_KEEP_DAYS,*,keep_ymds:set[str]|None=None)->None:
    if keep_ymds is None:cutoff=_keep_cutoff_ymd(keep_days)
    else:cutoff=min(keep_ymds)if keep_ymds else datetime.now(LOCAL_TZ).date().strftime("%Y%m%d")
    _log("INFO",f"[hf_prune] keep_ymds={sorted(keep_ymds)if keep_ymds else'None'} cutoff={cutoff}")
    local_dirs=0
    for kind in["tilecache","rawtilecache"]:
        root=CACHE/kind
        if not root.exists():continue
        for dom_dir in root.iterdir():
            if not dom_dir.is_dir():continue
            for hours_dir in dom_dir.iterdir():
                if not hours_dir.is_dir():continue
                for ymd_dir in hours_dir.iterdir():
                    if not ymd_dir.is_dir():continue
                    ymd=ymd_dir.name
                    if not re.fullmatch(r"[0-9]{8}",ymd):continue
                    if(keep_ymds and ymd not in keep_ymds)or(not keep_ymds and ymd<cutoff):
                        shutil.rmtree(ymd_dir,ignore_errors=True);local_dirs+=1
    local_files=0
    try:
        fc=_forecast_dir()
        if fc.exists():
            for p in fc.glob("*_cog.tif"):
                m=re.search(r"([0-9]{8})",p.name)
                if not m:continue
                ymd=m.group(1)
                if(keep_ymds and ymd not in keep_ymds)or(not keep_ymds and ymd<cutoff):_safe_unlink(p);local_files+=1
    except:pass
    local_markers=0;marker_dir=CACHE/"_tar_markers"
    if marker_dir.exists():
        for mk in marker_dir.glob("*.done"):
            m=re.search(r"_(\d{8})\.tar\.gz\.done$",mk.name)
            if not m:continue
            ymd=m.group(1)
            if(keep_ymds and ymd not in keep_ymds)or(not keep_ymds and ymd<cutoff):
                try:mk.unlink(missing_ok=True);local_markers+=1
                except:pass
    _log("INFO",f"[hf_prune] local done: dirs={local_dirs} files={local_files} markers={local_markers}")
    api=_hf_api();tok,repo=_hf_cfg()
    if api is None or not repo:return
    dom=(WARM_DOM or"zz").lower();folder_ops=[]
    for kind in["tilecache","rawtilecache"]:
        for h in WARM_HOURS_LIST:
            prefix=f"{kind}/{dom}/{h}/"
            try:
                items=list(api.list_repo_tree(repo_id=repo,repo_type="dataset",path_in_repo=prefix,recursive=False))
                for item in items:
                    name=getattr(item,"path","").rsplit("/",1)[-1]if hasattr(item,"path")else""
                    if not re.fullmatch(r"[0-9]{8}",name):continue
                    if(keep_ymds and name not in keep_ymds)or(not keep_ymds and name<cutoff):
                        folder_ops.append(CommitOperationDelete(path_in_repo=f"{prefix}{name}/"));_log("INFO",f"[hf_prune] will delete dir: {prefix}{name}/")
            except:pass
    try:
        items=list(api.list_repo_tree(repo_id=repo,repo_type="dataset",path_in_repo="archives/",recursive=False))
        for item in items:
            fname=getattr(item,"path","").rsplit("/",1)[-1]if hasattr(item,"path")else""
            if not fname.endswith(".tar.gz"):continue
            m=re.search(r"_(\d{8})\.tar\.gz$",fname)
            if not m:continue
            ymd=m.group(1)
            if(keep_ymds and ymd not in keep_ymds)or(not keep_ymds and ymd<cutoff):
                folder_ops.append(CommitOperationDelete(path_in_repo=f"archives/{fname}"));_log("INFO",f"[hf_prune] will delete archive: archives/{fname}")
    except:pass
    try:
        items=list(api.list_repo_tree(repo_id=repo,repo_type="dataset",path_in_repo="forecast/",recursive=False))
        for item in items:
            fname=getattr(item,"path","").rsplit("/",1)[-1]if hasattr(item,"path")else""
            m=re.search(r"([0-9]{8})",fname)
            if not m:continue
            ymd=m.group(1)
            if(keep_ymds and ymd not in keep_ymds)or(not keep_ymds and ymd<cutoff):
                folder_ops.append(CommitOperationDelete(path_in_repo=f"forecast/{fname}"))
    except:pass
    if not folder_ops:_log("INFO","[hf_prune] remote: nothing to delete");return
    _log("INFO",f"[hf_prune] remote: deleting {len(folder_ops)} items");deleted=0
    for op in folder_ops:
        for attempt in range(3):
            try:
                api.create_commit(repo_id=repo,repo_type="dataset",operations=[op],commit_message=f"prune: delete {op.path_in_repo}")
                _log("INFO",f"[hf_prune] deleted: {op.path_in_repo}");deleted+=1;time.sleep(1.0);break
            except Exception as e:
                err_str=repr(e)
                if"404"in err_str or"EntryNotFound"in err_str or"does not exist"in err_str:_log("INFO",f"[hf_prune] {op.path_in_repo}: already deleted");deleted+=1;break
                if"ReadTimeout"in err_str and attempt==0:_log("INFO",f"[hf_prune] {op.path_in_repo}: timeout (likely succeeded)");deleted+=1;time.sleep(2.0);break
                _log("WARN",f"[hf_prune] {op.path_in_repo} attempt {attempt+1} failed: {e!r}");time.sleep(5.0*(attempt+1))
    _log("INFO",f"[hf_prune] remote done: deleted {deleted}/{len(folder_ops)}")

def _ts_to_dt(ts10:str)->datetime:return datetime.strptime(ts10,"%Y%m%d%H").replace(tzinfo=timezone.utc)
def _yyyymmdd(d)->str:return d.strftime("%Y%m%d")

def _track_points_path()->Path:
    for p in[Path("web")/"cn_track_points.npz",Path("cn_track_points.npz"),Path("cache")/"cn_track_points.npz"]:
        if p.exists():return p
    return Path("web")/"cn_track_points.npz"

@lru_cache(maxsize=4)
def _load_track_points(max_points:int=35000)->list[tuple[float,float]]:
    p=_track_points_path()
    if not p.exists():_log("INFO",f"[prewarm] cn_track_points.npz not found at {p}");return[]
    try:
        data=np.load(str(p));arr=data["lonlat"]
        if arr.ndim!=2 or arr.shape[1]!=2:_log("INFO",f"[prewarm] cn_track_points.npz has unexpected shape {arr.shape}");return[]
        arr=arr.astype(np.float64,copy=False)
    except Exception as e:_log("INFO",f"[prewarm] failed to read cn_track_points.npz err={e!r}");return[]
    if arr.shape[0]==0:return[]
    if arr.shape[0]>int(max_points):step=max(1,int(arr.shape[0])//int(max_points));arr=arr[::step]
    return[(float(lon),float(lat))for lon,lat in arr]

def _lonlat_to_tile(lon:float,lat:float,z:int)->tuple[int,int]:
    lat=max(-85.05112878,min(85.05112878,float(lat)));lon=float(lon);n=2**int(z)
    x=int((lon+180.0)/360.0*n);lat_rad=np.deg2rad(lat)
    y=int((1.0-np.log(np.tan(lat_rad)+1.0/np.cos(lat_rad))/np.pi)/2.0*n)
    return max(0,min(n-1,x)),max(0,min(n-1,y))

def _collab_dirlist(ttl_sec:float=240.0)->str:
    now=datetime.utcnow().timestamp()
    if _DIRLIST["txt"]and(now-float(_DIRLIST["ts"]))<float(ttl_sec):return _DIRLIST["txt"]
    url=COLLAB_BASE+"/"
    try:
        r=_retry_get(url,timeout=(10,300),tries=3,headers={"User-Agent":"snodas-forecast/1.3"})
        _COLLAB_LAST.update({"ts":now,"status":r.status_code,"error":None,"len":len(r.text or"")})
        if r.status_code==200 and(r.text or"").strip():
            _DIRLIST["ts"]=now;_DIRLIST["txt"]=r.text or"";return _DIRLIST["txt"]
        _COLLAB_LAST["error"]=f"HTTP {r.status_code} (empty={not bool((r.text or'').strip())})";_log("INFO",f"[collab] dirlist bad: {_COLLAB_LAST['error']}");return""
    except Exception as e:_COLLAB_LAST.update({"ts":now,"status":None,"error":repr(e),"len":0});_log("INFO",f"[collab] dirlist EXC: {repr(e)}");return""

def _hrefs(html:str)->list[str]:return[m.group("u").strip()for m in _HREF_RE.finditer(html or"")if(m.group("u")or"").strip()]
def _ts_from_name(name:str)->Optional[str]:m=_TS_IN_NAME.search(Path(name).name);return m.group(1)if m else None
def _tt_from_name(name:str)->Optional[str]:m=_TT_IN_NAME.search(Path(name).name);return m.group(1)if m else None
def _dom_from_url(url:str)->str:n=Path(url).name.lower();return(n.split("_",1)[0]if"_"in n else n[:2]).strip()or"xx"

def _dom_pri(dom:str)->int:
    d=(dom or"").lower()
    if d=="zz":return 0
    if d=="dp":return 1
    if d=="qc":return 2
    if d=="us":return 3
    return 9

def _as_url(n:str)->str:return n if n.startswith("http")else f"{COLLAB_BASE}/{n}"
def _score_dom(url:str,dom_prefer:str)->int:d=_dom_from_url(url);return 0 if d==dom_prefer else(_dom_pri(d)+10)

def _valid_ts_for_days(days:int,now_utc:Optional[datetime]=None)->str:
    now_utc=now_utc or datetime.utcnow().replace(tzinfo=timezone.utc)
    return f"{(now_utc.date()+timedelta(days=int(days))).strftime('%Y%m%d')}05"

def _is_data_name(low:str)->bool:low=low.lower();return low.endswith(".nc")or low.endswith(".grz")or low.endswith(".grib2")

def _list_available_valid_ts_t0024(dom_prefer:str="zz")->list[str]:
    dom_prefer=(dom_prefer or"zz").lower();html=_collab_dirlist();names=_hrefs(html);ts_set:set[str]=set()
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low or"__t0024tt"not in low or not low.endswith(".nc"):continue
        ts=_ts_from_name(base)
        if ts and _TS10.fullmatch(ts):ts_set.add(ts)
    return sorted(ts_set)

def _lon0360_to_180(x:np.ndarray)->np.ndarray:return((x+180.0)%360.0)-180.0

def _best_state_nc_near_valid(code:str,valid:str,dom_prefer:str="zz",max_after_h:float=18.0,max_before_h:float=240.0)->dict:
    code=(code or"").lower();dom_prefer=(dom_prefer or"zz").lower();valid=_norm_ts(valid)
    html=_collab_dirlist();names=_hrefs(html);cands=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if code not in low or"__t0001ttnats"not in low or"hp001"not in low or not low.endswith(".nc"):continue
        ts=_ts_from_name(base)
        if not ts or not _TS10.fullmatch(ts):continue
        url=_as_url(base)
        try:dh=(_ts_to_dt(ts)-_ts_to_dt(valid)).total_seconds()/3600.0
        except:continue
        cands.append((dh,ts,_dom_from_url(url),url))
    if not cands:return{"picked_ts":None,"url":None,"delta_h":None}
    preferred=[r for r in cands if(r[2]or"").lower()==dom_prefer];pool=preferred if preferred else cands
    before=[r for r in pool if r[0]<=0];after=[r for r in pool if r[0]>0];pick=None
    if before:before.sort(key=lambda t:abs(t[0]));pick=before[0]if abs(before[0][0])<=float(max_before_h)else None
    if pick is None and after:after.sort(key=lambda t:t[0]);pick=after[0]if after[0][0]<=float(max_after_h)else None
    if pick is None:pool.sort(key=lambda t:abs(t[0]));pick=pool[0]
    dh,ts,dom,url=pick;return{"picked_ts":ts,"url":url,"delta_h":dh,"dom":dom}

def _find_fcst_11034_grib2_tt_ts(run_init:str,valid:str,dom_prefer:str="zz")->list[str]:
    run_init=_norm_ts(run_init);valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_tt=f"tt{run_init}";want_ts=f"ts{valid}";out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11034"not in low or"__t0001tt"not in low or"hp001"not in low:continue
        if want_tt not in low or want_ts not in low or not low.endswith(".grib2"):continue
        out.append(_as_url(base))
    return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))

def _pick_best_run_init_for_valid_t0024(valid:str,dom_prefer:str="zz")->Optional[str]:
    valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower();html=_collab_dirlist();names=_hrefs(html)
    want_ts=f"ts{valid}";rows=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low or"__t0024tt"not in low or want_ts not in low or not low.endswith(".nc"):continue
        tt=_tt_from_name(base)
        if not tt:continue
        rows.append((tt,_score_dom(_as_url(base),dom_prefer)))
    if not rows:return None
    rows.sort(key=lambda t:(t[0],-t[1]));return rows[-1][0]

def _find_melt_11044_t0024(run_init:str,valid:str,dom_prefer:str="zz")->list[str]:
    run_init=_norm_ts(run_init);valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_tt=f"tt{run_init}";want_ts=f"ts{valid}";out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low or"__t0024tt"not in low or want_tt not in low or want_ts not in low or not low.endswith(".nc"):continue
        out.append(_as_url(base))
    return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))

def _pick_forecast_for_valid(valid:str,dom_prefer:str="zz")->dict:
    dom_prefer=(dom_prefer or"zz").lower()
    # Extract date from valid (could be YYYYMMDD05 or YYYYMMDD11 or just YYYYMMDD)
    if len(valid)>=8:
        date_ymd=valid[:8]
    else:
        date_ymd=valid
    best=_find_best_melt_for_date(date_ymd,dom_prefer=dom_prefer)
    if not best.get("ok"):
        _log("DEBUG",f"[pick_forecast] no melt found for {date_ymd} dom={dom_prefer}: {best.get('error','unknown')}")
        return{"ok":False,"valid":valid,"error":best.get("error","no melt files found")}
    snowpack_urls=_find_snowpack_for_best_melt(best,dom_prefer=dom_prefer)
    actual_valid=best["valid"]
    return{
        "ok":True,
        "run_init":best["run_init"],
        "valid":actual_valid,
        "lead":"t0024",
        "melt_urls":best["melt_urls"],
        "snowpack_ts":actual_valid,
        "snowpack_urls":snowpack_urls or[],
        "snowpack_missing":not bool(snowpack_urls),
        "source_type":best["source_type"],
    }

def _pick_forecast_for_days(days:int,dom_prefer:str="zz")->dict:
    if days not in(2,3):raise HTTPException(status_code=400,detail="days must be 2 or 3")
    return{"ok":True,"run_init":run_init,"valid":valid,"lead":"t0024","melt_urls":melt_urls,"snowpack_ts":valid,"snowpack_urls":snowpack_urls or[],"snowpack_missing":not bool(snowpack_urls)}
    now=datetime.utcnow().replace(tzinfo=timezone.utc);valid=_valid_ts_for_days(days,now_utc=now)
    sel=_pick_forecast_for_valid(valid,dom_prefer=(dom_prefer or"zz").lower())
    if sel.get("ok"):sel["days"]=days;sel["now_utc"]=now.isoformat()
    return sel
def _find_melt_11044_ttna(valid:str,dom_prefer:str="zz")->list[str]:
    """Find TTNA (analysis) melt files for a given valid time."""
    valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low:continue
        if"__t0024ttnats"not in low:continue
        if want_ts not in low:continue
        if not low.endswith(".nc"):continue
        out.append(_as_url(base))
    return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))

def _find_snowpack_for_valid(valid:str,dom_prefer:str="zz")->list[str]:
    """Find snowpack files for a given valid time, checking both TTNA and forecast patterns."""
    valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";out=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11034"not in low and"ssmv11036"not in low:continue
        if"hp001"not in low:continue
        if want_ts not in low:continue
        if not(low.endswith(".grib2")or low.endswith(".nc")):continue
        out.append(_as_url(base))
    if out:return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10))
    # Fallback: try finding via the forecast TT approach
    run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom_prefer)
    if run_init:return _find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom_prefer)
    return[]
def _nsidc_month_folder(d)->str:return f"{d.month:02d}_{calendar.month_abbr[d.month].title()}"
def _nsidc_url(date,masked:bool)->str:return f"{NSIDC_BASE}/{date.year}/{_nsidc_month_folder(date)}/{_yyyymmdd(date)}/SNODAS_{'masked'if masked else'unmasked'}_{_yyyymmdd(date)}.tar"

def _nsidc_exists(date)->tuple[bool,str]:
    hdr={"User-Agent":"snodas/1.3"}
    for masked in(False,True):
        url=_nsidc_url(date,masked)
        try:
            r=requests.head(url,timeout=20,allow_redirects=True,headers=hdr)
            if r.status_code==200:return True,url
            if r.status_code in(403,405):
                rg=_retry_get(url,stream=True,timeout=(10,60),tries=2,headers=hdr);rg.close()
                if rg.status_code==200:return True,url
        except:pass
    return False,_nsidc_url(date,masked=False)

@lru_cache(maxsize=512)
def _nsidc_exists_cached(dstr:str)->tuple[bool,str]:return _nsidc_exists(datetime.strptime(dstr,"%Y%m%d").date())

def _http_get(url:str,timeout:int=180)->bytes:
    r=_retry_get(url,timeout=(15,timeout),tries=3,headers={"User-Agent":"snodas/1.3"});r.raise_for_status();return r.content

@lru_cache(maxsize=1)
def _resolve_latest_nsidc_day(days_back:int=120)->Optional[str]:
    utc_today=datetime.utcnow().date()
    for i in range(days_back+1):
        d=utc_today-timedelta(days=i);ok,_=_nsidc_exists(d)
        if ok:return _yyyymmdd(d)
    return None

def _download_day_tar(date)->Path:
    dstr=_yyyymmdd(date);out=CACHE/f"SNODAS_{dstr}.tar"
    if out.exists()and out.stat().st_size>0:return out
    ok,url=_nsidc_exists_cached(dstr)
    if not ok:raise HTTPException(status_code=404,detail=f"NSIDC not available for {dstr}")
    with _lock(f"nsidc_tar_{dstr}"):
        if out.exists()and out.stat().st_size>0:return out
        out.write_bytes(_http_get(url,timeout=360))
    return out

_KV_RE=re.compile(r"^\s*(?P<k>[^:=#;]+?)\s*(?:[:=]\s*|\s+)(?P<v>.+?)\s*$")

def _parse_hdr(path:Path)->tuple[dict,str]:
    txt=path.read_text(encoding="utf-8",errors="ignore");meta:dict[str,str]={}
    for raw in txt.splitlines():
        line=raw.strip()
        if not line or line.startswith(("#",";")):continue
        m=_KV_RE.match(line)
        if not m:continue
        k=m.group("k").strip().lower();v=m.group("v").strip().split("#",1)[0].split(";",1)[0].strip()
        meta[k]=v;meta[re.sub(r"[^a-z0-9]+","",k)]=v
    return meta,txt

def _dtype_from_hdr(nbits:int,pixeltype:str,byteorder:str)->np.dtype:
    pixeltype=(pixeltype or"").upper();byteorder=(byteorder or"I").upper()
    endian="<"if byteorder in("I","LSBFIRST","LITTLEENDIAN")else">"
    if pixeltype=="FLOAT":return np.dtype(endian+("f4"if nbits<=32 else"f8"))
    if pixeltype in("UNSIGNEDINT","UNSIGNED","UINTEGER"):return np.dtype(endian+("u1"if nbits<=8 else"u2"if nbits<=16 else"u4"))
    if pixeltype in("SIGNEDINT","SIGNED","INTEGER"):return np.dtype(endian+("i1"if nbits<=8 else"i2"if nbits<=16 else"i4"))
    return np.dtype(endian+"f4")

def _bil_to_geotiff(hdr:Path,bil:Path,out_tif:Path)->Path:
    meta,raw=_parse_hdr(hdr);raw_l=raw.lower()
    def gi(*keys,default=None):
        for k in keys:
            kk=re.sub(r"[^a-z0-9]+","",k.lower())
            for cand in(k.lower(),kk):
                v=meta.get(cand)
                if v is None:continue
                try:return int(float(str(v).strip()))
                except:pass
        return default
    def gf(*keys,default=None):
        for k in keys:
            kk=re.sub(r"[^a-z0-9]+","",k.lower())
            for cand in(k.lower(),kk):
                v=meta.get(cand)
                if v is None:continue
                try:return float(str(v).strip())
                except:pass
        return default
    def re_float(pat:str)->Optional[float]:
        m=re.search(pat,raw,re.I);return float(m.group(1))if m else None
    ncols=gi("ncols","samples","nsamples","columns")
    nrows=gi("nrows","lines","nlines","rows")
    if ncols is None:m=re.search(r"Number of columns:\s*([0-9]+)",raw,re.I);ncols=int(m.group(1))if m else None
    if nrows is None:m=re.search(r"Number of rows:\s*([0-9]+)",raw,re.I);nrows=int(m.group(1))if m else None
    if ncols is None or nrows is None:raise HTTPException(status_code=502,detail="BIL header missing ncols/nrows")
    bands=gi("bands","nbands","number of bands",default=1)or 1
    nbits=gi("nbits","bitspersample","bits per sample",default=32)or 32
    dbpp=gi("data bytes per pixel",default=None)
    if dbpp:nbits=int(dbpp)*8
    pixeltype=meta.get("pixeltype")or meta.get("datatype")or"FLOAT"
    byteorder=meta.get("byteorder")or"I"
    slope=gf("data slope","slope",default=1.0)or 1.0
    intercept=gf("data intercept","intercept",default=0.0)or 0.0
    nodata_raw=gf("no data value","nodata","nodata_value",default=None)
    ulx=gf("ulxmap","upper left x","ulx",default=None)
    uly=gf("ulymap","upper left y","uly",default=None)
    xll=gf("xllcorner","xllcenter","lower left x","xll",default=None)
    yll=gf("yllcorner","yllcenter","lower left y","yll",default=None)
    xres=gf("x-axis resolution","x axis resolution","xdim","x dim","cellsize",default=None)
    yres=gf("y-axis resolution","y axis resolution","ydim","y dim","cellsize",default=None)
    if xres is None:xres=re_float(r"X[- ]axis resolution:\s*([0-9.eE+-]+)")
    if yres is None:yres=re_float(r"Y[- ]axis resolution:\s*([0-9.eE+-]+)")
    xmin=gf("minimum x-axis coordinate","min x","minx",default=None)
    xmax=gf("maximum x-axis coordinate","max x","maxx",default=None)
    ymin=gf("minimum y-axis coordinate","min y","miny",default=None)
    ymax=gf("maximum y-axis coordinate","max y","maxy","maximumy",default=None)
    if xmin is None:xmin=re_float(r"Minimum x[- ]axis coordinate:\s*([0-9.eE+-]+)")
    if xmax is None:xmax=re_float(r"Maximum x[- ]axis coordinate:\s*([0-9.eE+-]+)")
    if ymin is None:ymin=re_float(r"Minimum y[- ]axis coordinate:\s*([0-9.eE+-]+)")
    if ymax is None:ymax=re_float(r"Maximum y[- ]axis coordinate:\s*([0-9.eE+-]+)")
    if xres is None and xmin is not None and xmax is not None:xres=(xmax-xmin)/float(ncols)
    if yres is None and ymin is not None and ymax is not None:yres=(ymax-ymin)/float(nrows)
    if ulx is None and xll is not None:ulx=float(xll)
    if uly is None and yll is not None and yres is not None:uly=float(yll)+abs(float(yres))*float(nrows)
    if ulx is None and xmin is not None:ulx=float(xmin)
    if uly is None and ymax is not None:uly=float(ymax)
    if ulx is None or uly is None or xres is None or yres is None:raise HTTPException(status_code=502,detail="BIL header missing geo params")
    units_line=(meta.get("data units")or meta.get("dataunits")or"").strip();denom=None
    m=re.search(r"Data units:\s*Meters\s*/\s*([0-9.]+)",raw,re.I)
    if m:
        try:denom=float(m.group(1))
        except:denom=None
    if denom is None and units_line and"/"in units_line:
        try:denom=float(units_line.split("/",1)[1].strip())
        except:denom=None
    is_nohrsc_gisrs="nohrsc gis/rs raster file"in raw_l
    file_bytes=bil.stat().st_size;bytes_per_sample=max(1,int(nbits)//8)
    expected_samples=int(nrows)*int(ncols)*max(1,int(bands));expected_bytes=expected_samples*bytes_per_sample
    if is_nohrsc_gisrs:dtp=np.dtype(">i2");expected_bytes=expected_samples*2
    else:dtp=_dtype_from_hdr(int(nbits),str(pixeltype),str(byteorder))
    if file_bytes!=expected_bytes:
        inferred=None
        for bps in(1,2,4):
            if file_bytes==expected_samples*bps:inferred=bps;break
        if inferred is None:raise HTTPException(status_code=502,detail=f"BIL byte mismatch file={file_bytes} expected={expected_bytes}")
        if is_nohrsc_gisrs and inferred!=2:raise HTTPException(status_code=502,detail="NOHRSC GIS/RS expected 2 bytes/sample")
        if not is_nohrsc_gisrs:
            if inferred==2:dtp=np.dtype("<i2")
            elif inferred==4:dtp=np.dtype("<f4")
    arr=np.fromfile(bil,dtype=dtp)
    if arr.size!=expected_samples:raise HTTPException(status_code=502,detail=f"BIL size mismatch expected={expected_samples} got={arr.size}")
    arr=arr.reshape((int(nrows),int(ncols)))if int(bands)==1 else arr.reshape((int(bands),int(nrows),int(ncols)))
    arr=arr.astype("float32")
    if slope!=1.0 or intercept!=0.0:arr=arr*float(slope)+float(intercept)
    if is_nohrsc_gisrs:
        if denom is None or denom<=0:raise HTTPException(status_code=502,detail="NOHRSC GIS/RS missing units denom")
        arr=arr*(1000.0/float(denom));nodata=(float(nodata_raw)*(1000.0/float(denom)))if nodata_raw is not None else None
    else:
        den=1.0
        if units_line and"/"in units_line:
            try:den=float(units_line.split("/",1)[1].strip())or 1.0
            except:den=1.0
        if den!=1.0:arr=arr/float(den)
        if"meter"in(units_line or"").lower():arr=arr*1000.0
        nodata=float(nodata_raw)if nodata_raw is not None else None
    arr[arr<=-9990.0]=nodata if nodata is not None else np.nan
    crs=CRS.from_epsg(4326);transform=Affine(float(xres),0.0,float(ulx),0.0,-abs(float(yres)),float(uly))
    profile={"driver":"GTiff","height":int(nrows),"width":int(ncols),"count":1 if arr.ndim==2 else int(arr.shape[0]),"dtype":"float32","crs":crs,"transform":transform,"compress":"DEFLATE","tiled":True}
    if nodata is not None:profile["nodata"]=float(nodata)
    with rasterio.open(out_tif,"w",**profile)as dst:
        if arr.ndim==2:dst.write(arr,1)
        else:
            for b in range(arr.shape[0]):dst.write(arr[b],b+1)
    return out_tif

def _sanitize_mm(da:xr.DataArray)->xr.DataArray:
    da=da.astype("float32");da=da.where(np.isfinite(da)).where(da>-9990.0)
    da=xr.where(da<0.0,0.0,da);da=xr.where(da>100000.0,np.nan,da);return da

def _open_grib2(path:Path)->xr.DataArray:
    idx=(CFGRIB_DIR/(path.name+".idx")).as_posix();ds=xr.open_dataset(path.as_posix(),engine="cfgrib",backend_kwargs={"indexpath":idx})
    try:
        da=_pick_grib_var(ds)
        if"step"in da.dims:da=da.isel(step=0)
        attrs=dict(da.attrs or{})
        for k in("_FillValue","missingValue","missing_value"):
            if k in attrs:
                try:da=da.where(da!=float(attrs[k]))
                except:pass
        da=da.where(np.isfinite(da)).where(da<1.0e20).astype("float32")
        units=str(attrs.get("units","")).strip().lower()
        if units in("m","meter","metre","m of water equivalent","meters"):da=da*1000.0
        dims_l={d.lower():d for d in da.dims};rename={}
        if"longitude"in dims_l:rename[dims_l["longitude"]]="x"
        if"latitude"in dims_l:rename[dims_l["latitude"]]="y"
        if rename:da=da.rename(rename)
        if"x"in da.coords:
            try:
                x=np.asarray(da["x"].values)
                if x.ndim==1 and x.size>2:
                    if float(np.nanmin(x))>=0.0 and float(np.nanmax(x))>180.0:da=da.assign_coords(x=_lon0360_to_180(x));da=da.sortby("x")
            except:pass
        if"x"in da.dims and"y"in da.dims:
            try:da=da.rio.set_spatial_dims(x_dim="x",y_dim="y",inplace=False)
            except:pass
            try:da=da.rio.write_crs("EPSG:4326",inplace=False)
            except:pass
            try:da=_ensure_xy_transform(da)
            except:pass
        return _sanitize_mm(da)
    finally:
        try:ds.close()
        except:pass

def _open_bil(hdr:Path,dat:Path)->xr.DataArray:
    tmp_tif=dat.with_suffix(".tif")
    if not tmp_tif.exists()or tmp_tif.stat().st_size==0:_bil_to_geotiff(hdr,dat,tmp_tif)
    da=xr.open_dataarray(tmp_tif.as_posix(),engine="rasterio").astype("float32")
    try:
        if da.rio.crs is None:da=da.rio.write_crs("EPSG:4326",inplace=False)
        da=da.rio.set_spatial_dims(x_dim="x",y_dim="y",inplace=False)
    except:pass
    return _sanitize_mm(da)

def _pick_netcdf_raster_var(ds:xr.Dataset)->xr.DataArray:
    bad=("bounds","bnds","crs","grid_mapping","lat_bounds","lon_bounds","time_bounds")
    bad2=("mask","qc","quality","flag","prob","probability","index","idx")
    good=("melt","snowmelt","smelt","we","water","equiv","mm","kg","swe")
    best=None;best_score=-10**9
    for name,da in ds.data_vars.items():
        n=name.lower()
        if any(t in n for t in bad)or any(t in n for t in bad2)or da.ndim<2:continue
        score=300 if any(t in n for t in good)else 0
        score+=200 if da.ndim==2 else 50
        dims=[d.lower()for d in da.dims]
        if("y"in dims and"x"in dims)or("south_north"in dims and"west_east"in dims):score+=200
        if("lat"in dims and"lon"in dims)or("latitude"in dims and"longitude"in dims):score+=150
        try:score+=int((da.shape[-1]*da.shape[-2])//50000)
        except:pass
        if score>best_score:best_score=score;best=da
    if best is None:raise HTTPException(status_code=502,detail=f"No suitable raster variable in NetCDF. vars={list(ds.data_vars)[:50]}")
    return best

def _open_netcdf(path:Path)->xr.DataArray:
    last=None;ds=None
    for eng in("netcdf4","h5netcdf","scipy"):
        try:ds=xr.open_dataset(path.as_posix(),engine=eng);break
        except Exception as e:last=e;ds=None
    if ds is None:raise HTTPException(status_code=502,detail=f"NetCDF backend unavailable/failed. last_err={last!r}")
    try:
        da=_pick_netcdf_raster_var(ds);dims_l={d.lower():d for d in da.dims};rename={}
        if"south_north"in dims_l and"west_east"in dims_l:rename[dims_l["south_north"]]="y";rename[dims_l["west_east"]]="x"
        elif"lat"in dims_l and"lon"in dims_l:rename[dims_l["lat"]]="y";rename[dims_l["lon"]]="x"
        elif"latitude"in dims_l and"longitude"in dims_l:rename[dims_l["latitude"]]="y";rename[dims_l["longitude"]]="x"
        elif"rlat"in dims_l and"rlon"in dims_l:rename[dims_l["rlat"]]="y";rename[dims_l["rlon"]]="x"
        if rename:da=da.rename(rename)
        spatial={"x","y"}
        for d in list(da.dims):
            if d not in spatial:da=da.isel({d:int(da.sizes.get(d,1)or 1)-1})
        da=da.astype("float32");units=str(da.attrs.get("units","")).strip().lower()
        if units in("m","meter","metre","m of water equivalent","meters"):da=da*1000.0
        try:
            lon=da.coords.get("longitude")or da.coords.get("lon");lat=da.coords.get("latitude")or da.coords.get("lat")
            if lon is not None and lat is not None:
                if lon.ndim==2 and lat.ndim==2:
                    h,w=da.shape[-2],da.shape[-1]
                    tfm=from_bounds(float(np.nanmin(lon.values)),float(np.nanmin(lat.values)),float(np.nanmax(lon.values)),float(np.nanmax(lat.values)),w,h)
                    da=da.rio.write_crs("EPSG:4326",inplace=False).rio.write_transform(tfm,inplace=False)
        except:pass
        # Set spatial dims FIRST
        try:
            if"x"in da.dims and"y"in da.dims:da=da.rio.set_spatial_dims(x_dim="x",y_dim="y",inplace=False)
        except:pass
        try:da=_ensure_xy_transform(da)
        except:pass
        # Force CRS — drop any conflicting spatial_ref first
        try:
            if da.rio.crs is None:
                if"spatial_ref"in da.coords:da=da.drop_vars("spatial_ref")
                da=da.rio.write_crs("EPSG:4326",inplace=False)
        except:pass
        try:
            if da.rio.crs is None:
                da=da.rio.write_crs("EPSG:4326",inplace=False)
        except Exception as e:
            _log("WARN",f"[open_netcdf] CRS write failed: {e!r}")
        return _sanitize_mm(da)
    finally:
        try:ds.close()
        except:pass

def _open_grid(desc:dict)->xr.DataArray:
    t=desc.get("type")
    if t=="grib2":return _open_grib2(desc["path"])
    if t=="bil":return _open_bil(desc["hdr"],desc["dat"])
    if t=="netcdf":return _open_netcdf(desc["path"])
    raise HTTPException(status_code=500,detail="unknown grid type")

def _is_probably_html(blob:bytes)->bool:head=blob[:512].lstrip();return head.startswith(b"<!DOCTYPE")or head.startswith(b"<html")or head.startswith(b"<HTML")


def _hf_try_pull_file(rel: str, timeout_sec: int = 20, cancel_check=None) -> bool:
    if cancel_check and cancel_check():
        return False

    tok, repo = _hf_cfg()
    if not tok or not repo:
        return False
    rel = (rel or "").lstrip("/")
    if not rel:
        return False
    dest = (CACHE / rel).resolve()
    dest.parent.mkdir(parents=True, exist_ok=True)
    result = {"tmp": None, "error": None}

    def do_download():
        try:
            result["tmp"] = hf_hub_download(
                repo_id=repo, repo_type="dataset", filename=rel,
                token=tok, local_dir=str(CACHE)
            )
        except Exception as e:
            result["error"] = e

    t = threading.Thread(target=do_download, daemon=True)
    t.start()
    poll_interval = 0.5
    elapsed = 0.0
    while t.is_alive() and elapsed < float(timeout_sec):
        if cancel_check and cancel_check():
            _log("DEBUG", f"[hf_pull] cancelled (gen changed) for {rel}")
            return False
        t.join(timeout=poll_interval)
        elapsed += poll_interval

    if t.is_alive():
        _log("DEBUG", f"[hf_pull] timeout after {timeout_sec}s for {rel}")
        return False
    if cancel_check and cancel_check():
        return False

    if result["error"] is not None:
        e = result["error"]
        err_str = str(e).lower()
        if "404" in err_str or "not found" in err_str:
            return False
        if "timeout" not in err_str and "connection" not in err_str:
            _log("WARN", f"[hf_pull] error for {rel}: {e!r}")
        return False

    try:
        tmp = result["tmp"]
        if tmp is None:
            return False
        p = Path(tmp)
        if not p.exists():
            return False
        b = p.read_bytes()
        if _is_lfs_pointer_bytes(b) or len(b) < 200:
            try:
                p.unlink(missing_ok=True)
                dest.unlink(missing_ok=True)
            except:
                pass
            return False
        if p.resolve() != dest:
            try:
                dest.write_bytes(b)
            except:
                pass
        return True
    except Exception as e:
        _log("WARN", f"[hf_pull] validation failed for {rel}: {e!r}")
        return False

def _is_netcdf_bytes(blob:bytes)->bool:
    if len(blob)>=4 and blob[:3]==b"CDF"and blob[3]in(1,2):return True
    return len(blob)>=8 and blob[:8]==b"\x89HDF\r\n\x1a\n"

_HF_OPS_LOCK=threading.Lock()
_HF_PENDING:dict[str,CommitOperationAdd]={}

def _hf_enqueue_files(paths:list[Path])->int:
    tok,repo=_hf_cfg()
    if not tok or not repo:return 0
    added=0
    with _HF_OPS_LOCK:
        for p in paths:
            try:
                if(not p.exists())or p.stat().st_size<=0:continue
                rel=p.relative_to(CACHE).as_posix()
                if not _is_cache_file_we_manage(rel):continue
                _HF_PENDING[rel]=CommitOperationAdd(path_in_repo=rel,path_or_fileobj=p.as_posix());added+=1
            except:continue
    return added

def _hf_flush_pending(message:str)->None:
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return
    with _HF_OPS_LOCK:ops=list(_HF_PENDING.values());_HF_PENDING.clear()
    if not ops:return
    for attempt in range(6):
        try:
            api.create_commit(repo_id=repo,repo_type="dataset",operations=ops,commit_message=message)
            _log("INFO",f"[hf_cache] flush OK files={len(ops)}");return
        except Exception as e:
            wait=(1.8**attempt)+random.uniform(0,1.0);_log("INFO",f"[hf_cache] flush retry attempt={attempt+1} wait={wait:.1f} err={e!r}");time.sleep(wait)
    _log("INFO",f"[hf_cache] flush FAILED files={len(ops)}")

def _download_url_to_grid(url:str,workdir:Path)->tuple[dict,str]:
    hdrs={"User-Agent":"snodas/1.3"};name=Path(url).name.lower();last_err=None
    for attempt in range(6):
        r=_retry_get(url,timeout=(10,360),tries=1,headers=hdrs);sc=int(r.status_code)
        if sc in(429,502,503,504):
            last_err=f"HTTP {sc}"
            if attempt<5:
                ra=r.headers.get("Retry-After")
                try:wait=float(ra)if ra else None
                except:wait=None
                time.sleep((wait if wait is not None else(1.6**attempt))+random.uniform(0,0.8));continue
            raise HTTPException(status_code=502,detail=f"{url} -> {last_err} after retries")
        if sc not in(200,206):raise HTTPException(status_code=404,detail=f"{url} -> HTTP {sc}")
        blob=r.content;ctype=(r.headers.get("Content-Type")or"").lower()
        if _is_probably_html(blob)or"text/html"in ctype:
            raise HTTPException(status_code=502,detail=f"Upstream returned HTML. url={url}")
        if len(blob)>=2 and blob[:2]==b"\x1f\x8b":
            try:blob=gzip.decompress(blob)
            except Exception as e:raise HTTPException(status_code=502,detail=f"gzip decompress failed: {e!r}")
        if blob.startswith(b"GRIB"):p=workdir/"data.grib2";p.write_bytes(blob);return{"type":"grib2","path":p},name
        if _is_netcdf_bytes(blob):p=workdir/"data.nc";p.write_bytes(blob);return{"type":"netcdf","path":p},name
        looks_tar=(len(blob)%512==0)or(b"ustar"in blob[:4096])
        if looks_tar:
            try:
                with tarfile.open(fileobj=io.BytesIO(blob),mode="r:*")as tf:
                    grib_member=hdr_member=dat_member=nc_member=None
                    for m in tf.getmembers():
                        n=m.name.lower()
                        if n.endswith(".grib2")and grib_member is None:grib_member=m
                        elif n.endswith(".nc")and nc_member is None:nc_member=m
                        elif n.endswith(".hdr")and hdr_member is None:hdr_member=m
                        elif(n.endswith(".dat")or n.endswith(".bil"))and dat_member is None:dat_member=m
                    if grib_member is not None:
                        p=workdir/"data.grib2";f=tf.extractfile(grib_member)
                        if f is None:raise HTTPException(status_code=502,detail="TAR member unreadable")
                        p.write_bytes(f.read(grib_member.size));return{"type":"grib2","path":p},name
                    if nc_member is not None:
                        p=workdir/"data.nc";f=tf.extractfile(nc_member)
                        if f is None:raise HTTPException(status_code=502,detail="TAR member unreadable")
                        p.write_bytes(f.read(nc_member.size));return{"type":"netcdf","path":p},name
                    if hdr_member is not None and dat_member is not None:
                        hdr=workdir/"data.hdr";dat=workdir/"data.bil"
                        fh=tf.extractfile(hdr_member);fd=tf.extractfile(dat_member)
                        if fh is None or fd is None:raise HTTPException(status_code=502,detail="TAR member unreadable")
                        hdr.write_bytes(fh.read(hdr_member.size));dat.write_bytes(fd.read(dat_member.size))
                        return{"type":"bil","hdr":hdr,"dat":dat},name
            except HTTPException:raise
            except Exception as e:raise HTTPException(status_code=502,detail=f"TAR parse failed: {e!r}")
        raise HTTPException(status_code=502,detail=f"payload unrecognized. url={url}")
    raise HTTPException(status_code=504,detail=f"upstream retries exhausted. url={url} last_err={last_err}")

def _da_to_cog(da:xr.DataArray,out:Path)->Path:
    out.parent.mkdir(parents=True,exist_ok=True)
    try:_=da.rio.transform();ok_transform=True
    except:ok_transform=False
    if not ok_transform and"x"in da.coords and"y"in da.coords:
        try:da=_ensure_xy_transform(da)
        except:pass
    nodata=np.float32(-9999.0)
    try:crs=da.rio.crs
    except:crs=None
    if crs is None:
        try:
            if"spatial_ref"in da.coords:da=da.drop_vars("spatial_ref")
            da=da.rio.write_crs("EPSG:4326",inplace=False)
        except:pass
    try:crs=da.rio.crs
    except:crs=None
    if crs is None:da=da.rio.write_crs("EPSG:4326",inplace=False)
    da=da.astype("float32").where(np.isfinite(da)).where(da>-9990.0);da=da.where(~da.isnull(),nodata)
    with tempfile.NamedTemporaryFile(suffix=".tif",dir=str(out.parent),delete=False)as tf:tmp=Path(tf.name)
    try:
        da.rio.to_raster(tmp.as_posix(),dtype="float32",compress="DEFLATE",nodata=float(nodata),tiled=True)
        # Force CRS into the written file if rio.to_raster didn't set it
        with rasterio.open(tmp.as_posix(),"r")as check:
            if check.crs is None:
                _log("INFO",f"[da_to_cog] patching missing CRS for {out.name}")
                data=check.read();prof=check.profile.copy()
                prof["crs"]=CRS.from_epsg(4326)
                tmp2=tmp.with_suffix(".fix.tif")
                with rasterio.open(tmp2.as_posix(),"w",**prof)as ds2:
                    ds2.write(data)
                tmp2.replace(tmp)
        _atomic_cog_translate(tmp,out)
    finally:tmp.unlink(missing_ok=True)
    return out

def _extract_var(tar_path:Path,code:str)->tuple[Path,Path]:
    outdir=CACHE/f"{tar_path.stem}_x";outdir.mkdir(exist_ok=True)
    hdr_path:Optional[Path]=None;dat_path:Optional[Path]=None
    with tarfile.open(tar_path)as tf:
        for m in tf.getmembers():
            name=Path(m.name).name;low=name.lower()
            if code not in low:continue
            if low.endswith(".hdr"):
                outp=outdir/name
                if not outp.exists():
                    f=tf.extractfile(m)
                    if f is not None:outp.write_bytes(f.read())
                hdr_path=outp
            elif low.endswith(".dat")or low.endswith(".bil"):
                outp=outdir/name
                if not outp.exists():
                    f=tf.extractfile(m)
                    if f is not None:outp.write_bytes(f.read())
                dat_path=outp
    if not hdr_path or not dat_path:raise HTTPException(status_code=500,detail=f"Missing .hdr or .dat for var {code}")
    return hdr_path,dat_path

def build_24h_cog(date)->Path:
    dstr=_yyyymmdd(date);out=CACHE/f"melt24h_{dstr}_cog.tif"
    if out.exists()and out.stat().st_size>0:return out
    _hf_try_pull_file(out.relative_to(CACHE).as_posix())
    if out.exists()and out.stat().st_size>0:return out
    with _lock(f"melt24h_{dstr}"):
        if out.exists()and out.stat().st_size>0:return out
        _hf_try_pull_file(out.relative_to(CACHE).as_posix())
        if out.exists()and out.stat().st_size>0:return out
        tar=_download_day_tar(date);hdr,dat=_extract_var(tar,"1044");da=_open_bil(hdr,dat);p=_da_to_cog(da,out)
        try:_hf_enqueue_files([p])
        except:pass
        return p

def build_72h_cog(end_date)->Path:
    dstr=_yyyymmdd(end_date);out=CACHE/f"melt72h_end_{dstr}_cog.tif"
    if out.exists()and out.stat().st_size>0:return out
    _hf_try_pull_file(out.relative_to(CACHE).as_posix())
    if out.exists()and out.stat().st_size>0:return out
    with _lock(f"melt72h_{dstr}"):
        if out.exists()and out.stat().st_size>0:return out
        _hf_try_pull_file(out.relative_to(CACHE).as_posix())
        if out.exists()and out.stat().st_size>0:return out
        cogs=[build_24h_cog(end_date-timedelta(days=2)),build_24h_cog(end_date-timedelta(days=1)),build_24h_cog(end_date)]
        with rasterio.open(cogs[0])as base:
            acc=base.read(1).astype("float32")
            for c in cogs[1:]:
                with rasterio.open(c)as src:
                    arr=src.read(1).astype("float32")
                    if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height):
                        tmp=np.full_like(acc,np.nan,dtype="float32")
                        reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest)
                        arr=tmp
                    m=np.isfinite(arr);acc[m]=np.nan_to_num(acc[m],nan=0.0)+arr[m]
            tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True})
            with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1)
        _atomic_cog_translate(tmp_tif,out);tmp_tif.unlink(missing_ok=True)
        try:_hf_enqueue_files([out])
        except:pass
        return out

def _forecast_dir()->Path:d=CACHE/"forecast";d.mkdir(parents=True,exist_ok=True);return d
def _forecast_melt_cog(run_init:str,valid:str,lead:str,dom:str)->Path:
    ri=run_init if run_init!="TTNA"else"TTNA"
    return _forecast_dir()/f"fcst_melt_{lead}_{ri}_{valid}_{dom}_cog.tif"
def _forecast_snow_cog(ts:str,dom:str,code:str)->Path:return _forecast_dir()/f"fcst_snow_{code}_{ts}_{dom}_cog.tif"

def _forecast_melt72h_end_cog(valid:str,dom_prefer:str="zz")->Path:
    valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower()
    out=_forecast_dir()/f"fcst_melt72h_end_{valid}_{dom_prefer}_cog.tif"
    if out.exists()and out.stat().st_size>0:return out
    _hf_try_pull_file(out.relative_to(CACHE).as_posix())
    if out.exists()and out.stat().st_size>0:return out
    with _lock(f"fcst_melt72h_{valid}_{dom_prefer}"):
        if out.exists()and out.stat().st_size>0:return out
        v0=valid;v1=_shift_ts(v0,-24);v2=_shift_ts(v0,-48)
        sels=[_pick_forecast_for_valid(v,dom_prefer=dom_prefer)for v in(v2,v1,v0)]
        if not all(s.get("ok")for s in sels):raise HTTPException(status_code=503,detail={"error":"forecast_selection_unavailable_for_72h","need":[v2,v1,v0],"sels":sels})
        day_cogs:list[Path]=[]
        for s,v in zip(sels,(v2,v1,v0)):
            melt_cogs=_build_forecast_melt_cogs(s["run_init"],v,"t0024",s["melt_urls"])
            pick=None
            for p in melt_cogs:
                if _cog_dom(p)==dom_prefer:pick=p;break
            if pick is None:pick=sorted(melt_cogs,key=lambda p:_dom_pri(_cog_dom(p)))[0]
            day_cogs.append(pick)
        with rasterio.open(day_cogs[0])as base:
            acc=base.read(1).astype("float32")
            for c in day_cogs[1:]:
                with rasterio.open(c)as src:
                    arr=src.read(1).astype("float32")
                    if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height):
                        tmp=np.full_like(acc,np.nan,dtype="float32")
                        reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest)
                        arr=tmp
                    m=np.isfinite(arr)&(arr>-9990.0);acc[m]=np.nan_to_num(acc[m],nan=0.0)+arr[m]
            tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True})
            with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1)
        _atomic_cog_translate(tmp_tif,out)
        try:_hf_enqueue_files([out])
        except:pass
        tmp_tif.unlink(missing_ok=True);_log("INFO",f"[build_forecast] melt72h OK valid={valid} dom_prefer={dom_prefer} -> {out.name}")
        return out

def _pick_grib_var(ds:xr.Dataset)->xr.DataArray:
    best=None;best_score=-10**9
    for name,da in ds.data_vars.items():
        a=dict(da.attrs or{});units=str(a.get("units","")).lower()
        txt=" ".join([name.lower(),str(a.get("GRIB_name","")).lower(),str(a.get("standard_name","")).lower(),str(a.get("long_name","")).lower()])
        score=0
        if"snow"in txt or"swe"in txt or"water equivalent"in txt:score+=500
        if"depth"in txt:score+=250
        if units in("m","meter","metre","mm"):score+=120
        if units in("k","kelvin","degc","c","degree_celsius"):score-=400
        if da.ndim>=2:score+=50
        if score>best_score:best_score=score;best=da
    if best is None:return ds[next(iter(ds.data_vars))]
    return best

def _ensure_xy_transform(da:xr.DataArray)->xr.DataArray:
    if not("x"in da.coords and"y"in da.coords):return da
    try:
        x=np.asarray(da["x"].values);y=np.asarray(da["y"].values)
        if x.ndim!=1 or y.ndim!=1 or x.size<2 or y.size<2:return da
        dx=float(np.nanmedian(np.diff(x)));dy=float(np.nanmedian(np.diff(y)))
        if not np.isfinite(dx)or not np.isfinite(dy)or dx==0.0 or dy==0.0:return da
        left=float(x[0])-dx/2.0;top=float(y[0])+(-dy)/2.0 if dy<0 else float(y[0])-dy/2.0
        return da.rio.write_transform(Affine(dx,0.0,left,0.0,dy,top),inplace=False)
    except:return da

def _build_forecast_melt_cogs(run_init:str,valid:str,lead:str,melt_urls:list[str])->list[Path]:
    run_init_safe=run_init if run_init!="TTNA"else"TTNA"
    run_init_norm=_norm_ts(run_init)if run_init!="TTNA"else"TTNA"
    valid=_norm_ts(valid);lead=(lead or"").lower()
    if not melt_urls:raise HTTPException(status_code=502,detail=f"No melt URLs provided for run_init={run_init} valid={valid} lead={lead}")
    dom2urls:dict[str,list[str]]={}
    for u in melt_urls:dom2urls.setdefault(_dom_from_url(u),[]).append(u)
    built:list[Path]=[];errors:list[str]=[]
    with _lock(f"melt_{lead}_{run_init_safe}_{valid}"):
        for dom in sorted(dom2urls.keys(),key=_dom_pri):
            out=_forecast_melt_cog(run_init_safe,valid,lead,dom)
            if out.exists()and out.stat().st_size>0:built.append(out);continue
            last_err=None
            for url in dom2urls[dom]:
                low=Path(url).name.lower()
                if lead and lead not in low:continue
                with tempfile.TemporaryDirectory()as td:
                    try:
                        desc,_=_download_url_to_grid(url,Path(td));da=_open_grid(desc);_da_to_cog(da,out)
                        try:_hf_enqueue_files([out])
                        except:pass
                        if out.exists()and out.stat().st_size>0:
                            _log("INFO",f"[build_forecast] melt OK run_init={run_init_safe} valid={valid} lead={lead} dom={dom} -> {out.name}")
                            built.append(out);break
                        last_err="empty COG after build"
                    except HTTPException as e:last_err=f"{e.status_code} {e.detail}"
                    except Exception as e:last_err=repr(e)
            if out not in built:errors.append(f"dom={dom} last_err={last_err}");_log("WARN",f"[build_forecast] melt FAIL run_init={run_init_safe} valid={valid} lead={lead} dom={dom}")
    if not built:raise HTTPException(status_code=502,detail={"error":"No usable MELT COGs","run_init":run_init,"valid":valid,"lead":lead,"errors":errors[:5]})
    return built

def _build_forecast_snowpack_cogs(ts:str,urls:list[str])->list[Path]:
    ts=_norm_ts(ts)if ts and _TS10.fullmatch(ts)else ts
    dom2urls:dict[str,list[str]]={}
    for u in urls:dom2urls.setdefault(_dom_from_url(u),[]).append(u)
    built:list[Path]=[]
    with _lock(f"snow_{ts}"):
        for dom in sorted(dom2urls.keys(),key=_dom_pri):
            last_err=None
            for url in dom2urls[dom]:
                low=Path(url).name.lower()
                code="ssmv11036"if"ssmv11036"in low else"ssmv11034"if"ssmv11034"in low else"ssmv11005"if"ssmv11005"in low else"ssmv_unknown"
                out=_forecast_snow_cog(ts or"na",dom,code)
                if out.exists()and out.stat().st_size>0:built.append(out);break
                with tempfile.TemporaryDirectory()as td:
                    try:
                        desc,_=_download_url_to_grid(url,Path(td));da=_open_grid(desc);_da_to_cog(da,out)
                        try:_hf_enqueue_files([out])
                        except:pass
                        if out.exists()and out.stat().st_size>0:
                            _log("INFO",f"[build_forecast] snow OK ts={ts} dom={dom} code={code} -> {out.name}")
                            built.append(out);break
                        last_err="empty COG"
                    except HTTPException as e:last_err=f"{e.status_code} {e.detail}"
                    except Exception as e:last_err=repr(e)
            if not any(f"_{dom}_"in p.name for p in built):_log("INFO",f"[build_forecast] snow FAIL ts={ts} dom={dom} last_err={last_err}")
    if not built:raise HTTPException(status_code=502,detail=f"No usable snow field for {ts}")
    return built

def _cog_dom(p:Path)->str:parts=p.name.split("_");return parts[-2]if len(parts)>=2 else"xx"

def _union_tile(cogs:list[Path],z:int,x:int,y:int,*,label:str,valid_pred:Optional[Callable[[np.ndarray,np.ndarray],np.ndarray]]=None)->tuple[Optional[np.ndarray],Optional[np.ndarray],str]:
    if not cogs:return None,None,f"{label}_no_cogs"
    out_arr:Optional[np.ndarray]=None;out_valid:Optional[np.ndarray]=None;used:list[str]=[]
    for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))):
        try:
            with COGReader(cog.as_posix())as r:data,mask=r.tile(x,y,z,tilesize=256,resampling_method="nearest")
            arr=data[0].astype("float32");m=mask[0];valid=(m>0)&np.isfinite(arr)&(arr>-9990.0)
            if valid_pred is not None:
                try:valid=valid&valid_pred(arr,m)
                except:pass
            if out_arr is None:out_arr=arr.copy();out_valid=valid.copy()
            else:
                fill=valid&(~out_valid)
                if fill.any():out_arr[fill]=arr[fill];out_valid[fill]=True
            used.append(cog.name)
        except TileOutsideBounds:continue
        except:continue
    if out_arr is None or out_valid is None:return None,None,f"{label}_unavailable_all_domains"
    return out_arr,out_valid,f"{label}_union:"+",".join(used[:4])+("..."if len(used)>4 else"")

def _read_mask_tile(ts:str,urls:list[str],z:int,x:int,y:int)->tuple[Optional[np.ndarray],Optional[np.ndarray],str]:
    ts=_norm_ts(ts)if ts else ts
    if not urls:return None,None,"mask_no_urls"
    try:cogs=_build_forecast_snowpack_cogs(ts or"0000000000",urls)
    except HTTPException as e:return None,None,f"mask_build_fail:{e.status_code}:{e.detail}"
    except Exception as e:return None,None,f"mask_build_fail:{e!r}"
    return _union_tile(cogs,z,x,y,label="mask")

def _cog_point_data_mask(r:COGReader,lon:float,lat:float)->tuple[np.ndarray,np.ndarray]:
    out=r.point(lon,lat)
    if isinstance(out,(tuple,list))and len(out)==2:return np.asarray(out[0]),np.asarray(out[1])
    data=getattr(out,"data",None);mask=getattr(out,"mask",None)
    if data is None or mask is None:raise ValueError(f"Unexpected COGReader.point() return type: {type(out)!r}")
    return np.asarray(data),np.asarray(mask)

INCH_TO_MM=25.4
BIN_EDGES_IN=np.array([0.01,0.02,0.04,0.08,0.12,0.16,0.20,0.39,0.59,0.79,0.98,1.4,2.0,3.0,4.0,5.0,9999.0],dtype="float32")
BIN_EDGES_72H_IN=np.array([0.01,0.04,0.08,0.16,0.24,0.31,0.39,0.79,1.4,2.0,3.0,3.9,5.0,6.5,8.0,9.8,9999.0],dtype="float32")
MELT_COLORS=np.array([[0xDC,0xDC,0xDC],[0x99,0x00,0x00],[0xCC,0x33,0x33],[0xFF,0x66,0x66],[0xCC,0xCC,0x66],[0xFF,0x8C,0x00],[0xFF,0xFF,0x66],[0x99,0xFF,0x99],[0x00,0xFF,0x00],[0x66,0xCC,0x66],[0x66,0x66,0xFF],[0x33,0x33,0xCC],[0x00,0x00,0x99],[0x99,0x00,0x99],[0x4B,0x00,0x82],[0x2F,0x00,0x4F],[0x00,0x00,0x00]],dtype="uint8")

@lru_cache(maxsize=32)
def _corridor_tiles_set_for_z(z:int)->set[tuple[int,int]]:
    pts=_load_track_points()
    if not pts:return set()
    tiles=_tiles_within_miles(pts,int(z),WARM_MILES,cap=WARM_CAP_PER_ZOOM)
    return set((x,y)for(_,x,y)in tiles)

def _is_corridor_tile(z:int,x:int,y:int)->bool:return(int(x),int(y))in _corridor_tiles_set_for_z(int(z))

def _effective_request_tile(z:int,x:int,y:int)->tuple[int,int,int,str]:
    z=int(z);x=int(x);y=int(y)
    if not _tile_allowed(z,x,y):return z,x,y,"none"
    if _is_corridor_tile(z,x,y):return z,x,y,"hi"
    z_lr=min(int(LOWRES_Z),z)
    if z<=z_lr:return z,x,y,"lo"
    shift=z-z_lr;return z_lr,(x>>shift),(y>>shift),"lo"

def _miles_to_meters(mi:float)->float:return float(mi)*1609.344

def _meters_per_pixel_webmercator(z:int,lat:float)->float:
    lat=max(-85.05112878,min(85.05112878,float(lat)))
    return 156543.03392*max(0.0,float(np.cos(np.deg2rad(lat))))/(2**int(z))

def _tile_radius_for_miles(z:int,lat:float,miles:float)->int:
    mpp=_meters_per_pixel_webmercator(int(z),float(lat))
    if not np.isfinite(mpp)or mpp<=0:return 0
    return max(0,int(np.ceil(_miles_to_meters(float(miles))/(mpp*256.0))))

def _tiles_within_miles(points:list[tuple[float,float]],z:int,miles:float,cap:int=8000)->list[tuple[int,int,int]]:
    if not points:return[]
    z=int(z);n=2**z;seen:set[tuple[int,int]]=set();out:list[tuple[int,int,int]]=[];miles=float(miles)
    for lon,lat in points:
        tx,ty=_lonlat_to_tile(lon,lat,z);r=_tile_radius_for_miles(z,lat,miles)
        for dx in range(-r,r+1):
            x=tx+dx
            if x<0 or x>=n:continue
            for dy in range(-r,r+1):
                y=ty+dy
                if y<0 or y>=n:continue
                k=(x,y)
                if k in seen:continue
                seen.add(k);out.append((z,x,y))
                if len(out)>=int(cap):return out
    return out

def _generate_forecast_by_date_raw(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str)->tuple[bytes,dict]:
    sem=CPU_SEM_VALUE if _mode_get()=="value"else CPU_SEM_TILES
    with sem:return _generate_forecast_by_date_raw_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom)

def _generate_forecast_by_date_raw_impl(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,cog_cache:CogCache|None=None)->tuple[bytes,dict]:
    dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05")
    if int(hours)==24:
        sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
        if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="forecast data may not be available yet for this date";raise HTTPException(status_code=503,detail=detail)
        melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom)
        melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt24_union,z,x,y,cog_cache=cog_cache)
        if oob or melt_mm is None or melt_mask is None:
            raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8"))
            return raw,{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":"snowpack_unavailable","X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24","X-Raw-Scale":"10"}
        melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255)
        snow_info="snowpack_unavailable"
        if sel.get("snowpack_urls"):
            try:
                snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom)
                if snow_union:snow_info=f"snowpack_union:{snow_union.name}"
            except Exception as e:snow_info=f"snowpack_unavailable:{e!r}"
        raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=melt_mask_u8)
        return raw,{"X-Allowed":"1","X-OOB":"0","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24","X-Raw-Scale":"10"}
    run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom)
    if not run_init:raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom,"suggestion":"forecast data may not be available yet for this date"})
    melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom)
    melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt72_cog,z,x,y,cog_cache=cog_cache)
    if oob or melt_mm is None or melt_mask is None:
        raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8"))
        return raw,{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":run_init,"X-Forecast-72h-COG":melt72_cog.name,"X-Raw-Scale":"10"}
    raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=melt_mask.astype("uint8",copy=False))
    return raw,{"X-Allowed":"1","X-OOB":"0","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":run_init,"X-Forecast-72h-COG":melt72_cog.name,"X-Raw-Scale":"10"}

_BAKING_MODE=threading.local()
def _is_baking_mode()->bool:return getattr(_BAKING_MODE,"active",False)

class _BakingMode:
    def __init__(self):self.prev=None
    def __enter__(self):self.prev=getattr(_BAKING_MODE,"active",False);_BAKING_MODE.active=True;return self
    def __exit__(self,*args):_BAKING_MODE.active=self.prev if self.prev is not None else False

def _bake_lowres_box_tiles_for_date(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,conc:int=2,push_every:int=300)->dict:
    dom=(dom or"zz").lower();_log("DEBUG",f"[bake_lowres] Starting for {date_yyyymmdd}")
    boxes=[(-170.0,49.0,-40.0,90.0),(-97.0,37.0,-63.0,49.0)]
    tiles=_tiles_for_boxes_zoom_range(boxes,zmin=int(zmin),zmax=int(zmax))
    if not tiles:return{"ok":False,"error":"no_lowres_tiles"}
    q=deque()
    for hours in hours_list:
        for(z,x,y)in tiles:q.append((int(hours),int(z),int(x),int(y)))
    total_jobs=len(q);ok=0;skip=0;fail=0;lk=threading.Lock();pending_paths_count=0
    def worker():
        nonlocal ok,skip,fail,pending_paths_count
        with _HfPullOnMiss(False):
            while True:
                with lk:
                    if not q:return
                    hours,z,x,y=q.popleft();processed=total_jobs-len(q)
                try:
                    p=_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y)
                    if _is_valid_png_file(p):
                        with lk:skip+=1
                        continue
                    png,_=_generate_forecast_by_date_png_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,max_in=None)
                    _tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,png)
                    with lk:ok+=1;pending_paths_count+=1
                    if processed%200==0:_log("DEBUG",f"[bake_lowres] {date_yyyymmdd}: {processed}/{total_jobs}")
                except Exception as e:
                    with lk:
                        if fail<5:_log("WARN",f"[bake_lowres] tile failed z={z} x={x} y={y} err={e!r}")
                        fail+=1
    threads=[threading.Thread(target=worker,daemon=True)for _ in range(int(conc))]
    for t in threads:t.start()
    for t in threads:t.join()
    _log("INFO",f"[bake_lowres] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, local_written={pending_paths_count}")
    return{"ok":True,"date":date_yyyymmdd,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":0,"remote_push":"deferred"}

def _bake_lowres_box_raw_tiles_for_date(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,conc:int=2,push_every:int=300)->dict:
    dom=(dom or"zz").lower();_log("DEBUG",f"[bake_lowres_raw] Starting for {date_yyyymmdd}")
    boxes=[(-170.0,49.0,-40.0,90.0),(-97.0,37.0,-63.0,49.0)]
    tiles=_tiles_for_boxes_zoom_range(boxes,zmin=int(zmin),zmax=int(zmax))
    if not tiles:return{"ok":False,"error":"no_lowres_tiles"}
    q=deque()
    for hours in hours_list:
        for(z,x,y)in tiles:q.append((int(hours),int(z),int(x),int(y)))
    total_jobs=len(q);ok=0;skip=0;fail=0;lk=threading.Lock();pending_paths_count=0
    def worker():
        nonlocal ok,skip,fail,pending_paths_count
        with _HfPullOnMiss(False):
            while True:
                with lk:
                    if not q:return
                    hours,z,x,y=q.popleft();processed=total_jobs-len(q)
                try:
                    p=_raw_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y)
                    if _is_valid_npz_file(p):
                        with lk:skip+=1
                        continue
                    raw,_=_generate_forecast_by_date_raw_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom)
                    _raw_tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,raw)
                    with lk:ok+=1;pending_paths_count+=1
                    if processed%200==0:_log("DEBUG",f"[bake_lowres_raw] {date_yyyymmdd}: {processed}/{total_jobs}")
                except Exception as e:
                    with lk:
                        if fail<5:_log("WARN",f"[bake_lowres_raw] tile failed z={z} x={x} y={y} err={e!r}")
                        fail+=1
    threads=[threading.Thread(target=worker,daemon=True)for _ in range(int(conc))]
    for t in threads:t.start()
    for t in threads:t.join()
    _log("INFO",f"[bake_lowres_raw] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, local_written={pending_paths_count}")
    return{"ok":True,"date":date_yyyymmdd,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":0,"remote_push":"deferred"}

def _hf_push_rawtilecache_for_window(window:list[str],dom:str,hours_list:list[int],zmin:int|None=None,zmax:int|None=None)->dict:
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"}
    dom=(dom or"zz").lower();root=_raw_tile_cache_dir()
    if not root.exists():return{"ok":True,"pushed":0,"skipped":0}
    want_dates=set(window);want_hours=set(int(h)for h in hours_list)
    ops:list[CommitOperationAdd]=[];pushed=0;skipped=0
    for p in root.rglob(f"bydate_*{RAW_TILE_EXT}"):
        name=p.name;m=re.match(rf"^bydate_([a-z0-9]+)_h(\d+)_(\d{{8}})_z\d+_\d+_\d+\{re.escape(RAW_TILE_EXT)}$",name)
        if not m:continue
        f_dom,f_hours,f_ymd=m.group(1),int(m.group(2)),m.group(3)
        if f_dom!=dom or f_hours not in want_hours or f_ymd not in want_dates:continue
        mz=_TILE_Z_RE.search(name);z=int(mz.group(1))if mz else None
        if z is not None and zmin is not None and z<int(zmin):continue
        if z is not None and zmax is not None and z>int(zmax):continue
        if not _is_valid_npz_file(p):skipped+=1;continue
        ops.append(CommitOperationAdd(path_in_repo=f"rawtilecache/{name}",path_or_fileobj=p.as_posix()))
        if len(ops)>=int(HF_MAX_OPS_PER_COMMIT):
            api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"rawtilecache: update dom={dom} files={len(ops)}")
            pushed+=len(ops);ops=[]
            if HF_PUSH_SLEEP_SEC>0:time.sleep(float(HF_PUSH_SLEEP_SEC))
    if ops:
        api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"rawtilecache: update dom={dom} files={len(ops)}")
        pushed+=len(ops)
    return{"ok":True,"pushed":pushed,"skipped":skipped,"dom":dom}

def _png(rgb:np.ndarray,a:np.ndarray)->bytes:return render(rgb,mask=a.astype("uint8"),img_format="PNG")

def _dilate_bool_1px(m:np.ndarray)->np.ndarray:
    m=m.astype(bool,copy=False)
    if m.size==0:return m
    p=np.pad(m,1,mode="constant",constant_values=False);out=p[1:-1,1:-1].copy()
    out|=p[:-2,1:-1];out|=p[2:,1:-1];out|=p[1:-1,:-2];out|=p[1:-1,2:]
    out|=p[:-2,:-2];out|=p[:-2,2:];out|=p[2:,:-2];out|=p[2:,2:]
    return out

def _melt_to_png(melt_mm:np.ndarray,melt_mask:np.ndarray,max_in:Optional[float],snow_mm:Optional[np.ndarray],snow_valid:Optional[np.ndarray],*,snow_allow_min_mm:float=0.0,snow_underlay_min_mm:float=0.01,dilate_px:int=1,bin_edges_in:Optional[np.ndarray]=None)->bytes:
    mm=melt_mm.astype("float32");melt_valid=(melt_mask>0)&np.isfinite(mm)&(mm>-9990.0)
    mm=np.where(melt_valid,np.maximum(mm,0.0),np.nan).astype("float32")
    snow_underlay=np.zeros_like(melt_valid,dtype=bool);snow_allow=None
    if snow_mm is not None and snow_valid is not None:
        snow=snow_mm.astype("float32");snow_ok=snow_valid&np.isfinite(snow)&(snow>-9990.0)
        snow_underlay=snow_ok&(snow>=float(snow_underlay_min_mm));snow_allow=snow_ok&(snow>=float(snow_allow_min_mm))
        if dilate_px and dilate_px>0:
            for _ in range(int(dilate_px)):snow_allow=_dilate_bool_1px(snow_allow)
        melt_valid=melt_valid&snow_allow
    inches=mm/INCH_TO_MM
    if max_in is not None:inches=np.minimum(inches,float(max_in))
    melt_vis=np.isfinite(inches)&(inches>0.0)&melt_valid;snow_only=snow_underlay&(~melt_vis)
    h,w=mm.shape;rgb=np.zeros((3,h,w),dtype="uint8");alpha=np.zeros((h,w),dtype="uint8")
    if snow_only.any():rgb[:,snow_only]=np.array(MELT_COLORS[0],dtype="uint8").reshape(3,1);alpha[snow_only]=255
    if melt_vis.any():
        filled=np.where(melt_vis,inches,0.0).astype("float32")
        edges=bin_edges_in if bin_edges_in is not None else BIN_EDGES_IN
        idx=np.clip(np.digitize(filled,edges,right=True),0,len(MELT_COLORS)-1)
        rgb[:,melt_vis]=np.moveaxis(MELT_COLORS[idx],-1,0)[:,melt_vis];alpha[melt_vis]=255
    return _png(rgb,alpha)

CACHE_CONTROL_TILES=os.environ.get("TILES_CACHE_CONTROL","public, max-age=604800, immutable")

def _resp_png(data:bytes,*,cache_control:str="public, max-age=31536000, immutable",**extra_headers)->Response:
    etag=hashlib.md5(data).hexdigest()[:16]
    headers={"Cache-Control":cache_control,"ETag":f'"{etag}"',"Content-Type":"image/png","X-Content-Type-Options":"nosniff"}
    for k,v in extra_headers.items():
        if v is not None:headers[k]=str(v)
    return Response(content=data,media_type="image/png",headers=headers)

_COG_READ_LOCKS: dict[str, threading.Lock] = {}
_COG_READ_LOCKS_LK = threading.Lock()

def _cog_read_lock(cog: Path) -> threading.Lock:
    key = cog.as_posix()
    with _COG_READ_LOCKS_LK:
        lk = _COG_READ_LOCKS.get(key)
        if lk is None:
            lk = threading.Lock(); _COG_READ_LOCKS[key] = lk
        return lk

def _tile_arrays_from_cog(cog: Path, z: int, x: int, y: int, *, cog_cache: CogCache = None) -> tuple[Optional[np.ndarray], Optional[np.ndarray], bool]:
    use_global = (cog_cache is None)
    if use_global: cog_cache = COG_CACHE
    if (not cog.exists()) or (cog.stat().st_size <= 0):
        raise HTTPException(status_code=500, detail=f"COG missing: {cog.name}")

    def _read_once():
        r = cog_cache.get(cog)
        data, mask = r.tile(x, y, z, tilesize=256, resampling_method="nearest")
        return data[0].astype("float32"), mask[0].astype("uint8"), False
    if use_global:
        lk = _cog_read_lock(cog)
        def _read_locked():
            with lk: return _read_once()
        read_fn = _read_locked
    else:
        read_fn = _read_once

    try: return read_fn()
    except TileOutsideBounds: return None, None, True
    except Exception:
        try: cog_cache.evict(cog)
        except: pass
        try: return read_fn()
        except TileOutsideBounds: return None, None, True
        except Exception as e2:
            msg = str(e2); _log("WARN", f"[cog_read] failed cog={cog.name} z={z} x={x} y={y} err={e2!r}")
            if ("TIFFReadEncodedTile" in msg) or ("ZIPDecode" in msg) or ("Decoding error" in msg):
                try: cog_cache.evict(cog)
                except: pass
                if use_global:
                    try: _safe_unlink(cog); _log("WARN", f"[cog_read] deleted corrupted COG: {cog.name}")
                    except: pass
            return None, None, True

def _tile_png_from_cog(cog:Path,z:int,x:int,y:int,max_in:Optional[float])->bytes:
    arr,mask,oob=_tile_arrays_from_cog(cog,z,x,y,cog_cache=COG_CACHE)
    if oob or arr is None or mask is None:return _transparent_png_256()
    return _melt_to_png(arr,mask,max_in,None,None)

def _serve_forecast_tile(run_init:str,valid:str,lead:str,melt_urls:list[str],snow_ts:str,snow_urls:list[str],z:int,x:int,y:int,max_in:float|None)->Response:
    run_init=_norm_ts(run_init);valid=_norm_ts(valid);lead=(lead or"").lower()
    melt_cogs=_build_forecast_melt_cogs(run_init,valid,lead,melt_urls)
    snow=snow_valid=None;snow_info="mask_unavailable"
    if snow_urls:snow,snow_valid,snow_info=_read_mask_tile(snow_ts or valid,snow_urls,z,x,y)
    melt_mm,melt_valid,melt_info=_union_tile(melt_cogs,z,x,y,label="melt")
    if melt_mm is None or melt_valid is None:return _resp_png(_transparent_png_256(),**{"X-Melt-Info":melt_info,"X-NoData":"1"})
    melt_mask_u8=(melt_valid.astype("uint8")*255)
    png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2)
    return _resp_png(png,**{"X-Forecast-RunInit-TT":run_init,"X-Forecast-Valid-TS":valid,"X-Forecast-Lead":lead,"X-Melt-Info":melt_info,"X-SnowMask-Info":snow_info,"X-SnowMask-TS":(snow_ts or valid or"")})

def _corridor_tiles_for_zoom_range(zmin:int,zmax:int,*,miles:float,cap_per_zoom:int)->list[tuple[int,int,int]]:
    pts=_load_track_points()
    if not pts:return[]
    out:list[tuple[int,int,int]]=[];seen:set[tuple[int,int,int]]=set()
    for z in range(int(zmin),int(zmax)+1):
        tiles=_tiles_within_miles(pts,z,float(miles),cap=int(cap_per_zoom))
        for(zz,x,y)in tiles:
            if not _tile_allowed(zz,x,y):continue
            t=(zz,x,y)
            if t in seen:continue
            seen.add(t);out.append(t)
    return out

def _point_mm_from_cog(cog:Path,lon:float,lat:float)->tuple[Optional[float],str]:
    if not cog.exists()or cog.stat().st_size<=0:return None,f"missing:{cog.name}"
    try:
        with COGReader(cog.as_posix())as r:data,mask=_cog_point_data_mask(r,lon,lat)
        v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0])
        if m<=0 or(not np.isfinite(v))or(v<=-9990.0):return None,f"nodata:{cog.name}"
        return v,f"ok:{cog.name}"
    except TileOutsideBounds:return None,f"oob:{cog.name}"
    except Exception as e:return None,f"error:{cog.name}:{e!r}"

def _union_point_mm(cogs:list[Path],lon:float,lat:float)->tuple[Optional[float],str]:
    for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))):
        v,info=_point_mm_from_cog(cog,lon,lat)
        if v is not None:return v,info
    return None,"all_domains_failed"

def _bake_corridor_tiles_for_date(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,miles:float,cap_per_zoom:int,max_tiles_total:int,conc:int=2,push_every:int=500)->dict:
    dom=(dom or"zz").lower();_log("DEBUG",f"[bake_corridor] Starting for {date_yyyymmdd}")
    tiles=_corridor_tiles_for_zoom_range(zmin,zmax,miles=float(miles),cap_per_zoom=int(cap_per_zoom))
    if not tiles:return{"ok":False,"error":"no_corridor_tiles"}
    tiles=tiles[:int(max_tiles_total)];_log("DEBUG",f"[bake_corridor] {len(tiles)} tiles to process")
    q=deque()
    for hours in hours_list:
        for(z,x,y)in tiles:q.append((int(hours),int(z),int(x),int(y)))
    total_jobs=len(q);ok=0;skip=0;fail=0;lk=threading.Lock();pending_push:list[Path]=[];total_pushed=0;push_lk=threading.Lock()
    def maybe_push():
        nonlocal pending_push,total_pushed
        with push_lk:
            if len(pending_push)>=push_every:
                batch=pending_push[:push_every];pending_push=pending_push[push_every:]
                result=_incremental_hf_push(batch,"tilecache",f"{date_yyyymmdd}_corridor_png_{total_pushed}")
                if result.get("ok"):total_pushed+=result.get("pushed",0);_log("DEBUG",f"[bake_corridor] Pushed {result.get('pushed',0)} tiles to HF (total: {total_pushed})")
                else:_log("WARN",f"[bake_corridor] Push failed: {result}")
    def worker():
        nonlocal ok,skip,fail,pending_push
        with _HfPullOnMiss(False):
            while True:
                with lk:
                    if not q:return
                    hours,z,x,y=q.popleft();processed=total_jobs-len(q)
                try:
                    p=_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y)
                    if _is_valid_png_file(p):
                        with lk:skip+=1
                        continue
                    png,_=_generate_forecast_by_date_png_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,max_in=None)
                    _tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,png)
                    with lk:ok+=1;pending_push.append(p)
                    maybe_push()
                    if processed%500==0:_log("DEBUG",f"[bake_corridor] {date_yyyymmdd}: {processed}/{total_jobs} (ok={ok}, skip={skip}, fail={fail})")
                except Exception as e:
                    with lk:
                        if fail<5:_log("WARN",f"[bake_corridor] tile failed z={z} x={x} y={y} err={e!r}")
                        fail+=1
    threads=[threading.Thread(target=worker,daemon=True)for _ in range(int(conc))]
    for t in threads:t.start()
    for t in threads:t.join()
    with push_lk:
        if pending_push:
            result=_incremental_hf_push(pending_push,"tilecache",f"{date_yyyymmdd}_corridor_png_final")
            if result.get("ok"):total_pushed+=result.get("pushed",0)
            _log("DEBUG",f"[bake_corridor] Final push: {result}");pending_push=[]
    _log("INFO",f"[bake_corridor] {date_yyyymmdd} complete: ok={ok}, skipped={skip}, fail={fail}, pushed={total_pushed}")
    return{"ok":True,"date":date_yyyymmdd,"dom":dom,"hours_list":hours_list,"tiles":len(tiles),"jobs":total_jobs,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":total_pushed,"zmin":zmin,"zmax":zmax}

def _rolling_dates_local_window(today_local=None, *, lookback: int = 1, lookahead: int = 3) -> list[str]:
    if today_local is None:
        today_local = datetime.now(LOCAL_TZ).date()
    return sorted([(today_local + timedelta(days=d)).strftime("%Y%m%d") for d in range(-int(lookback), int(lookahead) + 1)])

def _keep_cutoff_ymd(keep_days: int = _REMOTE_KEEP_DAYS) -> str:
    k = int(keep_days)
    today_local = datetime.now(LOCAL_TZ).date()
    if k <= 0:
        cutoff_date = today_local
    else:
        cutoff_date = today_local - timedelta(days=(k - 1))
    return cutoff_date.strftime("%Y%m%d")

def _prune_tile_cache(keep_ymds: set[str], keep_days_fallback: int = 0) -> int:
    deleted = 0
    root = _tile_cache_dir()
    if not root.exists():
        return 0
    for dom_dir in root.iterdir():
        if not dom_dir.is_dir():
            continue
        for hours_dir in dom_dir.iterdir():
            if not hours_dir.is_dir():
                continue
            for ymd_dir in hours_dir.iterdir():
                if not ymd_dir.is_dir():
                    continue
                ymd = ymd_dir.name
                if not re.fullmatch(r"[0-9]{8}", ymd):
                    continue
                if ymd in keep_ymds:
                    continue
                try:
                    shutil.rmtree(ymd_dir, ignore_errors=True)
                    deleted += 1
                    _log("INFO", f"[prune_tile] deleted local tilecache dir: {ymd_dir.relative_to(CACHE)}")
                except:
                    pass
    return deleted

def _prune_raw_tile_cache(keep_ymds: set[str], keep_days_fallback: int = 0) -> int:
    deleted = 0
    root = _raw_tile_cache_dir()
    if not root.exists():
        return 0
    for dom_dir in root.iterdir():
        if not dom_dir.is_dir():
            continue
        for hours_dir in dom_dir.iterdir():
            if not hours_dir.is_dir():
                continue
            for ymd_dir in hours_dir.iterdir():
                if not ymd_dir.is_dir():
                    continue
                ymd = ymd_dir.name
                if not re.fullmatch(r"[0-9]{8}", ymd):
                    continue
                # ONLY keep dates explicitly in the keep set
                if ymd in keep_ymds:
                    continue
                try:
                    shutil.rmtree(ymd_dir, ignore_errors=True)
                    deleted += 1
                    _log("INFO", f"[prune_raw] deleted local rawtilecache dir: {ymd_dir.relative_to(CACHE)}")
                except:
                    pass
    return deleted


def _prune_forecast_cogs(keep_days: int = 0, *, keep_ymds: set[str] | None = None) -> int:
    deleted = 0
    fc = _forecast_dir()
    if not fc.exists():
        return 0
    for p in fc.glob("*_cog.tif"):
        ymd = _file_ymd_from_name(p.name)
        if not ymd:
            continue
        if keep_ymds is not None:
            if ymd in keep_ymds:
                continue
        else:
            cutoff = _keep_cutoff_ymd(keep_days)
            if ymd >= cutoff:
                continue
        try:
            p.unlink(missing_ok=True)
            deleted += 1
            _log("INFO", f"[prune_cog] deleted forecast COG: {p.name}")
        except:
            pass
    return deleted

def _ensure_forecast_cogs_for_date(date_yyyymmdd:str,dom:str)->dict:
    dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05")
    sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
    if not sel.get("ok"):return{"ok":False,"date":date_yyyymmdd,"error":"pick_failed","detail":sel}
    try:_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"])
    except Exception as e:return{"ok":False,"date":date_yyyymmdd,"valid":sel["valid"],"error":f"melt_build_failed:{e!r}"}
    if sel.get("snowpack_urls"):
        try:_build_forecast_snowpack_cogs(sel.get("snowpack_ts")or sel["valid"],sel["snowpack_urls"])
        except:pass
    melt24_union=snow_union=melt72=None
    try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom)
    except Exception as e:_log("WARN",f"[ensure_forecast] 24h UNION melt failed for {sel['valid']}: {e!r}")
    try:snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom)
    except Exception as e:_log("WARN",f"[ensure_forecast] UNION snowpack failed for {sel['valid']}: {e!r}")
    try:melt72=_forecast_melt72h_end_cog(sel["valid"],dom_prefer=dom)
    except Exception as e:_log("WARN",f"[ensure_forecast] 72h COG failed for {sel['valid']}: {e!r}")
    return{"ok":True,"date":date_yyyymmdd,"valid":sel["valid"],"tt":sel["run_init"],"dom":dom,"melt24_union_cog":(melt24_union.name if melt24_union else None),"snow_union_cog":(snow_union.name if snow_union else None),"melt72_cog":(melt72.name if melt72 else None)}

def _ensure_forecast_for_valid(valid: str, dom_prefer: str = "zz") -> dict:
    dom_prefer = (dom_prefer or "zz").lower()
    valid = _norm_ts(valid)

    date_yyyymmdd = valid[:8]
    sel = _pick_forecast_for_valid(valid, dom_prefer=dom_prefer)
    if not sel.get("ok"):
        return {
            "ok": False,
            "date": date_yyyymmdd,
            "valid": valid,
            "error": "pick_failed",
            "detail": sel,
        }

    try:
        _build_forecast_melt_cogs(
            sel["run_init"],
            sel["valid"],
            "t0024",
            sel["melt_urls"],
        )
    except Exception as e:
        return {
            "ok": False,
            "date": date_yyyymmdd,
            "valid": valid,
            "error": f"melt_build_failed:{e!r}",
        }

    if sel.get("snowpack_urls"):
        try:
            _build_forecast_snowpack_cogs(
                sel.get("snowpack_ts") or sel["valid"],
                sel["snowpack_urls"],
            )
        except Exception:
            pass

    melt24_union = None
    snow_union = None
    melt72 = None

    try:
        melt24_union = _forecast_melt24h_union_end_cog(
            sel["valid"],
            dom_prefer=dom_prefer,
        )
    except Exception as e:
        _log("WARN", f"[ensure_forecast] 24h UNION melt failed for {valid}: {e!r}")

    try:
        snow_union = _forecast_snowpack_union_end_cog(
            sel.get("snowpack_ts") or sel["valid"],
            dom_prefer=dom_prefer,
        )
    except Exception as e:
        _log("WARN", f"[ensure_forecast] UNION snowpack failed for {valid}: {e!r}")

    try:
        melt72 = _forecast_melt72h_end_cog(
            sel["valid"],
            dom_prefer=dom_prefer,
        )
    except Exception as e:
        _log("WARN", f"[ensure_forecast] 72h COG failed for {valid}: {e!r}")

    return {
        "ok": True,
        "date": date_yyyymmdd,
        "valid": sel["valid"],
        "tt": sel["run_init"],
        "dom": dom_prefer,
        "melt24_union_cog": (melt24_union.name if melt24_union else None),
        "snow_union_cog": (snow_union.name if snow_union else None),
        "melt72_cog": (melt72.name if melt72 else None),
    }

def _next_run_local(now:datetime|None=None)->datetime:
    now=now or datetime.now(LOCAL_TZ);target=now.replace(hour=SCHEDULE_HOUR_LOCAL,minute=SCHEDULE_MIN_LOCAL,second=0,microsecond=0)
    return target+timedelta(days=1)if target<=now else target
    
def _bake_corridor_tiles_for_date_optimized(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,miles:float,cap_per_zoom:int,max_tiles_total:int,push_every:int=300)->dict:
    dom=(dom or"zz").lower();_log("INFO",f"[bake_corridor_opt] Starting for {date_yyyymmdd}")
    tiles=_corridor_tiles_for_zoom_range(zmin,zmax,miles=float(miles),cap_per_zoom=int(cap_per_zoom))
    if not tiles:return{"ok":False,"error":"no_tiles"}
    tiles=tiles[:int(max_tiles_total)];jobs=[(int(h),int(z),int(x),int(y))for h in hours_list for(z,x,y)in tiles]
    total_jobs=len(jobs);_log("INFO",f"[bake_corridor_opt] {total_jobs} tiles to process");ok=0;skip=0;fail=0;local_written=0
    with CogCache()as cog_cache:
        _log("INFO",f"[bake_corridor_opt] Pre-loading COGs for {date_yyyymmdd}");preloaded=_preload_forecast_for_date(date_yyyymmdd,dom,cog_cache)
        _log("INFO",f"[bake_corridor_opt] Preload complete: h24={preloaded.get('h24',{}).get('ok',False)}, h72={preloaded.get('h72',{}).get('ok',False)}")
        for i,(hours,z,x,y)in enumerate(jobs):
            try:
                p=_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y)
                if _is_valid_png_file(p):skip+=1;continue
                png,is_valid=_generate_tile_from_preloaded(z,x,y,hours,preloaded,cog_cache=cog_cache)
                _tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,png);ok+=1;local_written+=1
            except Exception as e:
                if fail<5:_log("WARN",f"[bake_corridor_opt] tile failed z={z} x={x} y={y}: {e!r}")
                fail+=1
            if(i+1)%500==0:_log("DEBUG",f"[bake_corridor_opt] {date_yyyymmdd}: {i+1}/{total_jobs} (ok={ok}, skip={skip}, fail={fail})")
    _log("INFO",f"[bake_corridor_opt] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, local_written={local_written}")
    return{"ok":True,"date":date_yyyymmdd,"jobs":total_jobs,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":0,"remote_push":"deferred"}

def _bake_corridor_raw_tiles_for_date_optimized(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,miles:float,cap_per_zoom:int,max_tiles_total:int,push_every:int=300)->dict:
    dom=(dom or"zz").lower();_log("INFO",f"[bake_corridor_raw_opt] Starting for {date_yyyymmdd}")
    tiles=_corridor_tiles_for_zoom_range(zmin,zmax,miles=float(miles),cap_per_zoom=int(cap_per_zoom))
    if not tiles:return{"ok":False,"error":"no_tiles"}
    tiles=tiles[:int(max_tiles_total)];jobs=[(int(h),int(z),int(x),int(y))for h in hours_list for(z,x,y)in tiles]
    total_jobs=len(jobs);_log("INFO",f"[bake_corridor_raw_opt] {total_jobs} tiles to process");ok=0;skip=0;fail=0;pending_push:list[Path]=[];total_pushed=0
    with CogCache()as cog_cache:
        _log("INFO",f"[bake_corridor_raw_opt] Pre-loading COGs for {date_yyyymmdd}")
        preloaded=_preload_forecast_for_date(date_yyyymmdd,dom,cog_cache)
        _log("INFO",f"[bake_corridor_raw_opt] Preload complete: h24={preloaded.get('h24',{}).get('ok',False)}, h72={preloaded.get('h72',{}).get('ok',False)}")
        for i,(hours,z,x,y)in enumerate(jobs):
            try:
                p=_raw_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y)
                if _is_valid_npz_file(p):skip+=1;continue
                raw,_=_generate_forecast_by_date_raw_preloaded(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,preloaded=preloaded,cog_cache=cog_cache)
                _raw_tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,raw);ok+=1;pending_push.append(p)
            except Exception as e:
                if fail<5:_log("WARN",f"[bake_corridor_raw_opt] tile failed z={z} x={x} y={y}: {e!r}")
                fail+=1
            if len(pending_push)>=push_every:
                result=_incremental_hf_push(pending_push,"rawtilecache",f"{date_yyyymmdd}_corridor_raw_{total_pushed}")
                if result.get("ok"):total_pushed+=result.get("pushed",0)
                pending_push=[]
            if(i+1)%500==0:_log("DEBUG",f"[bake_corridor_raw_opt] {date_yyyymmdd}: {i+1}/{total_jobs} (ok={ok}, skip={skip}, fail={fail}, pushed={total_pushed})")
        if pending_push:
            result=_incremental_hf_push(pending_push,"rawtilecache",f"{date_yyyymmdd}_corridor_raw_final")
            if result.get("ok"):total_pushed+=result.get("pushed",0)
    _log("INFO",f"[bake_corridor_raw_opt] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, pushed={total_pushed}")
    return{"ok":True,"date":date_yyyymmdd,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":total_pushed}

_PUSH_PAUSED = threading.Event()
_PUSH_PAUSED.clear()
def _hf_push_baked_dates(baked_ymds: list[str], dom: str, hours_list: list[int]) -> dict:
    tok, repo = _hf_cfg(); api = _hf_api()
    if not tok or not repo or api is None:
        return {"ok": False, "error": "hf_not_configured"}
    dom = (dom or "zz").lower(); pushed = 0; errors = 0
    for ymd in baked_ymds:
        for h in hours_list:
            for kind in ["tilecache", "rawtilecache"]:
                base = CACHE / kind / dom / str(int(h)) / ymd
                if not base.exists(): continue
                while _PUSH_PAUSED.is_set():
                    time.sleep(2.0)
                ext = ".png" if kind == "tilecache" else ".npz"
                ops = []; total_bytes = 0
                for p in base.rglob(f"*{ext}"):
                    sz = p.stat().st_size
                    if sz < 60: continue
                    try:
                        rel = p.relative_to(CACHE).as_posix()
                        ops.append(CommitOperationAdd(path_in_repo=rel, path_or_fileobj=str(p)))
                        total_bytes += sz
                    except: pass
                if not ops: continue
                _log("INFO", f"[hf_push] {kind}/{ymd}/h{h}: {len(ops)} files, {total_bytes/1024/1024:.1f} MB total")
                for i in range(0, len(ops),750):
                    while _PUSH_PAUSED.is_set():
                        time.sleep(2.0)
                    batch = ops[i:i+750]
                    batch_bytes = sum(Path(op.path_or_fileobj).stat().st_size for op in batch if hasattr(op, 'path_or_fileobj'))
                    batch_num = i // 750+ 1
                    ok = False
                    for attempt in range(3):
                        t0 = time.time()
                        try:
                            api.create_commit(repo_id=repo, repo_type="dataset", operations=batch,
                                commit_message=f"{kind}: {ymd} h{h} batch {batch_num} ({len(batch)} files)")
                            elapsed = time.time() - t0
                            pushed += len(batch)
                            ok = True
                            _log("INFO", f"[hf_push] {kind}/{ymd}/h{h} batch {batch_num}: OK {len(batch)} files, {batch_bytes/1024:.0f} KB in {elapsed:.1f}s")
                            break
                        except Exception as e:
                            elapsed = time.time() - t0
                            err_str = repr(e)
                            if "ReadTimeout" in err_str and elapsed >= 9.0:
                                _log("INFO", f"[hf_push] {kind}/{ymd}/h{h} batch {batch_num}: timeout after {elapsed:.1f}s — likely succeeded (attempt {attempt+1})")
                                pushed += len(batch); ok = True; time.sleep(2.0); break
                            _log("WARN", f"[hf_push] {kind}/{ymd}/h{h} batch {batch_num} attempt {attempt+1}: {type(e).__name__} after {elapsed:.1f}s — {len(batch)} files, {batch_bytes/1024:.0f} KB")
                            time.sleep(10.0 * (attempt + 1))
                    if not ok:
                        errors += len(batch)
                    else:
                        time.sleep(2.0)
                _log("INFO", f"[hf_push] {kind}/{ymd}/h{h}: done (pushed={pushed} errors={errors})")
    _log("INFO", f"[hf_push_baked] done: pushed={pushed} errors={errors}")
    return {"ok": True, "pushed": pushed, "errors": errors}
           
def _run_daily_roll_job_impl()->dict:
    if not _try_acquire_run_lock(_ROLL_LOCK_NAME,stale_seconds=_ROLL_LOCK_STALE_SECONDS):
        _log("INFO","[roll] skipped: already running");return{"ok":True,"skipped":True,"reason":"already_running"}
    try:
        import gc;started=datetime.now(LOCAL_TZ).isoformat();today_local=datetime.now(LOCAL_TZ).date()
        window=_rolling_dates_local_window(today_local=today_local,lookback=2,lookahead=3);keep_set=set(window)
        _log("INFO",f"[roll] local_today={today_local} window={window}")
        need=[ymd for ymd in window if not _date_is_baked(WARM_DOM,ymd)]
        already=[y for y in window if y not in need]
        _log("INFO",f"[roll] need={need} already_baked={already}")
        if not need:
            _log("INFO","[roll] all dates baked — prune only")
            try:_hf_prune_remote_and_local(keep_ymds=keep_set)
            except Exception as e:_log("WARN",f"[roll] prune failed: {e!r}")
            return{"ok":True,"skipped":True,"reason":"nothing_needed","window":window,"finished":datetime.now(LOCAL_TZ).isoformat()}
        _log("INFO","[roll] Step 2: Pulling forecast COGs + tar archives")
        try:_hf_pull_cache()
        except Exception as e:_log("WARN",f"[roll] HF pull failed: {e!r}")
        need=[ymd for ymd in need if not _date_is_baked(WARM_DOM,ymd)]
        _log("INFO",f"[roll] after pull, still need={need}")
        if not need:
            try:_hf_prune_remote_and_local(keep_ymds=keep_set)
            except:pass
            return{"ok":True,"skipped":True,"reason":"hf_pull_satisfied","window":window,"finished":datetime.now(LOCAL_TZ).isoformat()}
        _log("INFO","[roll] Step 3: Pruning old tiles")
        try:
            dt=_prune_tile_cache(keep_set);dr=_prune_raw_tile_cache(keep_set);dc=_prune_forecast_cogs(keep_ymds=keep_set)
            _log("INFO",f"[roll] pruned tiles={dt} raw={dr} cogs={dc}")
        except Exception as e:_log("WARN",f"[roll] prune failed: {e!r}")
        try:_tile_png_cached.cache_clear()
        except:pass
        today_str=today_local.strftime("%Y%m%d");tmrw_str=(today_local+timedelta(days=1)).strftime("%Y%m%d")
        need=sorted(need,key=lambda y:(0 if y==today_str else 1 if y==tmrw_str else 2,y))
        _log("INFO",f"[roll] Step 4: Baking {len(need)} date(s) (priority order: {need})")
        results={"ok":True,"started":started,"window":window,"need":need,"dates":{}};baked_ymds=[];failed_ymds=[]
        for i,ymd in enumerate(need):
            _log("INFO",f"[roll] === {ymd} ({i+1}/{len(need)}) ===");dr={"date":ymd}
            cog_ok=False
            for cog_attempt in range(5):
                try:
                    r=_ensure_forecast_for_valid(_norm_ts(f"{ymd}05"),dom_prefer=WARM_DOM)
                    if r.get("ok"):cog_ok=True;dr["cog"]={"ok":True};break
                    else:_log("WARN",f"[roll] {ymd}: COG attempt {cog_attempt+1}/5 failed")
                except Exception as e:_log("WARN",f"[roll] {ymd}: COG attempt {cog_attempt+1}/5 exception: {e!r}")
                if cog_attempt<4:sleep_sec=10*(2**cog_attempt);_log("INFO",f"[roll] {ymd}: retrying COG in {sleep_sec}s");time.sleep(sleep_sec)
            if not cog_ok:_log("WARN",f"[roll] {ymd}: COG failed after 5 attempts");dr["cog"]={"ok":False};results["dates"][ymd]=dr;failed_ymds.append(ymd);continue
            with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool:
                futs={}
                            # Bake sequentially: PNG first, then raw
                for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]:
                    try:
                        r=fn(date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra)
                        dr[label]={"ok":r.get("ok_count",0),"skip":r.get("skip_count",0),"fail":r.get("fail_count",0)}
                        _log("INFO",f"[roll] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
                    except Exception as e:_log("WARN",f"[roll] {ymd}: {label} exception: {e!r}");dr[label]={"error":repr(e)}
            for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
                try:
                    r=fn(date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra)
                    dr[label]={"ok":r.get("ok_count",0),"skip":r.get("skip_count",0),"fail":r.get("fail_count",0)}
                    _log("INFO",f"[roll] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
                except Exception as e:_log("WARN",f"[roll] {ymd}: {label} exception: {e!r}");dr[label]={"error":repr(e)}
            try:
                best=_find_best_melt_for_date(ymd,dom_prefer=WARM_DOM)
                if best.get("ok")and best.get("melt_urls"):_record_baked_sig(ymd,WARM_DOM,best["melt_urls"][0],source_type=best.get("source_type"))
            except:pass
            results["dates"][ymd]=dr;baked_ymds.append(ymd);_log("INFO",f"[roll] === {ymd} bake complete ===");gc.collect()
        if baked_ymds:
            _log("INFO",f"[roll] Step 5: Pushing {len(baked_ymds)} dates to HF via tar archives")
            push_result=_hf_push_baked_as_tar(baked_ymds,WARM_DOM,WARM_HOURS_LIST);results["push"]=push_result
            _log("INFO",f"[roll] Push complete: {push_result}")
        else:_log("INFO","[roll] Step 5: No dates to push")
        _log("INFO","[roll] Step 6: Pruning old data")
        try:_hf_prune_remote_and_local(keep_ymds=keep_set)
        except Exception as e:_log("WARN",f"[roll] prune failed: {e!r}")
        results["finished"]=datetime.now(LOCAL_TZ).isoformat()
        _log("INFO",f"[roll] ========== COMPLETE at {results['finished']} ==========")
        if failed_ymds:
            _log("INFO",f"[roll] Step 7: Starting retry loop for failed dates: {failed_ymds}")
            def _retry_failed():
                remaining=list(failed_ymds);max_retries=48
                for retry_round in range(max_retries):
                    if not remaining:break
                    _log("INFO",f"[roll_retry] round {retry_round+1}: waiting 30 min before retrying {remaining}");time.sleep(30*60)
                    still_failed=[]
                    for ymd in remaining:
                        if _date_is_baked(WARM_DOM,ymd):_log("INFO",f"[roll_retry] {ymd}: already baked");continue
                        cog_ok=False
                        for cog_attempt in range(5):
                            try:
                                r=_ensure_forecast_for_valid(_norm_ts(f"{ymd}05"),dom_prefer=WARM_DOM)
                                if r.get("ok"):cog_ok=True;break
                            except:pass
                            if cog_attempt<4:time.sleep(10*(2**cog_attempt))
                        if not cog_ok:_log("WARN",f"[roll_retry] {ymd}: COG still unavailable");still_failed.append(ymd);continue
                        _log("INFO",f"[roll_retry] {ymd}: COG OK, baking")
                        try:
                            with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool:
                                futs={}
                                for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]:
                                    futs[label]=pool.submit(fn,date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra)
                                for label,fut in futs.items():
                                    try:r=fut.result();_log("INFO",f"[roll_retry] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
                                    except Exception as e:_log("WARN",f"[roll_retry] {ymd}: {label} exception: {e!r}")
                            for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
                                try:r=fn(date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra);_log("INFO",f"[roll_retry] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
                                except Exception as e:_log("WARN",f"[roll_retry] {ymd}: {label} exception: {e!r}")
                            _hf_push_baked_as_tar([ymd],WARM_DOM,WARM_HOURS_LIST);_log("INFO",f"[roll_retry] {ymd}: bake + push complete")
                            try:
                                best=_find_best_melt_for_date(ymd,dom_prefer=WARM_DOM)
                                if best.get("ok")and best.get("melt_urls"):_record_baked_sig(ymd,WARM_DOM,best["melt_urls"][0],source_type=best.get("source_type"))
                            except:pass
                        except Exception as e:_log("WARN",f"[roll_retry] {ymd}: bake failed: {e!r}");still_failed.append(ymd)
                        gc.collect()
                    remaining=still_failed
                if remaining:_log("WARN",f"[roll_retry] giving up on: {remaining} after {max_retries} rounds")
                else:_log("INFO","[roll_retry] all failed dates recovered")
            threading.Thread(target=_retry_failed,daemon=True,name="roll_retry").start()
        return results
    finally:_release_run_lock(_ROLL_LOCK_NAME)
        
def _daily_scheduler_loop()->None:
    time.sleep(0.25);_log("INFO","[scheduler] start")
    while True:
        try:
            now=datetime.now(LOCAL_TZ);nxt=_next_run_local(now);sleep_s=max(1.0,(nxt-now).total_seconds())
            _log("INFO",f"[scheduler] next_run_local={nxt.isoformat()} sleep_s={int(sleep_s)}");time.sleep(sleep_s)
            r=_run_daily_roll_job_impl()
            if isinstance(r,dict)and r.get("skipped"):_log("INFO",f"[scheduler] roll skipped reason={r.get('reason')}")
        except Exception as e:_log("INFO",f"[scheduler] crashed err={e!r}");time.sleep(30.0)

def _preload_forecast_for_date(date_yyyymmdd:str,dom:str,cog_cache:CogCache)->dict:
    dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05")
    result={"date":date_yyyymmdd,"valid":valid,"dom":dom,"h24":{"ok":False},"h72":{"ok":False}}
    try:
        sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
        if sel.get("ok"):
            melt24_union=snow_union=None
            try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom)
            except Exception as e:_log("WARN",f"[preload] melt24 union failed for {date_yyyymmdd}: {e!r}");melt24_union=None
            if sel.get("snowpack_urls"):
                try:snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom)
                except:snow_union=None
            if melt24_union is not None:
                try:cog_cache.get(melt24_union)
                except:pass
            if snow_union is not None:
                try:cog_cache.get(snow_union)
                except:pass
            if melt24_union is not None and melt24_union.exists()and melt24_union.stat().st_size>0:
                result["h24"]={"ok":True,"run_init":sel["run_init"],"valid":sel["valid"],"melt24_union_cog":melt24_union,"snow_union_cog":snow_union}
    except Exception as e:_log("WARN",f"[preload] h24 failed for {date_yyyymmdd}: {e!r}")
    try:
        run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom)
        if run_init:
            melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom)
            if melt72_cog.exists()and melt72_cog.stat().st_size>0:
                try:cog_cache.get(melt72_cog)
                except:pass
                result["h72"]={"ok":True,"run_init":run_init,"valid":valid,"melt72_cog":melt72_cog,"snow_union_cog":result["h24"].get("snow_union_cog")if result["h24"].get("ok")else None}
    except Exception as e:_log("WARN",f"[preload] h72 failed for {date_yyyymmdd}: {e!r}")
    return result

def _generate_tile_from_preloaded(z:int,x:int,y:int,hours:int,preloaded:dict,max_in:float|None=None,*,cog_cache:CogCache|None=None)->tuple[bytes,bool]:
    if hours==24:
        h24=preloaded.get("h24",{})
        if not h24.get("ok"):return _transparent_png_256(),False
        melt_cog=h24.get("melt24_union_cog")
        if not melt_cog:return _transparent_png_256(),False
        melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt_cog,z,x,y,cog_cache=cog_cache)
        if oob or melt_mm is None or melt_mask is None:return _transparent_png_256(),False
        melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255)
        snow=snow_valid=None;snow_cog=h24.get("snow_union_cog")
        if snow_cog:
            snow_arr,snow_mask,oob2=_tile_arrays_from_cog(snow_cog,z,x,y,cog_cache=cog_cache)
            if(not oob2)and(snow_arr is not None)and(snow_mask is not None):snow_valid=(snow_mask>0)&np.isfinite(snow_arr)&(snow_arr>-9990.0);snow=snow_arr
        png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_IN)
        return png,True
    if hours==72:
        h72=preloaded.get("h72",{})
        if not h72.get("ok"):return _transparent_png_256(),False
        melt72_cog=h72.get("melt72_cog")
        if not melt72_cog:return _transparent_png_256(),False
        melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt72_cog,z,x,y,cog_cache=cog_cache)
        if oob or melt_mm is None or melt_mask is None:return _transparent_png_256(),False
        melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255)
        snow=snow_valid=None;snow_cog=h72.get("snow_union_cog")
        if snow_cog:
            snow_arr,snow_mask,oob2=_tile_arrays_from_cog(snow_cog,z,x,y,cog_cache=cog_cache)
            if(not oob2)and(snow_arr is not None)and(snow_mask is not None):snow_valid=(snow_mask>0)&np.isfinite(snow_arr)&(snow_arr>-9990.0);snow=snow_arr
        png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_72H_IN)
        return png,True
    return _transparent_png_256(),False

def _generate_forecast_by_date_raw_preloaded(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,preloaded:dict,cog_cache:CogCache)->tuple[bytes,dict]:
    dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05")
    if int(hours)==24:
        h24=preloaded.get("h24",{})
        if not h24.get("ok"):raise HTTPException(status_code=503,detail={"error":"forecast_unavailable","valid":valid,"dom":dom})
        sel=h24.get("sel")or{};melt_info="melt_unavailable"
        if h24.get("melt24_union_cog"):
            melt_mm,melt_mask,oob=_tile_arrays_from_cog(h24["melt24_union_cog"],z,x,y,cog_cache=cog_cache)
            if oob or melt_mm is None or melt_mask is None:melt_mm=None;melt_valid=None
            else:melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0)
            melt_info=f"melt24_union:{h24['melt24_union_cog'].name}"
        else:melt_mm,melt_valid,melt_info=_union_tile_cached(h24.get("melt_cogs",[]),z,x,y,label="melt",cog_cache=cog_cache)
        snow_info="snowpack_unavailable"
        try:
            if h24.get("snow_union_cog"):snow_info=f"snowpack_union:{h24['snow_union_cog'].name}"
            elif h24.get("snow_cogs"):_,_,snow_info=_union_tile_cached(h24["snow_cogs"],z,x,y,label="snowpack",cog_cache=cog_cache)
        except Exception as e:snow_info=f"snowpack_unavailable:{e!r}"
        if melt_mm is None or melt_valid is None:
            raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8"))
            return raw,{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":melt_info,"X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel.get("valid",valid),"X-Forecast-RunInit-TT":sel.get("run_init",""),"X-Forecast-Hours":"24","X-Raw-Scale":"10"}
        raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=(melt_valid.astype("uint8")*255))
        return raw,{"X-Allowed":"1","X-OOB":"0","X-Melt-Info":melt_info,"X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel.get("valid",valid),"X-Forecast-RunInit-TT":sel.get("run_init",""),"X-Forecast-Hours":"24","X-Raw-Scale":"10"}
    h72=preloaded.get("h72",{})
    if not h72.get("ok"):raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom,"suggestion":"forecast data may not be available yet for this date"})
    melt_mm,melt_mask_u8,oob=_tile_arrays_from_cog(h72["melt72_cog"],z,x,y,cog_cache=cog_cache)
    if oob or melt_mm is None or melt_mask_u8 is None:
        raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8"))
        return raw,{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":h72.get("run_init",""),"X-Forecast-72h-COG":h72["melt72_cog"].name,"X-Raw-Scale":"10"}
    raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=melt_mask_u8.astype("uint8",copy=False))
    return raw,{"X-Allowed":"1","X-OOB":"0","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":h72.get("run_init",""),"X-Forecast-72h-COG":h72["melt72_cog"].name,"X-Raw-Scale":"10"}

def _shift_ts(ts10:str,hours:int)->str:return(_ts_to_dt(_norm_ts(ts10))+timedelta(hours=hours)).strftime("%Y%m%d%H")

def _union_point(cogs:list[Path],lon:float,lat:float)->tuple[Optional[float],str]:
    if not cogs:return None,"no_cogs"
    used=[]
    for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))):
        try:
            with COGReader(cog.as_posix())as r:data,mask=_cog_point_data_mask(r,lon,lat)
            v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0]);used.append(cog.name)
            if m>0 and np.isfinite(v)and v>-9990.0:return v,"ok:"+cog.name
        except:continue
    return None,"nodata:"+",".join(used[:3])

def _rolling_window_dates(lookback_days: int = 1, lookahead_days: int = 3) -> list[str]:
    today = datetime.now(LOCAL_TZ).date()
    return sorted([(today + timedelta(days=d)).strftime("%Y%m%d") for d in range(-int(lookback_days), int(lookahead_days) + 1)])

def _check_date_has_local_tiles(dom: str, hours: int, ymd: str) -> bool:
    dom = (dom or "zz").lower()
    for kind, ext in [("tilecache", "*.png"), ("rawtilecache", "*.npz")]:
        d = CACHE / kind / dom / str(hours) / ymd
        if d.exists():
            try:
                next(d.rglob(ext))
                return True
            except StopIteration:
                pass
    return False

def _date_has_local_tiles_both(dom: str, ymd: str, min_tiles: int = 50) -> bool:
    dom = (dom or "zz").lower()
    for h in WARM_HOURS_LIST:
        d = CACHE / "tilecache" / dom / str(h) / ymd
        if not d.exists():
            return False
        try:
            count = sum(1 for _ in d.rglob("*.png"))
            if count < min_tiles:
                return False
        except:
            return False
    return True

_AVAILABLE_DATES_CACHE:dict={"ts":0.0,"data":None,"key":None}
_AVAILABLE_DATES_CACHE_TTL=120.0

@app.get("/forecast/sample_melt")
def forecast_sample_melt(date_yyyymmdd:str=Query(...,regex=r"^\d{8}$"),hours:int=Query(24,ge=24,le=72),lon:float=Query(...,ge=-180.0,le=180.0),lat:float=Query(...,ge=-90.0,le=90.0),dom:str=Query("zz")):
    dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05");mm=None;info=""
    if int(hours)==72:
        cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom)
        try:
            r=_cogreader_get(cog);data,mask=_cog_point_data_mask(r,lon,lat);v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0])
            if m>0 and np.isfinite(v)and v>-9990.0:mm=v;info="ok:"+cog.name
            else:info="nodata:"+cog.name
        except Exception as e:info=f"err:{e!r}"
    else:
        sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
        if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="forecast data may not be available yet for this date";raise HTTPException(status_code=503,detail=detail)
        try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom)
        except Exception as e:raise HTTPException(status_code=503,detail={"error":"melt24_union_cog_unavailable","date_yyyymmdd":date_yyyymmdd,"valid":sel.get("valid",valid),"dom":dom,"upstream_error":repr(e)})
        try:
            r=_cogreader_get(melt24_union);data,mask=_cog_point_data_mask(r,lon,lat);v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0])
            if m>0 and np.isfinite(v)and v>-9990.0:mm=v;info="ok:"+melt24_union.name
            else:info="nodata:"+melt24_union.name
        except Exception as e:info=f"err:{e!r}"
    if mm is None:return{"ok":True,"value_in":None,"info":info,"hours":int(hours),"date_yyyymmdd":date_yyyymmdd}
    inches=max(0.0,float(mm)/INCH_TO_MM);return{"ok":True,"value_in":round(inches,1),"info":info,"hours":int(hours),"date_yyyymmdd":date_yyyymmdd}


@app.on_event("startup")
def _startup()->None:
    _transparent_png_256();_zero_raw_npz_256()
    if _acquire_singleton_lock("hf_pull"):
        def _background_hf_pull_and_cn():
            try:_log("INFO","[startup] background hf_pull_cache starting");_hf_pull_cache();_log("INFO","[startup] background hf_pull_cache finished")
            except Exception as e:_log("WARN",f"[startup] background hf_pull_cache failed: {e!r}")
            try:_log("INFO","[startup] background ensure_cn_mbtiles starting");_ensure_cn_mbtiles();_log("INFO","[startup] background ensure_cn_mbtiles finished")
            except Exception as e:_log("WARN",f"[tracks] failed to ensure cn mbtiles (background): {e!r}")
        t_hf=threading.Thread(target=_background_hf_pull_and_cn,daemon=True,name="hf_pull_bg");t_hf.start();_log("INFO","[startup] hf pull scheduled in background (singleton)")
    else:_log("INFO","[startup] hf pull skipped (another worker holds lock)")
    if _acquire_singleton_lock("daily_scheduler"):t=threading.Thread(target=_daily_scheduler_loop,daemon=True);t.start();_log("INFO","[startup] daily scheduler enabled (singleton)")
    else:_log("INFO","[startup] daily scheduler disabled (another worker holds lock)")
    if(str(BOOT_ROLL_DEFAULT).strip()!="0")and _acquire_singleton_lock("boot_roll"):t2=threading.Thread(target=_run_daily_roll_job_impl,daemon=True);t2.start();_log("INFO","[startup] boot roll job kicked off")
    else:_log("INFO","[startup] boot roll job skipped")
    if _acquire_singleton_lock("hf_uploader"):t3=threading.Thread(target=_hf_tiles_uploader_loop,daemon=True);t3.start();_log("INFO","[startup] hf uploader enabled (singleton)")
    else:_log("INFO","[startup] hf uploader disabled (another worker holds lock)")
    if _acquire_singleton_lock("staleness_monitor"):
        t_stale = threading.Thread(target=_staleness_monitor_loop, daemon=True, name="staleness_monitor")
        t_stale.start()
        _log("INFO", "[startup] staleness monitor enabled (singleton)")
    else:
        _log("INFO", "[startup] staleness monitor disabled (another worker holds lock)")
        
@app.get("/",response_class=PlainTextResponse)
def root():return"SNODAS Snowmelt Tiles. See /web for demo and /docs for API."

@app.get("/healthz",response_class=PlainTextResponse)
def health():return"ok"

@app.get("/config.js",response_class=PlainTextResponse)
def config_js():return f"window.__SNODAS_CONFIG__ = {{ MAPBOX_TOKEN: '{os.environ.get('MAPBOX_TOKEN','')}' }};"

@app.get("/tiles/24h/latest/{z}/{x}/{y}.png")
def tiles_24h_latest(z:int,x:int,y:int,max:float|None=Query(None)):
    d=_resolve_latest_nsidc_day()
    if d is None:raise HTTPException(status_code=503,detail="No recent NSIDC day available")
    if max is None:
        try:
            hit=_tile_cache_get("nsidc",24,d,z,x,y)
            if hit is not None:return _resp_png(hit,cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"24h-latest","X-Cache":"tilecache-hit","X-Latest-Day":d})
        except:pass
    cog=build_24h_cog(datetime.strptime(d,"%Y%m%d").date());png=_tile_png_from_cog(cog,z,x,y,max)
    if max is None:
        try:_tile_cache_put("nsidc",24,d,z,x,y,png)
        except:pass
    return _resp_png(png,cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"24h-latest","X-Cache":("tilecache-store"if max is None else"bypass-max"),"X-Latest-Day":d})

@app.get("/tiles/24h/{date_yyyymmdd:regex(\\d{8})}/{z}/{x}/{y}.png")
def tiles_24h(date_yyyymmdd:str,z:int,x:int,y:int,max:float|None=Query(None)):
    try:d=datetime.strptime(date_yyyymmdd,"%Y%m%d").date()
    except:raise HTTPException(status_code=400,detail="Date must be YYYYMMDD")
    return _resp_png(_tile_png_from_cog(build_24h_cog(d),z,x,y,max),cache_control=CACHE_CONTROL_BY_DATE,**{"X-Route":"24h-date"})

@app.get("/tiles/72h/{end_yyyymmdd:regex(\\d{8})}/{z}/{x}/{y}.png")
def tiles_72h(end_yyyymmdd:str,z:int,x:int,y:int,max:float|None=Query(None)):
    try:d=datetime.strptime(end_yyyymmdd,"%Y%m%d").date()
    except:raise HTTPException(status_code=400,detail="Date must be YYYYMMDD")
    return _resp_png(_tile_png_from_cog(build_72h_cog(d),z,x,y,max),cache_control=CACHE_CONTROL_BY_DATE,**{"X-Route":"72h"})

@app.get("/forecast/ending")
def forecast_ending(days:int=Query(2,ge=2,le=3),hours:int=Query(24,ge=24,le=72)):
    now=datetime.utcnow().replace(tzinfo=timezone.utc);end_dt=_ts_to_dt(_valid_ts_for_days(days,now_utc=now))
    return{"days":int(days),"hours":int(hours),"valid":end_dt.strftime("%Y%m%d%H"),"end_iso":end_dt.isoformat(),"label":f"{int(hours)} Hour snowmelt ending on {end_dt.strftime('%Y-%m-%d %HZ')}","now_utc":now.isoformat(),"anchor_date_05z":now.date().isoformat()}

@app.get("/tiles/forecast/latest/{z}/{x}/{y}.png")
def tiles_forecast_latest(z:int,x:int,y:int,max:float|None=Query(None),days:int=Query(2,ge=2,le=3),hours:int=Query(24,ge=24,le=72)):
    z_eff,x_eff,y_eff,tier=_effective_request_tile(z,x,y)
    if tier=="none":return _resp_png(_transparent_png_256(),cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"forecast-latest","X-Allowed":"0"})
    sel=_pick_forecast_for_days(days,dom_prefer="zz")
    if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="latest forecast may not be available yet";raise HTTPException(status_code=503,detail=detail)
    valid_ymd=(sel["valid"]or"")[:8]
    if max is None and valid_ymd:
        try:
            hit=_tile_cache_get("zz",int(hours),valid_ymd,z_eff,x_eff,y_eff)
            if hit is not None:return _resp_png(hit,cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"forecast-latest","X-Allowed":"1","X-Cache":"hit","X-Forecast-Valid":sel["valid"],"X-Forecast-Hours":str(int(hours)),"X-Forecast-RunInit-TT":sel["run_init"],"X-Tier":tier})
        except:pass
    try:
        png,headers=_generate_forecast_by_date_png(z=z_eff,x=x_eff,y=y_eff,date_yyyymmdd=valid_ymd,hours=hours,dom="zz",max_in=max)
        cache_status="generated-no-cache"
        if max is None and valid_ymd:
            try:_tile_cache_put("zz",int(hours),valid_ymd,z_eff,x_eff,y_eff,png);_hf_enqueue_files([_tile_cache_path("zz",int(hours),valid_ymd,z_eff,x_eff,y_eff)]);cache_status="generated-and-cached"
            except:cache_status="generated-only"
        headers["X-Cache"]=cache_status;headers["X-Tier"]=tier;headers["X-Route"]="forecast-latest"
        return _resp_png(png,cache_control=CACHE_CONTROL_LATEST,**headers)
    except HTTPException as e:_log("WARN",f"[tiles] forecast/latest failed: {e.detail}");raise

def _is_gzip(data:bytes)->bool:return len(data)>=2 and data[0]==0x1F and data[1]==0x8B

@app.get("/cn_tiles/{z}/{x}/{y}.pbf")
def cn_tiles(z:int,x:int,y:int):
    conn=_cn_mbtiles_conn();zmax=_cn_mbtiles_maxzoom();z2,x2,y2=_mbtiles_overzoom(z,x,y,zmax)
    try:data=_mbtiles_get_tile(conn,z2,x2,y2)
    except Exception as e:_log("WARN",f"[cn_tiles] read fail z={z} x={x} y={y} -> z2={z2} x2={x2} y2={y2} err={e!r}");data=None
    if not data:return Response(status_code=404)
    headers={"Content-Type":"application/x-protobuf","Cache-Control":"public, max-age=86400"}
    if _is_gzip(data):headers["Content-Encoding"]="gzip"
    return FastResponse(content=data,headers=headers)

@app.get("/value/forecast/by_date")
def value_forecast_by_date(date_yyyymmdd:str=Query(...,regex=r"^\d{8}$"),hours:int=Query(24,ge=24,le=72),lon:float=Query(...),lat:float=Query(...),dom:str=Query("zz")):
    with _Mode("value"):
        dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05")
        if int(hours)==24:
            sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
            if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);raise HTTPException(status_code=503,detail=detail)
            melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]);mm,info=_union_point_mm(melt_cogs,float(lon),float(lat))
            inches=None if mm is None else(float(mm)/INCH_TO_MM);return{"ok":True,"hours":24,"valid":sel["valid"],"run_init":sel["run_init"],"mm":mm,"inches":inches,"info":info}
        run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom)
        if not run_init:raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom})
        melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom);mm,info=_point_mm_from_cog(melt72_cog,float(lon),float(lat))
        inches=None if mm is None else(float(mm)/INCH_TO_MM);inches_1dp=None if inches is None else round(inches,1)
        return{"ok":True,"hours":72,"valid":valid,"run_init":run_init,"mm":None if mm is None else float(mm),"inches":inches,"inches_1dp":inches_1dp,"info":info}

@app.post("/value/forecast/viewport_grid")
def value_forecast_viewport_grid(payload:dict):
    with _Mode("value"):
        try:
            date_yyyymmdd=payload.get("date_yyyymmdd");hours=int(payload.get("hours",24));dom=(payload.get("dom")or"zz").lower();bounds=payload.get("bounds");resolution=int(payload.get("resolution",20))
            if not date_yyyymmdd or not bounds or len(bounds)!=4:raise HTTPException(status_code=400,detail="Missing date_yyyymmdd or bounds")
            west,south,east,north=[float(b)for b in bounds];resolution=max(5,min(30,resolution));south=max(-85.0,min(85.0,south));north=max(-85.0,min(85.0,north))
            if east<west:west,east=east,west
            if north<south:south,north=north,south
            valid=_norm_ts(f"{date_yyyymmdd}05");qstep=0.10;qw,qs,qe,qn=_q(west,qstep),_q(south,qstep),_q(east,qstep),_q(north,qstep)
            cache_key=("v1",date_yyyymmdd,int(hours),dom,qw,qs,qe,qn,int(resolution))
            with _VIEWGRID_LOCK:hit=_mem_cache_get(_VIEWGRID_CACHE,cache_key,_VIEWGRID_TTL_S)
            if hit is not None:return hit
            lons=np.linspace(west,east,resolution);lats=np.linspace(south,north,resolution)
            if int(hours)==24:
                sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
                if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);raise HTTPException(status_code=503,detail=detail)
                melt_key=("24",dom,sel["run_init"],sel["valid"])
                with _MELTCOGS_LOCK:melt_cogs=_mem_cache_get(_MELTCOGS_CACHE,melt_key,_MELTCOGS_TTL_S)
                if melt_cogs is None:melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]);
                grid=[]
                for lat in lats:
                    row=[]
                    for lon in lons:mm,_=_union_point_mm(melt_cogs,float(lon),float(lat));row.append(None if mm is None else round(float(mm)/INCH_TO_MM,1))
                    grid.append(row)
                resp={"ok":True,"hours":24,"valid":sel["valid"],"run_init":sel["run_init"],"bounds":[west,south,east,north],"resolution":resolution,"lons":lons.tolist(),"lats":lats.tolist(),"grid":grid}
                with _VIEWGRID_LOCK:_mem_cache_put(_VIEWGRID_CACHE,cache_key,resp)
                return resp
            run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom)
            if not run_init:raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid"})
            melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom);grid=[]
            for lat in lats:
                row=[]
                for lon in lons:mm,_=_point_mm_from_cog(melt72_cog,float(lon),float(lat));row.append(None if mm is None else round(float(mm)/INCH_TO_MM,1))
                grid.append(row)
            resp={"ok":True,"hours":72,"valid":valid,"run_init":run_init,"bounds":[west,south,east,north],"resolution":resolution,"lons":lons.tolist(),"lats":lats.tolist(),"grid":grid}
            with _VIEWGRID_LOCK:_mem_cache_put(_VIEWGRID_CACHE,cache_key,resp)
            return resp
        except HTTPException:raise
        except Exception as e:_log("ERROR",f"[viewport_grid] error: {e!r}");raise HTTPException(status_code=500,detail=f"viewport_grid_error: {e!r}")

@app.get("/tiles/forecast/by_date/{z}/{x}/{y}.png")
def tiles_forecast_by_date(
    z: int, x: int, y: int,
    date_yyyymmdd: str = Query(..., regex=r"^\d{8}$"),
    hours: int = Query(24),
    dom: str = Query("zz"),
    max_in: float | None = Query(None),
    client_id: str | None = Query(None),
    gen: int | None = Query(None),
) -> Response:
    _pause_push_for_user()
    dom = (dom or "zz").lower()
    hours_i = int(hours)
    if not _tile_allowed(z, x, y):
        return _resp_png(
            _transparent_png_256(),
            cache_control="public, max-age=31536000, immutable",
            **{"X-Route": "forecast-by-date", "X-Allowed": "0"},
        )
    if client_id and gen is not None and not _client_is_active_gen(client_id, gen):
        return _resp_png(
            _transparent_png_256(),
            cache_control="no-store",
            **{"X-Route": "forecast-by-date", "X-Stale": "gen", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i)},
        )
    if client_id and not _client_is_active_date(client_id, date_yyyymmdd):
        return _resp_png(
            _transparent_png_256(),
            cache_control="no-store",
            **{"X-Route": "forecast-by-date", "X-Stale": "1", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i)},
        )
    cancel_check = _make_cancel_check(client_id, gen)
    try:
        cached = _tile_cache_get(dom, hours_i, date_yyyymmdd, z, x, y, cancel_check=cancel_check)
        if cached is not None:
            return _resp_png(
                cached,
                cache_control="public, max-age=31536000, immutable",
                **{"X-Route": "forecast-by-date", "X-Cache": "hit"},
            )
    except:
        pass
    if cancel_check and cancel_check():
        return _resp_png(
            _transparent_png_256(),
            cache_control="no-store",
            **{"X-Route": "forecast-by-date", "X-Stale": "gen-pre-gen", "X-Date": date_yyyymmdd},
        )

    #_log("INFO", f"[tile] generating on-demand: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}")
    png, headers = _generate_forecast_by_date_png(
        z=z, x=x, y=y, date_yyyymmdd=date_yyyymmdd,
        hours=hours_i, dom=dom, max_in=max_in,
    )
    if not _is_valid_png_bytes(png):
        _log("WARN", f"[tile] generated invalid PNG: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}")
        return _resp_png(
            _transparent_png_256(),
            cache_control="no-store",
            **{"X-Route": "forecast-by-date", "X-Bad-PNG": "1", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i)},
        )
    if cancel_check and cancel_check():
        return _resp_png(
            _transparent_png_256(),
            cache_control="no-store",
            **{"X-Route": "forecast-by-date", "X-Stale": "gen-post-gen", "X-Date": date_yyyymmdd},
        )
    try:
        _tile_cache_put(dom, hours_i, date_yyyymmdd, z, x, y, png)
    except:
        pass
    extra = dict(headers) if isinstance(headers, dict) else {}
    extra.setdefault("X-Route", "forecast-by-date")
    extra.setdefault("X-Cache", "miss")
    return _resp_png(png, cache_control="public, max-age=31536000, immutable", **extra)


@app.get("/tiles/forecast_raw/by_date/{z}/{x}/{y}.npz")
def tiles_forecast_by_date_raw(
    z: int, x: int, y: int,
    date_yyyymmdd: str = Query(..., regex=r"^\d{8}$"),
    hours: int = Query(24),
    dom: str = Query("zz"),
    client_id: str | None = Query(None),
    gen: int | None = Query(None),
) -> Response:
    _pause_push_for_user()
    if client_id and not _check_rate_limit(client_id):
        return Response(
            content=_zero_raw_npz_256(),
            media_type="application/octet-stream",
            headers={
                "Cache-Control": "no-store",
                "X-Route": "forecast-raw-by-date",
                "X-Rate-Limited": "1",
                "Retry-After": "5",
            },
        )

    dom = (dom or "zz").lower()
    hours_i = int(hours)
    if client_id and gen is not None and not _client_is_active_gen(client_id, gen):
        return Response(
            content=_zero_raw_npz_256(),
            media_type="application/octet-stream",
            headers={
                "Cache-Control": "no-store",
                "X-Route": "forecast-raw-by-date",
                "X-Stale": "gen",
                "X-Date": date_yyyymmdd,
                "X-Hours": str(hours_i),
            },
        )
    if client_id and not _client_is_active_date(client_id, date_yyyymmdd):
        return Response(
            content=_zero_raw_npz_256(),
            media_type="application/octet-stream",
            headers={
                "Cache-Control": "no-store",
                "X-Route": "forecast-raw-by-date",
                "X-Stale": "1",
                "X-Date": date_yyyymmdd,
                "X-Hours": str(hours_i),
            },
        )
    cancel_check = _make_cancel_check(client_id, gen)
    try:
        cached = _raw_tile_cache_get(dom, hours_i, date_yyyymmdd, z, x, y, cancel_check=cancel_check)
        if cached is not None:
            if not _validate_npz_bytes(cached):
                _log("WARN", f"[tile_raw] Cached NPZ failed header check: {date_yyyymmdd} z{z} x{x} y{y}")
                try:
                    _safe_unlink(_raw_tile_cache_path(dom, hours_i, date_yyyymmdd, z, x, y))
                except:
                    pass
            else:
                return Response(
                    content=cached,
                    media_type="application/octet-stream",
                    headers={
                        "Cache-Control": "public, max-age=31536000, immutable",
                        "X-Route": "forecast-raw-by-date",
                        "X-Cache": "hit",
                    },
                )
    except Exception as e:
        _log("WARN", f"[tile_raw] cache-read error: {e!r}")
    if cancel_check and cancel_check():
        return Response(
            content=_zero_raw_npz_256(),
            media_type="application/octet-stream",
            headers={
                "Cache-Control": "no-store",
                "X-Route": "forecast-raw-by-date",
                "X-Stale": "gen-pre-gen",
                "X-Date": date_yyyymmdd,
            },
        )
    #_log("INFO", f"[tile_raw] generating on-demand: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}")
    raw, headers = _generate_forecast_by_date_raw(
        z=z, x=x, y=y, date_yyyymmdd=date_yyyymmdd,
        hours=hours_i, dom=dom,
    )
    if not _validate_npz_bytes(raw):
        _log("WARN", f"[tile_raw] generated invalid NPZ header: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}")
        return Response(
            content=_zero_raw_npz_256(),
            media_type="application/octet-stream",
            headers={
                "Cache-Control": "no-store",
                "X-Route": "forecast-raw-by-date",
                "X-Bad-NPZ": "1",
            },
        )
    if cancel_check and cancel_check():
        return Response(
            content=_zero_raw_npz_256(),
            media_type="application/octet-stream",
            headers={
                "Cache-Control": "no-store",
                "X-Route": "forecast-raw-by-date",
                "X-Stale": "gen-post-gen",
                "X-Date": date_yyyymmdd,
            },
        )
    try:
        _raw_tile_cache_put(dom, hours_i, date_yyyymmdd, z, x, y, raw)
    except Exception as e:
        _log("WARN", f"[tile_raw] failed to write raw cache for {date_yyyymmdd} z{z} x{x} y{y}: {e!r}")
    extra = dict(headers) if isinstance(headers, dict) else {}
    h = {
        "Cache-Control": "public, max-age=31536000, immutable",
        "X-Route": "forecast-raw-by-date",
        "X-Cache": "miss",
    }
    for k, v in extra.items():
        try:
            h[str(k)] = str(v)
        except:
            pass
    return Response(content=raw, media_type="application/octet-stream", headers=h)

@app.get("/__debug/record_all_sigs")
def __debug_record_all_sigs():
    dom=(WARM_DOM or"zz").lower()
    window=_rolling_dates_local_window(lookback=2,lookahead=3)
    recorded=[]
    for ymd in window:
        best=_find_best_melt_for_date(ymd,dom_prefer=dom)
        if best.get("ok")and best.get("melt_urls"):
            _record_baked_sig(ymd,dom,best["melt_urls"][0],source_type=best.get("source_type"))
            recorded.append({"date":ymd,"source_type":best.get("source_type"),"url":best["melt_urls"][0]})
        else:
            recorded.append({"date":ymd,"error":best.get("error","no melt found")})
    return{"recorded":recorded}

@app.get("/tiles/forecast/{ts_key:regex(\\d{8}(?:[_-]?\\d{2}))}/{z}/{x}/{y}.png")
def tiles_forecast(ts_key:str,z:int,x:int,y:int,max:float|None=Query(None),lead:str=Query("t0024"),valid:str|None=Query(None),dom:str=Query("zz")):
    if not _tile_allowed(z,x,y):return _resp_png(_transparent_png_256(),**{"X-Route":"forecast-direct","X-Allowed":"0"})
    ts_key=_norm_ts(ts_key);lead=(lead or"").lower()
    if lead!="t0024":raise HTTPException(status_code=400,detail="This service now serves direct 24h melt files (t0024). Use lead=t0024.")
    if not valid:raise HTTPException(status_code=400,detail="valid is required (TSYYYYMMDDHH) for direct 24h melt")
    valid=_norm_ts(valid);melt_urls=_find_melt_11044_t0024(ts_key,valid,dom_prefer=dom)
    if not melt_urls:raise HTTPException(status_code=404,detail=f"No melt file for TT={ts_key} TS={valid} lead=t0024")
    sd=_best_state_nc_near_valid("ssmv11036",valid,dom_prefer=dom)
    if not sd.get("url"):raise HTTPException(status_code=404,detail="No snowdepth 11036 state .nc found")
    resp=_serve_forecast_tile(ts_key,valid,"t0024",melt_urls,sd.get("picked_ts")or"",[sd["url"]],z,x,y,max);resp.headers["X-Allowed"]="1";return resp

@app.get("/forecast/latest_ts")
def forecast_latest_ts(dom:str="zz"):
    now=datetime.utcnow().replace(tzinfo=timezone.utc);sel2=_pick_forecast_for_days(2,dom_prefer=dom);sel3=_pick_forecast_for_days(3,dom_prefer=dom)
    return{"now_utc":now.isoformat(),"d2":sel2,"d3":sel3,"collab_last":dict(_COLLAB_LAST)}

@app.get("/forecast/selection")
def forecast_selection(dom:str="zz"):return forecast_latest_ts(dom=dom)

@app.post("/value/forecast/by_date_batch")
def value_forecast_by_date_batch(payload:dict):
    with _Mode("value"):
        try:
            date_yyyymmdd=payload.get("date_yyyymmdd");hours=int(payload.get("hours",24));dom=(payload.get("dom")or"zz").lower();pts=payload.get("pts",[])
            if not date_yyyymmdd or not pts:raise HTTPException(status_code=400,detail="Missing date_yyyymmdd or pts")
            valid=_norm_ts(f"{date_yyyymmdd}05")
            if int(hours)==24:
                sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
                if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);raise HTTPException(status_code=503,detail=detail)
                melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]);values=[]
                for pt in pts:
                    if not isinstance(pt,(list,tuple))or len(pt)<2:values.append(None);continue
                    lon,lat=float(pt[0]),float(pt[1]);mm,info=_union_point_mm(melt_cogs,lon,lat)
                    if mm is None:values.append(None)
                    else:inches=float(mm)/INCH_TO_MM;values.append({"mm":float(mm),"inches":inches,"inches_1dp":round(inches,1)})
                return{"ok":True,"hours":24,"valid":sel["valid"],"run_init":sel["run_init"],"count":len(pts),"values":values}
            run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom)
            if not run_init:raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom})
            melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom);values=[]
            for pt in pts:
                if not isinstance(pt,(list,tuple))or len(pt)<2:values.append(None);continue
                lon,lat=float(pt[0]),float(pt[1]);mm,info=_point_mm_from_cog(melt72_cog,lon,lat)
                if mm is None:values.append(None)
                else:inches=float(mm)/INCH_TO_MM;values.append({"mm":float(mm),"inches":inches,"inches_1dp":round(inches,1)})
            return{"ok":True,"hours":72,"valid":valid,"run_init":run_init,"count":len(pts),"values":values}
        except HTTPException:raise
        except Exception as e:_log("ERROR",f"[batch] unexpected error: {e!r}");raise HTTPException(status_code=500,detail=f"batch_error: {e!r}")

@app.get("/forecast/avail")
def forecast_avail(lead:str="t0024"):
    html=_collab_dirlist();names=_hrefs(html);lead=(lead or"").lower();tts:list[str]=[]
    for n in names:
        base=n.split("?",1)[0].strip();low=base.lower()
        if"ssmv11044"not in low:continue
        if lead and lead not in low:continue
        if not _is_data_name(low):continue
        tt=_tt_from_name(base)
        if tt:tts.append(tt)
    tts=sorted(set(tts));return{"lead":lead,"count":len(tts),"min_tt":tts[0]if tts else None,"max_tt":tts[-1]if tts else None,"sample_tail":tts[-10:]if len(tts)>10 else tts}

_AVAIL_CACHE={"ts":0.0,"key":None,"resp":None}

def _available_dates_from_hf_repo(dom:str,hours:int,lookback_days:int)->list[str]:
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return[]
    try:files=api.list_repo_files(repo_id=repo,repo_type="dataset",token=tok)
    except Exception as e:_log("WARN",f"[available_days] hf list_repo_files failed: {e!r}");return[]
    now=datetime.utcnow().replace(tzinfo=timezone.utc);min_ymd=(now.date()-timedelta(days=int(lookback_days))).strftime("%Y%m%d")
    prefix=f"tilecache/{dom}/{int(hours)}/";out=set()
    for f in files:
        if not f.startswith(prefix):continue
        parts=f.split("/")
        if len(parts)<4:continue
        ymd=parts[3]
        if len(ymd)==8 and ymd.isdigit()and ymd>=min_ymd:out.add(ymd)
    return sorted(out)

_HF_FILESET_CACHE={"ts":0.0,"files":None}

def _hf_repo_files_cached(ttl_s:float=60.0)->list[str]:
    now=time.time()
    if _HF_FILESET_CACHE["files"]is not None and(now-_HF_FILESET_CACHE["ts"])<ttl_s:return _HF_FILESET_CACHE["files"]
    tok,repo=_hf_cfg();api=_hf_api()
    if not tok or not repo or api is None:return[]
    try:files=api.list_repo_files(repo_id=repo,repo_type="dataset",token=tok)
    except Exception as e:_log("WARN",f"[hf_inventory] list_repo_files failed: {e!r}");return[]
    _HF_FILESET_CACHE["ts"]=now;_HF_FILESET_CACHE["files"]=files;return files

def _hf_has_any_for_day(dom:str,hours:int,ymd:str)->bool:
    dom=(dom or"zz").lower();hours=int(hours);pref_png=f"tilecache/{dom}/{hours}/{ymd}/";pref_raw=f"rawtilecache/{dom}/{hours}/{ymd}/"
    files=_hf_repo_files_cached(ttl_s=120.0)or[]
    for f in files:
        if not f:continue
        f=f.strip()
        if f.startswith(pref_png)or f.startswith(pref_raw):return True
        parts=f.split("/",1)
        if len(parts)==2:
            tail=parts[1]
            if tail.startswith(pref_png)or tail.startswith(pref_raw):return True
    try:
        tok,repo=_hf_cfg()
        if not repo:return False
        api=HfApi()
        for c in[f"tilecache/{dom}/{hours}/{ymd}/0/0/0.png",f"tilecache/{dom}/{hours}/{ymd}/1/0/0.png",f"tilecache/{dom}/{hours}/{ymd}/2/0/0.png"]:
            try:hf_hub_download(repo_id=repo,filename=c,repo_type="dataset",token=tok,repo_type_kwargs=None);return True
            except:continue
    except Exception as e:_log("WARN",f"[hf_probe] error probing HF for day {ymd}: {e!r}")
    return False

@app.get("/forecast/available_days")
def forecast_available_days(dom: str = Query("zz"), lookback_days: int = Query(1, ge=0, le=7), lookahead_days: int = Query(3, ge=1, le=5)):
    cache_key = (dom, int(lookback_days), int(lookahead_days))
    if _AVAILABLE_DATES_CACHE.get("data") is not None and _AVAILABLE_DATES_CACHE.get("key") == cache_key and (time.time() - float(_AVAILABLE_DATES_CACHE.get("ts", 0.0))) < _AVAILABLE_DATES_CACHE_TTL:
        return _AVAILABLE_DATES_CACHE["data"]
    window_dates = _rolling_window_dates(lookback_days=int(lookback_days), lookahead_days=int(lookahead_days))
    local_status = {}
    for ymd in window_dates:
        has_24 = _check_date_has_local_tiles(dom, 24, ymd)
        has_72 = _check_date_has_local_tiles(dom, 72, ymd)
        local_status[ymd] = {"has_24h": has_24, "has_72h": has_72, "has_any": (has_24 or has_72)}
    now_local = datetime.now(LOCAL_TZ)
    resp = {"ok": True, "now_local": now_local.isoformat(), "today": now_local.strftime("%Y%m%d"), "lookback_days": int(lookback_days), "lookahead_days": int(lookahead_days), "available_dates": window_dates, "local_tile_status": local_status, "source": "rolling_window", "dom": dom}
    _AVAILABLE_DATES_CACHE["ts"] = time.time(); _AVAILABLE_DATES_CACHE["key"] = cache_key; _AVAILABLE_DATES_CACHE["data"] = resp
    return resp

@app.get("/__debug/inspect_snow")
def __debug_inspect_snow(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")):
    def _safe(v):
        try:
            f=float(v)
            if not math.isfinite(f):return None
            return f
        except:return None
    dom=(dom or"zz").lower()
    sel=_pick_forecast_for_valid(f"{date}05",dom_prefer=dom)
    if not sel.get("ok"):return{"ok":False,"error":"pick_failed","detail":str(sel)}
    snow_info={"snowpack_urls":sel.get("snowpack_urls",[]),"snowpack_ts":sel.get("snowpack_ts",""),"run_init":sel.get("run_init",""),"source_type":sel.get("source_type","")}
    # Check snowpack COG if it exists
    valid=sel.get("valid",f"{date}05")
    snow_cog_name=f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif"
    snow_cog=_forecast_dir()/snow_cog_name
    cog_info={"exists":snow_cog.exists(),"name":snow_cog_name}
    if snow_cog.exists():
        try:
            with rasterio.open(snow_cog.as_posix())as src:
                cog_info["crs"]=str(src.crs)
                cog_info["transform"]=str(src.transform)
                cog_info["width"]=src.width
                cog_info["height"]=src.height
                cog_info["bounds"]=list(src.bounds)
        except Exception as e:cog_info["error"]=repr(e)
    # Also check the melt COG
    melt_cog_name=f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif"
    melt_cog=_forecast_dir()/melt_cog_name
    melt_info={"exists":melt_cog.exists(),"name":melt_cog_name}
    if melt_cog.exists():
        try:
            with rasterio.open(melt_cog.as_posix())as src:
                melt_info["crs"]=str(src.crs)
                melt_info["transform"]=str(src.transform)
                melt_info["width"]=src.width
                melt_info["height"]=src.height
                melt_info["bounds"]=list(src.bounds)
        except Exception as e:melt_info["error"]=repr(e)
    return{"ok":True,"date":date,"selection":snow_info,"snow_cog":cog_info,"melt_cog":melt_info}

@app.get("/__debug/prewarm_cn")
def __debug_prewarm_cn(days:int=Query(2,ge=2,le=3),hours:int=Query(24,ge=24,le=72),dom:str=Query("zz"),zmin:int=Query(6,ge=0,le=14),zmax:int=Query(12,ge=0,le=14),miles:float=Query(8.0,ge=0.0,le=50.0),cap_per_zoom:int=Query(6000,ge=100,le=20000),max_tiles_total:int=Query(20000,ge=100,le=100000)):
    dom=(dom or"zz").lower();now=datetime.utcnow().replace(tzinfo=timezone.utc);valid=_valid_ts_for_days(int(days),now_utc=now)
    sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
    if not sel.get("ok"):return{"ok":False,"error":"forecast_selection_failed","detail":sel}
    try:return _bake_corridor_tiles_for_date_optimized(date_yyyymmdd=sel["valid"][:8],dom=dom,hours_list=[int(hours)],zmin=int(zmin),zmax=int(zmax),miles=float(miles),cap_per_zoom=int(cap_per_zoom),max_tiles_total=int(max_tiles_total))
    except Exception as e:return{"ok":False,"error":f"{e!r}"}

@app.get("/__debug/roll_now")
def __debug_roll_now():return _run_daily_roll_job_impl()

@app.get("/__logs")
def __logs(since_ms:int|None=None,limit:int=200):
    try:limit=max(1,min(int(limit),1000))
    except:limit=200
    rows=list(_LOG)
    if since_ms is not None:
        try:rows=[r for r in rows if r["ts"]>int(since_ms)]
        except:pass
    return{"count":len(rows[-limit:]),"items":rows[-limit:]}

@lru_cache(maxsize=8000)
def _tile_png_cached(sig: str, z: int, x: int, y: int, max_in: float | None) -> bytes:
    name = sig.split(":", 1)[0]
    p = _forecast_dir().parent / "forecast" / name if name.endswith(".tif") else None
    if p is None or not p.exists():
        raise HTTPException(status_code=500, detail="cached tile missing COG")
    return _tile_png_from_cog(p, z, x, y, max_in)

class ActiveDatePayload(BaseModel):
    client_id: str | None = None
    date_yyyymmdd: str | None = None
    gen: int | None = None

@app.post("/ui/active_date")
def ui_active_date(payload: ActiveDatePayload = Body(...)):
    client_id = (payload.client_id or "").strip()
    date_yyyymmdd = (payload.date_yyyymmdd or "").strip()
    gen = payload.gen
    if not client_id:
        raise HTTPException(status_code=422, detail="client_id required")
    if not re.match(r"^\d{8}$", date_yyyymmdd):
        raise HTTPException(status_code=422, detail="date_yyyymmdd must be YYYYMMDD")
    _pause_push_for_user()
    _client_mark_active(client_id, date_yyyymmdd, gen=gen)
    return {"ok": True, "client_id": client_id, "active_date": date_yyyymmdd, "gen": gen}

@app.get("/__debug/purge_bad_npz")
def __debug_purge_bad_npz(dry_run:bool=Query(True),check_hf_repo:bool=Query(False),min_valid_size:int=Query(100)):
    results={"ok":True,"dry_run":dry_run,"local_cache":{"path":None,"exists":False,"found":0,"files":[]},"hf_repo":{"enabled":check_hf_repo,"found":0,"files":[]}}
    raw_cache=CACHE/"rawtilecache";results["local_cache"]["path"]=str(raw_cache)
    if raw_cache.exists():
        results["local_cache"]["exists"]=True;bad_files=[]
        for p in raw_cache.rglob("*.npz"):
            try:
                size=p.stat().st_size
                if size<min_valid_size:bad_files.append({"path":str(p.relative_to(CACHE)),"size":size})
                if not dry_run:p.unlink(missing_ok=True)
            except Exception as e:bad_files.append({"path":str(p),"error":str(e)})
        results["local_cache"]["found"]=len(bad_files);results["local_cache"]["files"]=bad_files[:50]
    if check_hf_repo:
        tok,repo=_hf_cfg()
        if tok and repo:
            try:
                api=HfApi();files=api.list_repo_files(repo_id=repo,repo_type="dataset",token=tok)
                npz_files=[f for f in files if f.startswith("rawtilecache/")and f.endswith(".npz")]
                results["hf_repo"]["total_npz_files"]=len(npz_files);results["hf_repo"]["sample_files"]=npz_files[:20]
                results["hf_repo"]["message"]="To check file sizes in HF repo, you need to use the HuggingFace web UI or download files individually."
            except Exception as e:results["hf_repo"]["error"]=str(e)
        else:results["hf_repo"]["error"]="HF not configured (missing HF_TOKEN or HF_DATASET_REPO)"
    tile_cache=CACHE/"tilecache"
    if tile_cache.exists():
        bad_pngs=[]
        for p in tile_cache.rglob("*.png"):
            try:
                size=p.stat().st_size
                if size<67:bad_pngs.append({"path":str(p.relative_to(CACHE)),"size":size})
                if not dry_run:p.unlink(missing_ok=True)
            except:pass
        if bad_pngs:results["tilecache_png"]={"found":len(bad_pngs),"files":bad_pngs[:50]}
    return results

@app.get("/__debug/cache_paths")
def __debug_cache_paths():
    paths={"CACHE_ROOT":str(CACHE),"tilecache":str(CACHE/"tilecache"),"rawtilecache":str(CACHE/"rawtilecache"),"forecast":str(CACHE/"forecast"),"tracks":str(CACHE/"tracks")}
    status={}
    for name,path in paths.items():
        p=Path(path)
        if p.exists():
            if p.is_dir():
                try:file_count=sum(1 for _ in p.rglob("*")if _.is_file());status[name]={"exists":True,"is_dir":True,"file_count":file_count}
                except Exception as e:status[name]={"exists":True,"is_dir":True,"error":str(e)}
            else:status[name]={"exists":True,"is_dir":False,"size":p.stat().st_size}
        else:status[name]={"exists":False}
    tok,repo=_hf_cfg();hf_info={"configured":bool(tok and repo),"repo":repo,"has_token":bool(tok)}
    return{"paths":paths,"status":status,"hf_repo":hf_info}

@app.get("/__debug/rebuild_cog")
def __debug_rebuild_cog(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz"),wipe_tiles:bool=Query(True)):
    dom=(dom or"zz").lower();valid=f"{date}05";deleted=[]
    for pattern in[
        f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",
        f"fcst_melt72h_end_{valid}_{dom}_cog.tif",
        f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",
        f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",
        f"fcst_melt_t0024_*_{valid}_*_cog.tif",
    ]:
        fc=_forecast_dir()
        for p in fc.glob(pattern):
            try:COG_CACHE.evict(p)
            except:pass
            try:p.unlink(missing_ok=True);deleted.append(p.name)
            except Exception as e:deleted.append(f"{p.name}: FAIL {e!r}")
    tiles_deleted=0
    if wipe_tiles:
        for kind in["tilecache","rawtilecache"]:
            for h in WARM_HOURS_LIST:
                d=CACHE/kind/dom/str(h)/date
                if d.exists():
                    try:shutil.rmtree(d,ignore_errors=True);tiles_deleted+=1
                    except:pass
    return{"deleted_cogs":deleted,"tiles_dirs_deleted":tiles_deleted,"date":date,"valid":valid,"dom":dom,"message":"COGs and tiles deleted. Next request will rebuild everything."}


@app.get("/__debug/inspect_melt_nc")
def __debug_inspect_melt_nc(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")):
    def _safe(v):
        try:
            f=float(v)
            if not math.isfinite(f):return None
            return f
        except:return None
    dom=(dom or"zz").lower();valid=_norm_ts(f"{date}05")
    sel=_pick_forecast_for_valid(valid,dom_prefer=dom)
    if not sel.get("ok"):return{"ok":False,"error":"pick_failed","detail":str(sel)}
    url=sel["melt_urls"][0]if sel.get("melt_urls")else None
    if not url:return{"ok":False,"error":"no_melt_url"}
    with tempfile.TemporaryDirectory()as td:
        desc,name=_download_url_to_grid(url,Path(td))
        da=_open_grid(desc)
        attrs=dict(da.attrs or{})
        units=str(attrs.get("units","")).strip()
        vals=da.values.astype("float32");valid_mask=np.isfinite(vals)&(vals>-9990.0)&(vals<100000.0)
        valid_vals=vals[valid_mask]
        raw_info={"type":desc.get("type"),"note":"not netcdf"}
        if desc.get("type")=="netcdf":
            try:
                ds=xr.open_dataset(desc["path"].as_posix(),engine="netcdf4")
                raw_var=_pick_netcdf_raster_var(ds)
                raw_attrs=dict(raw_var.attrs or{})
                raw_vals=raw_var.values.astype("float32").ravel()
                raw_valid=raw_vals[np.isfinite(raw_vals)&(raw_vals>-9990.0)&(raw_vals<1e20)]
                raw_info={"var_name":raw_var.name,"units":str(raw_attrs.get("units","")),"raw_mean":_safe(np.mean(raw_valid)),"raw_max":_safe(np.max(raw_valid)),"raw_min":_safe(np.min(raw_valid)),"raw_median":_safe(np.median(raw_valid)),"shape":list(raw_var.shape),"all_attrs":{k:str(v)for k,v in raw_attrs.items()}}
                ds.close()
            except Exception as e:
                raw_info={"error":repr(e)}
        samples={}
        for label,lon,lat in[("Rouses_Point",-73.365,44.991),("Ogdensburg",-75.486,44.694),("Belleville_ON",-77.385,44.163),("Montreal",-73.567,45.502),("Ottawa",-75.698,45.421)]:
            try:
                px=da.sel(x=lon,y=lat,method="nearest").values
                v=float(px)
                samples[label]={"mm":_safe(v),"inches":_safe(v/25.4)if np.isfinite(v)else None}
            except Exception as e:
                samples[label]={"error":str(e)}
    return{
        "ok":True,"url":url,"filename":name,"valid":valid,"run_init":sel.get("run_init"),"source_type":"ttna"if"ttnats"in url.lower()else"forecast",
        "after_open_grid":{"units":units,"mean_mm":_safe(np.mean(valid_vals)),"max_mm":_safe(np.max(valid_vals)),"min_mm":_safe(np.min(valid_vals)),"median_mm":_safe(np.median(valid_vals)),"valid_pixels":int(valid_mask.sum()),"total_pixels":int(vals.size),"attrs":{k:str(v)for k,v in attrs.items()}},
        "raw_file":raw_info,
        "point_samples_after_processing":samples,
    }

# def _hot_swap_rebuild_date(date:str,dom:str)->None:
#     """Rebuild tiles from latest data, then swap out old tiles atomically.
#     Users always see baked tiles during the transition."""
#     dom=(dom or"zz").lower()
#     valid_05=f"{date}05";valid_11=f"{date}11"
    
#     # 1. Build new COGs from latest NOAA data (old tiles still being served)
#     _log("INFO",f"[hot_swap] {date}: building fresh COGs")
#     # Delete only COGs, not tiles
#     for valid in[valid_05,valid_11]:
#         for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]:
#             for p in _forecast_dir().glob(pattern):
#                 try:COG_CACHE.evict(p)
#                 except:pass
#                 try:p.unlink(missing_ok=True)
#                 except:pass
#     # Also delete 72h COGs for next 2 dates
#     for offset in[1,2]:
#         try:
#             fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
#             for fv in[f"{fd}05",f"{fd}11"]:
#                 for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"):
#                     try:COG_CACHE.evict(p)
#                     except:pass
#                     try:p.unlink(missing_ok=True)
#                     except:pass
#         except:pass
#     # Clear negative cache
#     with _HF_NEGATIVE_CACHE_LK:
#         to_del=[k for k in _HF_NEGATIVE_CACHE if date in k]
#         for k in to_del:del _HF_NEGATIVE_CACHE[k]
#     best=_find_best_melt_for_date(date,dom_prefer=dom)
#     _log("INFO",f"[hot_swap] {date}: selected source_type={best.get('source_type','?')} valid={best.get('valid','?')}")
#     actual_valid=best.get("valid",f"{date}05")
#     r=_ensure_forecast_for_valid(actual_valid,dom_prefer=dom)
#     if not r.get("ok"):_log("WARN",f"[hot_swap] {date}: COG build failed: {r}");return
    
#     # 2. Delete old tiles so bake doesn't skip them
#     _log("INFO",f"[hot_swap] {date}: deleting old tiles")
#     for kind in["tilecache","rawtilecache"]:
#         for h in WARM_HOURS_LIST:
#             d=CACHE/kind/dom/str(h)/date
#             if d.exists():shutil.rmtree(d,ignore_errors=True)
    
#     # 3. Bake NEW tiles
#     _log("INFO",f"[hot_swap] {date}: baking new tiles")

#     with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool:
#         futs={}
#         for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]:
#             futs[label]=pool.submit(fn,date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra)
#         for label,fut in futs.items():
#             try:
#                 r=fut.result();_log("INFO",f"[hot_swap] {date}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
#             except Exception as e:_log("WARN",f"[hot_swap] {date}: {label} failed: {e!r}")
#     for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
#         try:fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra)
#         except:pass

#     for offset in[1,2]:
#         try:
#             fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
#             fb=_find_best_melt_for_date(fd,dom_prefer=dom)
#             fv=fb.get("valid",f"{fd}05")if fb.get("ok")else f"{fd}05"
#             try:_forecast_melt72h_end_cog(fv,dom_prefer=dom)
#             except:pass
#             _log("INFO",f"[hot_swap] rebaking 72h for {fd}")
#             with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool:
#                 for fn,extra in[(_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),(_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]:
#                     pool.submit(fn,date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra)
#             for fn,extra in[(_bake_lowres_box_tiles_for_date,dict(conc=1)),(_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
#                 try:fn(date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra)
#                 except:pass
#         except Exception as e:_log("WARN",f"[hot_swap] 72h rebake +{offset} failed: {e!r}")
    
#     # 4. NOW delete old HF archives and push new ones
#     _log("INFO",f"[hot_swap] {date}: deleting old HF archives")
#     tok,repo=_hf_cfg();api=_hf_api()
#     if tok and repo and api is not None:
#         # Delete old archives
#         for kind in["tilecache","rawtilecache"]:
#             for h in WARM_HOURS_LIST:
#                 ap=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz"
#                 try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old {ap}")
#                 except:pass
#         # Delete 72h archives for next 2 dates
#         for offset in[1,2]:
#             try:
#                 fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
#                 for kind in["tilecache","rawtilecache"]:
#                     ap=f"archives/{kind}_{dom}_h72_{fd}.tar.gz"
#                     try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old 72h {ap}")
#                     except:pass
#             except:pass
#     # Delete tar markers
#     marker_dir=CACHE/"_tar_markers"
#     if marker_dir.exists():
#         for mk in marker_dir.glob(f"*_{date}.tar.gz.done"):
#             try:mk.unlink(missing_ok=True)
#             except:pass
    
#     # 5. Push new tiles
#     push_dates=[date]
#     for offset in[1,2]:
#         try:push_dates.append((datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d"))
#         except:pass
#     _log("INFO",f"[hot_swap] pushing new tiles for {push_dates}")
#     _hf_push_baked_as_tar(push_dates,dom,WARM_HOURS_LIST)
    
#     # 6. Record signature
#     if best.get("ok")and best.get("melt_urls"):
#         _record_baked_sig(date,dom,best["melt_urls"][0],source_type=best.get("source_type"))
#     _log("INFO",f"[hot_swap] {date}: complete (source={best.get('source_type','?')})")

def _hot_swap_rebuild_date(date:str,dom:str)->None:
    dom=(dom or"zz").lower()
    valid_05=f"{date}05";valid_11=f"{date}11"
    
    # 1. Build new COGs from latest NOAA data
    #    Old COGs are deleted so they get rebuilt from fresh NOAA data.
    #    Old TILES are NOT deleted yet — they keep serving users.
    _log("INFO",f"[hot_swap] {date}: building fresh COGs")
    for valid in[valid_05,valid_11]:
        for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]:
            for p in _forecast_dir().glob(pattern):
                try:COG_CACHE.evict(p)
                except:pass
                try:p.unlink(missing_ok=True)
                except:pass
    for offset in[1,2]:
        try:
            fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
            for fv in[f"{fd}05",f"{fd}11"]:
                for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"):
                    try:COG_CACHE.evict(p)
                    except:pass
                    try:p.unlink(missing_ok=True)
                    except:pass
        except:pass
    with _HF_NEGATIVE_CACHE_LK:
        to_del=[k for k in _HF_NEGATIVE_CACHE if date in k]
        for k in to_del:del _HF_NEGATIVE_CACHE[k]
    best=_find_best_melt_for_date(date,dom_prefer=dom)
    _log("INFO",f"[hot_swap] {date}: selected source_type={best.get('source_type','?')} valid={best.get('valid','?')}")
    actual_valid=best.get("valid",f"{date}05")
    r=_ensure_forecast_for_valid(actual_valid,dom_prefer=dom)
    if not r.get("ok"):_log("WARN",f"[hot_swap] {date}: COG build failed: {r}");return
    
    # 2. Bake NEW tiles into staging directories
    #    Old tiles remain on disk at the real paths, still serving users.
    _log("INFO",f"[hot_swap] {date}: baking new tiles to staging (old tiles still serving)")
    staging_root=CACHE/"_staging_swap"/date
    try:
        if staging_root.exists():shutil.rmtree(staging_root,ignore_errors=True)
    except:pass
    
    old_dirs=[]
    for kind in["tilecache","rawtilecache"]:
        for h in WARM_HOURS_LIST:
            d=CACHE/kind/dom/str(h)/date
            if d.exists():
                backup=CACHE/"_staging_swap"/f"{kind}_{dom}_{h}_{date}_old"
                backup.parent.mkdir(parents=True,exist_ok=True)
                try:
                    # Rename is atomic on same filesystem — instant, no downtime
                    d.rename(backup)
                    old_dirs.append((d,backup))
                    _log("DEBUG",f"[hot_swap] moved {d} -> {backup}")
                except Exception as e:
                    # If rename fails (cross-device?), fall back to just deleting
                    _log("WARN",f"[hot_swap] rename failed, deleting old tiles: {e!r}")
                    shutil.rmtree(d,ignore_errors=True)
    
    # 3. Bake tiles — they go to the real paths (which are now empty)
    #    PNG first so users see visual tiles sooner, then raw for mouse readouts
    _log("INFO",f"[hot_swap] {date}: baking corridor PNG tiles")
    try:
        r=_bake_corridor_tiles_for_date_optimized(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)
        _log("INFO",f"[hot_swap] {date}: corridor_png ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
    except Exception as e:_log("WARN",f"[hot_swap] {date}: corridor_png failed: {e!r}")
 
    _log("INFO",f"[hot_swap] {date}: baking corridor raw tiles")
    try:
        r=_bake_corridor_raw_tiles_for_date_optimized(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)
        _log("INFO",f"[hot_swap] {date}: corridor_raw ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}")
    except Exception as e:_log("WARN",f"[hot_swap] {date}: corridor_raw failed: {e!r}")
 
    for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
        try:fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra)
        except:pass
 
    # 4. Clean up old backup dirs (new tiles are now in place)
    for orig,backup in old_dirs:
        try:shutil.rmtree(backup,ignore_errors=True)
        except:pass
    try:
        if staging_root.exists():shutil.rmtree(staging_root,ignore_errors=True)
    except:pass
    
    # 5. Rebake 72h for future dates (+1, +2)
    for offset in[1,2]:
        try:
            fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
            fb=_find_best_melt_for_date(fd,dom_prefer=dom)
            fv=fb.get("valid",f"{fd}05")if fb.get("ok")else f"{fd}05"
            try:_forecast_melt72h_end_cog(fv,dom_prefer=dom)
            except:pass
            _log("INFO",f"[hot_swap] rebaking 72h for {fd}")
            # Delete old 72h tiles so bake doesn't skip them
            for kind in["tilecache","rawtilecache"]:
                d=CACHE/kind/dom/"72"/fd
                if d.exists():
                    shutil.rmtree(d,ignore_errors=True)
                    _log("DEBUG",f"[hot_swap] deleted stale 72h tiles: {d}")
            # Also delete old 72h lowres tiles
            for kind in["tilecache","rawtilecache"]:
                d=CACHE/kind/dom/"72"/fd
                # Lowres tiles are at the same path under different zoom levels,
                # so the rmtree above already handles them
            # Bake fresh 72h tiles sequentially
            for fn,extra in[(_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),(_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]:
                try:fn(date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra)
                except:pass
            for fn,extra in[(_bake_lowres_box_tiles_for_date,dict(conc=1)),(_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]:
                try:fn(date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra)
                except:pass
        except Exception as e:_log("WARN",f"[hot_swap] 72h rebake +{offset} failed: {e!r}")
    # 6. Delete old HF archives and push new ones
    _log("INFO",f"[hot_swap] {date}: deleting old HF archives")
    tok,repo=_hf_cfg();api=_hf_api()
    if tok and repo and api is not None:
        for kind in["tilecache","rawtilecache"]:
            for h in WARM_HOURS_LIST:
                ap=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz"
                try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old {ap}")
                except:pass
        for offset in[1,2]:
            try:
                fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
                for kind in["tilecache","rawtilecache"]:
                    ap=f"archives/{kind}_{dom}_h72_{fd}.tar.gz"
                    try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old 72h {ap}")
                    except:pass
            except:pass
    marker_dir=CACHE/"_tar_markers"
    if marker_dir.exists():
        for mk in marker_dir.glob(f"*_{date}.tar.gz.done"):
            try:mk.unlink(missing_ok=True)
            except:pass
    
    # 7. Push new tiles
    push_dates=[date]
    for offset in[1,2]:
        try:push_dates.append((datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d"))
        except:pass
    _log("INFO",f"[hot_swap] pushing new tiles for {push_dates}")
    _hf_push_baked_as_tar(push_dates,dom,WARM_HOURS_LIST)
    
    # 8. Record signature
    if best.get("ok")and best.get("melt_urls"):
        _record_baked_sig(date,dom,best["melt_urls"][0],source_type=best.get("source_type"))
    _log("INFO",f"[hot_swap] {date}: complete (source={best.get('source_type','?')})")
 
    

@app.get("/__debug/nuke_date")
def __debug_nuke_date(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")):
    dom=(dom or"zz").lower();deleted_cogs=[];deleted_archives=[]
    # 1. Delete local COGs (forces rebuild from NOAA)
    for valid in[f"{date}05",f"{date}11"]:
        for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]:
            for p in _forecast_dir().glob(pattern):
                try:COG_CACHE.evict(p)
                except:pass
                try:p.unlink(missing_ok=True);deleted_cogs.append(p.name)
                except:pass
    # 2. Do NOT delete tiles — they keep serving users until rebake replaces them
    tile_dirs_present=0
    for kind in["tilecache","rawtilecache"]:
        for h in WARM_HOURS_LIST:
            d=CACHE/kind/dom/str(h)/date
            if d.exists():tile_dirs_present+=1
    # 3. Delete local tar markers so they don't block re-pull
    marker_dir=CACHE/"_tar_markers"
    markers_cleared=0
    if marker_dir.exists():
        for mk in marker_dir.glob(f"*_{date}.tar.gz.done"):
            try:mk.unlink(missing_ok=True);markers_cleared+=1
            except:pass
    # 4. Delete tar archives from HF repo
    tok,repo=_hf_cfg();api=_hf_api()
    if tok and repo and api is not None:
        for kind in["tilecache","rawtilecache"]:
            for h in WARM_HOURS_LIST:
                archive_path=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz"
                try:
                    api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=archive_path)],commit_message=f"nuke: delete {archive_path}")
                    deleted_archives.append(archive_path);_log("INFO",f"[nuke] deleted HF archive: {archive_path}")
                except Exception as e:
                    err=repr(e)
                    if"404"in err or"EntryNotFound"in err or"does not exist"in err:
                        deleted_archives.append(f"{archive_path} (already gone)")
                    else:
                        deleted_archives.append(f"{archive_path}: FAIL {err}")
    # Also handle 72h for future dates
    for offset in[1,2]:
        try:
            fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")
            for fv in[f"{fd}05",f"{fd}11"]:
                for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"):
                    try:COG_CACHE.evict(p)
                    except:pass
                    try:p.unlink(missing_ok=True);deleted_cogs.append(p.name)
                    except:pass
            if tok and repo and api is not None:
                for kind in["tilecache","rawtilecache"]:
                    ap=f"archives/{kind}_{dom}_h72_{fd}.tar.gz"
                    try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"nuke: delete stale 72h {ap}")
                    except:pass
        except:pass
    # 5. Clear negative cache
    with _HF_NEGATIVE_CACHE_LK:
        to_del=[k for k in _HF_NEGATIVE_CACHE if date in k]
        for k in to_del:del _HF_NEGATIVE_CACHE[k]
    return{
        "deleted_cogs":deleted_cogs,
        "tile_dirs_kept":tile_dirs_present,
        "tiles_note":"Old tiles kept for serving — rebake will replace them",
        "hf_archives_deleted":deleted_archives,
        "tar_markers_cleared":markers_cleared,
        "negative_cache_cleared":len(to_del),
        "date":date,
        "message":f"Nuked COGs for {date} (tiles kept). Hit /__debug/rebake_date?date={date} to rebuild with fresh data."
    }

@app.get("/__debug/rebake_date")
def __debug_rebake_date(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")):
    dom=(dom or"zz").lower()
    def _do_rebake():
        try:_hot_swap_rebuild_date(date,dom)
        except Exception as e:_log("WARN",f"[rebake] {date}: fatal error: {e!r}")
    threading.Thread(target=_do_rebake,daemon=True,name=f"rebake_{date}").start()
    return{"ok":True,"date":date,"message":f"Hot-swap rebake started for {date}. Old tiles served until new ones are ready."}

@app.get("/__debug/noaa_sigs")
def __debug_noaa_sigs():
    with _NOAA_FILE_SIGS_LK:
        return{"sigs":{k:v for k,v in _NOAA_FILE_SIGS.items()}}
@app.get("/__debug/inspect_full_tile")
def __debug_inspect_full_tile(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")):
    dom=(dom or"zz").lower()
    sel=_pick_forecast_for_valid(f"{date}05",dom_prefer=dom)
    best=_find_best_melt_for_date(date,dom_prefer=dom)
    melt_valid=best.get("valid","?")
    snow_urls=sel.get("snowpack_urls",[])
    snow_ts=sel.get("snowpack_ts","?")
    # Check what snowpack COG exists
    snow_cogs=[]
    for v in[f"{date}11",f"{date}05"]:
        p=_forecast_dir()/f"fcst_snowpack_union_end_{v}_{dom}_cog.tif"
        if p.exists():snow_cogs.append({"name":p.name,"valid":v,"size":p.stat().st_size})
    return{
        "date":date,
        "melt_source":best.get("source_type","?"),
        "melt_valid":melt_valid,
        "melt_url":best.get("melt_urls",["?"])[0].split("/")[-1]if best.get("melt_urls")else"?",
        "snowpack_ts_from_sel":snow_ts,
        "snowpack_urls_count":len(snow_urls),
        "snowpack_urls":[u.split("/")[-1]for u in snow_urls[:3]],
        "snowpack_cogs_on_disk":snow_cogs,
        "run_init":best.get("run_init","?"),
    }
if __name__=="__main__":
    import uvicorn
    host=os.environ.get("HOST","0.0.0.0");port=int(os.environ.get("PORT","8000"));workers=int(os.environ.get("UVICORN_WORKERS","3")or"3")
    module=(os.environ.get("APP_MODULE","")or"").strip()
    if workers>1 and not module:raise SystemExit("UVICORN_WORKERS>1 requires APP_MODULE like 'main:app'. Example: APP_MODULE=main:app UVICORN_WORKERS=3 python main.py")
    if module:uvicorn.run(module,host=host,port=port,workers=workers,log_level="info",access_log=False)
    else:uvicorn.run(app,host=host,port=port,workers=1,log_level="info",access_log=False)