Graham McIntire
e6952a42c8
Flip the cron from every-3-hours to every-hour now that a full f00-f18 chain runs in ~45-60 min instead of ~170 min. With the :propagation queue's concurrency-of-2, a slow chain won't block the next one — they interleave, and the file store's last-writer -wins semantics give the newer run_time's analysis data naturally. Add ScoresFile.retain_window/2 that deletes any file whose valid_time falls outside [run_time, run_time + 18h]. Call it at fh=18 of the chain so leftover files from the previous chain (specifically the old f00 whose valid_time the new chain doesn't overwrite because it advanced by one hour) are discarded immediately instead of waiting for the 2h prune cron. The step transition path is extracted to handle_step_transition/2 to keep the nesting under credo's limit and make the final-step logic easier to read.
411 lines
14 KiB
Elixir
411 lines
14 KiB
Elixir
defmodule Microwaveprop.Propagation.ScoresFile do
|
||
@moduledoc """
|
||
Binary on-disk storage for propagation score grids. One file per
|
||
`(band_mhz, valid_time)` tuple, ~93 KB each. Written in the
|
||
forecast-hour hot path as a companion to the Postgres upsert, and
|
||
eventually replacing it.
|
||
|
||
## Layout
|
||
|
||
magic : 4 bytes "NTMS"
|
||
version : 1 byte 0x01
|
||
band_mhz : 4 bytes uint32 little-endian
|
||
valid_time : 8 bytes int64 unix seconds little-endian
|
||
lat_min : 4 bytes float32 little-endian
|
||
lon_min : 4 bytes float32 little-endian
|
||
step_deg : 4 bytes float32 little-endian
|
||
n_rows : 2 bytes uint16 little-endian (lat dimension)
|
||
n_cols : 2 bytes uint16 little-endian (lon dimension)
|
||
scores : n_rows × n_cols bytes (uint8, 0-100 or 255 = no-data)
|
||
|
||
A cell at `(lat, lon)` is at byte offset
|
||
`row * n_cols + col` in the scores body, where
|
||
`row = (lat - lat_min) / step` and `col = (lon - lon_min) / step`
|
||
(both rounded to the nearest integer).
|
||
|
||
## Atomic writes
|
||
|
||
Files are written through `rename(2)`: `path.tmp.<uniq>` → fsync →
|
||
rename to the final path. On the same NFSv4 filesystem the rename
|
||
is atomic, so a concurrent reader either sees the old file, the
|
||
new file, or the absence of either — never a partial file.
|
||
"""
|
||
|
||
alias Microwaveprop.Propagation.Grid
|
||
|
||
require Logger
|
||
|
||
@magic "NTMS"
|
||
@version 1
|
||
@no_data 255
|
||
|
||
@doc "Returns the root directory score files are written to."
|
||
@spec base_dir() :: String.t()
|
||
def base_dir do
|
||
Application.get_env(:microwaveprop, :propagation_scores_dir, "/data/scores")
|
||
end
|
||
|
||
@doc "Returns the full path for a `(band_mhz, valid_time)` score file."
|
||
@spec path_for(non_neg_integer(), DateTime.t()) :: String.t()
|
||
def path_for(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do
|
||
iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601()
|
||
Path.join([base_dir(), Integer.to_string(band_mhz), "#{iso}.ntms"])
|
||
end
|
||
|
||
@doc """
|
||
Write a score grid to disk for `(band_mhz, valid_time)`. `scores`
|
||
must be a list of `%{lat, lon, score}` maps. Missing cells are
|
||
stored as the no-data sentinel (255).
|
||
"""
|
||
@spec write!(non_neg_integer(), DateTime.t(), [map()]) :: :ok
|
||
def write!(band_mhz, %DateTime{} = valid_time, scores) when is_integer(band_mhz) and is_list(scores) do
|
||
path = path_for(band_mhz, valid_time)
|
||
File.mkdir_p!(Path.dirname(path))
|
||
|
||
binary = encode(band_mhz, valid_time, scores)
|
||
|
||
tmp = path <> ".tmp." <> unique_suffix()
|
||
File.write!(tmp, binary, [:binary])
|
||
File.rename!(tmp, path)
|
||
:ok
|
||
end
|
||
|
||
@doc """
|
||
Read a score grid from disk. Returns:
|
||
|
||
* `{:ok, %{band_mhz, valid_time, lat_min, lon_min, step, n_rows, n_cols, scores}}`
|
||
where `scores` is the raw binary (row-major, uint8 per cell)
|
||
* `{:error, :enoent}` when the file doesn't exist
|
||
* `{:error, :invalid_format}` when the file isn't a score grid
|
||
"""
|
||
@spec read(non_neg_integer(), DateTime.t()) ::
|
||
{:ok, map()} | {:error, :enoent | :invalid_format}
|
||
def read(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do
|
||
case File.read(path_for(band_mhz, valid_time)) do
|
||
{:ok, binary} -> decode(binary)
|
||
{:error, :enoent} -> {:error, :enoent}
|
||
{:error, other} -> {:error, other}
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Delete every score file whose `valid_time` is strictly before
|
||
`cutoff`. Returns the number of files deleted. Foreign files in the
|
||
scores tree (anything that doesn't parse as `<iso>.ntms`) are left
|
||
alone.
|
||
"""
|
||
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
||
def prune_older_than(%DateTime{} = cutoff) do
|
||
cutoff_unix = DateTime.to_unix(cutoff)
|
||
|
||
base_dir()
|
||
|> list_score_files()
|
||
|> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
|
||
if valid_time_unix < cutoff_unix do
|
||
_ = File.rm(path)
|
||
acc + 1
|
||
else
|
||
acc
|
||
end
|
||
end)
|
||
end
|
||
|
||
@doc """
|
||
Keep only the files whose `valid_time` falls inside the closed
|
||
window `[run_time, run_time + max_forecast_hour * 3600]`. Returns
|
||
the number of files deleted.
|
||
|
||
Called at the end of a `PropagationGridWorker` chain so any files
|
||
left over from the previous run_time (that the new chain didn't
|
||
overwrite by coincidence of valid_time) get cleaned up immediately
|
||
instead of waiting for the 2h prune cron.
|
||
"""
|
||
@spec retain_window(DateTime.t(), non_neg_integer()) :: non_neg_integer()
|
||
def retain_window(%DateTime{} = run_time, max_forecast_hour) when max_forecast_hour >= 0 do
|
||
lo = DateTime.to_unix(run_time)
|
||
hi = lo + max_forecast_hour * 3600
|
||
|
||
base_dir()
|
||
|> list_score_files()
|
||
|> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
|
||
if valid_time_unix < lo or valid_time_unix > hi do
|
||
_ = File.rm(path)
|
||
acc + 1
|
||
else
|
||
acc
|
||
end
|
||
end)
|
||
end
|
||
|
||
@doc """
|
||
List every `valid_time` that has a score file on disk for
|
||
`band_mhz`, ordered ascending. Empty list if the band directory
|
||
doesn't exist yet.
|
||
"""
|
||
@spec list_valid_times(non_neg_integer()) :: [DateTime.t()]
|
||
def list_valid_times(band_mhz) when is_integer(band_mhz) do
|
||
dir = Path.join(base_dir(), Integer.to_string(band_mhz))
|
||
|
||
case File.ls(dir) do
|
||
{:ok, files} ->
|
||
files
|
||
|> Enum.map(&parse_valid_time_dt/1)
|
||
|> Enum.reject(&is_nil/1)
|
||
|> Enum.sort(DateTime)
|
||
|
||
_ ->
|
||
[]
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Return the most recent `valid_time` across all bands on disk, or
|
||
`nil` if the scores tree is empty.
|
||
"""
|
||
@spec latest_valid_time() :: DateTime.t() | nil
|
||
def latest_valid_time do
|
||
case File.ls(base_dir()) do
|
||
{:ok, band_dirs} ->
|
||
band_dirs
|
||
|> Enum.flat_map(&band_times_or_empty/1)
|
||
|> max_datetime()
|
||
|
||
_ ->
|
||
nil
|
||
end
|
||
end
|
||
|
||
defp band_times_or_empty(band_dir) do
|
||
case Integer.parse(band_dir) do
|
||
{band_mhz, ""} -> list_valid_times(band_mhz)
|
||
_ -> []
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Read the file for `(band_mhz, valid_time)` and return every
|
||
scored cell as a list of `%{lat, lon, score}` maps. Optional
|
||
`bounds` map (`%{"north", "south", "east", "west"}` or
|
||
`%{north:, south:, east:, west:}`) filters the return to cells
|
||
inside that box. Cells with the no-data sentinel are excluded.
|
||
|
||
Returns `[]` when the file doesn't exist or is unreadable —
|
||
callers should treat this as "no scores available for this
|
||
band/time yet" rather than an error.
|
||
"""
|
||
@spec read_bounds(non_neg_integer(), DateTime.t(), map() | nil) ::
|
||
[%{lat: float(), lon: float(), score: non_neg_integer()}]
|
||
def read_bounds(band_mhz, %DateTime{} = valid_time, bounds \\ nil) do
|
||
case read(band_mhz, valid_time) do
|
||
{:ok, payload} -> extract_points(payload, bounds)
|
||
_ -> []
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Fetch the score for a single grid cell from the file for
|
||
`(band_mhz, valid_time)`. Returns `nil` if the file is missing or
|
||
if the cell is outside the grid / carries the no-data sentinel.
|
||
"""
|
||
@spec read_point(non_neg_integer(), DateTime.t(), float(), float()) :: non_neg_integer() | nil
|
||
def read_point(band_mhz, %DateTime{} = valid_time, lat, lon) do
|
||
case read(band_mhz, valid_time) do
|
||
{:ok, payload} -> point_score(payload, lat, lon)
|
||
_ -> nil
|
||
end
|
||
end
|
||
|
||
# ── Encoding ────────────────────────────────────────────────────
|
||
|
||
defp encode(band_mhz, valid_time, scores) do
|
||
%{lat_min: lat_min, lat_max: lat_max, lon_min: lon_min, lon_max: lon_max} = Grid.bounds()
|
||
step = Grid.step()
|
||
n_rows = round((lat_max - lat_min) / step) + 1
|
||
n_cols = round((lon_max - lon_min) / step) + 1
|
||
|
||
lookup =
|
||
Map.new(scores, fn s ->
|
||
{{Float.round(s.lat, 3), Float.round(s.lon, 3)}, s.score}
|
||
end)
|
||
|
||
body = build_body(lookup, lat_min, lon_min, step, n_rows, n_cols)
|
||
valid_time_unix = valid_time |> DateTime.truncate(:second) |> DateTime.to_unix()
|
||
|
||
<<
|
||
@magic::binary,
|
||
@version::unsigned-8,
|
||
band_mhz::unsigned-little-32,
|
||
valid_time_unix::signed-little-64,
|
||
lat_min::float-little-32,
|
||
lon_min::float-little-32,
|
||
step::float-little-32,
|
||
n_rows::unsigned-little-16,
|
||
n_cols::unsigned-little-16,
|
||
body::binary
|
||
>>
|
||
end
|
||
|
||
defp build_body(lookup, lat_min, lon_min, step, n_rows, n_cols) do
|
||
for row <- 0..(n_rows - 1),
|
||
col <- 0..(n_cols - 1),
|
||
into: <<>> do
|
||
lat = Float.round(lat_min + row * step, 3)
|
||
lon = Float.round(lon_min + col * step, 3)
|
||
<<score_byte(Map.get(lookup, {lat, lon}))::unsigned-8>>
|
||
end
|
||
end
|
||
|
||
defp score_byte(nil), do: @no_data
|
||
defp score_byte(s) when is_integer(s) and s >= 0 and s <= 100, do: s
|
||
defp score_byte(s) when is_integer(s) and s > 100, do: 100
|
||
defp score_byte(s) when is_integer(s) and s < 0, do: 0
|
||
defp score_byte(_), do: @no_data
|
||
|
||
# ── Decoding ────────────────────────────────────────────────────
|
||
|
||
defp decode(
|
||
<<@magic, @version::unsigned-8, band_mhz::unsigned-little-32, valid_time_unix::signed-little-64,
|
||
lat_min::float-little-32, lon_min::float-little-32, step::float-little-32, n_rows::unsigned-little-16,
|
||
n_cols::unsigned-little-16, body::binary>>
|
||
) do
|
||
expected = n_rows * n_cols
|
||
|
||
if byte_size(body) == expected do
|
||
case DateTime.from_unix(valid_time_unix) do
|
||
{:ok, valid_time} ->
|
||
{:ok,
|
||
%{
|
||
band_mhz: band_mhz,
|
||
valid_time: valid_time,
|
||
lat_min: lat_min,
|
||
lon_min: lon_min,
|
||
step: step,
|
||
n_rows: n_rows,
|
||
n_cols: n_cols,
|
||
scores: body
|
||
}}
|
||
|
||
_ ->
|
||
{:error, :invalid_format}
|
||
end
|
||
else
|
||
{:error, :invalid_format}
|
||
end
|
||
end
|
||
|
||
defp decode(_), do: {:error, :invalid_format}
|
||
|
||
# ── Read helpers ────────────────────────────────────────────────
|
||
|
||
defp extract_points(%{scores: body} = payload, bounds) do
|
||
%{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols} = payload
|
||
{row_lo, row_hi, col_lo, col_hi} = bounds_to_index_range(payload, bounds)
|
||
|
||
for row <- row_lo..row_hi,
|
||
col <- col_lo..col_hi,
|
||
byte = :binary.at(body, row * n_cols + col),
|
||
byte != @no_data do
|
||
# guard against the ignored variables warning
|
||
_ = n_rows
|
||
lat = Float.round(lat_min + row * step, 3)
|
||
lon = Float.round(lon_min + col * step, 3)
|
||
%{lat: lat, lon: lon, score: byte}
|
||
end
|
||
end
|
||
|
||
defp bounds_to_index_range(payload, nil) do
|
||
{0, payload.n_rows - 1, 0, payload.n_cols - 1}
|
||
end
|
||
|
||
defp bounds_to_index_range(payload, bounds) do
|
||
%{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols} = payload
|
||
|
||
south = fetch_bound(bounds, :south, "south", lat_min)
|
||
north = fetch_bound(bounds, :north, "north", lat_min + (n_rows - 1) * step)
|
||
west = fetch_bound(bounds, :west, "west", lon_min)
|
||
east = fetch_bound(bounds, :east, "east", lon_min + (n_cols - 1) * step)
|
||
|
||
row_lo = clamp(trunc((south - lat_min) / step), 0, n_rows - 1)
|
||
row_hi = clamp(ceil((north - lat_min) / step), 0, n_rows - 1)
|
||
col_lo = clamp(trunc((west - lon_min) / step), 0, n_cols - 1)
|
||
col_hi = clamp(ceil((east - lon_min) / step), 0, n_cols - 1)
|
||
|
||
{min(row_lo, row_hi), max(row_lo, row_hi), min(col_lo, col_hi), max(col_lo, col_hi)}
|
||
end
|
||
|
||
defp fetch_bound(bounds, atom_key, string_key, default) do
|
||
case Map.get(bounds, atom_key) || Map.get(bounds, string_key) do
|
||
nil -> default
|
||
v when is_number(v) -> v
|
||
v when is_binary(v) -> String.to_float(v)
|
||
end
|
||
end
|
||
|
||
defp clamp(n, lo, _hi) when n < lo, do: lo
|
||
defp clamp(n, _lo, hi) when n > hi, do: hi
|
||
defp clamp(n, _lo, _hi), do: n
|
||
|
||
defp point_score(payload, lat, lon) do
|
||
%{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols, scores: body} = payload
|
||
row = round((lat - lat_min) / step)
|
||
col = round((lon - lon_min) / step)
|
||
|
||
if row in 0..(n_rows - 1) and col in 0..(n_cols - 1) do
|
||
case :binary.at(body, row * n_cols + col) do
|
||
@no_data -> nil
|
||
score -> score
|
||
end
|
||
end
|
||
end
|
||
|
||
defp max_datetime([]), do: nil
|
||
defp max_datetime(times), do: Enum.max(times, DateTime)
|
||
|
||
defp parse_valid_time_dt(filename) do
|
||
with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename),
|
||
{:ok, dt, _} <- DateTime.from_iso8601(iso) do
|
||
dt
|
||
else
|
||
_ -> nil
|
||
end
|
||
end
|
||
|
||
# ── Prune helpers ───────────────────────────────────────────────
|
||
|
||
defp list_score_files(root) do
|
||
case File.ls(root) do
|
||
{:ok, entries} ->
|
||
for entry <- entries,
|
||
band_path = Path.join(root, entry),
|
||
File.dir?(band_path),
|
||
file <- files_in(band_path),
|
||
valid_time_unix = parse_valid_time(file),
|
||
valid_time_unix != nil do
|
||
{Path.join(band_path, file), valid_time_unix}
|
||
end
|
||
|
||
_ ->
|
||
[]
|
||
end
|
||
end
|
||
|
||
defp files_in(dir) do
|
||
case File.ls(dir) do
|
||
{:ok, entries} -> entries
|
||
_ -> []
|
||
end
|
||
end
|
||
|
||
defp parse_valid_time(filename) do
|
||
with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename),
|
||
{:ok, dt, _} <- DateTime.from_iso8601(iso) do
|
||
DateTime.to_unix(dt)
|
||
else
|
||
_ -> nil
|
||
end
|
||
end
|
||
|
||
defp unique_suffix do
|
||
"#{System.system_time(:nanosecond)}.#{:erlang.unique_integer([:positive])}"
|
||
end
|
||
end
|