dadosabertosbh_crawler

defmodule Crawler.BusCoordinates do
  @moduledoc """
  Representa o conjunto de dados tempo [Real Ônibus - Coordenada atualizada](https://dados.pbh.gov.br/dataset/tempo_real_onibus_-_coordenada)
  * [Dicionário de dados](https://ckan.pbh.gov.br/dataset/730aaa4b-d14c-4755-aed6-433cb0ad9430/resource/825337e5-8cd5-43d9-ac52-837d80346721/download/dicionario_arquivo.csv)
  * [Arquivode conversão das linhas do sistema concencional](https://dados.pbh.gov.br/dataset/tempo_real_onibus_-_coordenada/resource/150bddd0-9a2c-4731-ade9-54aa56717fb6)

  Dados disponíveis publicamente no [BigQuery](https://console.cloud.google.com/bigquery?project=dadosabertosdebh&p=dadosabertosdebh&page=table&d=dadosabertosdebh&t=coordenadas_onibus)\n

  """

  defstruct [
    :codigo_linha,
    :codigo_evento,
    :codigo_do_veiculo,
    :timestamp,
    :coordenadas,
    :velocidade_instantanea,
    :distancia_pecorrida,
    :direcao_do_veiculo,
    :sentindo_da_viagem,
    :numero_linha,
    :nome_linha
  ]

  @typedoc """
  `:codigo_linha` NL - Código interno da linha\n
  `:codigo_evento` EV - Código do evento, 105 representa o evento de coordenadas\n
  `:codigo_do_veiculo` NV - Código do veículo\n
  `:timestamp` HR - Timestamp do evento\n
  `:coordenadas` LT, LG - Coordenada do veículo\n
  `:velocidade_instantanea` VL - Velocidade instantânea do veículo\n
  `:distancia_pecorrida` DT - Distância percorrida\n
  `:direcao_do_veiculo` DG - Direção do veículo\n
  `:sentindo_da_viagem` SV - Sentido do veículo em uma viagem ((1) ida, (2) volta)\n
  `:numero_linha` - Número da linha do ônibus\n
  `:nome_linha` - Nome da linha do ônibus\n
  """
  @type t :: %__MODULE__{
          codigo_linha: String.t(),
          codigo_evento: String.t(),
          codigo_do_veiculo: String.t(),
          timestamp: float(),
          coordenadas: struct(),
          velocidade_instantanea: float(),
          distancia_pecorrida: float(),
          direcao_do_veiculo: String.t(),
          sentindo_da_viagem: String.t(),
          numero_linha: String.t(),
          nome_linha: String.t()
        }

  require Logger

  @doc """
  Starts the Crawler.
  opts
    * `real_time_url`- Url to fetch bus coordinates, default to `https://temporeal.pbh.gov.br/?param=C"`
    * `pull_interval`- Time in miliseconds that should be pulled", default to `60s`
    * `bus_line_provider`- Default to `Crawler.CachexBusLineProvider`
  """
  @type opts :: [
          bus_line_provider: Crawler.BusLineProvider | nil,
          real_time_url: String.t() | nil,
          pull_interval: integer | nil
        ]
  @spec watch(opts) :: Enumerable.t()
  def watch(opts) do
    default = [
      real_time_url: "https://temporeal.pbh.gov.br/?param=C",
      pull_interval: 60 * 1000,
      bus_line_provider: Crawler.CachexBusLineProvider
    ]

    options = Keyword.merge(default, opts)
    real_time_url = options[:real_time_url]
    pull_interval = options[:pull_interval]
    bus_line_provider = options[:bus_line_provider]

    Stream.interval(pull_interval)
    |> Stream.flat_map(fn _ -> HTTPStream.get(real_time_url) end)
    |> HTTPStream.lines()
    # |> Stream.each(&IO.puts/1)
    |> CSV.decode!(separator: ?;, headers: true, strip_fields: true, validate_row_length: false)
    |> Stream.map(fn row -> merge_with_bus_line(row, bus_line_provider) end)
  end

  defp merge_with_bus_line(row, bus_line_provider) do
    bus_coordinates = decode_bus_coordinates(row)

    case bus_line_provider.get(bus_coordinates.codigo_linha) do
      {:error, reason} ->
        Logger.info("Failed to find bus line for #{row["NL"]}, reason: #{reason}")
        bus_coordinates

      {:ok, nil} ->
        Logger.info("Found nil cache for line #{row["NL"]}")
        bus_coordinates

      {:ok, bus_line} ->
        %{bus_coordinates | numero_linha: bus_line["Linha"], nome_linha: bus_line["Nome"]}
    end
  end

  defp decode_bus_coordinates(row) do
    geoPoint = %Geo.Point{coordinates: {parse_decimal(row["LT"]), parse_decimal(row["LG"])}}

    %Crawler.BusCoordinates{
      codigo_linha: row["NL"],
      codigo_evento: row["EV"],
      codigo_do_veiculo: row["NV"],
      timestamp: parse_date(row["HR"]),
      coordenadas: Geo.JSON.encode!(geoPoint) |> Poison.encode!(),
      velocidade_instantanea: parse_decimal(row["VL"]),
      distancia_pecorrida: parse_decimal(row["DT"]),
      direcao_do_veiculo: row["DG"],
      sentindo_da_viagem: row["SV"]
    }
  end

  defp parse_date(date_str) do
    case Timex.parse(date_str, "%Y%m%d%H%M%S", :strftime) do
      {:ok, date} ->
        date

      {:error, reason} ->
        Logger.error("Failed to parse date #{date_str}, reason: #{reason}")
        NaiveDateTime.utc_now()
    end
  end

  defp parse_decimal(nil) do
    nil
  end

  defp parse_decimal(value) do
    case value |> String.replace(",", ".") |> Float.parse() do
      {decimal, _} ->
        decimal

      :error ->
        Logger.error("failed to parse decimal #{value}")
        0.0
    end
  end
end