Oban cron worker runs every 4 hours, finds QSOs without weather data, discovers nearby ASOS/sounding stations, and enqueues WeatherFetchWorker jobs. Migrations run automatically on app start in production.
167 lines
4.5 KiB
Elixir
167 lines
4.5 KiB
Elixir
defmodule Microwaveprop.Weather do
|
|
@moduledoc false
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.SolarIndex
|
|
alias Microwaveprop.Weather.Sounding
|
|
alias Microwaveprop.Weather.Station
|
|
alias Microwaveprop.Weather.SurfaceObservation
|
|
|
|
# Approximate km per degree latitude
|
|
@km_per_deg_lat 111.0
|
|
|
|
def find_or_create_station(attrs) do
|
|
code = attrs[:station_code] || attrs["station_code"]
|
|
type = attrs[:station_type] || attrs["station_type"]
|
|
|
|
if code && type do
|
|
case Repo.get_by(Station, station_code: code, station_type: type) do
|
|
nil ->
|
|
%Station{}
|
|
|> Station.changeset(attrs)
|
|
|> Repo.insert()
|
|
|
|
station ->
|
|
{:ok, station}
|
|
end
|
|
else
|
|
%Station{}
|
|
|> Station.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
end
|
|
|
|
def upsert_surface_observation(%Station{} = station, attrs) do
|
|
attrs = Map.put(attrs, :station_id, station.id)
|
|
|
|
%SurfaceObservation{}
|
|
|> SurfaceObservation.changeset(attrs)
|
|
|> Repo.insert(
|
|
on_conflict: {:replace_all_except, [:id, :station_id, :observed_at, :inserted_at]},
|
|
conflict_target: [:station_id, :observed_at],
|
|
returning: true
|
|
)
|
|
end
|
|
|
|
def upsert_sounding(%Station{} = station, attrs) do
|
|
attrs = Map.put(attrs, :station_id, station.id)
|
|
|
|
%Sounding{}
|
|
|> Sounding.changeset(attrs)
|
|
|> Repo.insert(
|
|
on_conflict: {:replace_all_except, [:id, :station_id, :observed_at, :inserted_at]},
|
|
conflict_target: [:station_id, :observed_at],
|
|
returning: true
|
|
)
|
|
end
|
|
|
|
def upsert_solar_index(attrs) do
|
|
%SolarIndex{}
|
|
|> SolarIndex.changeset(attrs)
|
|
|> Repo.insert(
|
|
on_conflict: {:replace_all_except, [:id, :date, :inserted_at]},
|
|
conflict_target: [:date],
|
|
returning: true
|
|
)
|
|
end
|
|
|
|
def has_surface_observations?(station_id, start_dt, end_dt) do
|
|
SurfaceObservation
|
|
|> where([o], o.station_id == ^station_id)
|
|
|> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt)
|
|
|> Repo.exists?()
|
|
end
|
|
|
|
def has_sounding?(station_id, observed_at) do
|
|
Sounding
|
|
|> where([s], s.station_id == ^station_id and s.observed_at == ^observed_at)
|
|
|> Repo.exists?()
|
|
end
|
|
|
|
def get_solar_index(date) do
|
|
Repo.get_by(SolarIndex, date: date)
|
|
end
|
|
|
|
def existing_solar_dates do
|
|
SolarIndex
|
|
|> select([s], s.date)
|
|
|> Repo.all()
|
|
|> MapSet.new()
|
|
end
|
|
|
|
def nearby_stations(lat, lon, station_type, radius_km) do
|
|
dlat = radius_km / @km_per_deg_lat
|
|
dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180))
|
|
|
|
Station
|
|
|> where([s], s.station_type == ^station_type)
|
|
|> where(
|
|
[s],
|
|
s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
|
|
s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
|
|
)
|
|
|> Repo.all()
|
|
end
|
|
|
|
def sounding_times_around(dt) do
|
|
date = DateTime.to_date(dt)
|
|
|
|
times =
|
|
if dt.hour < 12 do
|
|
[
|
|
DateTime.new!(Date.add(date, -1), ~T[12:00:00], "Etc/UTC"),
|
|
DateTime.new!(date, ~T[00:00:00], "Etc/UTC")
|
|
]
|
|
else
|
|
[
|
|
DateTime.new!(date, ~T[00:00:00], "Etc/UTC"),
|
|
DateTime.new!(date, ~T[12:00:00], "Etc/UTC")
|
|
]
|
|
end
|
|
|
|
Enum.uniq(times)
|
|
end
|
|
|
|
def weather_for_qso(qso_params, opts \\ []) do
|
|
lat = qso_params[:lat] || qso_params.lat
|
|
lon = qso_params[:lon] || qso_params.lon
|
|
timestamp = qso_params[:timestamp] || qso_params.timestamp
|
|
|
|
radius_km = Keyword.get(opts, :radius_km, 150)
|
|
time_window_hours = Keyword.get(opts, :time_window_hours, 6)
|
|
|
|
# Bounding box in degrees
|
|
dlat = radius_km / @km_per_deg_lat
|
|
dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180))
|
|
|
|
time_start = DateTime.add(timestamp, -time_window_hours * 3600, :second)
|
|
time_end = DateTime.add(timestamp, time_window_hours * 3600, :second)
|
|
|
|
station_ids =
|
|
Station
|
|
|> where(
|
|
[s],
|
|
s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
|
|
s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
|
|
)
|
|
|> select([s], s.id)
|
|
|
|
surface_observations =
|
|
SurfaceObservation
|
|
|> where([o], o.station_id in subquery(station_ids))
|
|
|> where([o], o.observed_at >= ^time_start and o.observed_at <= ^time_end)
|
|
|> preload(:station)
|
|
|> Repo.all()
|
|
|
|
soundings =
|
|
Sounding
|
|
|> where([s], s.station_id in subquery(station_ids))
|
|
|> where([s], s.observed_at >= ^time_start and s.observed_at <= ^time_end)
|
|
|> preload(:station)
|
|
|> Repo.all()
|
|
|
|
%{surface_observations: surface_observations, soundings: soundings}
|
|
end
|
|
end
|