Skip to content

Commit

Permalink
chore: black lint
Browse files Browse the repository at this point in the history
  • Loading branch information
@stigok
stigok committed Jun 6, 2021
1 parent 7e498fd commit 5385f0b
Show file tree
Hide file tree
Showing 8 changed files with 164 additions and 102 deletions.
156 changes: 105 additions & 51 deletions ruterstop/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,7 @@
__version__ = "0.4.2"

# Default settings
DEFAULTS = dict(
long_eta=59
)
DEFAULTS = dict(long_eta=59)

ENTUR_CLIENT_ID = __version__
ENTUR_STOP_PLACE_ENDPOINT = "https://api.entur.io/stop-places/v1/graphql"
Expand Down Expand Up @@ -74,22 +72,29 @@
webapp = bottle.Bottle()
log = logging.getLogger("ruterstop")


def not_found_error_handler(res):
res.set_header("Content-Type", "text/plain")
return "Ugyldig stoppested"


webapp.error(code=404)(not_found_error_handler)


def default_error_handler(res):
res.set_header("Content-Type", "text/plain")
log.error(res.traceback)
return "Feil på serveren"


webapp.default_error_handler = default_error_handler


class Departure(namedtuple("Departure", ["line", "name", "eta", "direction", "realtime"])):
class Departure(
namedtuple("Departure", ["line", "name", "eta", "direction", "realtime"])
):
"""Represents a transport departure"""

def __str__(self):
name = str(self.line)
if self.name:
Expand Down Expand Up @@ -119,11 +124,15 @@ def get_realtime_stop(*, stop_id=None):
headers = {
"Accept": "application/json",
"ET-Client-Name": "ruterstop - stigok/ruterstop",
"ET-Client-Id": ENTUR_CLIENT_ID
"ET-Client-Id": ENTUR_CLIENT_ID,
}
qry = ENTUR_GRAPHQL_QUERY % dict(stop_id=stop_id)
res = requests.post(ENTUR_GRAPHQL_ENDPOINT, headers=headers, timeout=5,
json=dict(query=qry, variables={}))
res = requests.post(
ENTUR_GRAPHQL_ENDPOINT,
headers=headers,
timeout=5,
json=dict(query=qry, variables={}),
)
res.raise_for_status()
return res.json()

Expand All @@ -138,11 +147,15 @@ def get_stop_search_result(*, name_search):
headers = {
"Accept": "application/json",
"ET-Client-Name": "ruterstop - stigok/ruterstop",
"ET-Client-Id": ENTUR_CLIENT_ID
"ET-Client-Id": ENTUR_CLIENT_ID,
}
qry = ENTUR_STOP_PLACE_QUERY % dict(stop_name=name_search)
res = requests.post(ENTUR_STOP_PLACE_ENDPOINT, headers=headers, timeout=5,
json=dict(query=qry, variables={}))
res = requests.post(
ENTUR_STOP_PLACE_ENDPOINT,
headers=headers,
timeout=5,
json=dict(query=qry, variables={}),
)
res.raise_for_status()
return res.json()

Expand All @@ -154,7 +167,8 @@ def parse_stops(raw_dict):
numid,
stop["name"]["value"],
stop["topographicPlace"]["name"]["value"],
stop["topographicPlace"]["parentTopographicPlace"]["name"]["value"])
stop["topographicPlace"]["parentTopographicPlace"]["name"]["value"],
)


def parse_departures(raw_dict, *, date_fmt="%Y-%m-%dT%H:%M:%S%z"):
Expand All @@ -169,8 +183,9 @@ def parse_departures(raw_dict, *, date_fmt="%Y-%m-%dT%H:%M:%S%z"):
"""
if raw_dict["data"]["stopPlace"]:
for dep in raw_dict["data"]["stopPlace"]["estimatedCalls"]:
eta = datetime.strptime(dep["expectedArrivalTime"],
date_fmt).replace(tzinfo=None)
eta = datetime.strptime(dep["expectedArrivalTime"], date_fmt).replace(
tzinfo=None
)
yield Departure(
line=dep["serviceJourney"]["line"]["publicCode"],
name=norwegian_ascii(dep["destinationDisplay"]["frontText"]),
Expand Down Expand Up @@ -213,7 +228,14 @@ def get_departures(*, stop_id=None):
return parse_departures(raw_stop)


def format_departure_list(departures, *, min_eta=0, long_eta=DEFAULTS["long_eta"], directions=None, grouped=False):
def format_departure_list(
departures,
*,
min_eta=0,
long_eta=DEFAULTS["long_eta"],
directions=None,
grouped=False
):
"""
Filters, formats and groups departures based on arguments passed.
"""
Expand All @@ -225,8 +247,9 @@ def format_departure_list(departures, *, min_eta=0, long_eta=DEFAULTS["long_eta"

# Filter departures with minimum time treshold
time_treshold = datetime.now() + timedelta(minutes=min_eta)
deps = filter(lambda d: d.eta >= time_treshold or
(min_eta == 0 and d.realtime), deps)
deps = filter(
lambda d: d.eta >= time_treshold or (min_eta == 0 and d.realtime), deps
)

# Group departures with same departure time
# TODO: The check for whether directions has filter might need more work
Expand All @@ -250,50 +273,79 @@ def format_departure_list(departures, *, min_eta=0, long_eta=DEFAULTS["long_eta"
newdeps.append(deps[0])
continue

newdeps.append(Departure(line=", ".join([d.line for d in deps]),
name="", eta=deps[0].eta,
direction=deps[0].direction))
newdeps.append(
Departure(
line=", ".join([d.line for d in deps]),
name="",
eta=deps[0].eta,
direction=deps[0].direction,
)
)
deps = newdeps

# Create pretty output
s = ""
for dep in deps:
if 0 < long_eta < delta(dep.eta):
s += dep.ts_str() + '\n'
s += dep.ts_str() + "\n"
else:
s += str(dep) + '\n'
s += str(dep) + "\n"
return s


def main(argv=sys.argv, *, stdout=sys.stdout):
"""Main function for CLI usage"""
# Parse command line arguments
par = argparse.ArgumentParser(prog="ruterstop")
par.add_argument('--search-stop', type=str, metavar="<name>",
help="search for a stop by name")
par.add_argument('--stop-id', metavar="<id>",
help="use --search-stop or official website to find stops " +
"https://stoppested.entur.org (guest:guest)")
par.add_argument('--direction', choices=["inbound", "outbound"],
help="filter direction of departures")
par.add_argument('--min-eta', type=int, default=0, metavar="<minutes>",
help="minimum ETA of departures to return")
par.add_argument('--long-eta', type=int, default=DEFAULTS["long_eta"], metavar="<minutes>",
help="show departure time when ETA is later than this limit" +
"(disable with -1)")
par.add_argument('--grouped', action="store_true",
help="group departures with same ETA together " +
"when --direction is also specified.")
par.add_argument('--server', action="store_true",
help="start a HTTP server")
par.add_argument('--host', type=str, default="0.0.0.0", metavar="<ip|hostname>",
help="HTTP server hostname")
par.add_argument('--port', type=int, default=4000, metavar="<port>",
help="HTTP server listen port")
par.add_argument('--debug', action="store_true",
help="enable debug logging")
par.add_argument('--version', action="store_true",
help="show version information")
par.add_argument(
"--search-stop", type=str, metavar="<name>", help="search for a stop by name"
)
par.add_argument(
"--stop-id",
metavar="<id>",
help="use --search-stop or official website to find stops https://stoppested.entur.org (guest:guest)",
)
par.add_argument(
"--direction",
choices=["inbound", "outbound"],
help="filter direction of departures",
)
par.add_argument(
"--min-eta",
type=int,
default=0,
metavar="<minutes>",
help="minimum ETA of departures to return",
)
par.add_argument(
"--long-eta",
type=int,
default=DEFAULTS["long_eta"],
metavar="<minutes>",
help="show departure time when ETA is later than this limit (disable with -1)",
)
par.add_argument(
"--grouped",
action="store_true",
help="group departures with same ETA together when --direction is also specified.",
)
par.add_argument("--server", action="store_true", help="start a HTTP server")
par.add_argument(
"--host",
type=str,
default="0.0.0.0",
metavar="<ip|hostname>",
help="HTTP server hostname",
)
par.add_argument(
"--port",
type=int,
default=4000,
metavar="<port>",
help="HTTP server listen port",
)
par.add_argument("--debug", action="store_true", help="enable debug logging")
par.add_argument("--version", action="store_true", help="show version information")

args = par.parse_args(argv[1:])

Expand Down Expand Up @@ -328,11 +380,13 @@ def main(argv=sys.argv, *, stdout=sys.stdout):

# Just print stop information
deps = get_departures(stop_id=args.stop_id)
formatted = format_departure_list(deps,
min_eta=args.min_eta,
long_eta=args.long_eta,
directions=directions,
grouped=args.grouped)
formatted = format_departure_list(
deps,
min_eta=args.min_eta,
long_eta=args.long_eta,
directions=directions,
grouped=args.grouped,
)

print(formatted, file=stdout)

Expand Down
2 changes: 1 addition & 1 deletion ruterstop/__main__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import sys

# Don't require installing this package in order to run this script
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

from ruterstop import main

Expand Down
19 changes: 10 additions & 9 deletions ruterstop/tests/test_cli.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,18 +11,19 @@

def run(args):
out = StringIO()
ruterstop.main(['TEST'] + args, stdout=out)
lines = out.getvalue().split('\n')
ruterstop.main(["TEST"] + args, stdout=out)
lines = out.getvalue().split("\n")
return lines


class CommandLineInterfaceTestCase(TestCase):
def setUp(self):
self.patches = []

p = os.path.realpath(os.path.dirname(__file__))
with open(os.path.join(p, 'test_data.json')) as fp:
with open(os.path.join(p, "test_data.json")) as fp:
departure_data = json.load(fp)
patcher = patch('ruterstop.get_realtime_stop', return_value=departure_data)
patcher = patch("ruterstop.get_realtime_stop", return_value=departure_data)
self.patches.append(patcher)
self.patched_get_realtime_stop = patcher.start()

Expand All @@ -44,7 +45,7 @@ def setUp(self):
"31 Grorud T 11 min",
"31 Snaroeya 12 min",
"25 Loerenskog 14 min",
"25 Majorstuen 15 min"
"25 Majorstuen 15 min",
]

def tearDown(self):
Expand All @@ -57,15 +58,15 @@ def test_simple_output(self):
out = run(["--stop-id", "1337"])
self.patched_get_realtime_stop.assert_called_once_with(stop_id="1337")

actual = filter(None, out) # remove empty lines
actual = filter(None, out) # remove empty lines
self.assertEqual(list(actual), self.expected_output)

def test_adjustable_minimum_time(self):
with freeze_time(self.first_departure_time):
# Call CLI with custom args
out = run(["--stop-id", "1337", "--min-eta", "2"])
lines = filter(None, out) # remove empty lines
self.assertEqual(list(lines), self.expected_output[3:]) # skip first 3
lines = filter(None, out) # remove empty lines
self.assertEqual(list(lines), self.expected_output[3:]) # skip first 3

def test_direction_arg_is_accounted_for(self):
with freeze_time(self.first_departure_time):
Expand All @@ -78,5 +79,5 @@ def test_direction_arg_is_accounted_for(self):
def test_returns_stop_id_by_name(self):
with patch("ruterstop.get_stop_search_result", return_value=self.raw_stop_data):
out = run(["--search-stop", "foobar"])
out = filter(None, out) # remove empty lines
out = filter(None, out) # remove empty lines
self.assertEqual(len(list(out)), 5)
Loading

0 comments on commit 5385f0b

Please sign in to comment.