GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

drought/pollen pushes

e5d0a50 · gmorell · 2026-05-11 14:53

61 files +1044 -77

Files changed

modified apps/room_air_quality/lib/room_air_quality/aqi_worker.ex
+123 −64
@@ -12,12 +12,6 @@ defmodule RoomAirQuality.Worker do
12 12
13 13 def start_link(opts) do
14 14 GenServer.start_link(__MODULE__, opts, name: via_tuple("aqi" <> opts[:name]))
15 −
16 − Periodic.start_link(
17 − every: :timer.seconds(60 * 60),
18 − run: fn -> RoomGitlab.Worker.query_projects(opts[:name]) end,
19 − initial_delay: :timer.seconds(60 * 30),
20 − )
21 15 end
22 16
23 17 # Public
@@ -38,6 +32,12 @@ defmodule RoomAirQuality.Worker do
38 32 end
39 33
40 34 def init(opts) do
35 + Periodic.start_link(
36 + every: :timer.seconds(60 * 60),
37 + run: fn -> RoomAirQuality.Worker.update_static_data(opts[:name]) end,
38 + initial_delay: :timer.seconds(30)
39 + )
40 +
41 41 {:ok, %{id: opts[:name]}}
42 42 end
43 43
@@ -54,6 +54,28 @@ defmodule RoomAirQuality.Worker do
54 54 defp intify(val) when is_float(val), do: val |> Kernel.trunc()
55 55 defp intify(val), do: val
56 56
57 + defp parse_float(""), do: nil
58 + defp parse_float(nil), do: nil
59 +
60 + defp parse_float(s) when is_binary(s) do
61 + case Float.parse(s) do
62 + {f, _} -> f
63 + :error -> nil
64 + end
65 + end
66 +
67 + defp parse_float(f) when is_float(f), do: f
68 + defp parse_float(_), do: nil
69 +
70 + # AirNow leaves unmeasured pollutant fields as empty strings ("").
71 + # Ecto's :float cast rejects "" (changeset becomes invalid → row silently dropped).
72 + # Normalize all blanks to nil before building the changeset.
73 + defp normalize_blanks(data) do
74 + data
75 + |> Enum.map(fn {k, v} -> {k, if(v == "", do: nil, else: v)} end)
76 + |> Map.new()
77 + end
78 +
57 79 defp write_data(result, type, id) do
58 80 datetime = NaiveDateTime.local_now()
59 81 Logger.info("AQI::#{id} writing bundle #{type}")
@@ -61,8 +83,9 @@ defmodule RoomAirQuality.Worker do
61 83 case type do
62 84 :hourly ->
63 85 result.body
64 − |> String.split("\r\n")
86 + |> String.split(~r/\r?\n/)
65 87 |> Enum.filter(fn x -> x != "" end)
88 + |> Enum.map(&(&1 <> "\n"))
66 89 |> CSV.decode(
67 90 headers: [
68 91 :valid_date,
@@ -98,9 +121,13 @@ defmodule RoomAirQuality.Worker do
98 121
99 122 :sites ->
100 123 result.body
101 − |> String.split("\r\n")
124 + |> String.split(~r/\r?\n/)
102 125 |> List.delete_at(0)
103 126 |> Enum.filter(fn x -> x != "" end)
127 + # The :csv library expects each stream element to end with a newline;
128 + # without it, consecutive lines are interpreted as quote-escaped
129 + # continuations and the entire file collapses into a single row.
130 + |> Enum.map(&(&1 <> "\n"))
104 131 |> CSV.decode(
105 132 headers: [
106 133 :station_id,
@@ -162,11 +189,16 @@ defmodule RoomAirQuality.Worker do
162 189 end)
163 190
164 191 :hourly_obs ->
192 + # Column order matches AirNow's current HourlyAQObs CSV header (verified live).
193 + # Positions 23-34 reorder: PM25, OZONE, NO2, CO, SO2, PM10 (each followed by its _Unit).
165 194 result.body
166 − |> String.split("\r\n")
195 + |> String.split(~r/\r?\n/)
167 196 |> List.delete_at(0)
168 − |> IO.inspect
169 197 |> Enum.filter(fn x -> x != "" end)
198 + # The :csv library expects each stream element to end with a newline;
199 + # without it, consecutive lines are interpreted as quote-escaped
200 + # continuations and the entire file collapses into a single row.
201 + |> Enum.map(&(&1 <> "\n"))
170 202 |> CSV.decode(
171 203 headers: [
172 204 :aqsid,
@@ -191,40 +223,44 @@ defmodule RoomAirQuality.Worker do
191 223 :pm10_measured,
192 224 :pm25_measured,
193 225 :no2_measured,
226 + :pm25,
227 + :pm25_unit,
228 + :ozone,
229 + :ozone_unit,
194 230 :no2,
195 231 :no2_unit,
196 232 :co,
197 233 :co_unit,
198 − :pm25,
199 − :pm25_unit,
200 234 :so2,
201 235 :so2_unit,
202 − :ozone,
203 − :ozone_unit,
204 236 :pm10,
205 237 :pm10_unit
206 238 ]
207 239 )
208 240 |> Enum.to_list()
209 − |> IO.inspect
210 241 |> Enum.map(fn {:ok, x} -> x end)
242 + |> Enum.map(&normalize_blanks/1)
211 243 |> Enum.map(fn data ->
212 − point = %Geo.Point{
213 − coordinates: {data.lat |> String.to_float(), data.lon |> String.to_float()},
214 − srid: 4326
215 − }
244 + lat = parse_float(data.lat)
245 + lon = parse_float(data.lon)
216 246
217 − {offset, _} = data.gmt_offset |> Float.parse()
247 + point =
248 + if lat && lon do
249 + %Geo.Point{coordinates: {lat, lon}, srid: 4326}
250 + end
218 251
219 252 RoomSanctum.Storage.change_hourly_obs_data(
220 253 %RoomSanctum.Storage.AirNow.HourlyObsData{},
221 254 data
222 255 |> Map.put(:source_id, id)
223 256 |> Map.put(:point, point)
224 − |> Map.put(:gmt_offset, offset)
257 + |> Map.put(:lat, lat)
258 + |> Map.put(:lon, lon)
259 + |> Map.put(:elevation, parse_float(data[:elevation]))
260 + |> Map.put(:gmt_offset, parse_float(data[:gmt_offset]))
225 261 |> Map.put(
226 262 :reporting_areas,
227 − data |> Map.get(:reporting_areas, "") |> String.split("|")
263 + (data[:reporting_areas] || "") |> String.split("|")
228 264 )
229 265 |> Map.put(
230 266 :valid_date,
@@ -235,7 +271,6 @@ defmodule RoomAirQuality.Worker do
235 271 data.valid_time |> Timex.parse!("%H:%M", :strftime) |> NaiveDateTime.to_time()
236 272 )
237 273 )
238 − |> IO.inspect
239 274 |> Map.put(:inserted_at, datetime)
240 275 |> Map.put(:updated_at, datetime)
241 276 |> Map.put(:ozone_aqi, data.ozone_aqi |> intify)
@@ -243,14 +278,34 @@ defmodule RoomAirQuality.Worker do
243 278 |> Map.put(:pm25_aqi, data.pm25_aqi |> intify)
244 279 |> Map.put(:no2_aqi, data.no2_aqi |> intify)
245 280 end)
246 − |> IO.inspect
247 − |> Enum.map(fn x ->
248 − Repo.insert(
249 − x,
250 − on_conflict: {:replace_all_except, [:id]},
251 − conflict_target: [:source_id, :aqsid]
252 − )
281 + |> Enum.reduce({0, 0, []}, fn cs, {ok, fail, sample_errs} ->
282 + case Repo.insert(cs,
283 + on_conflict: {:replace_all_except, [:id]},
284 + conflict_target: [:source_id, :aqsid]
285 + ) do
286 + {:ok, _} ->
287 + {ok + 1, fail, sample_errs}
288 +
289 + {:error, bad} ->
290 + new_sample =
291 + if length(sample_errs) < 3 do
292 + [{cs.changes[:aqsid] || cs.data.aqsid, bad.errors} | sample_errs]
293 + else
294 + sample_errs
295 + end
296 +
297 + {ok, fail + 1, new_sample}
298 + end
253 299 end)
300 + |> case do
301 + {ok, 0, _} ->
302 + Logger.info("AQI::#{id} inserted #{ok} rows")
303 +
304 + {ok, fail, sample} ->
305 + Logger.warning(
306 + "AQI::#{id} inserted #{ok} rows, #{fail} failed. First failures: #{inspect(sample)}"
307 + )
308 + end
254 309 end
255 310
256 311 Logger.info("AQI::#{id} completed writing bundle")
@@ -266,14 +321,41 @@ defmodule RoomAirQuality.Worker do
266 321 # "https://s3-us-west-1.amazonaws.com//files.airnowtech.org/airnow/today/HourlyData_#{file_str}.dat"
267 322 # end
268 323
269 − defp build_obs_url(ts \\ DateTime.utc_now()) do
270 − sh = ts |> DateTime.add(-1 * 60 * 60, :second)
324 + defp build_obs_url(hours_ago, ts \\ DateTime.utc_now()) do
325 + sh = ts |> DateTime.add(-hours_ago * 60 * 60, :second)
271 326 file_str = sh |> Timex.format!("%Y%m%d%H", :strftime)
272 327 date_str = sh |> Timex.format!("%Y%m%d", :strftime)
273 − "https://s3-us-west-1.amazonaws.com//files.airnowtech.org/airnow/2024/20240418/HourlyAQObs_2024041817.dat"
274 −# "https://s3-us-west-1.amazonaws.com//files.airnowtech.org/airnow/2024/20240418/HourlyAQObs_2024041816.dat"
275 −# "https://s3-us-west-1.amazonaws.com//files.airnowtech.org/airnow/#{sh.year}/#{date_str}/HourlyAQObs_#{file_str}.dat"
276 −# |> IO.inspect()
328 + "https://s3-us-west-1.amazonaws.com//files.airnowtech.org/airnow/#{sh.year}/#{date_str}/HourlyAQObs_#{file_str}.dat"
329 + end
330 +
331 + # AirNow publishes HourlyAQObs files with variable lag (typically ~1-3h).
332 + # Walk back from 1h to 6h and use the first 200 we find.
333 + defp fetch_latest_obs(id) do
334 + Enum.reduce_while(1..6, nil, fn hours_ago, _acc ->
335 + url = build_obs_url(hours_ago)
336 +
337 + case HTTPoison.get(url) do
338 + {:ok, %{status_code: 200} = result} ->
339 + Logger.info("AQI::#{id} fetched #{url}")
340 + {:halt, {:ok, result}}
341 +
342 + {:ok, %{status_code: code}} ->
343 + Logger.debug("AQI::#{id} #{code} on #{url}, trying older hour")
344 + {:cont, nil}
345 +
346 + {:error, info} ->
347 + Logger.warning("AQI::#{id} HTTP error on #{url}: #{inspect(info.reason)}")
348 + {:cont, nil}
349 + end
350 + end)
351 + |> case do
352 + {:ok, _} = ok ->
353 + ok
354 +
355 + _ ->
356 + Logger.warning("AQI::#{id} no AirNow file available in last 6 hours")
357 + :error
358 + end
277 359 end
278 360
279 361 def handle_cast(:refresh_db_cfg, state) do
@@ -283,37 +365,14 @@ defmodule RoomAirQuality.Worker do
283 365
284 366 def handle_cast(:update_static, state) do
285 367 Logger.info("AQI::#{state.id} updating data")
286 −
287 368 bcast(state.id, :downloading, 1, 10)
288 − # case HTTPoison.get(build_todays_url()) do
289 − # {:ok, result} ->
290 − # write_data(result, :hourly, state.id)
291 − # {:noreply, state}rec
292 − #
293 − # {:error, info} ->
294 − # Logger.info(info.reason)
295 − # {:noreply, state}
296 − # end
297 −
298 − case HTTPoison.get(build_obs_url()) do
299 − {:ok, result} ->
300 − write_data(result, :hourly_obs, state.id)
301 − {:noreply, state}
302 −
303 − {:error, info} ->
304 − Logger.info(info.reason)
305 − {:noreply, state}
369 +
370 + case fetch_latest_obs(state.id) do
371 + {:ok, result} -> write_data(result, :hourly_obs, state.id)
372 + :error -> :ok
306 373 end
307 374
308 − # case HTTPoison.get(@monitoring_site_url) do
309 − # {:ok, result} ->
310 − # write_data(result, :sites, state.id)
311 − # {:noreply, state}
312 − #
313 − # {:error, info} ->
314 − # Logger.info(info.reason)
315 − # {:noreply, state}
316 − # end
375 + {:noreply, state}
317 376 end
318 377
319 378 def handle_cast(_msg, state) do
added apps/room_drought/lib/room_drought.ex
+3 −0
@@ -0,0 +1,3 @@
1 +defmodule RoomDrought do
2 + @moduledoc false
3 +end
added apps/room_drought/lib/room_drought/worker.ex
+194 −0
@@ -0,0 +1,194 @@
1 +defmodule RoomDrought.Worker do
2 + @moduledoc false
3 + use GenServer
4 +
5 + require Logger
6 +
7 + alias RoomSanctum.Configuration
8 +
9 + @registry :zeus
10 + @drought_config_module RoomSanctum.Configuration.Configs.Drought
11 + @usdm_base "https://usdmdataservices.unl.edu/api"
12 +
13 + def start_link(opts) do
14 + GenServer.start_link(__MODULE__, opts, name: via_tuple("drought#{opts[:name]}"))
15 + end
16 +
17 + def init(opts) do
18 + Periodic.start_link(
19 + every: :timer.seconds(10),
20 + run: fn -> RoomDrought.Worker.refresh_db_cfg(opts[:name]) end,
21 + initial_delay: 0
22 + )
23 +
24 + Periodic.start_link(
25 + every: :timer.seconds(60 * 60 * 6),
26 + run: fn -> RoomDrought.Worker.refresh_usdm(opts[:name]) end,
27 + initial_delay: :timer.seconds(15)
28 + )
29 +
30 + {:ok,
31 + %{
32 + id: opts[:name],
33 + inst: %{},
34 + usdm: %{},
35 + targets: MapSet.new()
36 + }}
37 + end
38 +
39 + def pid(name) do
40 + "drought#{name}"
41 + |> via_tuple()
42 + |> GenServer.whereis()
43 + end
44 +
45 + # Public API
46 + def refresh_db_cfg(name), do: GenServer.cast(via("drought#{name}"), :refresh_db_cfg)
47 + def refresh_usdm(name), do: GenServer.cast(via("drought#{name}"), :refresh_usdm)
48 + def add_target(name, scope), do: GenServer.cast(via("drought#{name}"), {:add_target, scope})
49 + def read(name, query), do: GenServer.call(via("drought#{name}"), {:read, query})
50 +
51 + # Cast handlers
52 + def handle_cast(:refresh_db_cfg, state) do
53 + inst = Configuration.get_source!(state.id)
54 + {:noreply, state |> Map.put(:inst, inst)}
55 + end
56 +
57 + def handle_cast(:refresh_usdm, state) do
58 + if usdm_enabled?(state.inst) do
59 + Logger.info("Drought::#{state.inst.id} refreshing USDM (#{MapSet.size(state.targets)} targets)")
60 +
61 + new_usdm =
62 + state.targets
63 + |> Enum.reduce(state.usdm, fn target, acc ->
64 + case fetch_usdm(target) do
65 + {:ok, record} -> Map.put(acc, target, record)
66 + :error -> acc
67 + end
68 + end)
69 +
70 + {:noreply, state |> Map.put(:usdm, new_usdm)}
71 + else
72 + {:noreply, state}
73 + end
74 + end
75 +
76 + def handle_cast({:add_target, scope}, state) do
77 + target = normalize_scope(scope)
78 +
79 + if target do
80 + {:noreply, state |> Map.put(:targets, MapSet.put(state.targets, target))}
81 + else
82 + {:noreply, state}
83 + end
84 + end
85 +
86 + # Call handlers
87 + def handle_call({:read, query}, _from, state) do
88 + product = Map.get(query, :product) || Map.get(query, "product") || "usdm"
89 +
90 + result =
91 + case product do
92 + "usdm" ->
93 + case scope_from_query(query) do
94 + nil ->
95 + []
96 +
97 + target ->
98 + # Make sure we're tracking it for next refresh
99 + GenServer.cast(self(), {:add_target, target})
100 +
101 + case Map.get(state.usdm, target) do
102 + nil ->
103 + case fetch_usdm(target) do
104 + {:ok, record} -> [record]
105 + :error -> []
106 + end
107 +
108 + record ->
109 + [record]
110 + end
111 + end
112 +
113 + product when product in ["csi", "cpc", "gpcc"] ->
114 + [%{"product" => product, "status" => "not_yet_implemented"}]
115 +
116 + _ ->
117 + []
118 + end
119 +
120 + {:reply, result, state}
121 + end
122 +
123 + def handle_call(_msg, _from, state), do: {:reply, :ok, state}
124 +
125 + # Helpers
126 + defp via(name), do: via_tuple(name)
127 + defp via_tuple(name), do: {:via, Registry, {@registry, name}}
128 +
129 + defp usdm_enabled?(%{enabled: true, config: %{__struct__: @drought_config_module, products: products}})
130 + when is_list(products),
131 + do: "usdm" in products
132 +
133 + defp usdm_enabled?(_), do: false
134 +
135 + defp scope_from_query(query) do
136 + fips = Map.get(query, :fips) || Map.get(query, "fips")
137 + state = Map.get(query, :state) || Map.get(query, "state")
138 +
139 + cond do
140 + is_binary(fips) and fips != "" -> {:county, fips}
141 + is_binary(state) and state != "" -> {:state, String.upcase(state)}
142 + true -> nil
143 + end
144 + end
145 +
146 + defp normalize_scope({:county, fips}) when is_binary(fips), do: {:county, fips}
147 + defp normalize_scope({:state, st}) when is_binary(st), do: {:state, String.upcase(st)}
148 + defp normalize_scope(_), do: nil
149 +
150 + defp fetch_usdm({:county, fips}) do
151 + fetch_usdm_path("/CountyStatistics/GetDroughtSeverityStatisticsByAreaPercent", fips)
152 + end
153 +
154 + defp fetch_usdm({:state, st}) do
155 + fetch_usdm_path("/StateStatistics/GetDroughtSeverityStatisticsByAreaPercent", st)
156 + end
157 +
158 + defp fetch_usdm_path(path, aoi) do
159 + today = Date.utc_today()
160 + start_date = today |> Date.add(-60) |> format_usdm_date()
161 + end_date = today |> format_usdm_date()
162 +
163 + url =
164 + "#{@usdm_base}#{path}?aoi=#{URI.encode(aoi)}" <>
165 + "&startdate=#{start_date}&enddate=#{end_date}&statisticsType=1"
166 +
167 + case HTTPoison.get(url, [{"Accept", "application/json"}], recv_timeout: 15_000) do
168 + {:ok, %{status_code: 200, body: body}} ->
169 + case Poison.decode(body) do
170 + {:ok, [_ | _] = list} ->
171 + # API returns most-recent first
172 + {:ok, List.first(list)}
173 +
174 + {:ok, []} ->
175 + Logger.info("USDM empty list for #{aoi}")
176 + :error
177 +
178 + {:error, _} ->
179 + Logger.warning("USDM JSON decode failed for #{aoi}")
180 + :error
181 + end
182 +
183 + {:ok, %{status_code: code}} ->
184 + Logger.warning("USDM HTTP #{code} for #{aoi}")
185 + :error
186 +
187 + {:error, err} ->
188 + Logger.warning("USDM HTTP error for #{aoi}: #{inspect(err.reason)}")
189 + :error
190 + end
191 + end
192 +
193 + defp format_usdm_date(%Date{year: y, month: m, day: d}), do: "#{m}/#{d}/#{y}"
194 +end
added apps/room_drought/mix.exs
+28 −0
@@ -0,0 +1,28 @@
1 +defmodule RoomDrought.MixProject do
2 + use Mix.Project
3 +
4 + def project do
5 + [
6 + app: :room_drought,
7 + version: "0.1.0",
8 + build_path: "../../_build",
9 + config_path: "../../config/config.exs",
10 + deps_path: "../../deps",
11 + lockfile: "../../mix.lock",
12 + elixir: "~> 1.14",
13 + start_permanent: Mix.env() == :prod,
14 + deps: deps()
15 + ]
16 + end
17 +
18 + def application do
19 + [extra_applications: [:logger]]
20 + end
21 +
22 + defp deps do
23 + [
24 + {:httpoison, "~> 1.8"},
25 + {:poison, "~> 5.0", override: true}
26 + ]
27 + end
28 +end
added apps/room_pollen/lib/room_pollen.ex
+3 −0
@@ -0,0 +1,3 @@
1 +defmodule RoomPollen do
2 + @moduledoc false
3 +end
added apps/room_pollen/lib/room_pollen/worker.ex
+188 −0

Click to load diff…

added apps/room_pollen/mix.exs
+28 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum/condensers/basic_mqtt.ex
+6 −0

Click to load diff…

added apps/room_sanctum/lib/room_sanctum/configuration/configs/drought.ex
+22 −0

Click to load diff…

added apps/room_sanctum/lib/room_sanctum/configuration/configs/pollen.ex
+16 −0

Click to load diff…

added apps/room_sanctum/lib/room_sanctum/configuration/queries/drought.ex
+30 −0

Click to load diff…

added apps/room_sanctum/lib/room_sanctum/configuration/queries/pollen.ex
+18 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum/configuration/query.ex
+2 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum/configuration/source.ex
+4 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum/storage/air_now/hourly_obs_data.ex
+1 −11

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum/workers/keryx_item_cache.ex
+8 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum/workers/keryx_worker.ex
+12 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum/workers/vision_worker.ex
+12 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/icon_helpers.ex
+12 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_pythiae_live/public.html.heex
+2 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_pythiae_live/show.html.heex
+2 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_query_live/form_component.html.heex
+24 −1

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_query_live/show.ex
+12 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_query_live/show.html.heex
+2 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_scribus_live/public.html.heex
+2 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/form_component.html.heex
+32 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_vision_live/show.html.heex
+2 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/landing_live/index.html.heex
+2 −0

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/live_preview.ex
+251 −0

Click to load diff…

modified apps/room_zeus/lib/supervision_worker.ex
+2 −0

Click to load diff…

modified docker-compose.yml
+1 −1

Click to load diff…

Parents: 0134df0