MQTT Signal Generator

Mix.install([
  {:kino, "~> 0.8.0"},
  {:tortoise, "~> 0.9"},
  {:jason, "~> 1.2"}
])

Configuration

Which broker and prefix to work on:

kino_broker = Kino.Input.text("MQTT Broker:", default: "broker.hivemq.com")
# using public mqtt broker
kino_prefix = Kino.Input.text("MQTT Prefix:", default: "dk/sdu/iot/2023/anton")
# subscribe to dk/sdu/iot/2023/anton/#
Kino.Layout.grid([kino_broker, kino_prefix])

{broker, prefix} = {
  Kino.Input.read(kino_broker),
  Kino.Input.read(kino_prefix)
}

Next, we define which data should be replayed to which topics:

config = [
  %{
    topic: "siggen/room1/temp",
    samples: [
      %{time: 1.0, value: 20.0},
      %{time: 2.0, value: 20.5},
      %{time: 3.0, value: 21.0},
      %{time: 4.0, value: 21.5},
      %{time: 5.0, value: 22.0},
      %{time: 6.0, value: 22.5},
      %{time: 7.0, value: 23.0},
      %{time: 8.0, value: 24.5},
      %{time: 9.0, value: 24.3},
      %{time: 10.0, value: 22.7},
      %{time: 11.0, value: 19.9},
      %{time: 12.0, value: 17.3}
    ]
  },
  %{
    topic: "siggen/room1/rhum",
    samples: [
      %{time: 1.0, value: 50.0},
      %{time: 2.0, value: 55.3},
      %{time: 3.0, value: 63.1},
      %{time: 4.0, value: 65.6},
      %{time: 5.0, value: 77.2},
      %{time: 6.0, value: 76.8},
      %{time: 7.0, value: 84.9},
      %{time: 8.0, value: 81.0},
      %{time: 9.0, value: 89.7},
      %{time: 10.0, value: 93.3},
      %{time: 11.0, value: 74.8},
      %{time: 12.0, value: 59.4}
    ]
  },
  %{
    topic: "siggen/room2/temp",
    samples: [
      %{time: 1.0, value: 19.3},
      %{time: 2.0, value: 19.7},
      %{time: 3.0, value: 19.5},
      %{time: 4.0, value: 20.6},
      %{time: 5.0, value: 20.4},
      %{time: 6.0, value: 23.7},
      %{time: 7.0, value: 22.0},
      %{time: 8.0, value: 24.2},
      %{time: 9.0, value: 26.1},
      %{time: 10.0, value: 25.3},
      %{time: 11.0, value: 24.8},
      %{time: 12.0, value: 22.9}
    ]
  },
  %{
    topic: "siggen/room2/rhum",
    samples: [
      %{time: 1.0, value: 61.7},
      %{time: 2.0, value: 59.4},
      %{time: 3.0, value: 58.8},
      %{time: 4.0, value: 56.3},
      %{time: 5.0, value: 63.5},
      %{time: 6.0, value: 62.2},
      %{time: 7.0, value: 68.1},
      %{time: 8.0, value: 73.6},
      %{time: 9.0, value: 72.9},
      %{time: 10.0, value: 76.4},
      %{time: 11.0, value: 78.7},
      %{time: 12.0, value: 74.3}
    ]
  }
]

Producer Code

First we start an MQTT client:

t = :os.system_time(:milli_seconds)
client_id = "sdu_iot_mqtt_livebook_producer_#{t}"

Tortoise.Supervisor.start_child(
  client_id: client_id,
  handler: {Tortoise.Handler.Default, []},
  server: {Tortoise.Transport.Tcp, host: broker, port: 1883},
  subscriptions: []
)

First, we define a producer:

defmodule Producer do
  use GenServer

  # callbacks

  @impl true
  def init({client_id, topic, samples}) do
    samples = Enum.map(samples, fn %{time: t, value: v} -> %{time: t * 1000, value: v} end)

    t0 = :os.system_time(:milli_seconds)
    i = 0
    period = Enum.fetch!(samples, -1) |> Map.get(:time)

    register_timeout(t0, i, period, samples)

    state = [t0: t0, period: period, i: i, samples: samples, client: client_id, topic: topic]
    {:ok, state}
  end

  @impl true
  def handle_info({:emit, tnext, samples}, state) do
    [sample | rest] = samples
    value = Map.get(sample, :value)
    [t0: t0, period: period, i: i, samples: ss, client: client_id, topic: topic] = state

    message = '{"time": #{tnext}, "value": #{value}}'
    Tortoise.publish(client_id, topic, message, qos: 0)

    {i, rest} =
      case rest do
        [] -> {i + 1, ss}
        _ -> {i, rest}
      end

    register_timeout(t0, i, period, rest)

    state = [t0: t0, period: period, i: i, samples: ss, client: client_id, topic: topic]
    {:noreply, state}
  end

  # helpers

  defp register_timeout(t0, i, period, samples) do
    tsample = Enum.fetch!(samples, 0) |> Map.get(:time)
    tnext = t0 + i * period + tsample
    tnow = :os.system_time(:milli_seconds)
    Process.send_after(self(), {:emit, tnext, samples}, Kernel.trunc(tnext - tnow))
  end
end

Starting Things

Some convenience:

config_entry = Enum.fetch!(config, 0)
topic = prefix <> Map.get(config_entry, :topic)
samples = Map.get(config_entry, :samples)
{config_entry, topic, samples}

Uncomment to test single producer:

# {:ok, producer_pid} = GenServer.start_link(Producer, {client_id, topic, samples})

Start all producers:

starter = fn c ->
  topic = prefix <> "/" <> c[:topic]
  samples = c[:samples]
  GenServer.start_link(Producer, {client_id, topic, samples})
end

producers = Enum.map(config, starter)