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.

Credo: Cronos, the machine step and MQTT dispatch split into small functions

9675d32 · Gabe Morell · 2026-11-04 20:45

5 files +198 -216

Files changed

modified lib/lambent_ex/cronos.ex
+48 −62
@@ -75,71 +75,57 @@ defmodule LambentEx.Cronos do
75 75 Process.send_after(self(), :work, :timer.minutes(1))
76 76 curr = LambentEx.Meta.get_cronos() |> group(state)
77 77
78 − for i <- curr do
79 − case i[:action] do
80 − :machine_speed ->
81 − case i[:arg] do
82 − :faster ->
83 − i[:targets] |> Enum.map(&LambentEx.Machine.faster/1)
84 −
85 − :slower ->
86 − i[:targets] |> Enum.map(&LambentEx.Machine.slower/1)
87 −
88 − otherwise ->
89 − i[:targets] |> Enum.map(fn t -> LambentEx.Machine.set_speed(t, otherwise) end)
90 − end
91 −
92 − :machine_bright ->
93 − case i[:arg] do
94 − :dimmer ->
95 − i[:targets] |> Enum.map(&LambentEx.Machine.dimmer/1)
96 −
97 − :brighter ->
98 − i[:targets] |> Enum.map(&LambentEx.Machine.brighter/1)
99 −
100 − otherwise ->
101 − i[:targets] |> Enum.map(fn t -> LambentEx.Machine.set_brightness(t, otherwise) end)
102 − end
103 −
104 − # the global clamp has no targets, it is the one ceiling over everything
105 − :global_bright ->
106 − case i[:arg] do
107 − :dimmer -> LambentEx.Brightness.dimmer()
108 − :brighter -> LambentEx.Brightness.brighter()
109 − otherwise -> LambentEx.Brightness.set_global(otherwise)
110 − end
111 −
112 − # targets here are places, not machines
113 − :place_bright ->
114 − case i[:arg] do
115 − :dimmer ->
116 − i[:targets] |> Enum.map(&LambentEx.Brightness.dimmer/1)
117 −
118 − :brighter ->
119 − i[:targets] |> Enum.map(&LambentEx.Brightness.brighter/1)
120 −
121 − otherwise ->
122 − i[:targets] |> Enum.map(fn t -> LambentEx.Brightness.set_place(t, otherwise) end)
123 − end
124 −
125 − :link_on ->
126 − i[:targets] |> Enum.map(&LambentEx.Link.enable/1)
127 −
128 − :link_off ->
129 − i[:targets] |> Enum.map(&LambentEx.Link.disable/1)
130 −
131 − :link_toggle ->
132 − i[:targets] |> Enum.map(&LambentEx.Link.toggle/1)
133 −
134 − # an entry saved by a version that knew an action this one does not
135 − # should not take the whole scheduler down with it
136 − otherwise ->
137 − Logger.warning("#{@s} Skipping #{key_of(i)}, nothing here does #{inspect(otherwise)}")
138 − end
139 − end
78 + for i <- curr, do: run_entry(i)
140 79
141 80 {:noreply, state}
142 81 end
82 +
83 + defp run_entry(i), do: run_action(i[:action], i[:arg], i[:targets], i)
84 +
85 + defp run_action(:machine_speed, :faster, targets, _i),
86 + do: Enum.each(targets, &LambentEx.Machine.faster/1)
87 +
88 + defp run_action(:machine_speed, :slower, targets, _i),
89 + do: Enum.each(targets, &LambentEx.Machine.slower/1)
90 +
91 + defp run_action(:machine_speed, speed, targets, _i),
92 + do: Enum.each(targets, fn t -> LambentEx.Machine.set_speed(t, speed) end)
93 +
94 + defp run_action(:machine_bright, :dimmer, targets, _i),
95 + do: Enum.each(targets, &LambentEx.Machine.dimmer/1)
96 +
97 + defp run_action(:machine_bright, :brighter, targets, _i),
98 + do: Enum.each(targets, &LambentEx.Machine.brighter/1)
99 +
100 + defp run_action(:machine_bright, level, targets, _i),
101 + do: Enum.each(targets, fn t -> LambentEx.Machine.set_brightness(t, level) end)
102 +
103 + # the global clamp has no targets, it is the one ceiling over everything
104 + defp run_action(:global_bright, :dimmer, _targets, _i), do: LambentEx.Brightness.dimmer()
105 + defp run_action(:global_bright, :brighter, _targets, _i), do: LambentEx.Brightness.brighter()
106 + defp run_action(:global_bright, level, _targets, _i), do: LambentEx.Brightness.set_global(level)
107 +
108 + # targets here are places, not machines
109 + defp run_action(:place_bright, :dimmer, targets, _i),
110 + do: Enum.each(targets, &LambentEx.Brightness.dimmer/1)
111 +
112 + defp run_action(:place_bright, :brighter, targets, _i),
113 + do: Enum.each(targets, &LambentEx.Brightness.brighter/1)
114 +
115 + defp run_action(:place_bright, level, targets, _i),
116 + do: Enum.each(targets, fn t -> LambentEx.Brightness.set_place(t, level) end)
117 +
118 + defp run_action(:link_on, _arg, targets, _i), do: Enum.each(targets, &LambentEx.Link.enable/1)
119 + defp run_action(:link_off, _arg, targets, _i), do: Enum.each(targets, &LambentEx.Link.disable/1)
120 +
121 + defp run_action(:link_toggle, _arg, targets, _i),
122 + do: Enum.each(targets, &LambentEx.Link.toggle/1)
123 +
124 + # an entry saved by a version that knew an action this one does not
125 + # should not take the whole scheduler down with it
126 + defp run_action(otherwise, _arg, _targets, i) do
127 + Logger.warning("#{@s} Skipping #{key_of(i)}, nothing here does #{inspect(otherwise)}")
128 + end
143 129 end
144 130
145 131 ### Examples
modified lib/lambent_ex/machine.ex
+107 −116
@@ -2,6 +2,8 @@ defmodule LambentEx.Machine do
2 2 @moduledoc false
3 3 use Parent.GenServer
4 4
5 + require Logger
6 +
5 7 @registry :lambent_machine
6 8
7 9 @pubsub_name LambentEx.PubSub
@@ -28,10 +30,10 @@ defmodule LambentEx.Machine do
28 30 1000 => :ONES,
29 31 2000 => :TWOS,
30 32 5000 => :FIVES,
31 − 10000 => :TENS,
32 − 20000 => :TWENTYS,
33 − 30000 => :THIRTYS,
34 − 60000 => :MINS
33 + 10_000 => :TENS,
34 + 20_000 => :TWENTYS,
35 + 30_000 => :THIRTYS,
36 + 60_000 => :MINS
35 37 }
36 38
37 39 # check to see if machine-opts[:step]-name is better in machine-name
@@ -105,7 +107,7 @@ defmodule LambentEx.Machine do
105 107 step: opts[:step],
106 108 steps: pids,
107 109 name: opts[:name],
108 − # todo configurable
110 + # not configurable yet
109 111 speed: opts[:speed],
110 112 bright_tgt: 255,
111 113 bright_curr: 255,
@@ -312,57 +314,93 @@ defmodule LambentEx.Machine do
312 314 state = state |> Map.put(:bright_global, LambentEx.Brightness.global())
313 315
314 316 state[:steps]
315 − |> Enum.map(&GenServer.cast(&1, :step))
317 + |> Enum.each(&GenServer.cast(&1, :step))
318 +
319 + data = read_steps(state)
320 +
321 + # expand single to fill cnt. kept out here rather than rebound inside
322 + # broadcast_frame, so the firehose shows the strand that was sent and not
323 + # one read of the step
324 + frame = expand_frame(data, state)
325 +
326 + broadcast_frame(data, frame, state)
327 +
328 + # Phoenix.PubSub.broadcast(@pubsub_name, @pubsub_topic <> state[:name], {:publish, data})
329 +
330 + state = firehose(data, frame, state)
331 +
332 + {:noreply, state |> remember_fullkeys(data) |> bright_step()}
333 + end
334 +
335 + @impl true
336 + def handle_info(:publish, state) do
337 + Process.send_after(self(), :publish, 400)
316 338
339 + case state[:single] do
340 + :full ->
341 + state[:fullkeys]
342 + |> Enum.each(fn k ->
343 + Phoenix.PubSub.broadcast(
344 + @pubsub_name,
345 + @pubsub_topic_idx,
346 + {:machines_pub, publish(state, k)}
347 + )
348 + end)
349 +
350 + _otherwise ->
351 + Phoenix.PubSub.broadcast(@pubsub_name, @pubsub_topic_idx, {:machines_pub, publish(state)})
352 + end
353 +
354 + {:noreply, state}
355 + end
356 +
357 + def handle_info({:EXIT, pid, err}, state) do
358 + Logger.info("machine #{state[:name]}: a step exited, #{inspect(err)}")
359 + {:noreply, state}
360 + end
361 +
362 + defp read_steps(state) do
317 363 data =
318 364 state[:steps]
319 365 |> Enum.map(&GenServer.call(&1, :read))
320 366 |> Enum.map(fn x -> x |> Enum.map(fn y -> bmath(y, state) end) end)
321 367
322 − data =
323 − case state[:single] do
324 − :full ->
325 − data
326 − |> List.flatten()
327 − |> Enum.reduce(fn k, v -> Map.merge(k, v, fn _k, v1, v2 -> v2 ++ v1 end) end)
368 + case state[:single] do
369 + :full ->
370 + data
371 + |> List.flatten()
372 + |> Enum.reduce(fn k, v -> Map.merge(k, v, fn _k, v1, v2 -> v2 ++ v1 end) end)
328 373
329 − _otherwise ->
330 − data
331 − end
374 + _otherwise ->
375 + data
376 + end
377 + end
332 378
333 − # expand single to fill cnt. kept out here rather than rebound inside the
334 − # case below, so the firehose shows the strand that was sent and not one
335 − # read of the step
336 − frame =
337 − case state[:single] do
338 − true ->
339 − flat = data |> List.flatten()
379 + defp expand_frame(data, %{single: true} = state) do
380 + flat = data |> List.flatten()
340 381
341 − flat
342 − |> Stream.concat(
343 − if(state[:repeat], do: Stream.cycle(flat), else: Stream.repeatedly(fn -> 0 end))
344 − )
345 − |> Enum.take(state[:cnt] * 3)
346 − |> Enum.chunk_every(3)
382 + flat
383 + |> Stream.concat(
384 + if(state[:repeat], do: Stream.cycle(flat), else: Stream.repeatedly(fn -> 0 end))
385 + )
386 + |> Enum.take(state[:cnt] * 3)
387 + |> Enum.chunk_every(3)
388 + end
347 389
348 − _otherwise ->
349 − data
350 − end
390 + defp expand_frame(data, _state), do: data
351 391
392 + defp broadcast_frame(data, frame, state) do
352 393 case state[:single] do
353 394 true ->
354 395 Phoenix.PubSub.broadcast(@pubsub_name, @pubsub_topic <> state[:name], {:publish, frame})
355 396
356 − # do nothing
357 397 false ->
358 − data
359 −
360 398 Phoenix.PubSub.broadcast(@pubsub_name, @pubsub_topic <> state[:name], {:publish, data})
361 399
362 400 :full ->
363 401 # Bit hacky
364 402 data
365 − |> Enum.map(fn {k, datum} ->
403 + |> Enum.each(fn {k, datum} ->
366 404 Phoenix.PubSub.broadcast(
367 405 @pubsub_name,
368 406 # yolo etc
@@ -371,98 +409,51 @@ defmodule LambentEx.Machine do
371 409 )
372 410 end)
373 411 end
412 + end
374 413
375 − # Phoenix.PubSub.broadcast(@pubsub_name, @pubsub_topic <> state[:name], {:publish, data})
376 −
377 − state =
378 − cond do
379 − state[:single] == :full ->
380 − data
381 − |> Enum.map(fn {k, datum} ->
382 − Phoenix.PubSub.broadcast(
383 − @pubsub_name,
384 − @pubsub_topic_fh,
385 − {:firehose, {state[:name] <> Atom.to_string(k), datum |> preview_pixels()}}
386 − )
387 − end)
388 −
389 − state
390 −
391 − state[:speed] < 500 ->
392 − case 250 / state[:speed] <= state[:modpub] do
393 − true ->
394 − Phoenix.PubSub.broadcast(
395 − @pubsub_name,
396 − @pubsub_topic_fh,
397 − {:firehose, {state[:name], frame |> preview_pixels()}}
398 − )
399 −
400 − state |> Map.put(:modpub, 0)
401 −
402 − false ->
403 − state |> Map.put(:modpub, state[:modpub] + 1)
404 − end
405 −
406 − true ->
407 − Phoenix.PubSub.broadcast(
408 − @pubsub_name,
409 − @pubsub_topic_fh,
410 − {:firehose, {state[:name], frame |> preview_pixels()}}
411 − )
412 −
413 − state
414 − end
415 −
416 − case state[:single] do
417 − :full ->
418 − case state[:fullkeys] do
419 − val when length(val) == 0 ->
420 − {:noreply,
421 − bright_step(
422 − state
423 − |> Map.put(
424 − :fullkeys,
425 − data
426 − |> Map.keys()
427 − )
428 − )}
429 −
430 − _otherwise ->
431 − {:noreply, bright_step(state)}
432 − end
414 + defp firehose(data, _frame, %{single: :full} = state) do
415 + data
416 + |> Enum.each(fn {k, datum} ->
417 + Phoenix.PubSub.broadcast(
418 + @pubsub_name,
419 + @pubsub_topic_fh,
420 + {:firehose, {state[:name] <> Atom.to_string(k), datum |> preview_pixels()}}
421 + )
422 + end)
433 423
434 − _otherwise ->
435 − {:noreply, bright_step(state)}
436 − end
424 + state
437 425 end
438 426
439 − @impl true
440 − def handle_info(:publish, state) do
441 − Process.send_after(self(), :publish, 400)
442 −
443 − case state[:single] do
444 − :full ->
445 − state[:fullkeys]
446 − |> Enum.map(fn k ->
447 − Phoenix.PubSub.broadcast(
448 − @pubsub_name,
449 − @pubsub_topic_idx,
450 − {:machines_pub, publish(state, k)}
451 − )
452 − end)
427 + # a fast machine only shows on the firehose every few steps
428 + defp firehose(_data, frame, %{speed: speed} = state) when speed < 500 do
429 + case 250 / speed <= state[:modpub] do
430 + true ->
431 + firehose_frame(frame, state)
432 + state |> Map.put(:modpub, 0)
453 433
454 − _otherwise ->
455 − Phoenix.PubSub.broadcast(@pubsub_name, @pubsub_topic_idx, {:machines_pub, publish(state)})
434 + false ->
435 + state |> Map.put(:modpub, state[:modpub] + 1)
456 436 end
437 + end
457 438
458 − {:noreply, state}
439 + defp firehose(_data, frame, state) do
440 + firehose_frame(frame, state)
441 + state
459 442 end
460 443
461 − def handle_info({:EXIT, pid, err}, state) do
462 − IO.inspect(err)
463 − {:noreply, state}
444 + defp firehose_frame(frame, state) do
445 + Phoenix.PubSub.broadcast(
446 + @pubsub_name,
447 + @pubsub_topic_fh,
448 + {:firehose, {state[:name], frame |> preview_pixels()}}
449 + )
464 450 end
465 451
452 + defp remember_fullkeys(%{single: :full, fullkeys: []} = state, data),
453 + do: state |> Map.put(:fullkeys, data |> Map.keys())
454 +
455 + defp remember_fullkeys(state, _data), do: state
456 +
466 457 defp do_slower(current_val) do
467 458 sl = (@speeds |> Map.keys() |> length) - 1
468 459
modified lib/lambent_ex/mqtt.ex
+43 −38
@@ -201,7 +201,7 @@ defmodule LambentEx.MQTT do
201 201
202 202 @impl true
203 203 def handle_cast({:poke, id}, state) do
204 − IO.inspect("got poke")
204 + Logger.debug("#{state[:name]}: got poke")
205 205 {:noreply, state}
206 206 end
207 207
@@ -299,52 +299,57 @@ defmodule LambentEx.MQTT.Handler do
299 299 end
300 300
301 301 def handle_message(topic, payload, state) do
302 − case topic do
303 − ["tele", device, "LWT"] ->
304 − Logger.info("#{device}: got LWT, queueing discovery work, #{payload}")
302 + on_message(topic, payload, state)
303 + {:ok, state}
304 + end
305 305
306 − ["tele", device, "STATE"] ->
307 − Logger.info("#{device}: got STATE, updating device states, #{payload}")
308 − {:ok, p} = Jason.decode(payload)
306 + defp on_message(["tele", device, "LWT"], payload, _state),
307 + do: Logger.info("#{device}: got LWT, queueing discovery work, #{payload}")
309 308
310 − if is_lighting_entity?(:p, p) do
311 − publish_entity(state[:id], :light, device, p)
312 − end
309 + defp on_message(["tele", device, "STATE"], payload, state) do
310 + Logger.info("#{device}: got STATE, updating device states, #{payload}")
311 + publish_light(device, payload, state)
312 + end
313 313
314 − ["tele", device, "INFO1"] ->
315 − Logger.info("#{device}: got INFO1, updating device states, #{payload}")
314 + defp on_message(["tele", device, "INFO1"], payload, _state),
315 + do: Logger.info("#{device}: got INFO1, updating device states, #{payload}")
316 316
317 − ["tele", device, "INFO2"] ->
318 − Logger.info("#{device}: got INFO2, updating device states, #{payload}")
317 + defp on_message(["tele", device, "INFO2"], payload, _state),
318 + do: Logger.info("#{device}: got INFO2, updating device states, #{payload}")
319 319
320 − ["tele", device, "INFO3"] ->
321 − Logger.info("#{device}: got INFO3, updating device states, #{payload}")
320 + defp on_message(["tele", device, "INFO3"], payload, _state),
321 + do: Logger.info("#{device}: got INFO3, updating device states, #{payload}")
322 322
323 − ["stat", device, "POWER"] ->
324 − Logger.info("#{device}: got POWER message, #{payload}")
323 + defp on_message(["stat", device, "POWER"], payload, _state),
324 + do: Logger.info("#{device}: got POWER message, #{payload}")
325 +
326 + defp on_message(["stat", device, "RESULT"], payload, state) do
327 + Logger.info("#{device}: got RESULT message, #{payload}")
328 + publish_light(device, payload, state)
329 + end
325 330
326 − ["stat", device, "RESULT"] ->
327 − Logger.info("#{device}: got RESULT message, #{payload}")
328 − {:ok, p} = Jason.decode(payload)
331 + defp on_message([root, "audio", device, kind], payload, state)
332 + when kind in ["fft", "peaks", "beat"] do
333 + case Jason.decode(payload) do
334 + {:ok, p} -> Process.send(state[:id], {:audio, root, device, kind, p}, [])
335 + {:error, _reason} -> Logger.debug("#{device}: unreadable #{kind}, #{payload}")
336 + end
337 + end
329 338
330 − if is_lighting_entity?(:p, p) do
331 − publish_entity(state[:id], :light, device, p)
332 − end
339 + defp on_message([root, "audio", device, kind], payload, state)
340 + when kind in ["status", "ip"] do
341 + Process.send(state[:id], {:audio, root, device, kind, payload}, [])
342 + end
333 343
334 − [root, "audio", device, kind] when kind in ["fft", "peaks", "beat"] ->
335 − case Jason.decode(payload) do
336 − {:ok, p} -> Process.send(state[:id], {:audio, root, device, kind, p}, [])
337 − {:error, _reason} -> Logger.debug("#{device}: unreadable #{kind}, #{payload}")
338 − end
344 + defp on_message(topic, payload, _state),
345 + do: Logger.debug("xxx---#{Enum.join(topic, "/")} #{inspect(payload)}")
339 346
340 − [root, "audio", device, kind] when kind in ["status", "ip"] ->
341 − Process.send(state[:id], {:audio, root, device, kind, payload}, [])
347 + defp publish_light(device, payload, state) do
348 + {:ok, p} = Jason.decode(payload)
342 349
343 − _otherwise ->
344 − Logger.debug("xxx---#{Enum.join(topic, "/")} #{inspect(payload)}")
350 + if lighting_entity?(:p, p) do
351 + publish_entity(state[:id], :light, device, p)
345 352 end
346 −
347 − {:ok, state}
348 353 end
349 354
350 355 def terminate(reason, _state) do
@@ -352,12 +357,12 @@ defmodule LambentEx.MQTT.Handler do
352 357 :ok
353 358 end
354 359
355 − def is_lighting_entity?(:t, topic) do
360 + def lighting_entity?(:t, topic) do
356 361 "tasmota" in topic and
357 362 Enum.any?(@lighting_keywords, fn keyword -> Enum.member?(topic, keyword) end)
358 363 end
359 364
360 − def is_lighting_entity?(:p, payload) do
365 + def lighting_entity?(:p, payload) do
361 366 keys = payload |> Map.keys()
362 367
363 368 Enum.any?(["Color", "Fade", "White", "CT", "Dimmer"], fn keyword ->
@@ -365,6 +370,6 @@ defmodule LambentEx.MQTT.Handler do
365 370 end)
366 371 end
367 372
368 − def is_sensor_entity(:p, payload) do
373 + def sensor_entity?(:p, payload) do
369 374 end
370 375 end

Parents: f16dbff