weewx-proxy/lib/weewx_proxy/sdr/ecowitt.ex

defmodule WeewxProxy.Sdr.Ecowitt do
  @moduledoc false

  require Logger

  use Tortoise311.Handler

  alias WeewxProxy.HTTP.PurpleAir
  alias WeewxProxy.{Publisher, Utils}

  @type parsed_body :: %{required(String.t()) => String.t() | float() | integer()}

  # API

  @spec recently_uploaded_keys :: [atom()]
  def recently_uploaded_keys do
    current_timestamp = Utils.utc_timestamp()
    recently_uploaded_keys(current_timestamp)
  end

  @spec recently_uploaded_keys(non_neg_integer()) :: [atom()]
  def recently_uploaded_keys(current_timestamp) do
    wh65b_keys =
      case :ets.lookup(:sdr_ecowitt, {:wh65b, :last_update}) do
        [{_key, timestamp}] when current_timestamp - timestamp < 48 ->
          [:outTemp, :outHumidity, :windSpeed, :windGust, :windDir]

        _ ->
          []
      end

    wh32b_keys =
      case :ets.lookup(:sdr_ecowitt, {:wh32b, :last_update}) do
        [{_key, timestamp}] when current_timestamp - timestamp < 80 ->
          [:inTemp, :inHumidity, :pressure]

        _ ->
          []
      end

    wh65b_keys ++ wh32b_keys
  end

  # Callbacks

  @impl true
  def init(_opts) do
    _ = Logger.info("Initializing handler")
    {:ok, nil}
  end

  @impl true
  def connection(:up, state) do
    _ = Logger.info("Connection has been established")
    {:ok, state}
  end

  @impl true
  def connection(:down, state) do
    _ = Logger.warn("Connection has been dropped")
    {:ok, state}
  end

  @impl true
  def connection(:terminating, state) do
    _ = Logger.warn("Connection is terminating")
    {:ok, state}
  end

  @impl true
  def subscription(:up, topic, state) do
    _ = Logger.info("Subscribed to `#{topic}'")
    {:ok, state}
  end

  @impl true
  def subscription({:warn, [requested: req, accepted: qos]}, topic, state) do
    _ = Logger.warn("Subscribed to `#{topic}'; requested #{req} but got accepted with QoS #{qos}")
    {:ok, state}
  end

  @impl true
  def subscription({:error, reason}, topic, state) do
    _ = Logger.error("Error subscribing to `#{topic}'; #{inspect(reason)}")
    {:ok, state}
  end

  @impl true
  def subscription(:down, topic, state) do
    _ = Logger.info("Unsubscribed from `#{topic}'")
    {:ok, state}
  end

  @impl true
  def handle_message(topic, publish, state) do
    full_topic = Enum.join(topic, "/")
    parsed_message = parse_message(full_topic, publish)

    :ok = handle_reading(parsed_message)

    {:ok, state}
  end

  @impl true
  def terminate(reason, _state) do
    _ = Logger.warn("Client has been terminated with reason: `#{inspect(reason)}'")
    :ok
  end

  # Helper

  @spec parse_message(String.t(), String.t()) :: parsed_body() | nil
  defp parse_message("rtl433", message) do
    {:ok, body} = Jason.decode(message)

    if handle_reading?(body) do
      body
    else
      _ = Logger.warn("Ignoring reading: #{inspect(body)}")
      nil
    end
  end

  defp parse_message(_topic, _message), do: nil

  @spec handle_reading?(parsed_body()) :: boolean()
  defp handle_reading?(%{"model" => "Fineoffset-WH65B", "id" => 189}), do: true
  defp handle_reading?(%{"model" => "Fineoffset-WH32B", "id" => 173}), do: true
  defp handle_reading?(_reading), do: false

  @spec handle_reading(parsed_body() | nil) :: :ok
  defp handle_reading(nil), do: :ok

  defp handle_reading(body) do
    {type, data} = transform_data(body)

    :ok =
      if valid_data?(type, data) do
        purpleair_keys = PurpleAir.recently_uploaded_keys(data.dateTime)
        _ = Logger.debug("Removing keys: `#{inspect(purpleair_keys)}'")
        partial_data = Map.drop(data, purpleair_keys)

        true = :ets.insert(:sdr_ecowitt, {{type, :last_update}, data.dateTime})
        Publisher.publish("weewx/ingest_si", partial_data)
      else
        _ = Logger.error("Not publishing record because data appears invalid: #{inspect(data)}")
        :ok
      end

    :ok
  end

  @spec transform_data(parsed_body()) :: {:wh65b | :wh32b, Publisher.data()}
  defp transform_data(%{"model" => "Fineoffset-WH65B"} = data) do
    data = %{
      dateTime: format_date_time(data),
      outTemp: Utils.parse_float(data["temperature_C"]),
      outHumidity: Utils.parse_float(data["humidity"]),
      windSpeed: Utils.parse_float(data["wind_avg_m_s"]),
      windGust: Utils.parse_float(data["wind_max_m_s"]),
      windDir: Utils.parse_float(data["wind_dir_deg"]),
      luminosity: Utils.parse_float(data["light_lux"])
    }

    {:wh65b, data}
  end

  defp transform_data(%{"model" => "Fineoffset-WH32B"} = data) do
    data = %{
      dateTime: format_date_time(data),
      inTemp: Utils.parse_float(data["temperature_C"]),
      inHumidity: Utils.parse_float(data["humidity"]),
      pressure: Utils.parse_float(data["pressure_hPa"])
    }

    {:wh32b, data}
  end

  @spec format_date_time(parsed_body()) :: non_neg_integer()
  defp format_date_time(data) do
    {:ok, dt, 0} =
      data
      |> Map.get("time")
      |> String.replace(" ", "T")
      |> Utils.append_string("Z")
      |> DateTime.from_iso8601()

    DateTime.to_unix(dt)
  end

  @spec valid_data?(:wh65b | :wh32b, Publisher.data()) :: boolean()
  defp valid_data?(:wh65b, data) do
    Map.has_key?(data, :outTemp) and is_number(data.outTemp)
  end

  defp valid_data?(:wh32b, data) do
    Map.has_key?(data, :inTemp) and is_number(data.inTemp)
  end
end