[AAAAMMJJHH] * Sans run : le plus récent disponible et complet. Écrit les GRIB bruts dans /raw/// * et un fichier .complete quand le run est entier. Affiche « RUN » en cas de succès. */ declare(strict_types=1); require __DIR__ . '/lib.php'; const AROME_SLICES = ['00H06H', '07H12H', '13H18H', '19H24H', '25H30H', '31H36H', '37H42H', '43H48H', '49H51H']; const ARPEGE_SLICES = ['000H012H', '013H024H', '025H036H', '037H048H', '049H060H', '061H072H']; function raw_dir(string $model, string $run): string { $d = data_dir() . "/raw/$model/$run"; ensure_dir($d); return $d; } /** Heures d'échéance utiles : de 0 à la fin de la fenêtre, par pas de $step, plafonnées. */ function hours_needed(int $run, int $step, ?int $cap = null): array { $h = intdiv(fetch_end() - $run, 3600); if ($cap !== null) { $h = min($h, $cap); } return $h >= 0 ? range(0, $h, $step) : []; } /* ---------------------------------------------------------------- Météo-France */ function fetch_mf(string $model, int $run): bool { $base = 'https://public-api.meteofrance.fr/previnum'; [$slices, $url] = $model === 'arome' ? [AROME_SLICES, "$base/DPPaquetAROME/v1/models/AROME/grids/0.025/packages/SP1/productARO"] : [ARPEGE_SLICES, "$base/DPPaquetARPEGE/v1/models/ARPEGE/grids/0.1/packages/SP1/productARP"]; $maxh = intdiv(fetch_end() - $run, 3600); $dir = raw_dir($model, gmdate('YmdH', $run)); $token = mf_token(); foreach ($slices as $s) { [$a] = explode('H', $s); if ((int) $a > $maxh) { break; } $out = "$dir/SP1_$s.grib2"; if (is_file($out)) { continue; } $q = '?referencetime=' . gmdate('Y-m-d\TH:00:00\Z', $run) . "&time=$s&format=grib2"; $code = http_download($url . $q, ["Authorization: Bearer $token", 'Accept: */*'], $out, 1 << 30); if ($code === 401) { /* jeton renouvelé entre-temps par le collecteur : on relit le cache */ sleep(30); $token = mf_token(); $code = http_download($url . $q, ["Authorization: Bearer $token", 'Accept: */*'], $out, 1 << 30); } logm($model, gmdate('d/H', $run), $s, (string) $code); if ($code !== 200) { /* la dernière tranche (49H51H) n'existe pas pour tous les runs : run accepté sans elle */ return $code === 404 && $s === end($slices); } sleep(1); /* quota du portail : 100 requêtes par minute, partagées avec les autres collectes */ } return true; } /* ---------------------------------------------------------------- ECMWF (IFS, AIFS) */ function fetch_ecmwf(string $model, int $run): bool { $stream = $model === 'ifs' ? 'ifs' : 'aifs-single'; $params = $model === 'ifs' ? ['msl', '10fg'] : ['msl']; $step = $model === 'ifs' ? 3 : 6; $hh = gmdate('H', $run); $ymd = gmdate('Ymd', $run); $dir = raw_dir($model, gmdate('YmdH', $run)); $todo = []; foreach (hours_needed($run, $step, in_array($hh, ['06', '18'], true) ? 90 : null) as $h) { if (!is_file(sprintf('%s/f%03d.grib2', $dir, $h))) { $todo[$h] = sprintf('https://data.ecmwf.int/forecasts/%s/%sz/%s/0p25/oper/%s%s0000-%dh-oper-fc', $ymd, $hh, $stream, $ymd, $hh, $h); } } /* 1. index de chaque échéance (une ligne JSON par champ, avec position et longueur en octets) */ $idx = http_multi(array_map(fn($b) => [$b . '.index', null], $todo), 6, 4 << 20); $ranges = []; foreach ($todo as $h => $b) { [$code, $body] = $idx[$h]; if ($code !== 200) { return false; } foreach (explode("\n", trim($body)) as $line) { $j = json_decode($line, true); if (is_array($j) && in_array($j['param'] ?? '', $params, true) && ($j['levtype'] ?? '') === 'sfc') { $o = (int) $j['_offset']; $ranges["$h:$o"] = [$b . '.grib2', $o . '-' . ($o + (int) $j['_length'] - 1)]; } } } /* 2. seulement les octets des champs utiles */ $parts = http_multi($ranges, 8, 64 << 20); $byHour = []; ksort($parts, SORT_NATURAL); foreach ($parts as $k => [$code, $body]) { if ($code !== 206 && $code !== 200) { return false; } $byHour[(int) explode(':', $k)[0]][] = $body; } foreach ($byHour as $h => $bodies) { atomic_write(sprintf('%s/f%03d.grib2', $dir, $h), implode('', $bodies)); } logm($model, gmdate('d/H', $run), 'ok', (string) count($todo), 'échéances'); return true; } /* ---------------------------------------------------------------- GFS (NOAA, S3) */ function fetch_gfs(string $model, int $run): bool { $hh = gmdate('H', $run); $ymd = gmdate('Ymd', $run); $dir = raw_dir($model, gmdate('YmdH', $run)); $todo = []; foreach (hours_needed($run, 1, 120) as $h) { if (!is_file(sprintf('%s/f%03d.grib2', $dir, $h))) { $todo[$h] = sprintf('https://noaa-gfs-bdp-pds.s3.amazonaws.com/gfs.%s/%s/atmos/gfs.t%sz.pgrb2.0p25.f%03d', $ymd, $hh, $hh, $h); } } $idx = http_multi(array_map(fn($u) => [$u . '.idx', null], $todo), 8, 4 << 20); $ranges = []; foreach ($todo as $h => $u) { [$code, $body] = $idx[$h]; if ($code !== 200) { return false; } /* .idx : « n:octet:date:VAR:niveau:... » ; un champ s'arrête où commence le suivant */ $lines = explode("\n", trim($body)); foreach ($lines as $i => $line) { if ((str_contains($line, ':PRMSL:mean sea level:') || str_contains($line, ':GUST:surface:')) && isset($lines[$i + 1])) { $a = (int) explode(':', $line)[1]; $b = (int) explode(':', $lines[$i + 1])[1] - 1; $ranges["$h:$a"] = [$u, "$a-$b"]; } } } $parts = http_multi($ranges, 8, 64 << 20); ksort($parts, SORT_NATURAL); $byHour = []; foreach ($parts as $k => [$code, $body]) { if ($code !== 206 && $code !== 200) { return false; } $byHour[(int) explode(':', $k)[0]][] = $body; } foreach ($byHour as $h => $bodies) { atomic_write(sprintf('%s/f%03d.grib2', $dir, $h), implode('', $bodies)); } logm($model, gmdate('d/H', $run), 'ok', (string) count($todo), 'échéances'); return true; } /* ---------------------------------------------------------------- ICON-EU (DWD) */ function fetch_iconeu(string $model, int $run): bool { $hh = gmdate('H', $run); $ymdh = gmdate('YmdH', $run); $dir = raw_dir($model, $ymdh); $jobs = []; foreach (hours_needed($run, 1, 78) as $h) { foreach (['pmsl', 'vmax_10m'] as $v) { $out = sprintf('%s/%s_%03d.grib2', $dir, $v, $h); if (!is_file($out)) { $jobs[$out] = [sprintf('https://opendata.dwd.de/weather/nwp/icon-eu/grib/%s/%s/icon-eu_europe_regular-lat-lon_single-level_%s_%03d_%s.grib2.bz2', $hh, $v, $ymdh, $h, strtoupper($v)), null]; } } } foreach (http_multi($jobs, 8, 64 << 20) as $out => [$code, $body]) { if ($code !== 200) { return false; } $raw = bzdecompress_limited($body, 256 << 20); if ($raw === null) { return false; } atomic_write($out, $raw); } logm($model, gmdate('d/H', $run), 'ok', (string) count($jobs), 'fichiers'); return true; } /* décompression bzip2 bornée (l'extension bz2 n'étant pas compilée, on passe par bzip2 -dc) */ function bzdecompress_limited(string $data, int $max): ?string { $p = proc_open(['bzip2', '-dc'], [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['file', '/dev/null', 'w']], $pipes); if (!is_resource($p)) { return null; } stream_set_blocking($pipes[0], false); /* jamais bloqué en écriture : on lit la sortie entre deux envois */ stream_set_blocking($pipes[1], false); $out = ''; $in = $data; while (true) { if ($in !== '') { $w = fwrite($pipes[0], substr($in, 0, 1 << 16)); $in = $w ? substr($in, $w) : $in; if ($in === '') { fclose($pipes[0]); } } $chunk = fread($pipes[1], 1 << 16); if ($chunk !== false && $chunk !== '') { $out .= $chunk; if (strlen($out) > $max) { proc_terminate($p, 9); return null; } } if ($in === '' && feof($pipes[1])) { break; } if ($chunk === '' || $chunk === false) { usleep(1000); } } fclose($pipes[1]); return proc_close($p) === 0 ? $out : null; } /* ---------------------------------------------------------------- choix du run */ function candidates(string $model): array { $c = MODELS[$model]['cycle']; $t = time() - (int) (MODELS[$model]['delay'] * 3600); $t -= $t % ($c * 3600); return array_map(fn($k) => $t - $k * $c * 3600, range(0, 3)); } function fetch_run(string $model, int $run): bool { return match ($model) { 'arome', 'arpege' => fetch_mf($model, $run), 'ifs', 'aifs' => fetch_ecmwf($model, $run), 'gfs' => fetch_gfs($model, $run), 'iconeu' => fetch_iconeu($model, $run), }; } function main(array $argv): int { $model = $argv[1] ?? ''; if (!isset(MODELS[$model])) { fwrite(STDERR, "usage : fetch.php [AAAAMMJJHH]\n"); return 2; } $runs = isset($argv[2]) ? [run_time($argv[2])] : candidates($model); foreach ($runs as $run) { $d = data_dir() . "/raw/$model/" . gmdate('YmdH', $run); $done = data_dir() . "/derived/$model/" . gmdate('YmdH', $run) . '.json'; if (is_file("$d/.complete") || is_file("$d/.decoded") || is_file($done)) { echo 'RUN ', gmdate('YmdH', $run), "\n"; return 0; } if (fetch_run($model, $run)) { touch("$d/.complete"); echo 'RUN ', gmdate('YmdH', $run), "\n"; return 0; } logm($model, gmdate('YmdH', $run), 'indisponible, run précédent'); } return 1; } exit(main($argv));