bacstack

defmodule BACnet.Stack.ClientHelper do
  @moduledoc """
  BACnet stack client helper functions for executing commands/queries.
  """

  alias BACnet.Protocol.AccessSpecification
  alias BACnet.Protocol.APDU
  alias BACnet.Protocol.ApplicationTags.Encoding
  alias BACnet.Protocol.Constants
  alias BACnet.Protocol.BACnetDate
  alias BACnet.Protocol.BACnetTime
  alias BACnet.Protocol.ObjectIdentifier
  alias BACnet.Protocol.ObjectsUtility
  alias BACnet.Protocol.PropertyRef
  alias BACnet.Protocol.ReadAccessResult
  alias BACnet.Protocol.Services
  alias BACnet.Stack.Client

  import BACnet.Internal, only: [log_debug: 1]

  require Constants
  require Logger

  @doc """
  Sends an I-Am service request to the destination, or optionally using
  `:broadcast` (or the real broadcast address) as local broadcast.

  See also `BACnet.Protocol.Services.IAm`.

  The `Client.send/4` options are available.
  """
  @spec i_am(
          GenServer.server(),
          term() | :broadcast,
          ObjectIdentifier.t(),
          non_neg_integer(),
          Keyword.t()
        ) ::
          :ok | {:error, term()}
  def i_am(
        server,
        destination,
        %ObjectIdentifier{type: :device} = device,
        vendor_id,
        opts \\ []
      )
      when is_integer(vendor_id) and vendor_id >= 0 and is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "i_am/5 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    with {:ok, addr} <- get_address(server, destination),
         {trans_mod, _transport, _portal} <- Client.get_transport(server),
         {:ok, req} <-
           Services.IAm.to_apdu(
             %Services.IAm{
               device: device,
               max_apdu: trans_mod.max_apdu_length(),
               segmentation_supported:
                 Constants.macro_assert_name(:segmentation, :segmented_both),
               vendor_id: vendor_id
             },
             []
           ) do
      Client.send(server, addr, req, opts)
    end
  end

  @doc """
  Read a BACnet object from a remote BACnet device and transform it into an object.

  The required properties are always as a bare minimum read, only more properties can be read, never less.

  The value is casted through the `BACnet.Protocol.ObjectsUtility` module based on the object modules.
  As such object types or properties that are not supported, will fail.

  If you want to read a device object and don't know the proper device instance number,
  you can use `4_194_303` as instance number. By the BACnet specification that instance number will be
  treated by the remote BACnet device as if the instance number was locally correctly matched.

  The following options are available:
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.ReadPropertyMultiple.to_apdu/2`.
  - All options from `BACnet.Protocol.ObjectsUtility.cast_read_properties_ack/3`.
  - All options from `BACnet.Protocol.ObjectsUtility.cast_properties_to_object/3`.
  - `properties: [:all | :required | Constants.property_identifier()]` - Optional. Select the properties to read.
  - `read_level: :all | :required` - Optional. Select how many properties should be read (defaults to `:all`).

  `properties` and `read_level` are mutually excluse. If both are given, `properties` takes precedence.
  """
  @spec read_object(
          GenServer.server(),
          term(),
          ObjectIdentifier.t(),
          Keyword.t()
        ) ::
          {:ok, ObjectsUtility.bacnet_object()}
          | {:error, BACnet.Protocol.apdu()}
          | {:error, term()}
  def read_object(
        server,
        destination,
        %ObjectIdentifier{} = object,
        opts \\ []
      )
      when is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "read_object/4 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    properties =
      case Keyword.get(opts, :properties, nil) do
        nil ->
          read_level = Keyword.get(opts, :read_level, :all)

          unless read_level == :all or read_level == :required do
            raise ArgumentError,
                  "read_object/4 expected read_level to be :all or :required, " <>
                    "got: #{inspect(read_level)}"
          end

          [read_level]

        props when is_list(props) ->
          props
          |> Enum.map(&make_access_property_from_identifier/1)
          |> then(fn list ->
            # Assert we have the required properties covered
            if Enum.any?(list, &(&1 == :all or &1 == :required)) do
              list
            else
              [:required | list]
            end
          end)

        term ->
          raise ArgumentError,
                "read_object/4 expected properties to be a list, " <>
                  "got: #{inspect(term)}"
      end

    with {:ok, req} <-
           Services.ReadPropertyMultiple.to_apdu(
             %Services.ReadPropertyMultiple{
               list: [
                 %AccessSpecification{
                   object_identifier: object,
                   properties: properties
                 }
               ]
             },
             opts
           ),
         {:ok, %APDU.ComplexACK{} = resp} <- Client.send(server, destination, req, opts),
         {:ok, ack} <- Services.Ack.ReadPropertyMultipleAck.from_apdu(resp),
         # When using 4_194_303 as device ID, fetch the correct ID from the first result
         cast_object_id =
           (case object do
              %{type: :device, instance: 4_194_303} when ack.results != [] ->
                hd(ack.results).object_identifier

              _else ->
                object
            end),
         {:ok, values} <- ObjectsUtility.cast_read_properties_ack(cast_object_id, [ack], opts) do
      ObjectsUtility.cast_properties_to_object(cast_object_id, values, opts)
    else
      {:ok, apdu} -> {:error, apdu}
      {:error, _err} = err -> err
    end
  end

  @doc """
  Read a single property from a remote BACnet object and transform the value.

  The value is casted through the `BACnet.Protocol.ObjectsUtility` module based on the object modules.
  As such object types or properties that are not supported, will fail, unless you specify
  the `raw` options, which will give you the `Encoding` struct (or list of) to handle yourself.
  Array indexes of 0 will return the array size as `{:ok, non_neg_integer()}`, if successfully read.

  If you want to read a device object's property without needing to know before hand which instance number,
  you can use `4_194_303` as instance number. By the BACnet specification that instance number will be
  treated by the remote BACnet device as if the instance number was locally correctly matched.

  The following options are available:
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.ReadProperty.to_apdu/2`.
  - `raw: boolean()` - Optional. Returns the `t:Encoding.t/0` (or list of) instead of trying to transform the value.
  """
  @spec read_property(
          GenServer.server(),
          term(),
          ObjectIdentifier.t(),
          Constants.property_identifier() | non_neg_integer(),
          non_neg_integer() | nil,
          Keyword.t()
        ) ::
          {:ok, term()}
          | {:ok, Encoding.t() | [Encoding.t()]}
          | {:error, BACnet.Protocol.apdu()}
          | {:error, term()}
  def read_property(
        server,
        destination,
        %ObjectIdentifier{} = object,
        property,
        array_index \\ nil,
        opts \\ []
      )
      when ((is_atom(property) and property not in [:all, :required, :optional]) or
              (is_integer(property) and property >= 0)) and
             (is_nil(array_index) or (is_integer(array_index) and array_index >= 0)) and
             is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "read_property/6 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    with {:ok, req} <-
           Services.ReadProperty.to_apdu(
             %Services.ReadProperty{
               object_identifier: object,
               property_identifier: property,
               property_array_index: array_index
             },
             opts
           ),
         {:ok, %APDU.ComplexACK{} = resp} <- Client.send(server, destination, req, opts),
         {:ok, ack} <- Services.Ack.ReadPropertyAck.from_apdu(resp),
         {:ok, value} <-
           (case opts[:raw] do
              true ->
                {:ok, ack.property_value}

              _else ->
                if array_index == 0 and is_integer(ack.property_value.value) do
                  {:ok, ack.property_value.value}
                else
                  ObjectsUtility.cast_property_to_value(
                    object,
                    ack.property_identifier,
                    ack.property_value,
                    allow_partial: array_index != nil
                  )
                end
            end) do
      {:ok, value}
    else
      {:ok, apdu} -> {:error, apdu}
      {:error, _err} = err -> err
    end
  end

  @doc """
  Read multiple properties from a remote BACnet object at once and transform each value.

  The values are casted through the `BACnet.Protocol.ObjectsUtility` module based on the object modules.
  As such object types or properties that are not supported, will fail, unless you specify
  the `raw` options, which will give you a list of `ReadAccessResult`s to handle yourself.

  The following options are available:
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.ReadPropertyMultiple.to_apdu/2`.
  - `raw: boolean()` - Optional. Returns the results instead of trying to transform each value.
  """
  @spec read_property_multiple(
          GenServer.server(),
          term(),
          ObjectIdentifier.t(),
          [
            AccessSpecification.Property.t()
            | Constants.property_identifier()
            | :all
            | :required
            | :optional
          ],
          Keyword.t()
        ) ::
          {:ok, %{optional(Constants.property_identifier()) => term()}}
          | {:ok, [ReadAccessResult.t()]}
          | {:error, BACnet.Protocol.apdu()}
          | {:error, term()}
  def read_property_multiple(
        server,
        destination,
        %ObjectIdentifier{} = object,
        properties,
        opts \\ []
      )
      when is_list(properties) and is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "read_property_multiple/5 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    with {:ok, req} <-
           Services.ReadPropertyMultiple.to_apdu(
             %Services.ReadPropertyMultiple{
               list: [
                 %AccessSpecification{
                   object_identifier: object,
                   properties: Enum.map(properties, &make_access_property_from_identifier/1)
                 }
               ]
             },
             opts
           ),
         {:ok, %APDU.ComplexACK{} = resp} <- Client.send(server, destination, req, opts),
         {:ok, ack} <- Services.Ack.ReadPropertyMultipleAck.from_apdu(resp),
         {:ok, values} <-
           (case opts[:raw] do
              true ->
                {:ok, ack.results}

              _else ->
                ObjectsUtility.cast_read_properties_ack(object, [ack], opts)
            end) do
      {:ok, values}
    else
      {:ok, apdu} -> {:error, apdu}
      {:error, _err} = err -> err
    end
  end

  @doc """
  Send a Reinitialize-Device service request to a remote BACnet device.

  Password must be an ASCII string between 1 to 20 characters, inclusive, or nil.

  The following options are available:
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.ReinitializeDevice.to_apdu/2`.
  """
  @spec reinitialize_device(
          GenServer.server(),
          term(),
          Constants.reinitialized_state(),
          String.t() | nil,
          Keyword.t()
        ) :: :ok | {:error, BACnet.Protocol.apdu()} | {:error, term()}
  def reinitialize_device(
        server,
        destination,
        state \\ Constants.macro_assert_name(:reinitialized_state, :warmstart),
        password \\ nil,
        opts \\ []
      ) do
    with {:ok, req} <-
           Services.ReinitializeDevice.to_apdu(
             %Services.ReinitializeDevice{
               reinitialized_state: state,
               password: password
             },
             opts
           ) do
      Client.send(server, destination, req, opts)
    end
  end

  @doc """
  Scan the given device for available objects and read all objects. A map of objects will be returned on success.

  If you don't know the device object identifier of the BACnet device in question, but you know the
  BACnet network address (i.e. the IP address and port for BACnet/IP), you can use the Who-Is service
  with the destination address being the device's network address, to discover the object identifier.
  You can also use `read_property/6` to read the `:object_identifier` property.

  The scan process is parallelized through `Task.async_stream/3` and thus the `invoke_id` is
  automatically being set. Since this implementation simply uses `invoke_id` in the range of `0..max_concurrency-1`,
  it would be safest when the `BACnet.Stack.Client` implementation manages and overrides the `invoke_id`,
  so that an user does not have to care about possible collisions.
  The current "default" implementation of `BACnet.Stack.Client` does manage `invoke_id`s,
  but it can be deactivated, so care must be exercised if it done.
  You need to be aware to not invoke/have parallel other requests to the same destination,
  as the `invoke_id` could be duplicated.

  The values are casted through the `BACnet.Protocol.ObjectsUtility` module based on the object modules.
  As such object types or properties that are not supported, will fail the operation.

  The following options are available:
  - All options from `read_object/4`.
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.ReadPropertyMultiple.to_apdu/2`, except `invoke_id`.
  - All options from `BACnet.Protocol.ObjectsUtility.cast_read_properties_ack/3`.
  - All options from `BACnet.Protocol.ObjectsUtility.cast_properties_to_object/3`.
  - `exit_on_error: boolean()` - Optional. Whether to exit the process on first error.
  - `ignore_errors: boolean()` - Optional. Whether to ignore errors and continue with the rest.
  - `ignore_unsupported_object_types: boolean()` - Optional. Whether to ignore unknown/unsupported object types.
  - `task_max_concurrency: pos_integer()` - Optional. The maximum task concurrency to use (limited to 255).
  - `task_supervisor: Supervisor.supervisor()` - Optional. The task supervisor to use for spawning tasks.
  - `task_timeout: timeout()` - Optional. The timeout to use for the task async stream (defaults to `30_000`).

  `exit_on_error` and `ignore_errors` are mutually excluse. `ignore_errors` takes precedence, if set to `true`.
  """
  @spec scan_device(GenServer.server(), term(), ObjectIdentifier.t(), Keyword.t()) ::
          {:ok,
           %{
             optional(Constants.object_type()) => %{
               optional(non_neg_integer()) => ObjectsUtility.bacnet_object()
             }
           }}
          | {:error, {term(), ObjectIdentifier.t()}}
          | {:error, term()}
  def scan_device(server, destination, %ObjectIdentifier{type: :device} = device, opts \\ []) do
    exit_on_error = Keyword.get(opts, :exit_on_error, false)
    ignore_errors = Keyword.get(opts, :ignore_errors, false)
    ignore_unsupported_object_types = Keyword.get(opts, :ignore_unsupported_object_types, false)
    task_supervisor = Keyword.get(opts, :task_supervisor, nil)

    max_concurrency =
      min(255, Keyword.get_lazy(opts, :task_max_concurrency, &System.schedulers_online/0))

    task_spawn =
      if task_supervisor do
        &Task.Supervisor.async_stream(task_supervisor, &1, &2, &3)
      else
        &Task.async_stream/3
      end

    with {:ok, objects_list} <-
           read_property(
             server,
             destination,
             device,
             Constants.macro_assert_name(:property_identifier, :object_list),
             nil,
             raw: true
           ) do
      len = length(objects_list)
      pre_new_opts = Keyword.put(opts, :remote_device_id, device.instance)

      objects_stream =
        objects_list
        |> Stream.map(fn
          %Encoding{value: val} -> val
          term -> term
        end)
        # Do chunks by count chunks as max_concurrency (we only want max_concurrency chunks)
        |> Stream.chunk_every(trunc(Float.ceil(len / max_concurrency)))
        |> Stream.with_index()
        |> task_spawn.(
          # Make sure index/invoke_id is always in range 0..255
          fn {object_ids, index} when index >= 0 and index <= 255 ->
            Enum.reduce_while(object_ids, {:ok, []}, fn
              %ObjectIdentifier{} = object_id, {:ok, acc} ->
                new_opts = Keyword.put(pre_new_opts, :invoke_id, index)

                case read_object(server, destination, object_id, new_opts) do
                  {:ok, %{} = obj} ->
                    {:cont, {:ok, [{object_id, obj} | acc]}}

                  {:error, :unsupported_object_type} when ignore_unsupported_object_types ->
                    {:cont, {:ok, acc}}

                  {:error, err} ->
                    cond do
                      ignore_errors -> {:cont, {:ok, acc}}
                      exit_on_error -> exit({:error, {err, object_id}})
                      true -> {:halt, {:error, {err, object_id}}}
                    end
                end

              term, acc ->
                cond do
                  ignore_errors -> {:cont, acc}
                  exit_on_error -> exit({:error, {:invalid_object_identifier, term}})
                  true -> {:halt, {:error, {:invalid_object_identifier, term}}}
                end
            end)
          end,
          max_concurrency: max_concurrency,
          on_timeout: :kill_task,
          ordered: false,
          timeout: Keyword.get(opts, :task_timeout, 30_000)
        )
        |> Enum.map(fn
          {:exit, reason} when exit_on_error -> exit(reason)
          {:ok, {:ok, _val} = val} -> val
          {:ok, {:error, _err} = err} -> err
          term -> term
        end)

      # Find the first error to return, if not ignoring errors
      find_error =
        unless ignore_errors do
          Enum.find(
            objects_stream,
            &(match?({:exit, _reason}, &1) or match?({:error, _err}, &1))
          )
        end

      case find_error do
        {:error, _err} = err ->
          err

        {:exit, reason} ->
          {:error, reason}

        _else ->
          objects =
            objects_stream
            |> Stream.filter(&match?({:ok, _val}, &1))
            |> Stream.flat_map(fn {:ok, objects} -> objects end)
            |> Enum.group_by(fn {%{type: type}, _obj} -> type end)
            |> Map.new(fn {type, chunk} ->
              objs =
                Map.new(chunk, fn {%{instance: instance}, obj} ->
                  {instance, obj}
                end)

              {type, objs}
            end)

          # There is currently NO value in parallization of this process,
          # however we will keep it here for future usage (if the need arises)
          #
          # |> task_spawn.(
          #   fn [{%{type: type}, _obj} | _tl] = chunk ->
          #     objs = Map.new(chunk, fn {%{instance: instance}, obj} ->
          #       {instance, obj}
          #     end)

          #     {type, objs}
          #   end,
          #   ordered: false,
          #   timeout: Keyword.get(opts, :task_timeout, 30_000)
          # )
          # |> Stream.map(fn
          #   {:exit, reason} when exit_on_error -> exit(reason)
          #   term -> term
          # end)
          # |> Stream.filter(&match?({:ok, _val}, &1))
          # |> Map.new(fn {:ok, term} -> term end)

          {:ok, objects}
      end
    end
  end

  @doc """
  Send a (UTC) Time Synchronization service APDU to the destination.

  `:broadcast` will be resolved to the local broadcast address.

  The following options are available:
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.TimeSynchronziation.to_apdu/2` respectively
    `BACnet.Protocol.Services.UtcTimeSynchronziation.to_apdu/2`.
  - `datetime: DateTime.t()` - Optional. The timestamp to use for synchronization.
    It will be automatically shifted to UTC, if necessary.
    If omitted, `DateTime.now!/1` will be used with Time Synchronization -
    if the default timezone is "Etc/UTC", then UTC Time Synchronization will be used.
    The `utc` option overrides the behaviour of the default timezone -
    you may use a non-UTC timezone and still be able to use UTC.
  - `utc: boolean()` - Optional. Whether to use UTC Time Synchronization.
  """
  @spec send_time_synchronization(
          GenServer.server(),
          term(),
          Keyword.t()
        ) :: :ok | {:error, term()}
  def send_time_synchronization(server, destination \\ :broadcast, opts \\ [])

  def send_time_synchronization(server, destination, opts) when is_list(opts) do
    {new_utc, dt} =
      case Keyword.fetch(opts, :datetime) do
        {:ok, dt} ->
          if Keyword.get(opts, :utc, false) do
            {true, DateTime.shift_zone(dt, "Etc/UTC")}
          else
            {false, dt}
          end

        :error ->
          if Keyword.get(opts, :utc, false) do
            {true, DateTime.now!("Etc/UTC")}
          else
            tz = Application.get_env(:bacstack, :default_timezone, "Etc/UTC")
            {tz == "Etc/UTC", DateTime.now!(tz)}
          end
      end

    sync_apdu =
      if new_utc do
        Services.UtcTimeSynchronization.to_apdu(
          %Services.UtcTimeSynchronization{
            date: BACnetDate.from_date(DateTime.to_date(dt)),
            time: BACnetTime.from_time(DateTime.to_time(dt))
          },
          opts
        )
      else
        Services.TimeSynchronization.to_apdu(
          %Services.TimeSynchronization{
            date: BACnetDate.from_date(DateTime.to_date(dt)),
            time: BACnetTime.from_time(DateTime.to_time(dt))
          },
          opts
        )
      end

    with {:ok, apdu} <- sync_apdu,
         {:ok, addr} <- get_address(server, destination) do
      Client.send(server, addr, apdu, Keyword.drop(opts, [:datetime, :utc]))
    end
  end

  @doc """
  Subscribes for COV notification for a remote BACnet object property.

  When using confirmed COV notifications, the remote BACnet device requires
  you to send confirmations of the reception (`BACnet.Protocol.Services.SimpleACK`) -
  this is not done automatically.

  The following options are available:
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.SubscribeCovProperty.to_apdu/2`.
  - `confirmed: boolean()` - Optional. Request confirmed COV notifications.
    By default, COV notifications are requested to be unconfirmed.
  - `cov_increment: float()` - Optional. The COV increment to use for float properties.
  - `lifetime: non_neg_integer() | nil` - Optional. The COV subscription lifetime to use
    in seconds (defaults to 3600). To unsubscribe, use `nil`.
  - `pid: non_neg_integer()` - Optional. The process identifier to use. By default,
    this will be calculated from the caller PID (`node bits 0-3 << 28 + pid_number << 13 + pid_serial`).
  """
  @spec subscribe_cov_property(
          GenServer.server(),
          term(),
          ObjectIdentifier.t(),
          Constants.property_identifier(),
          Keyword.t()
        ) ::
          :ok
          | {:error, {term(), ObjectIdentifier.t()}}
          | {:error, term()}
  def subscribe_cov_property(
        server,
        destination,
        %ObjectIdentifier{} = object,
        property,
        opts \\ []
      )
      when ((is_atom(property) and property not in [:all, :required, :optional]) or
              (is_integer(property) and property >= 0)) and
             is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "subscribe_cov_property/5 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    confirmed = Keyword.get(opts, :confirmed, false)
    lifetime = Keyword.get(opts, :lifetime, 3600)
    cov_increment = if lifetime, do: Keyword.get(opts, :cov_increment), else: nil

    pid =
      Keyword.get_lazy(opts, :pid, fn ->
        [node, pid, pid2] =
          self()
          |> :erlang.pid_to_list()
          |> :binary.list_to_bin()
          |> then(&Regex.scan(~r/<(\d+)\.(\d+)\.(\d+)>/, &1))
          # Get the first match list in the list
          |> hd()
          # Remove the full match from the list
          |> tl()
          |> Enum.map(&String.to_integer/1)

        Bitwise.bsl(Bitwise.band(node, 0x0F), 28) + Bitwise.bsl(pid, 13) + pid2
      end)

    with {:ok, req} <-
           Services.SubscribeCovProperty.to_apdu(
             %Services.SubscribeCovProperty{
               process_identifier: pid,
               monitored_object: object,
               issue_confirmed_notifications: confirmed,
               lifetime: lifetime,
               monitored_property: %PropertyRef{
                 property_identifier: property,
                 property_array_index: opts[:array_index]
               },
               cov_increment: cov_increment
             },
             opts
           ),
         {:ok, %APDU.SimpleACK{} = _apdu} <- Client.send(server, destination, req, opts) do
      :ok
    else
      {:ok, apdu} -> {:error, apdu}
      {:error, _err} = err -> err
    end
  end

  @doc """
  Sends a Who-Is service request to the network (local broadcast).
  The I-Am responses will be collected and returned.

  See also `BACnet.Protocol.Services.WhoIs`.

  By default, it will collect all responses received until `timeout`.
  By using `max` opts, one can tell the function how many to receive
  and then stop prematurely. Either `max` or `timeout` will stop
  the collecting. Timeout must be minimum `10`ms.

  This function will by default spawn a new task and subscribe for
  BACnet notification messages and afterwards unsubscribe.
  This behaviour can be disabled through `no_subscribe` opts.

  The following options are available, in addition the `Client.send/4` options:
  - `apdu_destination: term()` - Optional. Overrides the APDU destination address.
  - `high_limit: pos_integer()` - Optional. The maximum BACnet device ID for the Who-Is query.
  - `low_limit: pos_integer()` - Optional. The minimum BACnet device ID for the Who-Is query.
  - `max: pos_integer()` - Optional. The maximum amount of IAm responses to collect.
  - `no_subscribe: boolean()` - Optional. Whether to spawn a new task.
  """
  @spec who_is(GenServer.server(), pos_integer(), Keyword.t()) ::
          {:ok, [{source_address :: term(), Services.IAm.t()}]} | {:error, term()}
  def who_is(server, timeout \\ 5000, opts \\ [])
      when is_integer(timeout) and timeout >= 10 and is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "who_is/3 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    who_is = %Services.WhoIs{
      device_id_low_limit: opts[:low_limit],
      device_id_high_limit: opts[:high_limit]
    }

    dest =
      case Keyword.fetch(opts, :apdu_destination) do
        {:ok, _val} = val -> val
        _else -> GenServer.call(server, :get_broadcast_address)
      end

    with {:ok, broadcast} <- dest,
         req_opts =
           Keyword.drop(opts, [:high_limit, :low_limit, :no_subscribe, :apdu_destination]),
         {:ok, req} <- Services.WhoIs.to_apdu(who_is, []) do
      if opts[:no_subscribe] do
        do_who_is(server, timeout, req, broadcast, req_opts, opts)
      else
        task = Task.async(fn -> do_who_is(server, timeout, req, broadcast, req_opts, opts) end)
        Task.await(task, trunc(timeout * 1.5))
      end
    end
  end

  @doc """
  Write to a single property from a remote BACnet object.

  Either the actual value of the property can be given and then the value will
  be automatically encoded through `BACnet.Protocol.ObjectsUtility`.
  Or an `Encoding` struct (or list of) can be given, which will be used
  directly without validation.

  The following options are available:
  - All options from `BACnet.Stack.Client.send/4`.
  - All options from `BACnet.Protocol.Services.WriteProperty.to_apdu/2`.
  - `array_index: non_neg_integer() | nil` - Optional. The property array index to write to.
  - `priority: 1..16 | nil` - Optional. The BACnet priority to write to.
  """
  @spec write_property(
          GenServer.server(),
          term(),
          ObjectIdentifier.t(),
          Constants.property_identifier() | non_neg_integer(),
          term() | Encoding.t() | [Encoding.t()],
          Keyword.t()
        ) ::
          :ok
          | {:error, BACnet.Protocol.apdu()}
          | {:error, term()}
  def write_property(server, destination, object, property, value, opts \\ [])

  def write_property(
        server,
        destination,
        %ObjectIdentifier{} = object,
        property,
        %Encoding{} = value,
        opts
      )
      when ((is_atom(property) and property not in [:all, :required, :optional]) or
              (is_integer(property) and property >= 0)) and
             is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "write_property/6 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    do_write_property(server, destination, object, property, value, opts)
  end

  def write_property(
        server,
        destination,
        %ObjectIdentifier{} = object,
        property,
        value,
        opts
      )
      when ((is_atom(property) and property not in [:all, :required, :optional]) or
              (is_integer(property) and property >= 0)) and
             is_list(value) and
             is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "write_property/6 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    if Enum.all?(value, &is_struct(&1, Encoding)) do
      do_write_property(server, destination, object, property, value, opts)
    else
      cast_opts =
        if property == :present_value and opts[:priority] do
          Keyword.put(opts, :allow_nil, true)
        else
          opts
        end

      cast_opts = Keyword.put(cast_opts, :allow_partial, opts[:array_index] != nil)

      with {:ok, result} <-
             ObjectsUtility.cast_value_to_property(object, property, value, cast_opts) do
        do_write_property(server, destination, object, property, result, opts)
      end
    end
  end

  def write_property(
        server,
        destination,
        %ObjectIdentifier{} = object,
        property,
        value,
        opts
      )
      when is_atom(property) and property not in [:all, :required, :optional] and
             is_list(opts) do
    unless Keyword.keyword?(opts) do
      raise ArgumentError,
            "write_property/6 expected opts to be a keyword list, " <>
              "got: #{inspect(opts)}"
    end

    cast_opts =
      if property == :present_value and opts[:priority] do
        Keyword.put(opts, :allow_nil, true)
      else
        opts
      end

    cast_opts = Keyword.put(cast_opts, :allow_partial, opts[:array_index] != nil)

    with {:ok, result} <-
           ObjectsUtility.cast_value_to_property(object, property, value, cast_opts) do
      do_write_property(server, destination, object, property, result, opts)
    end
  end

  defp do_write_property(
         server,
         destination,
         %ObjectIdentifier{} = object,
         property,
         value,
         opts
       ) do
    with {:ok, req} <-
           Services.WriteProperty.to_apdu(
             %Services.WriteProperty{
               object_identifier: object,
               property_identifier: property,
               property_array_index: opts[:array_index],
               property_value: value,
               priority: opts[:priority]
             },
             opts
           ),
         {:ok, %APDU.SimpleACK{}} <- Client.send(server, destination, req, opts) do
      :ok
    else
      {:ok, apdu} -> {:error, apdu}
      {:error, _err} = err -> err
    end
  end

  defp get_address(server, :broadcast), do: GenServer.call(server, :get_broadcast_address)
  defp get_address(_server, addr), do: {:ok, addr}

  @spec make_access_property_from_identifier(
          AccessSpecification.Property.t()
          | Constants.property_identifier()
          | non_neg_integer()
          | term()
        ) :: AccessSpecification.Property.t() | no_return()
  defp make_access_property_from_identifier(%AccessSpecification.Property{} = identifier),
    do: identifier

  defp make_access_property_from_identifier(identifier)
       when identifier in [:all, :required, :optional],
       do: identifier

  defp make_access_property_from_identifier(identifier)
       when is_atom(identifier) or (is_integer(identifier) and identifier >= 0) do
    %AccessSpecification.Property{
      property_identifier: identifier,
      property_array_index: nil,
      property_value: nil
    }
  end

  defp make_access_property_from_identifier(identifier) do
    raise ArgumentError, "Invalid property identifier, got: #{inspect(identifier)}"
  end

  defp do_who_is(server, timeout, req, broadcast, req_opts, opts) do
    no_subscribe = opts[:no_subscribe]

    with :ok <-
           if(no_subscribe,
             do: :ok,
             else: Client.subscribe(server, self())
           ),
         :ok <- Client.send(server, broadcast, req, req_opts) do
      ref = make_ref()
      timer = Process.send_after(self(), {__MODULE__, :stop_who_is, ref}, timeout)

      max_items = opts[:max]

      iams =
        Enum.reduce_while(1..10_000_000, [], fn _index, acc ->
          if max_items > 0 and length(acc) >= max_items do
            {:halt, acc}
          else
            receive do
              {:bacnet_client, _ref,
               %APDU.UnconfirmedServiceRequest{
                 service: Constants.macro_assert_name(:unconfirmed_service_choice, :i_am)
               } = apdu, {source_addr, _bvlc, _npci}, _pid} ->
                case Services.IAm.from_apdu(apdu) do
                  {:ok, service} ->
                    {:cont, [{source_addr, service} | acc]}

                  {:error, err} ->
                    log_debug(fn ->
                      "ClientHelper.who_is/3 encountered an error during " <>
                        "APDU to service transformation, error: #{inspect(err)}"
                    end)

                    {:cont, acc}
                end

              {__MODULE__, :stop_who_is, ^ref} ->
                {:halt, acc}
            end
          end
        end)

      # Cleanup subscription (if done)
      unless no_subscribe do
        Client.unsubscribe(server, self())
      end

      # Cleanup timer
      Process.cancel_timer(timer)

      # Receive timer in case it was sent before cancellation (and after reduce)
      receive do
        {__MODULE__, :stop_who_is, ^ref} -> :ok
      after
        0 -> :ok
      end

      {:ok, iams}
    end
  end
end