neiam
/apollos_crib
It's Apollo's Crib
public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages
⭐
Log in to mark this repository.
GTFS importer rewrite w/ fixed broadcast and completion handlers, store last update time, display last update time, fix gbfs choking on addtl files, make gtfs_route::route_desc varchar(1024)
8893dd9 · gmorell · 2023-07-28 03:43
Files changed
modified
apps/room_gbfs/lib/room_gbfs/gbfs_worker.ex
+3
−0
@@ -255,6 +255,9 @@ defmodule RoomGbfs.Worker do
| 255 | 255 | "system_alerts" -> |
| 256 | 256 | write_data(url, :sys_alerts, state.id) |
| 257 | 257 | bcast(state.id, :system_information, 10, 10) |
| 258 | + | |
| 259 | + otherwise -> | |
| 260 | + Logger.info("GBFS::#{state.id} System has unhandled file #{otherwise}") | |
| 258 | 261 | end |
| 259 | 262 | end) |
| 260 | 263 |
modified
apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+308
−20
@@ -9,8 +9,8 @@ defmodule RoomGtfs.Worker do
| 9 | 9 | alias RoomSanctum.Repo |
| 10 | 10 | |
| 11 | 11 | @registry :zeus |
| 12 | − # 1 week | |
| 13 | − @default_refresh_seconds 604_800 | |
| 12 | + # 4 weeks | |
| 13 | + @default_refresh_seconds 604_800 * 4 | |
| 14 | 14 | |
| 15 | 15 | def start_link(opts) do |
| 16 | 16 | Parent.GenServer.start_link(__MODULE__, opts, name: via_tuple("gtfs" <> opts[:name])) |
@@ -35,6 +35,14 @@ defmodule RoomGtfs.Worker do
| 35 | 35 | |> GenServer.cast(:update_static) |
| 36 | 36 | end |
| 37 | 37 | |
| 38 | + def update_static_data(name, :str) do | |
| 39 | + IO.puts("qqq") | |
| 40 | + | |
| 41 | + "gtfs#{name}" | |
| 42 | + |> via_tuple() | |
| 43 | + |> GenServer.cast(:update_static_str) | |
| 44 | + end | |
| 45 | + | |
| 38 | 46 | def update_realtime_data(name) do |
| 39 | 47 | "gtfs#{name}" |
| 40 | 48 | |> via_tuple() |
@@ -49,7 +57,7 @@ defmodule RoomGtfs.Worker do
| 49 | 57 | |
| 50 | 58 | def query_stop(id, query) do |
| 51 | 59 | inst = Configuration.get_source!(id) |
| 52 | − res = Storage.get_upcoming_arrivals_for_stop(id, query.stop) | |
| 60 | + res = Storage.get_upcoming_arrivals_for_stop(id, query.stop) |> Storage.fix_arrival_times | |
| 53 | 61 | |
| 54 | 62 | case inst.config.url_rt_tu do |
| 55 | 63 | nil -> |
@@ -345,23 +353,25 @@ defmodule RoomGtfs.Worker.RT do
| 345 | 353 | |
| 346 | 354 | # filter out the protobuf for all relevant trips and then the relevant stop on that trip, nice and small |
| 347 | 355 | case state.rt_tu do |
| 348 | − nil -> {:reply, [], state} | |
| 349 | − _otherwise -> relevant_trips = | |
| 350 | − state.rt_tu.entity | |
| 351 | − |> Enum.filter(fn x -> Enum.member?(trips, x.trip_update.trip.trip_id) end) | |
| 352 | − |> Enum.map(fn x -> | |
| 353 | − x | |
| 354 | − |> Kernel.put_in( | |
| 355 | − [Access.key(:trip_update, %{}), Access.key(:stop_time_update, %{})], | |
| 356 | − x.trip_update.stop_time_update | |
| 357 | − |> Enum.filter(fn x -> x.stop_id == stop end) | |
| 358 | − |> List.first() | |
| 359 | − ) | |
| 360 | − end) | |
| 361 | − | |
| 362 | − {:reply, relevant_trips, state} | |
| 356 | + nil -> | |
| 357 | + {:reply, [], state} | |
| 358 | + | |
| 359 | + _otherwise -> | |
| 360 | + relevant_trips = | |
| 361 | + state.rt_tu.entity | |
| 362 | + |> Enum.filter(fn x -> Enum.member?(trips, x.trip_update.trip.trip_id) end) | |
| 363 | + |> Enum.map(fn x -> | |
| 364 | + x | |
| 365 | + |> Kernel.put_in( | |
| 366 | + [Access.key(:trip_update, %{}), Access.key(:stop_time_update, %{})], | |
| 367 | + x.trip_update.stop_time_update | |
| 368 | + |> Enum.filter(fn x -> x.stop_id == stop end) | |
| 369 | + |> List.first() | |
| 370 | + ) | |
| 371 | + end) | |
| 372 | + | |
| 373 | + {:reply, relevant_trips, state} | |
| 363 | 374 | end |
| 364 | − | |
| 365 | 375 | end |
| 366 | 376 | end |
| 367 | 377 |
@@ -373,6 +383,7 @@ defmodule RoomGtfs.Worker.Static do
| 373 | 383 | alias RoomSanctum.Configuration |
| 374 | 384 | alias RoomSanctum.Storage |
| 375 | 385 | alias RoomSanctum.Repo |
| 386 | + alias RoomSanctum.Storage.GTFS | |
| 376 | 387 | |
| 377 | 388 | def start_link(opts) do |
| 378 | 389 | GenServer.start_link(__MODULE__, opts, name: via_tuple("gtfs-st" <> opts[:name])) |
@@ -386,6 +397,10 @@ defmodule RoomGtfs.Worker.Static do
| 386 | 397 | Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, id, :disabled}) |
| 387 | 398 | end |
| 388 | 399 | |
| 400 | + defp bcast(id, :done) do | |
| 401 | + Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, id, :done}) | |
| 402 | + end | |
| 403 | + | |
| 389 | 404 | def init(opts) do |
| 390 | 405 | {:ok, |
| 391 | 406 | %{ |
@@ -395,6 +410,185 @@ defmodule RoomGtfs.Worker.Static do
| 395 | 410 | |
| 396 | 411 | defp via_tuple(name), do: {:via, Registry, {@registry, name}} |
| 397 | 412 | |
| 413 | + defp get_cols(schema) do | |
| 414 | + schema.__schema__(:fields) | |
| 415 | + |> Enum.map(&Atom.to_string/1) | |
| 416 | + |> List.delete("id") | |
| 417 | + # |> List.delete("updated_at") | |
| 418 | + # |> List.delete("inserted_at") | |
| 419 | + |> Enum.join(", ") | |
| 420 | + end | |
| 421 | + | |
| 422 | + defp get_cols(schema, cols) do | |
| 423 | + schema.__schema__(:fields) | |
| 424 | + |> Enum.map(&Atom.to_string/1) | |
| 425 | + |> Enum.filter(fn f -> Enum.member?(cols, f) end) | |
| 426 | + |> List.delete("id") | |
| 427 | + |> Enum.join(", ") | |
| 428 | + end | |
| 429 | + | |
| 430 | + defp as_pg(type) do | |
| 431 | + case type do | |
| 432 | + :id -> "bigint" | |
| 433 | + :string -> "varchar" | |
| 434 | + :naive_datetime -> "timestamp" | |
| 435 | + :time -> "time" | |
| 436 | + :integer -> "integer" | |
| 437 | + :float -> "double precision" | |
| 438 | + :date -> "date" | |
| 439 | + EctoInterval -> "interval" | |
| 440 | + end | |
| 441 | + end | |
| 442 | + | |
| 443 | + def get_cols_pgtypes(schema) do | |
| 444 | + schema.__schema__(:fields) | |
| 445 | + |> List.delete(:id) | |
| 446 | + |> Enum.map(fn f -> {f, schema.__schema__(:type, f) |> as_pg} end) | |
| 447 | + |> Enum.map(fn {k,v} -> "#{k}::#{v}" end) | |
| 448 | + |> Enum.join(", ") | |
| 449 | + end | |
| 450 | + | |
| 451 | + def get_cols_pgtypes(schema, cols) do | |
| 452 | + schema.__schema__(:fields) | |
| 453 | + |> List.delete(:id) | |
| 454 | + |> Enum.filter(fn f -> Enum.member?(cols |> Enum.map(&String.to_atom/1), f) end) | |
| 455 | + |> Enum.map(fn f -> {f, schema.__schema__(:type, f) |> as_pg} end) | |
| 456 | + |> Enum.map(fn {k,v} -> "#{k}::#{v}" end) | |
| 457 | + |> Enum.join(", ") | |
| 458 | + end | |
| 459 | + | |
| 460 | + def csv_cols_to_tmp_cols(cols) do | |
| 461 | + cols | |
| 462 | + |> Enum.map(fn x -> "#{x} varchar" end) | |
| 463 | + |> Enum.join(", ") | |
| 464 | + end | |
| 465 | + | |
| 466 | + def csv_cols_to_tmp_cols(cols, :add) do | |
| 467 | + cols | |
| 468 | + |> Kernel.++(["inserted_at", "updated_at", "source_id"]) | |
| 469 | + |> Enum.map(fn x -> "#{x} varchar" end) | |
| 470 | + |> Enum.join(", ") | |
| 471 | +end | |
| 472 | + | |
| 473 | + defp write_file(contents, type, id, via: :copy) do | |
| 474 | + datetime = NaiveDateTime.local_now() | |
| 475 | + Logger.info("GTFS::#{id} writing #{type} (c)") | |
| 476 | + | |
| 477 | + cols_j = | |
| 478 | + contents | |
| 479 | + |> Stream.chunk_every(500) | |
| 480 | + |> Stream.map(&String.split(&1 |> List.flatten() |> List.first(), "\n")) | |
| 481 | + |> Stream.take(1) | |
| 482 | + |> Enum.to_list() | |
| 483 | + |> List.flatten() | |
| 484 | + |> List.first() | |
| 485 | + |> String.strip | |
| 486 | + |> String.split(",") | |
| 487 | + | |
| 488 | +# cols_j = ["source_id" |cols_j] | |
| 489 | + cols_j_plus = cols_j ++ ["inserted_at", "updated_at", "source_id"] | |
| 490 | + | |
| 491 | +# contents |> Stream.take(500) |> Enum.to_list() |> IO.inspect | |
| 492 | + | |
| 493 | + | |
| 494 | + # add truncation here as necessary | |
| 495 | + case type do | |
| 496 | + :agencies -> RoomSanctum.Storage.truncate_agency(id) | |
| 497 | + :calendars -> RoomSanctum.Storage.truncate_calendar(id) | |
| 498 | + :directions -> RoomSanctum.Storage.truncate_direction(id) | |
| 499 | + :routes -> RoomSanctum.Storage.truncate_route(id) | |
| 500 | + :stops -> RoomSanctum.Storage.truncate_stop(id) | |
| 501 | + :stop_times -> RoomSanctum.Storage.truncate_stop_time(id) | |
| 502 | + :trips -> RoomSanctum.Storage.truncate_trip(id) | |
| 503 | + _ -> :ok | |
| 504 | + end | |
| 505 | + | |
| 506 | + # set our variables based on the type | |
| 507 | + {table, columns, pg_cols} = | |
| 508 | + case type do | |
| 509 | + :agencies -> {:gtfs_agencies, [GTFS.Agency |> get_cols(cols_j_plus)], GTFS.Agency |> get_cols_pgtypes(cols_j_plus)} | |
| 510 | + :calendars -> {:gtfs_calendars, [GTFS.Calendar |> get_cols(cols_j_plus)], GTFS.Calendar |> get_cols_pgtypes(cols_j_plus)} | |
| 511 | + :directions -> {:gtfs_directions, [GTFS.Direction |> get_cols(cols_j_plus)], GTFS.Direction |> get_cols_pgtypes(cols_j_plus)} | |
| 512 | + :routes -> {:gtfs_routes, [GTFS.Route |> get_cols(cols_j_plus)], GTFS.Route |> get_cols_pgtypes(cols_j_plus)} | |
| 513 | + :stops -> {:gtfs_stops, [GTFS.Stop |> get_cols(cols_j_plus)], GTFS.Stop |> get_cols_pgtypes(cols_j_plus)} | |
| 514 | + :stop_times -> {:gtfs_stop_times, [GTFS.StopTime |> get_cols(cols_j_plus)], GTFS.StopTime |> get_cols_pgtypes(cols_j_plus)} | |
| 515 | + :trips -> {:gtfs_trips, [GTFS.Trip |> get_cols(cols_j_plus)], GTFS.Trip |> get_cols_pgtypes(cols_j_plus)} | |
| 516 | + end | |
| 517 | + | |
| 518 | +# IO.inspect({table, columns, pg_cols}) | |
| 519 | + opts = RoomSanctum.Repo.config() | |
| 520 | + {:ok, pid} = Postgrex.start_link(opts) | |
| 521 | + | |
| 522 | + tmp_table_name = "tmp_#{type}_#{id}" | |
| 523 | + | |
| 524 | + Postgrex.transaction( | |
| 525 | + pid, | |
| 526 | + fn conn -> | |
| 527 | + # temp table | |
| 528 | + qt = | |
| 529 | + Postgrex.prepare!( | |
| 530 | + conn, | |
| 531 | + "", | |
| 532 | + "CREATE TEMPORARY TABLE #{tmp_table_name} (#{cols_j |> csv_cols_to_tmp_cols})" | |
| 533 | + ) | |
| 534 | + | |
| 535 | + Postgrex.execute(conn, qt, []) | |
| 536 | + | |
| 537 | + qt2 = | |
| 538 | + Postgrex.prepare!( | |
| 539 | + conn, | |
| 540 | + "", | |
| 541 | + "CREATE TEMPORARY TABLE #{tmp_table_name}_allcols (#{cols_j |> csv_cols_to_tmp_cols(:add)})" | |
| 542 | + ) | |
| 543 | + | |
| 544 | + Postgrex.execute(conn, qt2, []) | |
| 545 | + | |
| 546 | + # write csv | |
| 547 | + stream = | |
| 548 | + Postgrex.stream( | |
| 549 | + conn, | |
| 550 | + "COPY #{tmp_table_name}(#{cols_j |> Enum.join(",")}) FROM STDIN CSV HEADER DELIMITER ','", | |
| 551 | + [] | |
| 552 | + ) | |
| 553 | + | |
| 554 | + Enum.into(contents, stream) | |
| 555 | + | |
| 556 | + qtc = | |
| 557 | + Postgrex.prepare!( | |
| 558 | + conn, | |
| 559 | + "", | |
| 560 | + "INSERT INTO #{tmp_table_name}_allcols (#{cols_j |> Enum.join(",")}) SELECT * FROM #{tmp_table_name}" | |
| 561 | + ) | |
| 562 | + | |
| 563 | + Postgrex.execute(conn, qtc, []) | |
| 564 | + | |
| 565 | + # update fields | |
| 566 | + qtu = | |
| 567 | + Postgrex.prepare!( | |
| 568 | + conn, | |
| 569 | + "", | |
| 570 | + "UPDATE #{tmp_table_name}_allcols SET inserted_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP, source_id = #{id}" | |
| 571 | + ) | |
| 572 | + | |
| 573 | + Postgrex.execute(conn, qtu, []) | |
| 574 | + | |
| 575 | + # write into dest table | |
| 576 | + qs = | |
| 577 | + Postgrex.prepare!( | |
| 578 | + conn, | |
| 579 | + "", | |
| 580 | + "INSERT INTO #{atom_to_table(type)} (#{columns}) SELECT #{pg_cols} FROM #{tmp_table_name}_allcols" | |
| 581 | + ) | |
| 582 | + | |
| 583 | + Postgrex.execute(conn, qs, []) | |
| 584 | + | |
| 585 | +# qd = Postgrex.prepare!(conn, "", "DROP TABLE #{tmp_table_name}") | |
| 586 | +# Postgrex.execute(conn, qd, []) | |
| 587 | + end, | |
| 588 | + timeout: :infinity | |
| 589 | + ) |> IO.inspect | |
| 590 | + end | |
| 591 | + | |
| 398 | 592 | defp write_file(contents, type, id) do |
| 399 | 593 | datetime = NaiveDateTime.local_now() |
| 400 | 594 | Logger.info("GTFS::#{id} writing #{type}") |
@@ -510,8 +704,102 @@ defmodule RoomGtfs.Worker.Static do
| 510 | 704 | DateTime.utc_now() |
| 511 | 705 | end |
| 512 | 706 | |
| 707 | + defp file_to_atom(filename) do | |
| 708 | + case filename do | |
| 709 | + "agency.txt" -> | |
| 710 | + :agencies | |
| 711 | + | |
| 712 | + "calendar.txt" -> | |
| 713 | + :calendars | |
| 714 | + | |
| 715 | + "directions.txt" -> | |
| 716 | + :directions | |
| 717 | + | |
| 718 | + "routes.txt" -> | |
| 719 | + :routes | |
| 720 | + | |
| 721 | + "stops.txt" -> | |
| 722 | + :stops | |
| 723 | + | |
| 724 | + "stop_times.txt" -> | |
| 725 | + :stop_times | |
| 726 | + | |
| 727 | + "trips.txt" -> | |
| 728 | + :trips | |
| 729 | + end | |
| 730 | + end | |
| 731 | + | |
| 732 | + def file_to_order(filename) do | |
| 733 | + case filename do | |
| 734 | + "agency.txt" -> 3 | |
| 735 | + "calendar.txt" -> 4 | |
| 736 | + "directions.txt" -> 5 | |
| 737 | + "routes.txt" -> 6 | |
| 738 | + "stops.txt" -> 7 | |
| 739 | + "stop_times.txt" -> 8 | |
| 740 | + "trips.txt" -> 9 | |
| 741 | + end | |
| 742 | + end | |
| 743 | + | |
| 744 | + defp atom_to_table(atom) do | |
| 745 | + "gtfs_#{atom}" | |
| 746 | + end | |
| 747 | + | |
| 748 | + @impl true | |
| 513 | 749 | def handle_cast(:update_static, state) do |
| 514 | 750 | cfg = Configuration.get_source!(state.id) |
| 751 | + IO.puts("kkk") | |
| 752 | + | |
| 753 | + case cfg.enabled do | |
| 754 | + true -> | |
| 755 | + Logger.info("GTFS::#{state.id} updating static info") | |
| 756 | + bcast(state.id, :downloading, 1, 9) | |
| 757 | + | |
| 758 | + case HTTPoison.get(cfg.config.url) do | |
| 759 | + {:ok, result} -> | |
| 760 | + bcast(state.id, :extracting, 2, 9) | |
| 761 | + | |
| 762 | + case result.body |> Unzip.InMem.new() |> Unzip.new() do | |
| 763 | + {:ok, unzip} -> | |
| 764 | + files = Unzip.list_entries(unzip) | |
| 765 | + | |
| 766 | + files | |
| 767 | + |> Enum.map(fn e -> | |
| 768 | + if Enum.member?( | |
| 769 | + [ | |
| 770 | + "agency.txt", | |
| 771 | + "calendar.txt", | |
| 772 | + "directions.txt", | |
| 773 | + "routes.txt", | |
| 774 | + "stops.txt", | |
| 775 | + "stop_times.txt", | |
| 776 | + "trips.txt", | |
| 777 | + ], | |
| 778 | + e.file_name | |
| 779 | + ) do | |
| 780 | + | |
| 781 | + bcast(state.id, file_to_atom(e.file_name), file_to_order(e.file_name), 9) | |
| 782 | + | |
| 783 | + Unzip.file_stream!(unzip, e.file_name) | |
| 784 | + |> write_file(file_to_atom(e.file_name), state.id, via: :copy) | |
| 785 | + | |
| 786 | + # bcast(state.id, :stop_times, 8, 9) | |
| 787 | + end | |
| 788 | + end) | |
| 789 | + Configuration.update_source(cfg, %{meta: %{last_run: DateTime.utc_now()}}) | |
| 790 | + bcast(state.id, :done) | |
| 791 | + {:error, term} -> | |
| 792 | + Logger.error(term) | |
| 793 | + end | |
| 794 | + end | |
| 795 | + end | |
| 796 | + | |
| 797 | + {:noreply, state} | |
| 798 | + end | |
| 799 | + | |
| 800 | + @impl true | |
| 801 | + def handle_cast(:update_static_old, state) do | |
| 802 | + cfg = Configuration.get_source!(state.id) | |
| 515 | 803 | |
| 516 | 804 | case cfg.enabled do |
| 517 | 805 | true -> |
@@ -555,7 +843,7 @@ defmodule RoomGtfs.Worker.Static do
| 555 | 843 | bcast(state.id, :stops, 7, 9) |
| 556 | 844 | |
| 557 | 845 | 'stop_times.txt' -> |
| 558 | − write_file(as_csv, :stop_times, state.id) | |
| 846 | + write_file(data, :stop_times, state.id, via: :copy) | |
| 559 | 847 | bcast(state.id, :stop_times, 8, 9) |
| 560 | 848 | |
| 561 | 849 | 'trips.txt' -> |
added
apps/room_gtfs/lib/room_gtfs/unzip.ex
+18
−0
@@ -0,0 +1,18 @@
| 1 | +defmodule Unzip.InMem do | |
| 2 | + defstruct [:blob] | |
| 3 | + alias __MODULE__ | |
| 4 | + | |
| 5 | + def new(blob) do | |
| 6 | + %Unzip.InMem{blob: blob} | |
| 7 | + end | |
| 8 | +end | |
| 9 | + | |
| 10 | +defimpl Unzip.FileAccess, for: Unzip.InMem do | |
| 11 | + def size(file) do | |
| 12 | + {:ok, file.blob |> byte_size} | |
| 13 | + end | |
| 14 | + | |
| 15 | + def pread(file, offset, length) do | |
| 16 | + {:ok, file.blob |> binary_part(offset, length)} | |
| 17 | + end | |
| 18 | +end |
modified
apps/room_gtfs/room_gtfs.iml
+13
−0
@@ -44,5 +44,18 @@
| 44 | 44 | <orderEntry type="library" name="protobuf" level="project" /> |
| 45 | 45 | <orderEntry type="library" name="csv" level="project" /> |
| 46 | 46 | <orderEntry type="library" name="protobuf" level="project" /> |
| 47 | + <orderEntry type="library" name="unzip" level="project" /> | |
| 48 | + <orderEntry type="library" name="eflame" level="project" /> | |
| 49 | + <orderEntry type="library" name="nimble_csv" level="project" /> | |
| 50 | + <orderEntry type="library" name="ecto_interval" level="project" /> | |
| 51 | + <orderEntry type="library" name="postgrex" level="project" /> | |
| 52 | + <orderEntry type="library" name="table" level="project" /> | |
| 53 | + <orderEntry type="library" name="decimal" level="project" /> | |
| 54 | + <orderEntry type="library" name="db_connection" level="project" /> | |
| 55 | + <orderEntry type="library" name="ecto" level="project" /> | |
| 56 | + <orderEntry type="library" name="phoenix_html" level="project" /> | |
| 57 | + <orderEntry type="library" name="plug" level="project" /> | |
| 58 | + <orderEntry type="library" name="mime" level="project" /> | |
| 59 | + <orderEntry type="library" name="plug_crypto" level="project" /> | |
| 47 | 60 | </component> |
| 48 | 61 | </module> |
modified
apps/room_sanctum/lib/room_sanctum/storage.ex
+63
−25
@@ -635,18 +635,20 @@ defmodule RoomSanctum.Storage do
| 635 | 635 | |
| 636 | 636 | def list_stops(source_id, search_term) do |
| 637 | 637 | from(p in Stop, |
| 638 | − where: p.source_id == ^source_id and fragment( | |
| 639 | − "searchable @@ websearch_to_tsquery(?)", | |
| 640 | − ^search_term | |
| 641 | − ), | |
| 642 | − order_by: { | |
| 638 | + where: | |
| 639 | + p.source_id == ^source_id and | |
| 640 | + fragment( | |
| 641 | + "searchable @@ websearch_to_tsquery(?)", | |
| 642 | + ^search_term | |
| 643 | + ), | |
| 644 | + order_by: { | |
| 643 | 645 | :desc, |
| 644 | 646 | fragment( |
| 645 | 647 | "ts_rank_cd(searchable, websearch_to_tsquery(?), 4)", |
| 646 | 648 | ^search_term |
| 647 | 649 | ) |
| 648 | 650 | } |
| 649 | − ) | |
| 651 | + ) | |
| 650 | 652 | |> Repo.all() |
| 651 | 653 | end |
| 652 | 654 |
@@ -880,25 +882,59 @@ defmodule RoomSanctum.Storage do
| 880 | 882 | end |
| 881 | 883 | |
| 882 | 884 | defp conditional_where(queryable, source_id, stop_id, timestamp_time_hour, timestamp_time) do |
| 883 | − case timestamp_time < timestamp_time_hour do | |
| 884 | − true -> | |
| 885 | +# case timestamp_time < timestamp_time_hour do | |
| 886 | +# true -> | |
| 885 | 887 | where( |
| 886 | 888 | queryable, |
| 887 | 889 | [st], |
| 888 | − st.source_id == ^source_id and st.arrival_time >= ^timestamp_time and | |
| 889 | − st.arrival_time <= ^timestamp_time_hour and st.stop_id == ^stringify(stop_id) | |
| 890 | + st.source_id == ^source_id and st.arrival_time >= ^time_to_interval(timestamp_time) and | |
| 891 | + st.arrival_time <= ^time_to_interval(timestamp_time_hour) and st.stop_id == ^stringify(stop_id) | |
| 890 | 892 | ) |
| 891 | 893 | |
| 892 | − false -> | |
| 893 | − where( | |
| 894 | − queryable, | |
| 895 | − [st], | |
| 896 | − st.source_id == ^source_id and | |
| 897 | − ((st.arrival_time >= ^timestamp_time and st.arrival_time <= ^Time.new!(23, 59, 50)) or | |
| 898 | − (^Time.new!(0, 0, 0) <= st.arrival_time and st.arrival_time <= ^timestamp_time_hour)) and | |
| 899 | − st.stop_id == ^stringify(stop_id) | |
| 900 | − ) | |
| 894 | +# false -> | |
| 895 | +# where( | |
| 896 | +# queryable, | |
| 897 | +# [st], | |
| 898 | +# st.source_id == ^source_id and | |
| 899 | +# ((st.arrival_time >= ^timestamp_time and st.arrival_time <= ^Time.new!(23, 59, 50)) or | |
| 900 | +# (^Time.new!(0, 0, 0) <= st.arrival_time and st.arrival_time <= ^timestamp_time_hour)) and | |
| 901 | +# st.stop_id == ^stringify(stop_id) | |
| 902 | +# ) | |
| 903 | +# end | |
| 904 | + end | |
| 905 | + | |
| 906 | + defp divrem(num, den) do | |
| 907 | + {div(num, den), rem(num, den)} | |
| 908 | + end | |
| 909 | + | |
| 910 | + def time_to_interval(time) do | |
| 911 | + as_interval = | |
| 912 | + if time.hour < 5 do | |
| 913 | + %{hours: time.hour + 24, minutes: time.minute, seconds: time.second} | |
| 914 | + else | |
| 915 | + %{hours: time.hour, minutes: time.minute, seconds: time.second} | |
| 916 | + end | |
| 917 | + as_seconds = as_interval.hours * 60 * 60 + as_interval.minutes * 60 + as_interval.seconds | |
| 918 | + struct(Postgrex.Interval, %{secs: as_seconds}) | |
| 919 | + end | |
| 920 | + | |
| 921 | + def interval_to_time(interval) do | |
| 922 | + {h, r1} = divrem(interval, (60*60)) | |
| 923 | + {m, s} = divrem(r1, 60) | |
| 924 | + h = if h > 23 do | |
| 925 | + h - 24 | |
| 926 | + else | |
| 927 | + h | |
| 901 | 928 | end |
| 929 | + Time.new!(h,m,s) | |
| 930 | + end | |
| 931 | + | |
| 932 | + def fix_arrival_times(res) do | |
| 933 | + res |> Enum.map(fn map -> map |> Map.put(:arrival_time, map.arrival_time.secs |> interval_to_time) end) | |
| 934 | + end | |
| 935 | + | |
| 936 | + def get_all_arrivals_for_stop(source_id, stop_id) do | |
| 937 | + | |
| 902 | 938 | end |
| 903 | 939 | |
| 904 | 940 | def get_upcoming_arrivals_for_stop(source_id, stop_id, limit \\ 16, timestamp \\ :now) do |
@@ -1199,18 +1235,20 @@ defmodule RoomSanctum.Storage do
| 1199 | 1235 | |
| 1200 | 1236 | def list_gbfs_station_information(source_id, search_term) do |
| 1201 | 1237 | from(p in StationInfo, |
| 1202 | − where: p.source_id == ^source_id and fragment( | |
| 1203 | − "searchable @@ websearch_to_tsquery(?)", | |
| 1204 | − ^search_term | |
| 1205 | − ), | |
| 1206 | − order_by: { | |
| 1238 | + where: | |
| 1239 | + p.source_id == ^source_id and | |
| 1240 | + fragment( | |
| 1241 | + "searchable @@ websearch_to_tsquery(?)", | |
| 1242 | + ^search_term | |
| 1243 | + ), | |
| 1244 | + order_by: { | |
| 1207 | 1245 | :desc, |
| 1208 | 1246 | fragment( |
| 1209 | 1247 | "ts_rank_cd(searchable, websearch_to_tsquery(?), 4)", |
| 1210 | 1248 | ^search_term |
| 1211 | 1249 | ) |
| 1212 | 1250 | } |
| 1213 | − ) | |
| 1251 | + ) | |
| 1214 | 1252 | |> Repo.all() |
| 1215 | 1253 | end |
| 1216 | 1254 |
modified
apps/room_sanctum/lib/room_sanctum/storage/gtfs/stop_time.ex
+3
−3
@@ -4,11 +4,11 @@ defmodule RoomSanctum.Storage.GTFS.StopTime do
| 4 | 4 | |
| 5 | 5 | schema "gtfs_stop_times" do |
| 6 | 6 | belongs_to :source, RoomSanctum.Configuration.Source |
| 7 | − field :arrival_time, :time | |
| 7 | + field :arrival_time, EctoInterval | |
| 8 | 8 | field :checkpoint_id, :string |
| 9 | − field :continuous_dropoff, :string | |
| 9 | + field :continuous_drop_off, :string | |
| 10 | 10 | field :continuous_pickup, :string |
| 11 | − field :departure_time, :time | |
| 11 | + field :departure_time, EctoInterval | |
| 12 | 12 | field :drop_off_type, :integer |
| 13 | 13 | field :pickup_type, :integer |
| 14 | 14 | field :stop_headsign, :string |
modified
apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/show.ex
+14
−0
@@ -138,6 +138,13 @@ defmodule RoomSanctumWeb.SourceLive.Show do
| 138 | 138 | {:noreply, socket |> put_flash(:info, "Updates will not fire while disabled")} |
| 139 | 139 | end |
| 140 | 140 | |
| 141 | + @impl true | |
| 142 | + def handle_info({type, id, :done} = info, socket) do | |
| 143 | + source = Configuration.get_source!(socket.assigns.source_id) | |
| 144 | + {:noreply, socket |> assign(:source, source) |> put_flash(:info, "Completed")} | |
| 145 | + end | |
| 146 | + | |
| 147 | + | |
| 141 | 148 | @impl true |
| 142 | 149 | def handle_info({:gtfs, id, file, complete, total} = info, socket) do |
| 143 | 150 | if socket.assigns.source_id == |
@@ -226,4 +233,11 @@ defmodule RoomSanctumWeb.SourceLive.Show do
| 226 | 233 | defp icon_code(source_type) do |
| 227 | 234 | RoomSanctumWeb.IconHelpers.icon_code(source_type) |
| 228 | 235 | end |
| 236 | + | |
| 237 | + defp is_updated(source_meta) do | |
| 238 | + case source_meta |> Map.get(:last_run) do | |
| 239 | + nil -> "NEVER" | |
| 240 | + val -> val # |> Timex.format!("%Y-%m-%d @ %H:%M", :strftime) | |
| 241 | + end | |
| 242 | + end | |
| 229 | 243 | end |
modified
apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/show.html.heex
+11
−2
@@ -49,9 +49,18 @@
| 49 | 49 | <div class="flex"> |
| 50 | 50 | <div class="card card-compact w-1/4 bg-base-100 shadow-xl m-2"> |
| 51 | 51 | <div class="card-body"> |
| 52 | − <h3 class="card-title">Control</h3> | |
| 53 | − <%= case @source.type do %> | |
| 52 | + <div class="flex justify-between"> | |
| 53 | + <div class="card-title">Control</div> | |
| 54 | + <div class="py-1"> | |
| 55 | + <%= case @source.type do %> | |
| 54 | 56 | <% :gtfs -> %> |
| 57 | + Last Updated <%= @source.meta |> is_updated %> | |
| 58 | + <% _default -> %> | |
| 59 | + <% end %> | |
| 60 | + </div> | |
| 61 | + </div> | |
| 62 | + <%= case @source.type do %> | |
| 63 | + <% :gtfs -> %> | |
| 55 | 64 | <button class="btn btn-secondary" phx-click="do-update" phx-value-type="gtfs" phx-value-id={@source.id}> |
| 56 | 65 | Update |
| 57 | 66 | </button> |
modified
apps/room_sanctum/priv/repo/migrations/20220605151050_create_gtfs_stop_times.exs
+1
−1
@@ -15,7 +15,7 @@ defmodule RoomSanctum.Repo.Migrations.CreateStopTimes do
| 15 | 15 | add :timepoint, :integer |
| 16 | 16 | add :checkpoint_id, :string |
| 17 | 17 | add :continuous_pickup, :string |
| 18 | − add :continuous_dropoff, :string | |
| 18 | + add :continuous_drop_off, :string | |
| 19 | 19 | |
| 20 | 20 | timestamps() |
| 21 | 21 | end |
added
apps/room_sanctum/priv/repo/migrations/20230728031231_route_desc_embiggen.exs
+28
−0
@@ -0,0 +1,28 @@
| 1 | +defmodule RoomSanctum.Repo.Migrations.RouteDescEmbiggen do | |
| 2 | + use Ecto.Migration | |
| 3 | + | |
| 4 | + def change do | |
| 5 | + execute """ | |
| 6 | + ALTER TABLE gtfs_routes | |
| 7 | + DROP COLUMN searchable; | |
| 8 | + """ | |
| 9 | + | |
| 10 | + execute """ | |
| 11 | + alter table gtfs_routes | |
| 12 | + alter column route_desc type varchar(1024) using route_desc::varchar(1024); | |
| 13 | + """ | |
| 14 | + | |
| 15 | + execute """ | |
| 16 | + ALTER TABLE gtfs_routes | |
| 17 | + ADD COLUMN searchable tsvector | |
| 18 | + GENERATED ALWAYS AS ( | |
| 19 | + setweight(to_tsvector('english', coalesce(route_short_name, '')), 'A') || | |
| 20 | + setweight(to_tsvector('english', coalesce(route_long_name, '')), 'A') || | |
| 21 | + setweight(to_tsvector('english', coalesce(route_desc, '')), 'B') || | |
| 22 | + setweight(to_tsvector('english', coalesce(route_type, '')), 'C') || | |
| 23 | + setweight(to_tsvector('english', coalesce(route_url, '')), 'C') || | |
| 24 | + setweight(to_tsvector('english', coalesce(listed_route, '')), 'C') | |
| 25 | + ) STORED; | |
| 26 | + """ | |
| 27 | + end | |
| 28 | +end |
Parents: d56716b