GITGUD

neiam neiam /lambent_ex

Lambent Aether 5

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

⭐

Log in to mark this repository.

android prep

18ced42 · Gabriel Morell · 2026-08-17 23:11

27 files +811 -1

Files changed

modified lib/lambent_ex/scan/8266-7777.ex
+6 −0
@@ -92,6 +92,12 @@ defmodule LambentEx.Scan.ESP8266x7777 do
92 92 via_tuple("scan_8266") |> GenServer.call(:get_devices)
93 93 end
94 94
95 + # devices as the ui gets them off the wire, for anything that needs a snapshot
96 + # rather than waiting for the next publish
97 + def tagged_devices() do
98 + devices() |> add_type()
99 + end
100 +
95 101 def manual(ip) do
96 102 via_tuple("scan_8266") |> GenServer.cast({:manual, ip})
97 103 end
added lib/lambent_ex_web/channels/devices_channel.ex
+139 −0
@@ -0,0 +1,139 @@
1 +defmodule LambentExWeb.DevicesChannel do
2 + @moduledoc """
3 + Mirrors `LambentExWeb.DevicesLive.Index`: the discovered device map, plus
4 + rename / place / order / poke / probe.
5 +
6 + Pushes
7 + * `"devices"` — `%{devices: %{mac => device}}`, a snapshot on join and a
8 + delta on every publish. Merge them by mac, the way the liveview does.
9 + * `"probe_result"` — `%{ip: ip, status: "lambent" | "not_lambent" | "no_iface", sox: n}`
10 + """
11 +
12 + use Phoenix.Channel
13 + require Logger
14 +
15 + alias LambentEx.Scan.ESP8266x7777, as: S
16 + alias LambentExWeb.Channels.Payload
17 +
18 + @pubsub LambentEx.PubSub
19 +
20 + @impl true
21 + def join("devices", _params, socket) do
22 + Phoenix.PubSub.subscribe(@pubsub, "scan-82667777")
23 + Phoenix.PubSub.subscribe(@pubsub, "scan-manual")
24 + send(self(), :snapshot)
25 + {:ok, assign(socket, :devices, %{})}
26 + end
27 +
28 + # mqtt devices only exist in the publishing process, so the snapshot covers
29 + # the 7777 ones and the rest land on the next tick
30 + @impl true
31 + def handle_info(:snapshot, socket) do
32 + devices = S.tagged_devices()
33 + push(socket, "devices", %{devices: Payload.devices(devices)})
34 + {:noreply, assign(socket, :devices, devices)}
35 + end
36 +
37 + @impl true
38 + def handle_info({:devices_pub, devices}, socket) do
39 + push(socket, "devices", %{devices: Payload.devices(devices)})
40 + {:noreply, assign(socket, :devices, Map.merge(socket.assigns.devices, devices))}
41 + end
42 +
43 + @impl true
44 + def handle_info({:manual_result, ip, status}, socket) do
45 + payload =
46 + case status do
47 + {:lambent, sox} -> %{ip: ip, status: "lambent", sox: max(sox, 1)}
48 + other -> %{ip: ip, status: to_string(other)}
49 + end
50 +
51 + push(socket, "probe_result", payload)
52 + {:noreply, socket}
53 + end
54 +
55 + @impl true
56 + def handle_in("rename", %{"id" => id} = params, socket) do
57 + name = params |> Map.get("name") |> blank_to_nil()
58 +
59 + case type_of(socket, id) do
60 + t when t in ["7777-8266", "7777-flex"] -> S.rename(:mac, id, name)
61 + "mqtt-tasmota" -> LambentEx.MQTT.rename(dname(socket, id), id, name)
62 + other -> unhandled("rename", other)
63 + end
64 +
65 + {:reply, :ok, socket}
66 + end
67 +
68 + @impl true
69 + def handle_in("replace", %{"id" => id} = params, socket) do
70 + place = params |> Map.get("place") |> blank_to_nil()
71 +
72 + case type_of(socket, id) do
73 + t when t in ["7777-8266", "7777-flex"] -> S.replace(:mac, id, place)
74 + "mqtt-tasmota" -> LambentEx.MQTT.replace(dname(socket, id), id, place)
75 + other -> unhandled("replace", other)
76 + end
77 +
78 + {:reply, :ok, socket}
79 + end
80 +
81 + @impl true
82 + def handle_in("reorder", %{"id" => id, "ord" => ord}, socket) do
83 + case type_of(socket, id) do
84 + t when t in ["7777-8266", "7777-flex"] -> S.reorder(:mac, id, ord)
85 + other -> unhandled("reorder", other)
86 + end
87 +
88 + {:reply, :ok, socket}
89 + end
90 +
91 + @impl true
92 + def handle_in("poke", %{"id" => id} = params, socket) do
93 + case type_of(socket, id) do
94 + t when t in ["7777-8266", "7777-flex"] -> S.poke(id, poke_mode(params))
95 + "mqtt-tasmota" -> LambentEx.MQTT.do_poke(dname(socket, id), id)
96 + other -> unhandled("poke", other)
97 + end
98 +
99 + {:reply, :ok, socket}
100 + end
101 +
102 + @impl true
103 + def handle_in("probe", %{"ip" => ip}, socket) do
104 + ip = String.trim(ip)
105 +
106 + case IP.from_string(ip) do
107 + {:ok, addr} when tuple_size(addr) == 4 ->
108 + S.manual(ip)
109 + {:reply, :ok, socket}
110 +
111 + _otherwise ->
112 + {:reply, {:error, %{reason: "#{ip} is not a valid ipv4 address"}}, socket}
113 + end
114 + end
115 +
116 + defp poke_mode(params) do
117 + case params |> Map.get("mode") do
118 + "indicator" -> :indicator
119 + "strands" -> :strands
120 + _otherwise -> :auto
121 + end
122 + end
123 +
124 + defp type_of(socket, id) do
125 + socket.assigns.devices |> Map.get(id, %{}) |> Map.get("type")
126 + end
127 +
128 + defp dname(socket, id) do
129 + socket.assigns.devices |> Map.get(id, %{}) |> Map.get("dname")
130 + end
131 +
132 + defp blank_to_nil(""), do: nil
133 + defp blank_to_nil(other), do: other
134 +
135 + defp unhandled(cmd, type) do
136 + Logger.info("Need handler for #{cmd} on this type: #{type}")
137 + :ok
138 + end
139 +end
added lib/lambent_ex_web/channels/links_channel.ex
+65 −0
@@ -0,0 +1,65 @@
1 +defmodule LambentExWeb.LinksChannel do
2 + @moduledoc """
3 + Mirrors `LambentExWeb.LinksLive.Index`: running links, toggle / persist /
4 + quit, and both create forms.
5 +
6 + Pushes
7 + * `"link"` — one link's published state, keyed by `name`, including the
8 + `chan` it drives and whether it is currently enabled.
9 + """
10 +
11 + use Phoenix.Channel
12 + require Logger
13 +
14 + alias LambentEx.Schema.Links
15 + alias LambentEx.Schema.LinksB
16 + alias LambentEx.Scan.ESP8266x7777, as: S
17 + alias LambentExWeb.Channels.Payload
18 +
19 + @pubsub LambentEx.PubSub
20 +
21 + @impl true
22 + def join("links", _params, socket) do
23 + Phoenix.PubSub.subscribe(@pubsub, "links_idx")
24 + {:ok, socket}
25 + end
26 +
27 + @impl true
28 + def handle_info({:links_pub, link}, socket) do
29 + push(socket, "link", Payload.wire(link))
30 + {:noreply, socket}
31 + end
32 +
33 + @impl true
34 + def handle_in(cmd, %{"id" => id}, socket) when cmd in ~w(toggle persist quit) do
35 + apply(LambentEx.Link, String.to_existing_atom(cmd), [id])
36 + {:reply, :ok, socket}
37 + end
38 +
39 + # %{"link" => %{"name" =>, "source_id" =>, "target_id" =>, "chan" => "all" | "0"}}
40 + # chan is optional, and is forced to 0 for anything with a single channel
41 + @impl true
42 + def handle_in("create", %{"link" => params}, socket) do
43 + case Links.update_links(%Links{}, params, S.tagged_devices()) do
44 + {:ok, _link} -> {:reply, :ok, socket}
45 + {:error, changeset} -> {:reply, {:error, %{errors: errors(changeset)}}, socket}
46 + end
47 + end
48 +
49 + # same, with "target_ids" as a list
50 + @impl true
51 + def handle_in("create_bulk", %{"link" => params}, socket) do
52 + case LinksB.update_links(%LinksB{}, params, S.tagged_devices()) do
53 + {:ok, _links} -> {:reply, :ok, socket}
54 + {:error, changeset} -> {:reply, {:error, %{errors: errors(changeset)}}, socket}
55 + end
56 + end
57 +
58 + defp errors(changeset) do
59 + Ecto.Changeset.traverse_errors(changeset, fn {msg, opts} ->
60 + Regex.replace(~r"%{(\w+)}", msg, fn _, key ->
61 + opts |> Keyword.get(String.to_existing_atom(key), "") |> to_string()
62 + end)
63 + end)
64 + end
65 +end
added lib/lambent_ex_web/channels/machines_channel.ex
+112 −0
@@ -0,0 +1,112 @@
1 +defmodule LambentExWeb.MachinesChannel do
2 + @moduledoc """
3 + Mirrors `LambentExWeb.MachinesLive.Index`: running machines, their preview
4 + firehose, the speed/brightness/persist/quit controls, and both ways of
5 + starting one (library entry or a full machine definition).
6 +
7 + Pushes
8 + * `"machine"` — one machine's published state, keyed by `id`
9 + * `"preview"` — `%{name: name, pixels: [[r,g,b], ...]}`, first 32 pixels,
10 + up to ~4Hz per machine. Join with `%{"previews" => false}` to skip it.
11 + """
12 +
13 + use Phoenix.Channel
14 + require Logger
15 +
16 + alias LambentExWeb.Channels.Payload
17 +
18 + @pubsub LambentEx.PubSub
19 +
20 + @impl true
21 + def join("machines", params, socket) do
22 + Phoenix.PubSub.subscribe(@pubsub, "machines_idx")
23 +
24 + previews = Map.get(params, "previews", true)
25 + if previews, do: Phoenix.PubSub.subscribe(@pubsub, "machine:")
26 +
27 + send(self(), :after_join)
28 + {:ok, socket}
29 + end
30 +
31 + @impl true
32 + def handle_info(:after_join, socket) do
33 + push(socket, "library", %{library: library()})
34 + {:noreply, socket}
35 + end
36 +
37 + @impl true
38 + def handle_info({:machines_pub, machine}, socket) do
39 + push(socket, "machine", Payload.wire(machine))
40 + {:noreply, socket}
41 + end
42 +
43 + @impl true
44 + def handle_info({:firehose, {name, pixels}}, socket) do
45 + push(socket, "preview", %{name: name, pixels: Payload.wire(pixels)})
46 + {:noreply, socket}
47 + end
48 +
49 + @impl true
50 + def handle_in(cmd, %{"id" => id}, socket)
51 + when cmd in ~w(faster slower brighter dimmer persist quit) do
52 + apply(LambentEx.Machine, String.to_existing_atom(cmd), [id])
53 + {:reply, :ok, socket}
54 + end
55 +
56 + # pushed on join too, so a client can just listen rather than ask
57 + @impl true
58 + def handle_in("library", _params, socket) do
59 + lib = library()
60 + push(socket, "library", %{library: lib})
61 + {:reply, {:ok, %{library: lib}}, socket}
62 + end
63 +
64 + @impl true
65 + def handle_in("library_add", %{"cls" => cls, "entry" => entry}, socket) do
66 + spec =
67 + LambentEx.Machines.Library.machines()
68 + |> get_in([safe_atom(cls), safe_atom(entry)])
69 +
70 + case spec do
71 + nil ->
72 + {:reply, {:error, %{reason: "no library entry #{cls}/#{entry}"}}, socket}
73 +
74 + spec ->
75 + LambentEx.MachineSupervisor.start_child(
76 + spec[:machine],
77 + spec[:args],
78 + entry,
79 + spec[:mach_opts] || []
80 + )
81 +
82 + {:reply, :ok, socket}
83 + end
84 + end
85 +
86 + # same params the web form posts: %{"name" => _, "type" => _, "class" => %{...}}
87 + @impl true
88 + def handle_in("create", %{"machine" => params}, socket) do
89 + case LambentEx.Schema.Machines.update_machines(%LambentEx.Schema.Machines{}, params) do
90 + {:ok, _machine} -> {:reply, :ok, socket}
91 + {:error, changeset} -> {:reply, {:error, %{errors: errors(changeset)}}, socket}
92 + end
93 + end
94 +
95 + defp library, do: Payload.wire(LambentEx.Machines.Library.machines())
96 +
97 + # the library is keyed by atoms, and a client should not be able to grow the
98 + # atom table by asking for entries that do not exist
99 + defp safe_atom(str) do
100 + String.to_existing_atom(str)
101 + rescue
102 + ArgumentError -> nil
103 + end
104 +
105 + defp errors(changeset) do
106 + Ecto.Changeset.traverse_errors(changeset, fn {msg, opts} ->
107 + Regex.replace(~r"%{(\w+)}", msg, fn _, key ->
108 + opts |> Keyword.get(String.to_existing_atom(key), "") |> to_string()
109 + end)
110 + end)
111 + end
112 +end
added lib/lambent_ex_web/channels/payload.ex
+54 −0
@@ -0,0 +1,54 @@
1 +defmodule LambentExWeb.Channels.Payload do
2 + @moduledoc """
3 + The pubsub payloads the liveviews consume are built for elixir, not for the
4 + wire: ip addresses are tuples, machine and link options are keyword lists,
5 + and map keys are a mix of atoms and strings. None of that survives a json
6 + encoder, so everything a channel pushes goes through here first.
7 + """
8 +
9 + @doc """
10 + Recursively makes a term json-safe. Keyword lists become objects, other
11 + tuples become arrays, atoms and structs become strings or plain maps.
12 + """
13 + def wire(%DateTime{} = dt), do: DateTime.to_iso8601(dt)
14 + def wire(%NaiveDateTime{} = dt), do: NaiveDateTime.to_iso8601(dt)
15 + def wire(%_struct{} = s), do: s |> Map.from_struct() |> wire()
16 +
17 + def wire(%{} = map) do
18 + Map.new(map, fn {k, v} -> {key(k), wire(v)} end)
19 + end
20 +
21 + def wire(list) when is_list(list) do
22 + case Keyword.keyword?(list) and list != [] do
23 + true -> list |> Map.new() |> wire()
24 + false -> Enum.map(list, &wire/1)
25 + end
26 + end
27 +
28 + def wire(tuple) when is_tuple(tuple), do: tuple |> Tuple.to_list() |> wire()
29 +
30 + def wire(atom) when is_atom(atom) and not is_boolean(atom) and not is_nil(atom),
31 + do: to_string(atom)
32 +
33 + def wire(pid) when is_pid(pid), do: inspect(pid)
34 + def wire(other), do: other
35 +
36 + @doc """
37 + Devices carry their address as an `:inet` tuple, which reads better on the
38 + wire as the dotted string the ui shows.
39 + """
40 + def device(device) when is_map(device) do
41 + case Map.get(device, "ip") do
42 + ip when is_tuple(ip) -> device |> Map.put("ip", ip |> Tuple.to_list() |> Enum.join("."))
43 + _otherwise -> device
44 + end
45 + |> wire()
46 + end
47 +
48 + def devices(devices) when is_map(devices) do
49 + Map.new(devices, fn {mac, device} -> {mac, device(device)} end)
50 + end
51 +
52 + defp key(k) when is_binary(k), do: k
53 + defp key(k), do: to_string(k)
54 +end
added lib/lambent_ex_web/channels/services_channel.ex
+77 −0

Click to load diff…

added lib/lambent_ex_web/channels/user_socket.ex
+24 −0

Click to load diff…

modified lib/lambent_ex_web/components/layouts.ex
+56 −0

Click to load diff…

modified lib/lambent_ex_web/components/layouts/app.html.heex
+32 −1

Click to load diff…

modified lib/lambent_ex_web/endpoint.ex
+6 −0

Click to load diff…

modified mix.exs
+1 −0

Click to load diff…

modified mix.lock
+1 −0

Click to load diff…

added test/lambent_ex_web/channels/channels_test.exs
+196 −0

Click to load diff…

added test/lambent_ex_web/live/pairing_test.exs
+25 −0

Click to load diff…

added test/support/channel_case.ex
+17 −0

Click to load diff…

Parents: e27d884