A behaviour module for implementing KNXnet/IP tunnel clients.

The behaviour wraps Connection and attempts to ensure that the client is always connected to the KNX network.

The callback module is invoked when telegrams are received from the tunnel server, and when the connection status changes.

To send a telegram to the tunnel server, the callback module must return a :send_telegram tuple when a callback is invoked. The behaviour will then wrap the encoded telegram in a TUNNELLING_REQUEST and send it to the tunnel server. When the tunnel server responds with a TUNNELLING_ACK, the callback on_telegram_ack/1 is invoked.

The callback module must wait for on_telegram_ack/1 to be invoked before sending any further telegrams. If the callback module tries to send a new telegram while the client is waiting to receive a TUNNELLING_ACK from a previous TUNNELLING_REQUEST, the new telegram will be discarded.

The telegram is also discarded if the callback module tries to send a telegram while the client is not connected.

The behaviour deals with telegrams as raw binaries. KNXnetIP.Telegram defines a data structure to represent telegrams, as well as functions for encoding and decoding telegrams.

Example

This example will bind to the interface given by my_ip and connect to the server at server_ip.

The process registers itself with the name KnxApp.Tunnel, and other processes can send group writes to the KNX bus by using the send_group_write function. In addition, the tunnel process will forward any group writes it receives to the process that started the tunnel.

Note the map called group_addresses. This maps a certain group address to a datapoint type, and is used when encoding and decoding datapoints. This is necessary to determine the correct codec for the value in a given telegram, as the datapoint type is not present in the telegram.

defmodule KnxApp.Tunnel do
  require Logger
  alias KNXnetIP.{Datapoint, Telegram, Tunnel}
  @behaviour Tunnel

  def start_link(my_ip, server_ip) do
    knxnet_ip_opts = [
      ip: my_ip,
      server_ip: server_ip
    ]

    # 5.001 is DPT_Scaling, an 8-bit unsigned integer with unit and resolution of 0.4 %.
    # 14.056 is DPT_Value_Power, a 32-bit float with unit and resolution of 1 W.
    group_addresses = %{
      "2/0/2" => "5.001",
      "2/0/3" => "5.001",
      "2/0/4" => "5.001",
      "4/4/52" => "14.056",
      "4/4/56" => "14.056",
      "4/4/60" => "14.056"
    }

    opts = [parent: self(), group_addresses: group_addresses]

    Tunnel.start_link(__MODULE__, opts, knxnet_ip_opts, name: __MODULE__)
  end

  @doc "Encodes `value` according to the DPT of the `group_address`, and sends it in a `GroupValueWrite` to `group_address`"
  @spec send_group_write(binary(), term()) :: :ok | {:error, :unknown_group_address}
  def send_group_write(group_address, value) do
    Tunnel.call(__MODULE__, {:group_write, group_address, value})
  end

  @doc "Sends a `GroupValueRead` to `group_address`"
  @spec send_group_read(binary()) :: :ok | {:error, :unknown_group_address}
  def send_group_read(group_address) do
    Tunnel.call(__MODULE__, {:group_read, group_address})
  end

  @impl true
  def init(opts) do
    state = %{
      parent: Keyword.fetch!(opts, :parent),
      group_addresses: Keyword.fetch!(opts, :group_addresses)
    }

    {:ok, state}
  end

  @impl true
  def handle_call({:group_write, group_address, value}, _from, state) do
    case state.group_addresses[group_address] do
      nil ->
        {:reply, {:error, :unknown_group_address}, state}

      datapoint_type ->
        {:ok, value} = Datapoint.encode(value, datapoint_type)

        {:ok, telegram} =
          Telegram.encode(%Telegram{
            source: "0.0.0",
            destination: group_address,
            service: :group_write,
            type: :request,
            value: value
          })

        {:send_telegram, telegram, :ok, state}
    end
  end

  def handle_call({:group_read, group_address}, _from, state) do
    case state.group_addresses[group_address] do
      nil ->
        {:reply, {:error, :unknown_group_address}, state}

      _ ->
        {:ok, telegram} =
          Telegram.encode(%Telegram{
            source: "0.0.0",
            destination: group_address,
            service: :group_read,
            type: :request,
            value: <<0::6>>
          })

        {:send_telegram, telegram, :ok, state}
    end
  end

  @impl true
  def code_change(_vsn, state, _extra), do: {:ok, state}

  @impl true
  def terminate(_reason, state), do: state

  @impl true
  def on_connect(state), do: {:ok, state}

  @impl true
  def on_disconnect(reason, state) do
    case reason do
      :disconnect_requested ->
        {:backoff, 0, state}

      {:tunnelling_ack_error, _} ->
        {:backoff, 0, state}

      {:connectionstate_response_error, _} ->
        {:backoff, 0, state}

      {:connect_response_error, _} ->
        {:backoff, 5_000, state}
    end
  end

  @impl true
  def on_telegram_ack(state), do: {:ok, state}

  @impl true
  def on_telegram(telegram, state) do
    {:ok, telegram} = Telegram.decode(telegram)
    handle_telegram(telegram, state)
    {:ok, state}
  end

  defp handle_telegram(%Telegram{service: service} = telegram, state)
      when service in [:group_write, :group_response] do
    case state.group_addresses[telegram.destination] do
      nil ->
        Logger.info(fn -> "Ignoring unspecified group address: #{telegram.destination}" end)

      datapoint_type ->
        {:ok, value} = Datapoint.decode(telegram.value, datapoint_type)
        send(state.parent, {service, telegram.destination, value})
    end
  end

  # Ignore telegrams which are not group writes or group responses
  defp handle_telegram(_telegram, _state), do: :ok
end