Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_diff_for_humans_other_and_future_minute():
with pendulum.test(pendulum.today().at(12, 34, 56)):
now = pendulum.now().time()
assert now.diff_for_humans(now.subtract(minutes=1)) == "1 minute after"
def test_diff_for_humans_other_and_nearly_hour():
with pendulum.test(pendulum.today().at(12, 34, 56)):
now = pendulum.now().time()
assert now.diff_for_humans(now.add(minutes=59)) == "59 minutes before"
def test_diff_for_humans_now_and_second():
with pendulum.test(pendulum.today().at(12, 34, 56)):
now = pendulum.now().time()
assert now.diff_for_humans() == "a few seconds ago"
def test_diff_for_humans_other_and_nearly_future_minute():
with pendulum.test(pendulum.today().at(12, 34, 56)):
now = pendulum.now().time()
assert now.diff_for_humans(now.subtract(seconds=59)) == "59 seconds after"
def test_diff_for_humans_now_and_hours():
with pendulum.test(pendulum.today().at(12, 34, 56)):
now = pendulum.now().time()
assert now.subtract(hours=2).diff_for_humans() == "2 hours ago"
def test_diff_for_humans_other_and_future_hours():
with pendulum.test(pendulum.today().at(12, 34, 56)):
now = pendulum.now().time()
assert now.diff_for_humans(now.subtract(hours=2)) == "2 hours after"
def test_diff_in_seconds_vs_default_now():
with pendulum.test(pendulum.today().at(12, 34, 56)):
now = pendulum.now().time()
assert now.subtract(hours=1).diff().in_seconds() == 3600
def test_diff_for_humans_now_and_future_month():
with pendulum.test(pendulum.datetime(2016, 3, 1)):
today = pendulum.today().date()
assert "in 4 weeks" == today.add(weeks=4).diff_for_humans()
assert "in 1 month" == today.add(months=1).diff_for_humans()
with pendulum.test(pendulum.datetime(2017, 3, 31)):
today = pendulum.today().date()
assert "in 1 month" == today.add(months=1).diff_for_humans()
with pendulum.test(pendulum.datetime(2017, 4, 30)):
today = pendulum.today().date()
assert "in 1 month" == today.add(months=1).diff_for_humans()
with pendulum.test(pendulum.datetime(2017, 1, 31)):
today = pendulum.today().date()
def test_days_ago(self):
today = pendulum.today()
today_midnight = pendulum.instance(datetime.fromordinal(today.date().toordinal()))
self.assertTrue(dates.days_ago(0) == today_midnight)
self.assertTrue(dates.days_ago(100) == today_midnight + timedelta(days=-100))
self.assertTrue(dates.days_ago(0, hour=3) == today_midnight + timedelta(hours=3))
self.assertTrue(dates.days_ago(0, minute=3) == today_midnight + timedelta(minutes=3))
self.assertTrue(dates.days_ago(0, second=3) == today_midnight + timedelta(seconds=3))
self.assertTrue(dates.days_ago(0, microsecond=3) == today_midnight + timedelta(microseconds=3))
"""
The raw GTFS data includes multiple days worth of schedule information, and the entire schedule including all lines is contained in the stop_times.txt file, which we read into "full_schedule". This script identifies the trips running on a particular day and places them in the appropriate schedule table (one for each line).
The GTFS reference documentation will be useful here https://developers.google.com/transit/gtfs/reference/
The output of this script is a CSV for each line that includes the following fields:
"datetime", "trip_id", "stop_id", "stop_sequence", "direction_id"
The datetimes in both stop_times.txt and the output CSVs are in local LA time - so that we can group all scheduled services on one day together.
"""
import os
import pendulum
import pandas as pd
from pandas import read_csv
from analyzer.calendar import Calendar
from analyzer.schedule import scheduleTimeToDateTime
start_datetime = pendulum.today("America/Los_Angeles")
start_date = start_datetime.format("YYYY-MM-DD")
agency = "lametro-rail"
# Load all data
full_schedule = read_csv("data/GTFS/stop_times.txt")
calendar = Calendar("data/GTFS/calendar.txt")
trips = read_csv("data/GTFS/trips.txt")
# pre-processing (operations on full datasets)
services_running_today = calendar.services_running_on(start_date).service_id
trips_running_today = trips[trips["service_id"].isin(services_running_today)]
trips_and_directions = trips_running_today[["trip_id", "direction_id"]]
for line_no in range(801, 807):
line_trips = trips_running_today[trips_running_today["route_id"] == line_no]
line_schedule = full_schedule[full_schedule["trip_id"].isin(line_trips["trip_id"])]