Untitled notebook
Section
import Ecto.Query
defmodule Fetcher do
def fetch(url) do
%{status_code: 200, body: body} =
Transport.Shared.Wrapper.HTTPoison.impl().get!(url, [], follow_redirect: true)
%{
header: %{
gtfs_realtime_version: _version,
incrementality: :FULL_DATASET,
timestamp: _timestamp
},
entity: entity
} = TransitRealtime.FeedMessage.decode(body)
entity
end
end
DB.Resource
|> where(format: "gtfs-rt")
|> preload(:dataset)
|> DB.Repo.all(log: false)
|> Enum.filter(& &1.dataset.is_active)
# |> Enum.take(1)
|> Task.async_stream(
fn resource ->
try do
vehicles =
resource.url
|> Fetcher.fetch()
|> Enum.filter(& &1.vehicle)
|> Enum.map(& &1.vehicle)
|> Enum.map(fn v ->
%{
vehicle_id: v.vehicle.id,
latitude: v.position.latitude,
longitude: v.position.longitude,
bearing: v.position.bearing,
timestamp: v.timestamp
}
end)
%{
resource_id: resource.id,
vehicles: vehicles
}
rescue
e -> %{error: e}
end
end,
max_concurrency: 50,
on_timeout: :kill_task,
timeout: 10000
)
|> Enum.map(fn {:ok, r} -> r end)
|> Enum.reject(fn r -> Map.has_key?(r, :error) end)
|> Enum.reject(fn r -> r.vehicles |> length() == 0 end)
|> Enum.map(fn x -> {x.resource_id, x.vehicles |> length()} end)
|> Enum.into([])