Graham McIntire
de986bddf6
When the Oban queue backs up, the 60-second uniqueness window expires and duplicate jobs stack up per device. Switch to period: :infinity so only one poll/monitor job exists per device at any time. Add replace option to supersede stale scheduled jobs with updated scheduled_at. Remove :executing from unique states so self-scheduling works while the current job runs. Set max_attempts: 1 since retrying stale polls is pointless. Also fix all credo --strict issues across the codebase.
240 lines
7.1 KiB
Elixir
240 lines
7.1 KiB
Elixir
defmodule Towerops.Weather do
|
|
@moduledoc """
|
|
Weather context for site-correlated weather data.
|
|
|
|
Provides current conditions, historical observations, and weather alerts
|
|
tied to tower site locations. Used for outage correlation and operational
|
|
awareness.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Repo
|
|
alias Towerops.Weather.Alert, as: WeatherAlert
|
|
alias Towerops.Weather.Client
|
|
alias Towerops.Weather.Observation
|
|
|
|
require Logger
|
|
|
|
# ── Observations ─────────────────────────────────────────────────────
|
|
|
|
@doc """
|
|
Records a weather observation for a site.
|
|
"""
|
|
def create_observation(attrs) do
|
|
%Observation{}
|
|
|> Observation.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Gets the latest weather observation for a site.
|
|
"""
|
|
def latest_observation(site_id) do
|
|
Observation
|
|
|> where(site_id: ^site_id)
|
|
|> order_by(desc: :observed_at)
|
|
|> limit(1)
|
|
|> Repo.one()
|
|
end
|
|
|
|
@doc """
|
|
Gets the latest weather observation for each site in an organization.
|
|
Returns a map of site_id => observation.
|
|
"""
|
|
def latest_observations_for_org(organization_id) do
|
|
# Subquery to get latest observation per site
|
|
latest_ids =
|
|
from(o in Observation,
|
|
join: s in assoc(o, :site),
|
|
where: s.organization_id == ^organization_id,
|
|
group_by: o.site_id,
|
|
select: %{site_id: o.site_id, max_id: max(o.id)}
|
|
)
|
|
|
|
from(o in Observation,
|
|
join: l in subquery(latest_ids),
|
|
on: o.id == l.max_id
|
|
)
|
|
|> Repo.all()
|
|
|> Map.new(fn obs -> {obs.site_id, obs} end)
|
|
end
|
|
|
|
@doc """
|
|
Gets weather observations for a site within a time range.
|
|
Useful for correlating weather with outage events.
|
|
"""
|
|
def observations_for_site(site_id, opts \\ []) do
|
|
after_time = opts[:after] || DateTime.add(DateTime.utc_now(), -24, :hour)
|
|
limit = opts[:limit] || 100
|
|
|
|
Observation
|
|
|> where(site_id: ^site_id)
|
|
|> where([o], o.observed_at >= ^after_time)
|
|
|> order_by(desc: :observed_at)
|
|
|> limit(^limit)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Cleans up old weather observations. Keeps the last `keep_days` days.
|
|
"""
|
|
def cleanup_old_observations(keep_days \\ 30) do
|
|
cutoff = DateTime.add(DateTime.utc_now(), -keep_days, :day)
|
|
|
|
{count, _} = Repo.delete_all(from(o in Observation, where: o.observed_at < ^cutoff))
|
|
|
|
Logger.info("Cleaned up #{count} old weather observations")
|
|
{:ok, count}
|
|
end
|
|
|
|
# ── Weather Alerts ───────────────────────────────────────────────────
|
|
|
|
@doc """
|
|
Creates or updates a weather alert for a site.
|
|
"""
|
|
def upsert_alert(attrs) do
|
|
%WeatherAlert{}
|
|
|> WeatherAlert.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Gets active weather alerts for a site (not yet expired).
|
|
"""
|
|
def active_alerts_for_site(site_id) do
|
|
now = DateTime.utc_now()
|
|
|
|
WeatherAlert
|
|
|> where(site_id: ^site_id)
|
|
|> where([a], is_nil(a.ends_at) or a.ends_at > ^now)
|
|
|> order_by(desc: :inserted_at)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Gets active weather alerts for all sites in an organization.
|
|
"""
|
|
def active_alerts_for_org(organization_id) do
|
|
now = DateTime.utc_now()
|
|
|
|
Repo.all(
|
|
from(a in WeatherAlert,
|
|
join: s in assoc(a, :site),
|
|
where: s.organization_id == ^organization_id,
|
|
where: is_nil(a.ends_at) or a.ends_at > ^now,
|
|
order_by: [desc: a.inserted_at],
|
|
preload: [:site]
|
|
)
|
|
)
|
|
end
|
|
|
|
# ── Fetching ─────────────────────────────────────────────────────────
|
|
|
|
@doc """
|
|
Fetches and stores current weather for a site.
|
|
Requires the site to have latitude and longitude set.
|
|
"""
|
|
def fetch_and_store(site) do
|
|
if site.latitude && site.longitude do
|
|
case Client.get_current_weather(site.latitude, site.longitude) do
|
|
{:ok, data} ->
|
|
attrs = Observation.from_owm(site.id, data)
|
|
create_observation(attrs)
|
|
|
|
{:error, reason} ->
|
|
Logger.warning("Failed to fetch weather for site #{site.name}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
else
|
|
{:error, :no_coordinates}
|
|
end
|
|
end
|
|
|
|
# ── Correlation Helpers ──────────────────────────────────────────────
|
|
|
|
@doc """
|
|
Checks if there was severe weather at a site around a given time.
|
|
Returns observations with high wind, heavy rain, or storms.
|
|
|
|
Useful for determining if an outage was likely weather-related.
|
|
"""
|
|
def severe_weather_near?(site_id, time, window_minutes \\ 60) do
|
|
start_time = DateTime.add(time, -window_minutes, :minute)
|
|
end_time = DateTime.add(time, window_minutes, :minute)
|
|
|
|
observations =
|
|
Observation
|
|
|> where(site_id: ^site_id)
|
|
|> where([o], o.observed_at >= ^start_time and o.observed_at <= ^end_time)
|
|
|> Repo.all()
|
|
|
|
Enum.any?(observations, &severe?/1)
|
|
end
|
|
|
|
@doc """
|
|
Returns a weather summary for a site at a point in time.
|
|
Finds the closest observation within 30 minutes.
|
|
"""
|
|
def weather_at(site_id, time) do
|
|
start_time = DateTime.add(time, -30, :minute)
|
|
end_time = DateTime.add(time, 30, :minute)
|
|
|
|
Observation
|
|
|> where(site_id: ^site_id)
|
|
|> where([o], o.observed_at >= ^start_time and o.observed_at <= ^end_time)
|
|
|> order_by([o], fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", o.observed_at, ^time))
|
|
|> limit(1)
|
|
|> Repo.one()
|
|
end
|
|
|
|
@doc """
|
|
Determines if an observation represents severe weather conditions.
|
|
|
|
Thresholds relevant to WISP operations:
|
|
- Wind > 15 m/s (33 mph) — affects radio alignment
|
|
- Wind gusts > 20 m/s (45 mph) — risk of hardware damage
|
|
- Rain > 10mm/h — signal attenuation on wireless links
|
|
- Thunderstorm conditions
|
|
- Snow > 5mm/h — ice loading on equipment
|
|
"""
|
|
def severe?(%Observation{} = obs) do
|
|
high_wind?(obs) or strong_gust?(obs) or heavy_rain?(obs) or
|
|
heavy_snow?(obs) or thunderstorm?(obs)
|
|
end
|
|
|
|
def severe?(_), do: false
|
|
|
|
@doc """
|
|
Returns a severity level for weather conditions.
|
|
:clear, :mild, :moderate, :severe
|
|
"""
|
|
def severity(%Observation{} = obs) do
|
|
cond do
|
|
severe_severity?(obs) -> :severe
|
|
moderate_severity?(obs) -> :moderate
|
|
mild_severity?(obs) -> :mild
|
|
true -> :clear
|
|
end
|
|
end
|
|
|
|
def severity(_), do: :clear
|
|
|
|
defp high_wind?(obs), do: (obs.wind_speed_ms || 0) > 15
|
|
defp strong_gust?(obs), do: (obs.wind_gust_ms || 0) > 20
|
|
defp heavy_rain?(obs), do: (obs.rain_1h_mm || 0) > 10
|
|
defp heavy_snow?(obs), do: (obs.snow_1h_mm || 0) > 5
|
|
defp thunderstorm?(obs), do: obs.condition in ["Thunderstorm"]
|
|
|
|
defp severe_severity?(obs) do
|
|
(obs.wind_gust_ms || 0) > 25 or thunderstorm?(obs) or (obs.rain_1h_mm || 0) > 20
|
|
end
|
|
|
|
defp moderate_severity?(obs) do
|
|
high_wind?(obs) or strong_gust?(obs) or heavy_rain?(obs) or heavy_snow?(obs)
|
|
end
|
|
|
|
defp mild_severity?(obs) do
|
|
obs.condition in ["Rain", "Drizzle", "Snow"] or (obs.clouds_pct || 0) > 80
|
|
end
|
|
end
|