singletone input device support

.travis.yml

@@ -6,7 +6,5 @@ env:
 
 matrix:
   include:
-    - otp_release: 19.0
-      elixir: 1.4
-    - otp_release: 20.0
-      elixir: 1.5.1
+    - otp_release: 20
+      elixir: 1.5

README.md

@@ -6,13 +6,13 @@ Elixir library to read the DHT series of humidity and temperature sensors on a R
 The library is supposed to be included in a [nerves project](http://nerves-project.org/).
 
 If you want to build your project directly on a Raspberry (not in a crosscompiling nerves project)
-just export MIX_TARGET environment variable to you mix build.
-Valid values for MIX_TARGET are rpi | rp2 | rp3.
+just export `MIX_TARGET` environment variable to you mix build.
+Valid values for `MIX_TARGET` are `rpi`, `rp2`, `rp3`.
 
 * Supported sensors: DHT11, DHT22, AM2302
-* Supported boards: Raspberry 1 / 2 / 3
+* Supported boards: Raspberry 1, 2, 3
 
-Note: the library has no external dependencies and use a C executable to read the sensors data.
+**Note**: the library has no external dependencies and use a C executable to read the sensors data.
 
 ## Installation
 
@@ -21,7 +21,7 @@ The package can be installed by adding `nerves_dht` to your list of dependencies
 ```elixir
 def deps do
   [
-    {:nerves_dht, git: "https://github.com/visciang/nerves_dht.git", tag: "1.0.0"}
+    {:nerves_dht, git: "https://github.com/visciang/nerves_dht.git", tag: "1.1.0"}
   ]
 end
 ```
@@ -35,3 +35,20 @@ iex> NervesDHT.read(:am2302, 17)
 iex> NervesDHT.stream(:am2302, 17) |> Enum.take(2)
 [{:ok, 55.1, 24.719}, {:ok, 55.12, 24.9}]
 ```
+
+If you plan to read concurrently from the sensor, add `NervesDHT` to your application supervisor tree:
+
+```elixir
+children = [
+  {NervesDHT, [name: :my_sensor, sensor: :am2302, pin: 17]},
+  ...
+]
+Supervisor.start_link(children, strategy: :one_for_one)
+```
+
+and read with:
+
+```elixir
+NervesDHT.device_read(:my_sensor)
+NervesDHT.device_stream(:my_sensor)
+```

lib/nerves_dht.ex

@@ -1,8 +1,18 @@
 defmodule NervesDHT do
   @moduledoc """
   Elixir library to read the DHT series of humidity and temperature sensors on a Raspberry Pi.
+
+  The library expose direct access to the sensors via `read/4` and `stream/3`.
+
+  If your application has multiple processes reading the same sensor concurrently you should
+  add `NervesDHT` under your supervisor (see `child_spec/1`).
+  It ensures that only one read operation at the time is executed.
+  In case of multiple sources asking for a read while the operation is in progress, they will
+  get the result of the ongoing read.
   """
 
+  @typedoc "Device identifier"
+  @type device_id :: atom
   @typedoc "Sensor model"
   @type sensor :: :dht11 | :dht22 | :am2302
   @typedoc "GPIO pin (using BCM numbering)"
@@ -15,15 +25,15 @@ defmodule NervesDHT do
   @type interval :: non_neg_integer
   @typedoc "Error reason"
   @type reason :: :timeout | :checksum | :argument | :gpio
-  @typedoc "Successful reading humidity and temperature values"
+  @typedoc "Reading result"
   @type result :: {:ok, humidity :: float(), temperature :: float()} | {:error, reason}
 
-  # executable name injection (tests use "dht_exe.sh" fake)
-  @dht_exe Application.get_env(:nerves_dht, :dht_exe, "dht")
+  @retries 3
+  @delay 2000
 
   @doc """
   Read DHT sensor values.
-  
+
   Read from the specified `sensor` type (DHT11, DHT22, or AM2302) on
   specified `pin` and return humidity (as a floating point value in
   percent) and temperature (as a floating point value in Celsius) as
@@ -44,10 +54,10 @@ defmodule NervesDHT do
 
   """
   @spec read(sensor, pin, retries, delay) :: result
-  def read(sensor, pin, retries \\ 3, delay \\ 2000)
+  def read(sensor, pin, retries \\ @retries, delay \\ @delay)
 
   def read(sensor, pin, retries, delay) do
-    result = dht_read(sensor, pin)
+    result = NervesDHT.Driver.dht_read(sensor, pin)
 
     case result do
       {:ok, humidity, temperature} ->
@@ -60,10 +70,9 @@ defmodule NervesDHT do
   @doc """
   Return a Stream of sensor readings.
 
-  The reading `interval` defaults to 2 seconds.
-  Interval is the wait period beetwen two consecutive read attempts.
-  Since the device takes some X time to transmit the reading, the Stream
-  will push data with a `period >= X+interval`.
+  `interval` is the wait period beetwen two consecutive read attempts and
+  defaults to 2 seconds. Since the device takes some `x` time to transmit
+  the reading, the Stream will push data with a `period >= (interval + x)`.
 
   ## Examples
 
@@ -72,11 +81,66 @@ defmodule NervesDHT do
 
   """
   @spec stream(sensor, pin, interval) :: Enumerable.t
-  def stream(sensor, pin, interval \\ 2000) do
+  def stream(sensor, pin, interval \\ @delay) do
     Stream.interval(interval)
     |> Stream.map(fn(_) -> read(sensor, pin, 0, 0) end)
   end
 
+  @doc """
+  Return the child specification to put the a named device under your supervisor tree.
+  The device can be used to read concurrently from the sensor.
+
+  Add to you supervisor:
+
+  ```elixir
+  children = [
+    {NervesDHT, [name: :my_sensor, sensor: :am2302, pin: 17]},
+    ...
+  ]
+  Supervisor.start_link(children, strategy: :one_for_one)
+  ```
+
+  Read from the named `:my_sensor` device:
+
+  ```elixir
+  NervesDHT.device_read(:my_sensor)
+  ```
+  """
+  @spec child_spec([name: device_id, sensor: sensor, pin: pin]) :: Supervisor.child_spec
+  def child_spec([name: name, sensor: sensor, pin: pin]) do
+    fun = fn -> __MODULE__.read(sensor, pin) end
+    timeout = @retries * @delay
+    %{
+      id: "#{__MODULE__}_#{sensor}_#{pin}",
+      start: {NervesDHT.Device, :start_link, [name, fun, timeout]},
+      restart: :permanent,
+      shutdown: 5000,
+      type: :worker
+    }
+  end
+
+  @doc """
+  Read DHT sensor values from the named device `device_id`.
+  The underlying read/4 operation will apply the default `retries`, `delay` strategy.
+
+  See `child_spec/1`, `read/4`.
+  """
+  @spec device_read(device_id) :: result
+  def device_read(device_id) do
+    NervesDHT.Device.read(device_id, (@retries + 1) * @delay)
+  end
+
+  @doc """
+  Return a Stream of sensor readings from the named device `device_id`.
+
+  See `child_spec/1`, stream/3.
+  """
+  @spec device_stream(device_id, interval) :: Enumerable.t
+  def device_stream(device_id, interval \\ @delay) do
+    Stream.interval(interval)
+    |> Stream.map(fn(_) -> device_read(device_id) end)
+  end
+
   defp read_again(_sensor, _pin, retries, _delay, error) when retries <= 0 do
     {:error, error}
   end
@@ -90,39 +154,4 @@ defmodule NervesDHT do
       {:error, error}
     end
   end
-
-  defp dht_read(sensor, pin) do
-    cmd = Application.app_dir(:nerves_dht, Path.join("priv", @dht_exe))
-    args = dht_cmd_args(sensor, pin)
-
-    {result, exit_status} = System.cmd(cmd, args)
-
-    if exit_status == 0 do
-      [humidity_str, temperature_str] = String.split(result)
-
-      {humidity, ""} = Float.parse(humidity_str)
-      {temperature, ""} = Float.parse(temperature_str)
-
-      {:ok, humidity, temperature}
-    else
-      {:error, dht_error_code_to_reason(exit_status)}
-    end
-  end
-
-  defp dht_cmd_args(sensor, pin) do
-    [
-      case sensor do
-        :dht11 -> "11"
-        :dht22 -> "22"
-        :am2302 -> "2302"
-      end,
-      to_string(pin)
-    ]
-  end
-
-  # -1, -2, -3, -4
-  defp dht_error_code_to_reason(255), do: :timeout
-  defp dht_error_code_to_reason(254), do: :checksum
-  defp dht_error_code_to_reason(253), do: :argument
-  defp dht_error_code_to_reason(252), do: :gpio
 end

lib/nerves_dht/device.ex

@@ -0,0 +1,76 @@
+defmodule NervesDHT.Device do
+  @moduledoc false
+
+  defmodule Runner do
+    @moduledoc false
+
+    use GenServer
+
+    defmodule State do
+      @moduledoc false
+
+      defstruct [:father, :fun, :timeout]
+    end
+
+    def start_link(father, fun, timeout) do
+      state = %State{father: father, fun: fun, timeout: timeout}
+      GenServer.start_link(__MODULE__, state)
+    end
+
+    @impl GenServer
+    def handle_cast(:read, %State{father: father, fun: fun, timeout: timeout}=state) do
+      task = Task.async(fun)
+      case Task.yield(task, timeout) do
+        {:ok, result} ->
+          GenServer.cast(father, {:result, result})
+        nil ->
+          Task.shutdown(task)
+          GenServer.cast(father, {:result, {:error, :timeout}})
+      end
+      {:noreply, state}
+    end
+  end
+
+  use GenServer
+
+  defmodule State do
+    @moduledoc false
+    defstruct [:runner, :callers_queue]
+  end
+
+  def start_link(name, fun, timeout \\ 4000) do
+    GenServer.start_link(__MODULE__, [fun, timeout], name: name)
+  end
+
+  def read(ref, timeout \\ 5000) do
+    GenServer.call(ref, :read, timeout)
+  end
+
+  @impl GenServer
+  def init([fun, timeout]) do
+    {:ok, runner} = Runner.start_link(self(), fun, timeout)
+    {:ok, %State{runner: runner, callers_queue: []}}
+  end
+
+  @impl GenServer
+  def handle_call(:read, from, %State{runner: runner, callers_queue: []}=state) do
+    GenServer.cast(runner, :read)
+    {:noreply, %State{state | callers_queue: [from]}}
+  end
+
+  @impl GenServer
+  def handle_call(:read, from, %State{callers_queue: callers_queue}=state) do
+    {:noreply, %State{state | callers_queue: [from | callers_queue]}}
+  end
+
+  @impl GenServer
+  def handle_cast({:result, result}, %State{callers_queue: callers_queue}=state)
+      when callers_queue != [] do
+    notify(callers_queue, result)
+    {:noreply, %State{state | callers_queue: []}}
+  end
+
+  defp notify(callers_queue, reply) do
+    Enum.each(callers_queue, &(GenServer.reply(&1, reply)))
+  end
+end

lib/nerves_dht/driver.ex

@@ -0,0 +1,41 @@
+defmodule NervesDHT.Driver do
+  @moduledoc false
+
+  # executable name injection (tests use "dht_exe.sh" fake)
+  @dht_exe Application.get_env(:nerves_dht, :dht_exe, "dht")
+
+  def dht_read(sensor, pin) do
+    cmd = Application.app_dir(:nerves_dht, Path.join("priv", @dht_exe))
+    args = dht_cmd_args(sensor, pin)
+
+    {result, exit_status} = System.cmd(cmd, args)
+
+    if exit_status == 0 do
+      [humidity_str, temperature_str] = String.split(result)
+
+      {humidity, ""} = Float.parse(humidity_str)
+      {temperature, ""} = Float.parse(temperature_str)
+
+      {:ok, humidity, temperature}
+    else
+      {:error, dht_error_code_to_reason(exit_status)}
+    end
+  end
+
+  defp dht_cmd_args(sensor, pin) do
+    [
+      case sensor do
+        :dht11 -> "11"
+        :dht22 -> "22"
+        :am2302 -> "2302"
+      end,
+      to_string(pin)
+    ]
+  end
+
+  # -1, -2, -3, -4
+  defp dht_error_code_to_reason(255), do: :timeout
+  defp dht_error_code_to_reason(254), do: :checksum
+  defp dht_error_code_to_reason(253), do: :argument
+  defp dht_error_code_to_reason(252), do: :gpio
+end

mix.exs

@@ -5,7 +5,7 @@ defmodule NervesDHT.Mixfile do
     [
       app: :nerves_dht,
       version: "1.0.0",
-      elixir: "~> 1.4",
+      elixir: "~> 1.5",
       start_permanent: Mix.env == :prod,
       compilers: [:elixir_make] ++ Mix.compilers,
       deps: deps(),
@@ -14,7 +14,11 @@ defmodule NervesDHT.Mixfile do
   end
 
   def application do
-    []
+    if Mix.env == :test do
+      [extra_applications: [:logger]]
+    else
+      []
+    end
   end
 
   defp deps do