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.

Home Assistant MQTT discovery for boards and Keryxiae

edb5c66 · Gabriel Morell · 2026-10-14 21:16

✓ android ✓ image
12 files +621 -9
Message 
{commit_body(@commit)}

Files changed

added apps/room_sanctum/lib/room_sanctum/hass.ex
+359 −0
@@ -0,0 +1,359 @@
1 +defmodule RoomSanctum.Hass do
2 + @moduledoc """
3 + Home Assistant's MQTT discovery, for what an Ankyra or a Keryx already
4 + publishes.
5 +
6 + Nothing new is published as state. Every entity here is a template over a
7 + payload that is already on the broker -- the board on an Ankyra's topic, or a
8 + Keryx query's `.meta` -- so Home Assistant reads exactly what a panel or a
9 + phone reads, and a board that changes shape changes for all of them at once.
10 +
11 + One device per query, with an entity for each thing in it worth a sensor: a
12 + route and direction at a stop, a dock, a pollutant. Each is picked out of the
13 + payload by what it is rather than where it sits, since a route with nothing
14 + due drops out of the list and everything after it would move up.
15 +
16 + The discovery prefix lives under the Ankyra's own topic, `<topic>/ha`. A
17 + rabbit user may read only its own topic and the Keryxiae its account owns,
18 + so `homeassistant/` is not somewhere Home Assistant could subscribe to here;
19 + it is told to look in `<topic>/ha` instead. Clients subscribe to the topic
20 + itself rather than to `<topic>/#`, so nothing else sees these.
21 + """
22 +
23 + @origin %{name: "Apollo's Crib"}
24 +
25 + @doc "Where the discovery prefix sits under an Ankyra's routing key."
26 + def prefix(ankyra_topic), do: ankyra_topic <> ".ha"
27 +
28 + @doc "The routing key one device's config is published on."
29 + def config_key(ankyra_topic, object_id), do: "#{prefix(ankyra_topic)}.device.#{object_id}.config"
30 +
31 + @doc """
32 + Devices for a board an Ankyra is publishing.
33 +
34 + `reading` says which topic the board went out on -- the Plus board is on
35 + `<topic>.plus` and is the one read wherever a Pythiae publishes it, since it
36 + carries everything Basic does and more.
37 +
38 + Ids are built from the query rather than the board key, so a Pythiae that
39 + turns Plus on keeps its devices: `gtfs-3` and `gtfs_plus-3` are the same stop.
40 + """
41 + def board_devices(ankyra_id, ankyra_topic, board, reading \\ :basic) do
42 + state_topic = mqtt_topic(if reading == :plus, do: ankyra_topic <> ".plus", else: ankyra_topic)
43 +
44 + for {key, entry} <- board, entry not in [nil, %{}] do
45 + key = to_string(key)
46 + [type, id] = String.split(key, "-", parts: 2)
47 +
48 + device(
49 + "apollos_a#{ankyra_id}_#{base_type(type)}_#{slug(id)}",
50 + type,
51 + entry,
52 + %{state_topic: state_topic},
53 + "{% set e = value_json.get(#{lit(key)}) %}"
54 + )
55 + end
56 + end
57 +
58 + @doc """
59 + The device for one Keryx query, read off its `.meta` topic.
60 +
61 + A Keryx publishes every thirty seconds whatever has changed, so these expire
62 + when it stops: a stale bus time is worse than none.
63 + """
64 + def keryx_device(keryx_id, query_id, type, meta) do
65 + device(
66 + "apollos_k#{keryx_id}_#{slug(query_id)}",
67 + to_string(type),
68 + meta,
69 + %{
70 + state_topic: mqtt_topic("#{RoomSanctum.Configuration.Keryx.topic(keryx_id)}.#{type}.#{query_id}.meta"),
71 + expire_after: 300
72 + },
73 + "{% set e = value_json %}"
74 + )
75 + end
76 +
77 + # What a routing key looks like to an MQTT client.
78 + defp mqtt_topic(key), do: String.replace(key, ".", "/")
79 +
80 + defp base_type("gtfs_plus"), do: "gtfs"
81 + defp base_type(type), do: type
82 +
83 + defp device(object_id, type, entry, shared, root) do
84 + query = get(entry, :query) || %{}
85 + data = List.wrap(get(entry, :data))
86 +
87 + components =
88 + type
89 + |> components(data, root)
90 + |> Map.new(fn {id, cfg} ->
91 + {id, Map.merge(%{platform: "sensor", unique_id: "#{object_id}_#{id}"}, cfg)}
92 + end)
93 +
94 + {object_id,
95 + Map.merge(shared, %{
96 + device: %{
97 + identifiers: [object_id],
98 + name: get(query, :name) || object_id,
99 + manufacturer: "Apollo's Crib",
100 + model: model(type)
101 + },
102 + origin: @origin,
103 + components: components
104 + })}
105 + end
106 +
107 + # -- Per type ---------------------------------------------------------------
108 +
109 + # A route, a destination and a direction: one line on a stop's sign. The next
110 + # departure is live where the feed has an estimate and scheduled where not,
111 + # which is what the board itself would show -- and the first that has not
112 + # gone, since a board can still be listing one that left a minute ago.
113 + defp components(type, data, root) when type in ["gtfs", "gtfs_plus"] do
114 + for r <- data, is_map(r) do
115 + pick =
116 + "{% set d = (e.data | selectattr('route', 'eq', #{lit(get(r, :route))})" <>
117 + " | selectattr('dest', 'eq', #{lit(get(r, :dest))})" <>
118 + " | selectattr('dir', 'eq', #{lit(get(r, :dir))}) | list | first) if e else none %}"
119 +
120 + each =
121 + if type == "gtfs_plus",
122 + do: "{% for a in d.arrivals or [] %}{% set x = a.get('time_live') or a.get('time') %}",
123 + else:
124 + "{% set lv = d.get('times_live') or [] %}{% for s in d.get('times') or [] %}" <>
125 + "{% set x = (lv[loop.index0] if lv | length > loop.index0 else none) or s %}"
126 +
127 + next =
128 + "{% set ns = namespace(t=none) %}" <>
129 + each <>
130 + "{% if ns.t is none and x %}" <>
131 + clock("x", wrap: true) <>
132 + "{% if t > now() - timedelta(minutes=1) %}{% set ns.t = t %}{% endif %}{% endif %}{% endfor %}"
133 +
134 + {slug("#{get(r, :route)}_#{get(r, :dest)}_#{get(r, :dir)}"),
135 + %{
136 + name: "#{get(r, :route_name) || get(r, :route)} to #{get(r, :dest)}",
137 + icon: mode_icon(get(r, :mode)),
138 + device_class: "timestamp",
139 + value_template:
140 + root <> pick <> "{% if d %}" <> next <>
141 + "{{ ns.t.isoformat() if ns.t else None }}{% else %}None{% endif %}",
142 + json_attributes_template: root <> pick <> "{{ (d or {}) | tojson }}"
143 + }}
144 + end
145 + end
146 +
147 + # A dock: what can be taken from it and what can be left at it. A loose bike
148 + # is not here -- it has moved on by the next board.
149 + defp components("gbfs", data, root) do
150 + for s <- data, is_map(s), get(s, :kind) not in [:free_bike, "free_bike"], get(s, :id) != nil,
151 + {field, label, icon} <- [
152 + {:avail, "bikes", "mdi:bicycle"},
153 + {:avail_elec, "e-bikes", "mdi:bicycle-electric"},
154 + {:docks_avail, "docks", "mdi:parking"}
155 + ] do
156 + pick = "{% set d = (e.data | selectattr('id', 'eq', #{lit(get(s, :id))}) | list | first) if e else none %}"
157 +
158 + {slug("#{get(s, :id)}_#{field}"),
159 + %{
160 + name: "#{get(s, :name)} #{label}",
161 + icon: icon,
162 + state_class: "measurement",
163 + value_template: root <> pick <> "{{ d.get(#{lit(field)}) if d else None }}",
164 + json_attributes_template: root <> pick <> "{{ (d or {}) | tojson }}"
165 + }}
166 + end
167 + end
168 +
169 + # Each reporting area, each pollutant it measures. AirNow publishes these as
170 + # an index rather than a concentration, which is HA's `aqi`.
171 + defp components("aqi", data, root) do
172 + for a <- data, is_map(a), {k, v} <- a, is_number(v) do
173 + pick = "{% set d = (e.data | selectattr('name', 'eq', #{lit(get(a, :name))}) | list | first) if e else none %}"
174 +
175 + {slug("#{get(a, :name)}_#{k}"),
176 + %{
177 + name: "#{get(a, :name)} #{aqi_label(k)}",
178 + device_class: "aqi",
179 + state_class: "measurement",
180 + value_template: root <> pick <> "{{ d.get(#{lit(k)}) if d else None }}"
181 + }}
182 + end
183 + end
184 +
185 + defp components("weather", data, root) do
186 + for w <- data, is_map(w),
187 + {field, label, extra} <- weather_fields(get(w, :units)) do
188 + pick = "{% set d = (e.data | selectattr('name', 'eq', #{lit(get(w, :name))}) | list | first) if e else none %}"
189 +
190 + value =
191 + case field do
192 + :wind -> "{{ (d.get('wind') or {}).get('speed') if d else None }}"
193 + f -> "{{ d.get(#{lit(f)}) if d else None }}"
194 + end
195 +
196 + {slug("#{get(w, :name)}_#{field}"),
197 + Map.merge(%{name: "#{get(w, :name)} #{label}", value_template: root <> pick <> value}, extra)}
198 + end
199 + end
200 +
201 + # The day's highs and lows, as times today. Not wrapped onto tomorrow: once
202 + # the second high has passed there is no third in the payload to show.
203 + defp components("tidal", [tides | _], root) when is_map(tides) do
204 + for {time_key, label} <- [first_h: "First high tide", second_h: "Second high tide",
205 + first_l: "First low tide", second_l: "Second low tide"],
206 + get(tides, time_key) != nil do
207 + height = "#{time_key}v"
208 +
209 + {to_string(time_key),
210 + %{
211 + name: label,
212 + icon: "mdi:waves",
213 + device_class: "timestamp",
214 + value_template:
215 + root <> "{% set d = (e.data or [{}])[0] if e else {} %}{% set x = d.get(#{lit(time_key)}) %}" <>
216 + "{% if x %}" <> clock("x") <> "{{ t.isoformat() }}{% else %}None{% endif %}",
217 + json_attributes_template:
218 + root <> "{% set d = (e.data or [{}])[0] if e else {} %}{{ {'height': d.get(#{lit(height)})} | tojson }}"
219 + }}
220 + end
221 + end
222 +
223 + # Sun and moon. Home Assistant knows the sun already; the moon's phase and
224 + # times are the part it does not.
225 + defp components("ephem", [ephem | _], root) when is_map(ephem) do
226 + root = root <> "{% set d = (e.data or [{}])[0] if e else {} %}"
227 +
228 + phase =
229 + if get(ephem, :phase),
230 + do: [{"phase", %{name: "Moon phase", icon: "mdi:moon-waxing-crescent", value_template: root <> "{{ d.get('phase') }}"}}],
231 + else: []
232 +
233 + times =
234 + for k <- [:sunrise, :sunset, :moonrise, :moonset], get(ephem, k) != nil do
235 + {to_string(k),
236 + %{
237 + name: k |> to_string() |> String.capitalize(),
238 + device_class: "timestamp",
239 + value_template:
240 + root <> "{% set x = d.get(#{lit(k)}) %}{% if x %}" <> clock("x") <>
241 + "{{ t.isoformat() }}{% else %}None{% endif %}"
242 + }}
243 + end
244 +
245 + phase ++ times
246 + end
247 +
248 + # Everything else is a list of things without one obvious number in it: how
249 + # many there are, and the first few of them to read.
250 + defp components(_type, _data, root) do
251 + [
252 + {"items",
253 + %{
254 + name: "Items",
255 + icon: "mdi:format-list-bulleted",
256 + state_class: "measurement",
257 + value_template: root <> "{{ (e.data or []) | length if e else None }}",
258 + json_attributes_template: root <> "{{ {'items': ((e.data or []) if e else [])[:20]} | tojson }}"
259 + }}
260 + ]
261 + end
262 +
263 + defp weather_fields(units) do
264 + {temp, speed} =
265 + case to_string(units) do
266 + "imperial" -> {"°F", "mph"}
267 + "metric" -> {"°C", "m/s"}
268 + _ -> {"K", "m/s"}
269 + end
270 +
271 + measured = %{state_class: "measurement"}
272 +
273 + [
274 + {:weather, "conditions", %{icon: "mdi:weather-partly-cloudy"}},
275 + {:temp, "temperature", Map.merge(measured, %{device_class: "temperature", unit_of_measurement: temp})},
276 + {:feel, "feels like", Map.merge(measured, %{device_class: "temperature", unit_of_measurement: temp})},
277 + {:hum, "humidity", Map.merge(measured, %{device_class: "humidity", unit_of_measurement: "%"})},
278 + {:pressure, "pressure",
279 + Map.merge(measured, %{device_class: "atmospheric_pressure", unit_of_measurement: "hPa"})},
280 + {:wind, "wind", Map.merge(measured, %{device_class: "wind_speed", unit_of_measurement: speed})}
281 + ]
282 + end
283 +
284 + # -- Templates --------------------------------------------------------------
285 +
286 + @doc """
287 + Jinja that turns the clock time in `var` -- "17:14:12", "18:04:07.753",
288 + "14:25" -- into `t`, a datetime today.
289 +
290 + Built from midnight rather than with `today_at`, which refuses a GTFS time
291 + past 24:00 and anything with a fraction of a second. `wrap: true` moves a
292 + time more than twelve hours off onto the neighbouring day, for a departure
293 + list that crosses midnight: 00:10 read at 23:50 is tomorrow, not this
294 + morning.
295 +
296 + The times are the feed's own local time, and Home Assistant reads them in
297 + its own. They agree as long as the two are in the same place.
298 + """
299 + def clock(var, opts \\ []) do
300 + base =
301 + "{% set p = (#{var} ~ '').split(':') %}" <>
302 + "{% set t = today_at() + timedelta(hours=p[0] | int, minutes=p[1] | int, " <>
303 + "seconds=(p[2] if p | length > 2 else 0) | float) %}"
304 +
305 + if opts[:wrap] do
306 + base <>
307 + "{% set t = t + timedelta(days=(1 if t < now() - timedelta(hours=12) " <>
308 + "else (-1 if t > now() + timedelta(hours=12) else 0))) %}"
309 + else
310 + base
311 + end
312 + end
313 +
314 + # A Jinja string literal. JSON's escaping is a subset Jinja reads the same.
315 + defp lit(value) when is_atom(value) and not is_nil(value), do: lit(Atom.to_string(value))
316 + defp lit(nil), do: "none"
317 + defp lit(value) when is_number(value), do: to_string(value)
318 + defp lit(value), do: Jason.encode!(to_string(value))
319 +
320 + defp slug(value) do
321 + value
322 + |> to_string()
323 + |> String.downcase()
324 + |> String.replace(~r/[^a-z0-9]+/, "_")
325 + |> String.trim("_")
326 + end
327 +
328 + # Condensed in Elixir with atom keys, or decoded from somewhere with string
329 + # ones -- the types that pass their source's answer straight through do.
330 + defp get(map, key) when is_map(map), do: Map.get(map, key) || Map.get(map, to_string(key))
331 + defp get(_, _), do: nil
332 +
333 + defp aqi_label(k) do
334 + case to_string(k) do
335 + "pm25" -> "PM2.5"
336 + "pm10" -> "PM10"
337 + "ozone" -> "ozone"
338 + other -> String.upcase(other)
339 + end
340 + end
341 +
342 + defp mode_icon(mode) do
343 + case to_string(mode) do
344 + "Bus" -> "mdi:bus"
345 + "Trolleybus" -> "mdi:bus-electric"
346 + "Subway" -> "mdi:subway-variant"
347 + "LightRail" -> "mdi:tram"
348 + "Rail" -> "mdi:train"
349 + "Ferry" -> "mdi:ferry"
350 + _ -> "mdi:transit-connection-variant"
351 + end
352 + end
353 +
354 + defp model("gtfs_plus"), do: "Transit (Plus)"
355 + defp model("gtfs"), do: "Transit"
356 + defp model("gbfs"), do: "Bike share"
357 + defp model("aqi"), do: "Air quality"
358 + defp model(type), do: type |> String.capitalize()
359 +end
modified apps/room_sanctum/lib/room_sanctum/workers/ankyra_worker.ex
+92 −1
@@ -22,6 +22,12 @@ defmodule RoomSanctum.Worker.Ankyra do
22 22 @positions_ttl_s 1800
23 23 @positions_kept 64
24 24
25 + # How often a Home Assistant discovery config is sent again when nothing in
26 + # it has changed. RabbitMQ keeps no retained message for anything published
27 + # over AMQP, so a Home Assistant that has only just connected learns its
28 + # devices from this, or from asking -- see `discover/2`.
29 + @hass_refresh_s 600
30 +
25 31 @registry :zeus
26 32 def start_link(opts) do
27 33 GenServer.start_link(__MODULE__, opts, name: via_tuple("ankyra" <> opts[:id]))
@@ -47,7 +53,7 @@ defmodule RoomSanctum.Worker.Ankyra do
47 53 )
48 54
49 55 {:ok,
50 − %{id: opts[:id], ankyra: nil, requests: nil, last_request: nil, positions: []}}
56 + %{id: opts[:id], ankyra: nil, requests: nil, last_request: nil, positions: [], hass: %{}}}
51 57 end
52 58
53 59 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
@@ -85,6 +91,35 @@ defmodule RoomSanctum.Worker.Ankyra do
85 91 |> GenServer.cast({:publish_plus, data})
86 92 end
87 93
94 + @doc """
95 + Home Assistant discovery for a board, built here because the topic it is read
96 + from is this Ankyra's. `reading` is the topic the board went out on.
97 + """
98 + def discover_board(name, board, reading) do
99 + "ankyra#{name}"
100 + |> via_tuple()
101 + |> GenServer.cast({:discover_board, board, reading})
102 + end
103 +
104 + @doc """
105 + Home Assistant discovery, as `{object_id, config}` pairs from
106 + `RoomSanctum.Hass`, published under this Ankyra's discovery prefix.
107 +
108 + Sent when a device is new or has changed, and again every ten minutes when
109 + it has not, since there is no retained copy for a Home Assistant that
110 + connects in between. One that asks -- its birth message, pointed at
111 + `<topic>/publish/ha` -- gets everything at once instead of waiting.
112 +
113 + A device's entities only accumulate. A route with nothing due drops out of
114 + the board until something is, and dropping its sensor with it would have
115 + the dashboard rearrange itself every night; it reads unknown instead.
116 + """
117 + def discover(name, devices) do
118 + "ankyra#{name}"
119 + |> via_tuple()
120 + |> GenServer.cast({:discover, devices})
121 + end
122 +
88 123 @doc """
89 124 Where a client last said it was, or nil.
90 125
@@ -176,6 +211,19 @@ defmodule RoomSanctum.Worker.Ankyra do
176 211 {:noreply, state}
177 212 end
178 213
214 + def handle_cast({:discover_board, _board, _reading}, %{ankyra: nil} = state), do: {:noreply, state}
215 +
216 + def handle_cast({:discover_board, board, reading}, state) do
217 + devices = RoomSanctum.Hass.board_devices(state.id, state.ankyra.topic, board, reading)
218 + handle_cast({:discover, devices}, state)
219 + end
220 +
221 + def handle_cast({:discover, _devices}, %{ankyra: nil} = state), do: {:noreply, state}
222 +
223 + def handle_cast({:discover, devices}, state) do
224 + {:noreply, Enum.reduce(devices, state, &announce/2)}
225 + end
226 +
179 227 def handle_cast({:publish_img, data}, state) do
180 228 IO.puts("and here")
181 229
@@ -308,6 +356,10 @@ defmodule RoomSanctum.Worker.Ankyra do
308 356 # more than enough to feel immediate, and the Pythiae's own tick covers
309 357 # everything else.
310 358 defp serve_request(state, key) do
359 + # Home Assistant saying it has (re)started: it has none of its devices
360 + # until they are sent again, so forget having sent them. The board that
361 + # follows announces its own; a Keryx does on its next tick.
362 + state = if String.ends_with?(key, ".ha"), do: %{state | hass: %{}}, else: state
311 363 now = DateTime.utc_now()
312 364
313 365 if too_soon?(state.last_request, now) do
@@ -394,6 +446,45 @@ defmodule RoomSanctum.Worker.Ankyra do
394 446 )
395 447 end
396 448
449 + # One device's config, if it is new, has changed, or is due again. What was
450 + # sent is kept so the next board's components can be merged into it rather
451 + # than replacing it.
452 + defp announce({object_id, config}, state) do
453 + now = System.monotonic_time(:second)
454 +
455 + {config, due?} =
456 + case Map.get(state.hass, object_id) do
457 + nil ->
458 + {config, true}
459 +
460 + %{config: sent, at: at} ->
461 + merged = Map.update!(config, :components, &Map.merge(sent.components, &1))
462 + {merged, merged != sent or now - at >= @hass_refresh_s}
463 + end
464 +
465 + if due? do
466 + publish_discovery(state, object_id, config)
467 + put_in(state, [:hass, object_id], %{config: config, at: now})
468 + else
469 + state
470 + end
471 + end
472 +
473 + defp publish_discovery(state, object_id, config) do
474 + case AMQP.Application.get_channel(:default) do
475 + {:ok, chan} ->
476 + AMQP.Basic.publish(
477 + chan,
478 + "amq.topic",
479 + RoomSanctum.Hass.config_key(state.ankyra.topic, object_id),
480 + Jason.encode!(config)
481 + )
482 +
483 + {:error, error} ->
484 + Logger.warning("ankyra #{state.id}: could not publish discovery: #{inspect(error)}")
485 + end
486 + end
487 +
397 488 defp too_soon?(nil, _now), do: false
398 489
399 490 defp too_soon?(last, now), do: DateTime.diff(now, last, :millisecond) < @request_floor_ms
modified apps/room_sanctum/lib/room_sanctum/workers/keryx_worker.ex
+42 −8
@@ -302,18 +302,45 @@ defmodule RoomSanctum.Worker.Keryx do
302 302
303 303 # Private helper to query all workers and publish to MQTT
304 304 defp query_all_workers(keryx_q, keryx) do
305 − keryx_q
306 − |> Enum.map(fn q ->
307 − r = execute_query(q)
305 + results =
306 + keryx_q
307 + |> Enum.map(fn q ->
308 + r = execute_query(q)
308 309
309 − # Publish to MQTT: full array + individual items
310 − publish_query_data(keryx.id, keryx.name, q.source.type, q.id, r, q)
310 + # Publish to MQTT: full array + individual items
311 + meta = publish_query_data(keryx.id, keryx.name, q.source.type, q.id, r, q)
311 312
312 − {{q.id, q.source.type}, r}
313 − end)
313 + {{q.id, q.source.type}, r, meta}
314 + end)
315 +
316 + discover(keryx, results)
317 +
318 + results
319 + |> Enum.map(fn {key, r, _meta} -> {key, r} end)
314 320 |> Enum.into(%{})
315 321 end
316 322
323 + # Home Assistant discovery for each query's `.meta`, through every Ankyra the
324 + # Keryx's account has: those are the rabbit users allowed to read it, and so
325 + # the only places a Home Assistant could be listening from. The Ankyra
326 + # decides whether anything is new enough to send.
327 + defp discover(%Keryx{user_id: nil}, _results), do: :ok
328 +
329 + defp discover(keryx, results) do
330 + devices =
331 + for {{query_id, type}, _r, meta} <- results, meta do
332 + RoomSanctum.Hass.keryx_device(keryx.id, query_id, type, meta)
333 + end
334 +
335 + if devices != [] do
336 + for ankyra <- RoomSanctum.Accounts.list_users_rabbit({:user, keryx.user_id}) do
337 + RoomSanctum.Worker.Ankyra.discover(ankyra.id, devices)
338 + end
339 + end
340 +
341 + :ok
342 + end
343 +
317 344 defp publish_query_data(keryx_id, _name, type, query_id, data, query) when is_list(data) do
318 345 base_topic = query_topic(keryx_id, type, query_id)
319 346
@@ -326,7 +353,7 @@ defmodule RoomSanctum.Worker.Keryx do
326 353 AMQP.Basic.publish(chan, "amq.topic", base_topic, data |> Poison.encode!())
327 354
328 355 # Publish metadata using condense function
329 − publish_metadata(chan, base_topic, type, query_id, data, query)
356 + metadata = publish_metadata(chan, base_topic, type, query_id, data, query)
330 357
331 358 # For GTFS, group by route and publish hierarchically
332 359 if type == :gtfs do
@@ -341,8 +368,11 @@ defmodule RoomSanctum.Worker.Keryx do
341 368 end)
342 369 end
343 370
371 + metadata
372 +
344 373 {:error, error} ->
345 374 Logger.error("Failed to publish to MQTT: #{inspect(error)}")
375 + nil
346 376 end
347 377 end
348 378
@@ -353,9 +383,11 @@ defmodule RoomSanctum.Worker.Keryx do
353 383 case AMQP.Application.get_channel(:default) do
354 384 {:ok, chan} ->
355 385 AMQP.Basic.publish(chan, "amq.topic", topic, data |> Poison.encode!())
386 + nil
356 387
357 388 {:error, error} ->
358 389 Logger.error("Failed to publish to MQTT: #{inspect(error)}")
390 + nil
359 391 end
360 392 end
361 393
@@ -369,6 +401,8 @@ defmodule RoomSanctum.Worker.Keryx do
369 401
370 402 meta_topic = "#{base_topic}.meta"
371 403 AMQP.Basic.publish(chan, "amq.topic", meta_topic, metadata |> Poison.encode!())
404 +
405 + metadata
372 406 end
373 407
374 408 # Publish GTFS data hierarchically by route and destination
modified apps/room_sanctum/lib/room_sanctum/workers/pythiae_worker.ex
+14 −0
@@ -156,9 +156,23 @@ defmodule RoomSanctum.Worker.Pythiae do
156 156 for a <- state.pythiae.ankyra do
157 157 RoomSanctum.Worker.Ankyra.publish(a, basic)
158 158 if plus, do: RoomSanctum.Worker.Ankyra.publish_plus(a, plus)
159 + if discoverable?(state.pythiae), do: discover(a, basic, plus)
159 160 end
160 161 end
161 162
163 + # Home Assistant reads the richest board there is -- Plus where this
164 + # Pythiae publishes it, since it is the whole of Basic and more.
165 + defp discover(ankyra, _basic, plus) when not is_nil(plus),
166 + do: RoomSanctum.Worker.Ankyra.discover_board(ankyra, plus, :plus)
167 +
168 + defp discover(ankyra, basic, nil),
169 + do: RoomSanctum.Worker.Ankyra.discover_board(ankyra, basic, :basic)
170 +
171 + # Not a Plani's board. Its entries are whatever is near its client this
172 + # minute, and every one would be a device in Home Assistant for ever after.
173 + defp discoverable?(%{curr_plani: plani}), do: is_nil(plani)
174 + defp discoverable?(_), do: false
175 +
162 176 defp plus?(%{plus: true}), do: true
163 177 defp plus?(_), do: false
164 178
added apps/room_sanctum/test/room_sanctum/hass_test.exs
+114 −0
@@ -0,0 +1,114 @@
1 +defmodule RoomSanctum.HassTest do
2 + use ExUnit.Case, async: true
3 +
4 + alias RoomSanctum.Hass
5 +
6 + defp board do
7 + %{
8 + "gtfs-2" => %{
9 + query: %{name: "Broadway @ Holland St", meta: %{}},
10 + data: [
11 + %{
12 + route: "88",
13 + route_name: "88",
14 + dest: "Lechmere",
15 + dir: "Inbound",
16 + mode: "Bus",
17 + times: ["16:55:00"],
18 + times_live: ["16:56:25"]
19 + }
20 + ]
21 + },
22 + "gbfs-4" => %{
23 + query: %{name: "Teele Square", meta: %{}},
24 + data: [
25 + %{name: "Teele Square", id: "f834", avail: 1, avail_elec: 0, docks_avail: 14},
26 + %{kind: :free_bike, name: "E-bike", id: "b1", lat: 0, lon: 0}
27 + ]
28 + },
29 + "aqi-34" => %{
30 + query: %{name: "CAMBERVILLE", meta: %{}},
31 + data: [%{pm25: 34, no2: 6, combined: "PM2.5: 34, NO2: 6, ", name: "Boston Metro"}]
32 + },
33 + "cronos-9" => %{query: %{name: "Timers", meta: %{}}, data: [%{id: 1}]}
34 + }
35 + end
36 +
37 + defp devices(board, reading \\ :basic),
38 + do: board |> then(&Hass.board_devices(7, "Velvety", &1, reading)) |> Map.new()
39 +
40 + test "a device per query, read from the Ankyra's own topic" do
41 + devices = devices(board())
42 +
43 + assert Map.keys(devices) |> Enum.sort() ==
44 + ~w(apollos_a7_aqi_34 apollos_a7_cronos_9 apollos_a7_gbfs_4 apollos_a7_gtfs_2)
45 +
46 + stop = devices["apollos_a7_gtfs_2"]
47 + assert stop.state_topic == "Velvety"
48 + assert stop.device.name == "Broadway @ Holland St"
49 + assert %{name: "88 to Lechmere", device_class: "timestamp"} = stop.components["88_lechmere_inbound"]
50 + end
51 +
52 + test "picks a route out of the payload by what it is, not where it sits" do
53 + template = devices(board())["apollos_a7_gtfs_2"].components["88_lechmere_inbound"].value_template
54 +
55 + assert template =~ ~s|value_json.get("gtfs-2")|
56 + assert template =~ ~s|selectattr('route', 'eq', "88")|
57 + assert template =~ ~s|selectattr('dest', 'eq', "Lechmere")|
58 + end
59 +
60 + test "Plus is read from its own topic, and keeps the devices Basic made" do
61 + plus =
62 + board()
63 + |> Map.delete("gtfs-2")
64 + |> Map.put("gtfs_plus-2", put_in(board()["gtfs-2"], [:data], [%{route: "88", dest: "Lechmere", dir: "Inbound", arrivals: []}]))
65 +
66 + stop = devices(plus, :plus)["apollos_a7_gtfs_2"]
67 +
68 + assert stop.state_topic == "Velvety/plus"
69 + assert stop.components["88_lechmere_inbound"].value_template =~ "d.arrivals"
70 + end
71 +
72 + test "a dock is three sensors, and a loose bike is none" do
73 + components = devices(board())["apollos_a7_gbfs_4"].components
74 +
75 + assert Map.keys(components) |> Enum.sort() == ~w(f834_avail f834_avail_elec f834_docks_avail)
76 + assert components["f834_avail"].name == "Teele Square bikes"
77 + end
78 +
79 + test "each pollutant measured, and not the summary string" do
80 + components = devices(board())["apollos_a7_aqi_34"].components
81 +
82 + assert Map.keys(components) |> Enum.sort() == ~w(boston_metro_no2 boston_metro_pm25)
83 + assert components["boston_metro_pm25"].device_class == "aqi"
84 + end
85 +
86 + test "a type with no obvious number is counted" do
87 + assert %{"items" => %{name: "Items"}} = devices(board())["apollos_a7_cronos_9"].components
88 + end
89 +
90 + test "unique ids are scoped to the Ankyra, so two can feed one Home Assistant" do
91 + ids =
92 + for {_, d} <- devices(board()), {_, c} <- d.components, do: c.unique_id
93 +
94 + assert Enum.all?(ids, &String.starts_with?(&1, "apollos_a7_"))
95 + assert ids == Enum.uniq(ids)
96 + end
97 +
98 + test "a Keryx query is read from its .meta, and expires when it stops" do
99 + {object_id, device} = Hass.keryx_device(5, 12, :gtfs, board()["gtfs-2"])
100 +
101 + assert object_id == "apollos_k5_12"
102 + assert device.state_topic == "apollos/keryx/5/gtfs/12/meta"
103 + assert device.expire_after == 300
104 + assert device.components["88_lechmere_inbound"].value_template =~ "{% set e = value_json %}"
105 + end
106 +
107 + test "discovery sits under the Ankyra's topic, where its rabbit user may read" do
108 + assert Hass.config_key("Velvety", "apollos_a7_gtfs_2") == "Velvety.ha.device.apollos_a7_gtfs_2.config"
109 + end
110 +
111 + test "every config encodes" do
112 + for {_, d} <- devices(board()), do: assert({:ok, _} = Jason.encode(d))
113 + end
114 +end

Parents: 2cdebbe