diff --git a/config/releases.exs b/config/releases.exs index dd49ed5e..3589a7c4 100644 --- a/config/releases.exs +++ b/config/releases.exs @@ -50,3 +50,5 @@ if System.get_env("ENABLE_LOGGER_TELEGRAM") == "true" do chat_id: System.fetch_env!("CHAT_ID"), token: System.fetch_env!("TOKEN") end + +config :teslamate, :srtm_cache, System.get_env("SRTM_CACHE", "/tmp/srtm_cache") diff --git a/lib/teslamate/log.ex b/lib/teslamate/log.ex index 89db743b..5b4ce513 100644 --- a/lib/teslamate/log.ex +++ b/lib/teslamate/log.ex @@ -100,12 +100,22 @@ defmodule TeslaMate.Log do |> Repo.one() end - def get_positions_without_elevation(min_id \\ 0) do + def get_positions_without_elevation(min_id \\ 0, opts \\ []) do + limit = Keyword.get(opts, :limit, 100) + Position |> where([p], p.id > ^min_id and is_nil(p.elevation)) |> order_by(asc: :id) - |> limit(100) + |> limit(^limit) |> Repo.all() + |> Enum.reverse() + |> case do + [%Position{id: next} | _] = positions -> + {Enum.reverse(positions), next} + + [] -> + {[], nil} + end end def update_position(%Position{} = position, attrs) do diff --git a/lib/teslamate/mapping.ex b/lib/teslamate/mapping.ex index 583fb546..e77fb540 100644 --- a/lib/teslamate/mapping.ex +++ b/lib/teslamate/mapping.ex @@ -1,5 +1,5 @@ defmodule TeslaMate.Mapping do - use GenServer + use GenStateMachine require Logger import Core.Dependency, only: [call: 3] @@ -7,23 +7,25 @@ defmodule TeslaMate.Mapping do alias TeslaMate.Log defstruct [:client, :timeout, :blocked_on, :deps, :name] - alias __MODULE__, as: State + alias __MODULE__, as: Data @name __MODULE__ # API def start_link(opts) do - GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, @name)) + GenStateMachine.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, @name)) end def get_elevation(name \\ @name, coordinates) do - GenServer.call(name, {:get_elevation, coordinates}, 1000) + GenStateMachine.call(name, {:get_elevation, coordinates}, 2000) end # TODO # * Add Docker Volume for caches # * Grafana: m to ft + # * Test get_positions_without_elevation + # * Test update_position # Callbacks @@ -39,90 +41,115 @@ defmodule TeslaMate.Mapping do log: Keyword.get(opts, :deps_log, Log) } - {:ok, %State{client: client, timeout: timeout, deps: deps, name: name}, - {:continue, {:add_elevation_to_positions, 0}}} + {:ok, :ready, %Data{client: client, timeout: timeout, deps: deps, name: name}, + {:next_event, :internal, {:fetch_positions, 0}}} end + ## Calls + @impl true - def handle_continue({:add_elevation_to_positions, min_id}, state) do - case call(state.deps.log, :get_positions_without_elevation, [min_id]) |> Enum.reverse() do - [%Position{id: next_min_id} | _] = positions -> - {:noreply, %State{state | blocked_on: :update_positions}, - {:continue, {:update_positions, positions, next_min_id}}} + def handle_event({:call, from}, {:get_elevation, {lat, lng}}, :ready, %Data{} = data) do + task = Task.async(fn -> do_get_elevation({lat, lng}, data) end) - [] -> - Process.send_after(self(), :add_elevation_to_positions, :timer.hours(6)) - {:noreply, state} - end - end - - def handle_continue({:update_positions, positions, next_min_id}, %State{} = state) do - state = - Enum.reduce(positions, state, fn %Position{latitude: lat, longitude: lng} = pos, state -> - with {:ok, el, client} <- do_get_elevation({lat, lng}, state), - {:ok, _pos} <- call(state.deps.log, :update_position, [pos, %{elevation: el}]) do - %State{state | client: client} - else - {:error, reason} -> - log_warning(reason) - state - end - end) - - {:noreply, %State{state | blocked_on: nil}, - {:continue, {:add_elevation_to_positions, next_min_id}}} - end - - @impl true - def handle_call({:get_elevation, {_, _}}, _from, %State{blocked_on: blocker} = state) - when is_reference(blocker) or blocker == :update_positions do - {:reply, nil, state} - end - - def handle_call({:get_elevation, {lat, lng}}, _from, %State{} = state) do - task = Task.async(fn -> do_get_elevation({lat, lng}, state) end) - - case Task.yield(task, state.timeout) do + case Task.yield(task, data.timeout) do {:ok, {:ok, elevation, client}} -> - {:reply, elevation, %State{state | client: client}} + {:keep_state, %Data{data | client: client}, {:reply, from, elevation}} {:ok, {:error, :unavailable}} -> - {:reply, nil, state} + {:keep_state_and_data, {:reply, from, nil}} {:ok, {:error, reason}} -> log_warning(reason) - {:reply, nil, state} + {:keep_state_and_data, {:reply, from, nil}} nil -> - {:reply, nil, %State{state | blocked_on: task.ref}} + {:next_state, {:waiting, task.ref}, data, {:reply, from, nil}} end end - @impl true - def handle_info({task, result}, %State{blocked_on: task} = state) when is_reference(task) do + def handle_event({:call, from}, {:get_elevation, _coords}, _state, _data) do + {:keep_state_and_data, {:reply, from, nil}} + end + + ## Internal + + def handle_event(event, {:fetch_positions, min_id}, :ready, %Data{} = data) + when event in [:internal, :state_timeout] do + Logger.info("Updating positions ...") + + case call(data.deps.log, :get_positions_without_elevation, [min_id, [limit: 100]]) do + {[], nil} -> + {:keep_state_and_data, schedule_fetch()} + + {positions, next} -> + :ok = GenStateMachine.cast(self(), :process) + {:next_state, {:update, positions, next, nil}, data} + end + end + + ## Cast + + def handle_event(:cast, :process, {:update, [], next, nil}, data) do + {:next_state, :ready, data, {:next_event, :internal, {:fetch_positions, next}}} + end + + def handle_event(:cast, :process, {:update, [%Position{} = p | rest], next, nil}, data) do + task = Task.async(fn -> do_get_elevation({p.latitude, p.longitude}, data) end) + + case Task.yield(task, data.timeout) do + {:ok, {:ok, elevation, client}} -> + {:ok, _pos} = call(data.deps.log, :update_position, [p, %{elevation: elevation}]) + :ok = GenStateMachine.cast(self(), :process) + {:next_state, {:update, rest, next, nil}, %Data{data | client: client}} + + {:ok, {:error, :unavailable}} -> + :ok = GenStateMachine.cast(self(), :process) + {:next_state, {:update, rest, next, nil}, data} + + {:ok, {:error, reason}} -> + log_warning(reason) + :ok = GenStateMachine.cast(self(), :process) + {:next_state, {:update, rest, next, nil}, data} + + nil -> + {:next_state, {:update, [p | rest], next, task.ref}, data} + end + end + + ## Info + + def handle_event(:info, {ref, result}, {:waiting, ref}, data) do case result do {:ok, _elevation, %SRTM.Client{} = client} -> - {:noreply, %State{state | client: client, blocked_on: nil}} + {:next_state, :ready, %Data{data | client: client}, schedule_fetch()} {:error, reason} -> log_warning(reason) - {:noreply, %State{state | blocked_on: nil}} + {:next_state, :ready, data, schedule_fetch()} end end - def handle_info({:DOWN, _ref, :process, _pid, :normal}, state) do - {:noreply, state} + def handle_event(:info, {ref, result}, {:update, [%Position{} = p | rest], next, ref}, data) do + case result do + {:ok, elevation, %SRTM.Client{} = client} -> + {:ok, _pos} = call(data.deps.log, :update_position, [p, %{elevation: elevation}]) + :ok = GenStateMachine.cast(self(), :process) + {:next_state, {:update, rest, next, nil}, %Data{data | client: client}} + + {:error, reason} -> + log_warning(reason) + :ok = GenStateMachine.cast(self(), :process) + {:next_state, {:update, rest, next, nil}, data} + end end - def handle_info(:add_elevation_to_positions, state) do - {:noreply, state, {:continue, {:add_elevation_to_positions, 0}}} + def handle_event(:info, {:DOWN, _ref, :process, _pid, :normal}, _state, _data) do + :keep_state_and_data end # Private - defp do_get_elevation({lat, lng}, %State{} = state) do - %State{client: client, deps: %{srtm: srtm}, name: name} = state - + defp do_get_elevation({lat, lng}, %Data{client: client, deps: %{srtm: srtm}, name: name} = data) do case :fuse.ask(name, :sync) do :ok -> with {:error, reason} <- call(srtm, :get_elevation, [client, lat, lng]) do @@ -135,11 +162,19 @@ defmodule TeslaMate.Mapping do {:error, :not_found} -> :fuse.install(name, {{:standard, 2, :timer.minutes(2)}, {:reset, :timer.minutes(5)}}) - do_get_elevation({lat, lng}, state) + do_get_elevation({lat, lng}, data) end end - defp log_warning(reason), do: Logger.warn("Elevation query failed: #{inspect(reason)}") + defp schedule_fetch do + {:state_timeout, :timer.hours(6), {:fetch_positions, 0}} + end - defp cache_path, do: Application.fetch_env!(:teslamate, :srtm_cache) + defp log_warning(reason) do + Logger.warn("Elevation query failed: #{inspect(reason)}") + end + + defp cache_path do + Application.fetch_env!(:teslamate, :srtm_cache) + end end diff --git a/mix.exs b/mix.exs index 25fe7f97..72cde5ad 100644 --- a/mix.exs +++ b/mix.exs @@ -4,7 +4,7 @@ defmodule TeslaMate.MixProject do def project do [ app: :teslamate, - version: "1.6.2", + version: "1.7.0-beta", elixir: "~> 1.9", elixirc_paths: elixirc_paths(Mix.env()), compilers: [:phoenix, :gettext] ++ Mix.compilers(), diff --git a/test/support/mocks/log.ex b/test/support/mocks/log.ex index 4d3d4b3f..f901d0b5 100644 --- a/test/support/mocks/log.ex +++ b/test/support/mocks/log.ex @@ -45,8 +45,8 @@ defmodule LogMock do GenServer.call(name, {:insert_charge, car_id, attrs}) end - def get_positions_without_elevation(name, min_id) do - GenServer.call(name, {:get_positions_without_elevation, min_id}) + def get_positions_without_elevation(name, min_id, opts) do + GenServer.call(name, {:get_positions_without_elevation, min_id, opts}) end # Callbacks @@ -106,9 +106,9 @@ defmodule LogMock do {:reply, {:ok, %Update{}}, state} end - def handle_call({:get_positions_without_elevation, _} = action, _from, %State{pid: pid} = state) do - send(pid, action) - {:reply, [], state} + def handle_call({:get_positions_without_elevation, min_id, _opts}, _from, state) do + send(state.pid, {:get_positions_without_elevation, min_id}) + {:reply, {[], nil}, state} end def handle_call(action, _from, %State{pid: pid} = state) do diff --git a/test/teslamate/mapping/update_positions_test.exs b/test/teslamate/mapping/update_positions_test.exs index 67ce07fe..dd946396 100644 --- a/test/teslamate/mapping/update_positions_test.exs +++ b/test/teslamate/mapping/update_positions_test.exs @@ -9,7 +9,8 @@ defmodule TeslaMate.Mapping.UpdatePositionsTest do {:ok, _pid} = start_supervised({SRTMMock, name: srtm_name, pid: self(), responses: responses}) - {:ok, _} = start_supervised({Mapping, name: name, deps_srtm: {SRTMMock, srtm_name}}) + {:ok, _} = + start_supervised({Mapping, name: name, timeout: 100, deps_srtm: {SRTMMock, srtm_name}}) :ok end @@ -20,31 +21,56 @@ defmodule TeslaMate.Mapping.UpdatePositionsTest do positions = %{date: DateTime.utc_now(), latitude: 0, longitude: 0} |> List.duplicate(201) + |> List.replace_at(50, %{date: DateTime.utc_now(), latitude: 1, longitude: 1}) + |> List.replace_at(150, %{date: DateTime.utc_now(), latitude: 1, longitude: 1}) |> Enum.map(fn position -> {:ok, pos} = Log.insert_position(car_id, position) pos end) - :ok = start_mapping(name, %{{0.0, 0.0} => fn -> {:ok, 42} end}) + :ok = + start_mapping(name, %{ + {0.0, 0.0} => fn -> {:ok, 42} end, + {1.0, 1.0} => fn -> + :timer.sleep(100) + {:ok, 420} + end + }) - for _ <- positions do - assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 0.0, 0.0}} + # blocked + assert Mapping.get_elevation(name, {0, 0}) == nil + assert Mapping.get_elevation(name, {0, 0}) == nil + assert Mapping.get_elevation(name, {0, 0}) == nil + + for {_, i} <- Enum.with_index(positions) do + if i in [50, 150] do + assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 1.0, 1.0}} + else + assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 0.0, 0.0}} + end end - for position <- positions do - assert %Position{elevation: 42.0} = TeslaMate.Repo.get(Position, position.id) + for {position, i} <- Enum.with_index(positions) do + if i in [50, 150] do + assert %Position{elevation: 420.0} = TeslaMate.Repo.get(Position, position.id) + else + assert %Position{elevation: 42.0} = TeslaMate.Repo.get(Position, position.id) + end end refute_receive _ end @tag :capture_log - test "handles errors during update", %{test: name} do + test "handles errors during update!", %{test: name} do assert %Car{id: car_id} = car_fixture() - [p0, p1, p2] = + [p0, p1, p2, p3, p4, p5] = [ %{date: DateTime.utc_now(), latitude: 0, longitude: 0}, + %{date: DateTime.utc_now(), latitude: 1, longitude: 1}, + %{date: DateTime.utc_now(), latitude: 99, longitude: 99}, + %{date: DateTime.utc_now(), latitude: 99, longitude: 99}, %{date: DateTime.utc_now(), latitude: 99, longitude: 99}, %{date: DateTime.utc_now(), latitude: 0, longitude: 0} ] @@ -56,46 +82,27 @@ defmodule TeslaMate.Mapping.UpdatePositionsTest do :ok = start_mapping(name, %{ {0.0, 0.0} => fn -> {:ok, 42} end, - {99.0, 99.0} => fn -> {:error, :kaputt} end + {1.0, 1.0} => fn -> {:error, :boom} end, + {99.0, 99.0} => fn -> + :timer.sleep(100) + {:error, :kaputt} + end }) assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 0.0, 0.0}} + assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 1.0, 1.0}} assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 99.0, 99.0}} - assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 0.0, 0.0}} + assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 99.0, 99.0}} + # 4th and 5th are :unavailable - :timer.sleep(10) + :timer.sleep(300) assert %Position{elevation: 42.0} = TeslaMate.Repo.get(Position, p0.id) assert %Position{elevation: nil} = TeslaMate.Repo.get(Position, p1.id) - assert %Position{elevation: 42.0} = TeslaMate.Repo.get(Position, p2.id) - - refute_receive _ - end - - test "reacts to :add_elevation_to_positions messages", %{test: name} do - assert %Car{id: car_id} = car_fixture() - - :ok = start_mapping(name, %{{0.0, 0.0} => fn -> {:ok, 42} end}) - - refute_receive _, 40 - - positions = - %{date: DateTime.utc_now(), latitude: 0, longitude: 0} - |> List.duplicate(5) - |> Enum.map(fn position -> - {:ok, pos} = Log.insert_position(car_id, position) - pos - end) - - send(name, :add_elevation_to_positions) - - for _ <- positions do - assert_receive {SRTM, {:get_elevation, %SRTM.Client{}, 0.0, 0.0}} - end - - for position <- positions do - assert %Position{elevation: 42.0} = TeslaMate.Repo.get(Position, position.id) - end + assert %Position{elevation: nil} = TeslaMate.Repo.get(Position, p2.id) + assert %Position{elevation: nil} = TeslaMate.Repo.get(Position, p3.id) + assert %Position{elevation: nil} = TeslaMate.Repo.get(Position, p4.id) + assert %Position{elevation: nil} = TeslaMate.Repo.get(Position, p5.id) refute_receive _ end