public_transport_agent.rb 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133
  1. require 'date'
  2. require 'cgi'
  3. module Agents
  4. class PublicTransportAgent < Agent
  5. cannot_receive_events!
  6. default_schedule "every_2m"
  7. description <<-MD
  8. The Public Transport Request Agent generates Events based on NextBus GPS transit predictions.
  9. Specify the following user settings:
  10. * agency (string)
  11. * stops (array)
  12. * alert_window_in_minutes (integer)
  13. First, select an agency by visiting [http://www.nextbus.com/predictor/adaAgency.jsp](http://www.nextbus.com/predictor/adaAgency.jsp) and finding your transit system. Once you find it, copy the part of the URL after `?a=`. For example, for the San Francisco MUNI system, you would end up on [http://www.nextbus.com/predictor/adaDirection.jsp?a=**sf-muni**](http://www.nextbus.com/predictor/adaDirection.jsp?a=sf-muni) and copy "sf-muni". Put that into this Agent's agency setting.
  14. Next, find the stop tags that you care about.
  15. Select your destination and lets use the n-judah route. The link should be [http://www.nextbus.com/predictor/adaStop.jsp?a=sf-muni&r=N](http://www.nextbus.com/predictor/adaStop.jsp?a=sf-muni&r=N) Once you find it, copy the part of the URL after `r=`.
  16. The link may not work, but we're just trying to get the part after the r=, so even if it gives an error, continue to the next step.
  17. To find the tags for the sf-muni system, for the N route, visit this URL:
  18. [http://webservices.nextbus.com/service/publicXMLFeed?command=routeConfig&a=sf-muni&r=**N**](http://webservices.nextbus.com/service/publicXMLFeed?command=routeConfig&a=sf-muni&r=N)
  19. The tags are listed as tag="1234". Copy that number and add the route before it, separated by a pipe '&#124;' symbol. Once you have one or more tags from that page, add them to this Agent's stop list. E.g,
  20. agency: "sf-muni"
  21. stops: ["N|5221", "N|5215"]
  22. Remember to pick the appropriate stop, which will have different tags for in-bound and out-bound.
  23. This Agent will generate predictions by requesting a URL similar to the following:
  24. [http://webservices.nextbus.com/service/publicXMLFeed?command=predictionsForMultiStops&a=sf-muni&stops=N&#124;5221&stops=N&#124;5215](http://webservices.nextbus.com/service/publicXMLFeed?command=predictionsForMultiStops&a=sf-muni&stops=N&#124;5221&stops=N&#124;5215)
  25. Finally, set the arrival window that you're interested in. E.g., 5 minutes. Events will be created by the agent anytime a new train or bus comes into that time window.
  26. alert_window_in_minutes: 5
  27. MD
  28. event_description <<-MD
  29. Events look like this:
  30. { "routeTitle":"N-Judah",
  31. "stopTag":"5215",
  32. "prediction":
  33. {"epochTime":"1389622846689",
  34. "seconds":"3454","minutes":"57","isDeparture":"false",
  35. "affectedByLayover":"true","dirTag":"N__OB4KJU","vehicle":"1489",
  36. "block":"9709","tripTag":"5840086"
  37. }
  38. }
  39. MD
  40. def check_url
  41. stop_query = URI.encode(interpolated["stops"].collect{|a| "&stops=#{a}"}.join)
  42. "http://webservices.nextbus.com/service/publicXMLFeed?command=predictionsForMultiStops&a=#{interpolated["agency"]}#{stop_query}"
  43. end
  44. def stops
  45. interpolated["stops"].collect{|a| a.split("|").last}
  46. end
  47. def check
  48. hydra = Typhoeus::Hydra.new
  49. request = Typhoeus::Request.new(check_url, :followlocation => true)
  50. request.on_success do |response|
  51. page = Nokogiri::XML response.body
  52. predictions = page.css("//prediction")
  53. predictions.each do |pr|
  54. parent = pr.parent.parent
  55. vals = {"routeTitle" => parent["routeTitle"], "stopTag" => parent["stopTag"]}
  56. if pr["minutes"] && pr["minutes"].to_i < interpolated["alert_window_in_minutes"].to_i
  57. vals = vals.merge Hash.from_xml(pr.to_xml)
  58. if not_already_in_memory?(vals)
  59. create_event(:payload => vals)
  60. log "creating event..."
  61. update_memory(vals)
  62. else
  63. log "not creating event since already in memory"
  64. end
  65. end
  66. end
  67. end
  68. hydra.queue request
  69. hydra.run
  70. end
  71. def update_memory(vals)
  72. add_to_memory(vals)
  73. cleanup_old_memory
  74. end
  75. def cleanup_old_memory
  76. self.memory["existing_routes"] ||= []
  77. self.memory["existing_routes"].reject!{|h| h["currentTime"].to_time <= (Time.now - 2.hours)}
  78. end
  79. def add_to_memory(vals)
  80. self.memory["existing_routes"] ||= []
  81. self.memory["existing_routes"] << {"stopTag" => vals["stopTag"], "tripTag" => vals["prediction"]["tripTag"], "epochTime" => vals["prediction"]["epochTime"], "currentTime" => Time.now}
  82. end
  83. def not_already_in_memory?(vals)
  84. m = self.memory["existing_routes"] || []
  85. m.select{|h| h['stopTag'] == vals["stopTag"] &&
  86. h['tripTag'] == vals["prediction"]["tripTag"] &&
  87. h['epochTime'] == vals["prediction"]["epochTime"]
  88. }.count == 0
  89. end
  90. def default_options
  91. {
  92. agency: "sf-muni",
  93. stops: ["N|5221", "N|5215"],
  94. alert_window_in_minutes: 5
  95. }
  96. end
  97. def validate_options
  98. errors.add(:base, 'agency is required') unless options['agency'].present?
  99. errors.add(:base, 'alert_window_in_minutes is required') unless options['alert_window_in_minutes'].present?
  100. errors.add(:base, 'stops are required') unless options['stops'].present?
  101. end
  102. def working?
  103. event_created_within?(2) && !recent_error_logs?
  104. end
  105. end
  106. end