Merge pull request #230 from transitland/master

production release 4.3.0

app/controllers/api/v1/schedule_stop_pairs_controller.rb

@@ -1,3 +1,60 @@
+# == Schema Information
+#
+# Table name: current_schedule_stop_pairs
+#
+#  id                                 :integer          not null, primary key
+#  origin_id                          :integer
+#  destination_id                     :integer
+#  route_id                           :integer
+#  trip                               :string
+#  created_or_updated_in_changeset_id :integer
+#  version                            :integer
+#  trip_headsign                      :string
+#  origin_arrival_time                :string
+#  origin_departure_time              :string
+#  destination_arrival_time           :string
+#  destination_departure_time         :string
+#  frequency_start_time               :string
+#  frequency_end_time                 :string
+#  frequency_headway_seconds          :string
+#  tags                               :hstore
+#  service_start_date                 :date
+#  service_end_date                   :date
+#  service_added_dates                :date             default([]), is an Array
+#  service_except_dates               :date             default([]), is an Array
+#  service_days_of_week               :boolean          default([]), is an Array
+#  created_at                         :datetime         not null
+#  updated_at                         :datetime         not null
+#  block_id                           :string
+#  trip_short_name                    :string
+#  shape_dist_traveled                :float
+#  origin_timezone                    :string
+#  destination_timezone               :string
+#  window_start                       :string
+#  window_end                         :string
+#  origin_timepoint_source            :string
+#  destination_timepoint_source       :string
+#  operator_id                        :integer
+#  wheelchair_accessible              :boolean
+#  bikes_allowed                      :boolean
+#  pickup_type                        :string
+#  drop_off_type                      :string
+#  active                             :boolean
+#
+# Indexes
+#
+#  c_ssp_cu_in_changeset                                       (created_or_updated_in_changeset_id)
+#  c_ssp_destination                                           (destination_id)
+#  c_ssp_origin                                                (origin_id)
+#  c_ssp_route                                                 (route_id)
+#  c_ssp_service_end_date                                      (service_end_date)
+#  c_ssp_service_start_date                                    (service_start_date)
+#  c_ssp_trip                                                  (trip)
+#  index_current_schedule_stop_pairs_on_operator_id            (operator_id)
+#  index_current_schedule_stop_pairs_on_origin_departure_time  (origin_departure_time)
+#  index_current_schedule_stop_pairs_on_updated_at             (updated_at)
+#
+
 class Api::V1::ScheduleStopPairsController < Api::V1::BaseApiController
   include Geojson
   include JsonCollectionPagination
@@ -37,7 +94,7 @@ def index
   private
 
   def set_schedule_stop_pairs
-    @ssps = ScheduleStopPair.where('')
+    @ssps = ScheduleStopPair.where(active: true)
     # Service on a date
     if params[:date].present?
       @ssps = @ssps.where_service_on_date(params[:date])

app/models/change_payload.rb

@@ -42,21 +42,22 @@ class ChangePayload < ActiveRecord::Base
   })
 
   def apply!
+    cache = {}
+    entity_types = {
+      feed: Feed,
+      stop: Stop,
+      operator: Operator,
+      route: Route,
+      schedule_stop_pair: ScheduleStopPair
+    }
     (payload_as_ruby_hash[:changes] || []).each do |change|
-      if change[:feed].present?
-        Feed.apply_change(changeset: changeset, attrs: change[:feed], action: change[:action])
-      end
-      if change[:stop].present?
-        Stop.apply_change(changeset: changeset, attrs: change[:stop], action: change[:action])
-      end
-      if change[:operator].present?
-        Operator.apply_change(changeset: changeset, attrs: change[:operator], action: change[:action])
-      end
-      if change[:route].present?
-        Route.apply_change(changeset: changeset, attrs: change[:route], action: change[:action])
-      end
-      if change[:schedule_stop_pair].present?
-        ScheduleStopPair.apply_change(changeset: changeset, attrs: change[:schedule_stop_pair], action: change[:action])
+      (entity_types.keys & change.keys).each do |entity_type|
+        entity_types[entity_type].apply_change(
+          changeset: changeset,
+          attrs: change[entity_type],
+          action: change[:action],
+          cache: cache
+        )
       end
     end
   end

app/models/concerns/current_tracked_by_changeset.rb

@@ -13,11 +13,11 @@ module ClassMethods
     attr_reader :kind_of_model_tracked,
                 :virtual_attributes
 
-    def apply_change(changeset: nil, attrs: {}, action: nil)
+    def apply_change(changeset: nil, attrs: {}, action: nil, cache: {})
       existing_model = find_existing_model(attrs)
       case action
       when 'createUpdate'
-        attrs_to_apply = attrs.select { |key, value| self.changeable_attributes.include?(key) }
+        attrs_to_apply = apply_params(attrs, cache)
         if existing_model
           existing_model.update_making_history(changeset: changeset, new_attrs: attrs_to_apply)
         else
@@ -34,6 +34,11 @@ def apply_change(changeset: nil, attrs: {}, action: nil)
       end
     end
 
+    def apply_params(params, cache={})
+      # Filter changeset params
+      params.select { |key, value| self.changeable_attributes.include?(key) }
+    end
+
     def before_create_making_history(instantiated_model, changeset)
       # this is available for overriding in models
       super(instantiated_model, changeset) if defined?(super)

app/models/concerns/is_an_entity_imported_from_feeds.rb

@@ -8,14 +8,11 @@ module IsAnEntityImportedFromFeeds
   end
 
   def imported_from_feed=(imported_from_feed_params)
-    imported_from_feed_params = HashHelpers::update_keys(imported_from_feed_params, :underscore)
-    eiff_params = {}
-    feed = Feed.find_by!(onestop_id: imported_from_feed_params[:onestop_id])
-    eiff_params[:feed] = feed
-    if imported_from_feed_params[:sha1].present?
-      feed_version = feed.feed_versions.find_by!(sha1: imported_from_feed_params[:sha1])
-      eiff_params[:feed_version] = feed_version
+    params = HashHelpers::update_keys(imported_from_feed_params, :underscore)
+    feed = params[:feed] || Feed.find_by!(onestop_id: params[:onestop_id])
+    if params[:sha1].present?
+      feed_version = params[:feed_version] || feed.feed_versions.find_by!(sha1: params[:sha1])
     end
-    self.entities_imported_from_feed.find_or_initialize_by(eiff_params)
+    self.entities_imported_from_feed.find_or_initialize_by(feed: feed, feed_version: feed_version)
   end
 end

app/models/feed.rb

@@ -21,10 +21,13 @@
 #  geometry                           :geography({:srid geometry, 4326
 #  latest_fetch_exception_log         :text
 #  license_attribution_text           :text
+#  active_feed_version_id             :integer
 #
 # Indexes
 #
+#  index_current_feeds_on_active_feed_version_id              (active_feed_version_id)
 #  index_current_feeds_on_created_or_updated_in_changeset_id  (created_or_updated_in_changeset_id)
+#  index_current_feeds_on_geometry                            (geometry)
 #
 
 class BaseFeed < ActiveRecord::Base
@@ -69,6 +72,7 @@ def to_s
 
   has_many :feed_versions, -> { order 'created_at DESC' }, dependent: :destroy, as: :feed
   has_many :feed_version_imports, -> { order 'created_at DESC' }, through: :feed_versions
+  belongs_to :active_feed_version, class_name: 'FeedVersion'
 
   has_many :operators_in_feed
   has_many :operators, through: :operators_in_feed
@@ -187,6 +191,19 @@ def fetch_and_return_feed_version
     end
   end
 
+  def activate_feed_version(feed_version_sha1)
+    self.transaction do
+      feed_version = self.feed_versions.find_by!(sha1: feed_version_sha1)
+      raise Exception.new('Cannot activate already active feed') if feed_version == self.active_feed_version
+      feed_version.activate_schedule_stop_pairs!
+      self.active_feed_version.delete_schedule_stop_pairs! if self.active_feed_version
+      self.update(
+        active_feed_version: feed_version,
+        last_imported_at: feed_version.imported_at
+      )
+    end
+  end
+
   def self.async_fetch_all_feeds
     workers = []
     Feed.find_each do |feed|

app/models/feed_version.rb

@@ -39,6 +39,38 @@ class FeedVersion < ActiveRecord::Base
 
   before_validation :compute_and_set_hashes, :read_gtfs_calendar_dates, :read_gtfs_feed_info
 
+  def succeeded(timestamp)
+    self.update(imported_at: timestamp)
+    self.feed.activate_feed_version(self.sha1)
+  end
+
+  def failed
+    self.delete_schedule_stop_pairs!
+  end
+
+  def activate_schedule_stop_pairs!
+    self.imported_schedule_stop_pairs.update_all(active: true)
+  end
+
+  def deactivate_schedule_stop_pairs!
+    self.imported_schedule_stop_pairs.update_all(active: false)
+  end
+
+  def delete_schedule_stop_pairs!
+    # A call to "imported_schedule_stop_pairs.delete_all" deletes the records
+    # in the EIFF join table, not the SSPs. So, follow through join to delete.
+    # http://api.rubyonrails.org/classes/ActiveRecord/Associations/ClassMethods.html
+    # PostgreSQL supports joins in delete with "USING".
+    # http://www.postgresql.org/docs/9.0/static/sql-delete.html
+    ScheduleStopPair
+      .joins(:entities_imported_from_feed)
+      .where(entities_imported_from_feed: {feed_version: self, feed: self.feed, entity_type: 'ScheduleStopPair'})
+      .delete_all
+    EntityImportedFromFeed
+      .where(feed_version: self, feed: self.feed, entity_type: 'ScheduleStopPair')
+      .delete_all
+  end
+
   private
 
   def compute_and_set_hashes

app/models/feed_version_import.rb

@@ -31,11 +31,11 @@ def failed(exception_log)
       success: false,
       exception_log: exception_log
     )
+    self.feed_version.failed
   end
 
   def succeeded
     self.update(success: true)
-    self.feed_version.update(imported_at: self.updated_at)
-    self.feed.update(last_imported_at: self.updated_at)
+    self.feed_version.succeeded(self.updated_at)
   end
 end

app/models/operator.rb

@@ -22,6 +22,7 @@
 # Indexes
 #
 #  #c_operators_cu_in_changeset_id_index   (created_or_updated_in_changeset_id)
+#  index_current_operators_on_geometry     (geometry)
 #  index_current_operators_on_identifiers  (identifiers)
 #  index_current_operators_on_onestop_id   (onestop_id) UNIQUE
 #  index_current_operators_on_tags         (tags)

app/models/route.rb

@@ -17,6 +17,7 @@
 # Indexes
 #
 #  c_route_cu_in_changeset              (created_or_updated_in_changeset_id)
+#  index_current_routes_on_geometry     (geometry)
 #  index_current_routes_on_identifiers  (identifiers)
 #  index_current_routes_on_operator_id  (operator_id)
 #  index_current_routes_on_tags         (tags)

app/models/schedule_stop_pair.rb

@@ -39,6 +39,7 @@
 #  bikes_allowed                      :boolean
 #  pickup_type                        :string
 #  drop_off_type                      :string
+#  active                             :boolean
 #
 # Indexes
 #
@@ -49,6 +50,7 @@
 #  c_ssp_service_end_date                                      (service_end_date)
 #  c_ssp_service_start_date                                    (service_start_date)
 #  c_ssp_trip                                                  (trip)
+#  index_current_schedule_stop_pairs_on_active                 (active)
 #  index_current_schedule_stop_pairs_on_operator_id            (operator_id)
 #  index_current_schedule_stop_pairs_on_origin_departure_time  (origin_departure_time)
 #  index_current_schedule_stop_pairs_on_updated_at             (updated_at)
@@ -194,6 +196,27 @@ def self.find_by_attributes(attrs = {})
       find(attrs[:id])
     end
   end
+  def self.apply_params(params, cache={})
+    params = super(params, cache)
+    {
+      origin_onestop_id: :origin,
+      destination_onestop_id: :destination,
+      route_onestop_id: :route
+    }.each do |k,v|
+      cache[params[k]] ||= OnestopId.find!(params[k])
+      params[v] = cache[params.delete(k)]
+    end
+    if params[:imported_from_feed]
+      feed_onestop_id = params[:imported_from_feed][:onestop_id]
+      feed_version_id = params[:imported_from_feed][:sha1]
+      cache[feed_onestop_id] ||= OnestopId.find!(feed_onestop_id)
+      cache[feed_version_id] ||= cache[feed_onestop_id].feed_versions.find_by!(sha1: feed_version_id)
+      params[:imported_from_feed][:feed] = cache[feed_onestop_id]
+      params[:imported_from_feed][:feed_version] = cache[feed_version_id]
+    end
+    params[:operator] = params[:route].operator
+    params
+  end
 
   # Interpolate
   def self.interpolate(ssps, method=:linear)

app/models/stop.rb

@@ -13,10 +13,12 @@
 #  version                            :integer
 #  identifiers                        :string           default([]), is an Array
 #  timezone                           :string
+#  last_conflated_at                  :datetime
 #
 # Indexes
 #
 #  #c_stops_cu_in_changeset_id_index   (created_or_updated_in_changeset_id)
+#  index_current_stops_on_geometry     (geometry)
 #  index_current_stops_on_identifiers  (identifiers)
 #  index_current_stops_on_onestop_id   (onestop_id)
 #  index_current_stops_on_tags         (tags)
@@ -167,6 +169,11 @@ def before_destroy_making_history(changeset, old_model)
     joins{operators_serving_stop.operator}.where{operators_serving_stop.operator_id == operator.id}
   }
 
+  # Last conflated before
+  scope :last_conflated_before, -> (last_conflated_at) {
+    where('last_conflated_at <= ?', last_conflated_at)
+  }
+
   # Similarity search
   def self.find_by_similarity(point, name, radius=100, threshold=0.75)
     # Similarity search. Returns a score,stop tuple or nil.
@@ -208,6 +215,16 @@ def queue_conflate_with_osm
     end
   end
 
+  def self.re_conflate_with_osm(last_conflated_at=nil)
+      if last_conflated_at.nil?
+        max_hours = Float(Figaro.env.max_hours_since_last_conflate.presence || 84)
+        last_conflated_at = max_hours.hours.ago
+      end
+      Stop.last_conflated_before(last_conflated_at).ids.each_slice(1000) do |slice|
+        ConflateStopsWithOsmWorker.perform_async(slice)
+      end
+  end
+
   def self.conflate_with_osm(stops)
     stops.in_groups_of(TyrService::MAX_LOCATIONS_PER_REQUEST, false).each do |group|
       Stop.transaction do
@@ -218,11 +235,13 @@ def self.conflate_with_osm(stops)
           }
         end
         tyr_locate_response = TyrService.locate(locations: locations)
+        now = DateTime.now
         group.each_with_index do |stop, index|
           way_id = tyr_locate_response[index][:edges][0][:way_id]
           stop_tags = stop.tags.try(:clone) || {}
           stop_tags[:osm_way_id] = way_id
           stop.update(tags: stop_tags)
+          stop.update(last_conflated_at: now)
         end
       end
     end

app/serializers/feed_serializer.rb

@@ -21,10 +21,13 @@
 #  geometry                           :geography({:srid geometry, 4326
 #  latest_fetch_exception_log         :text
 #  license_attribution_text           :text
+#  active_feed_version_id             :integer
 #
 # Indexes
 #
+#  index_current_feeds_on_active_feed_version_id              (active_feed_version_id)
 #  index_current_feeds_on_created_or_updated_in_changeset_id  (created_or_updated_in_changeset_id)
+#  index_current_feeds_on_geometry                            (geometry)
 #
 
 class FeedSerializer < ApplicationSerializer

app/serializers/operator_serializer.rb

@@ -22,6 +22,7 @@
 # Indexes
 #
 #  #c_operators_cu_in_changeset_id_index   (created_or_updated_in_changeset_id)
+#  index_current_operators_on_geometry     (geometry)
 #  index_current_operators_on_identifiers  (identifiers)
 #  index_current_operators_on_onestop_id   (onestop_id) UNIQUE
 #  index_current_operators_on_tags         (tags)

app/serializers/route_serializer.rb

@@ -17,6 +17,7 @@
 # Indexes
 #
 #  c_route_cu_in_changeset              (created_or_updated_in_changeset_id)
+#  index_current_routes_on_geometry     (geometry)
 #  index_current_routes_on_identifiers  (identifiers)
 #  index_current_routes_on_operator_id  (operator_id)
 #  index_current_routes_on_tags         (tags)