From 7e498fd87d0b88430b4581942111a0159c256fee Mon Sep 17 00:00:00 2001 From: Stig Otnes Kolstad Date: Sun, 6 Jun 2021 23:46:22 +0200 Subject: [PATCH 1/2] chore: bump version to 0.4.2 --- ruterstop/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ruterstop/__init__.py b/ruterstop/__init__.py index 45bed52..dfcfa49 100644 --- a/ruterstop/__init__.py +++ b/ruterstop/__init__.py @@ -20,7 +20,7 @@ from ruterstop.utils import delta, human_delta, norwegian_ascii, timed_cache -__version__ = "0.4.0" +__version__ = "0.4.2" # Default settings DEFAULTS = dict( From 5385f0bca53513ea54d60b62ab6b2e564694dbd0 Mon Sep 17 00:00:00 2001 From: Stig Otnes Kolstad Date: Sun, 6 Jun 2021 23:51:59 +0200 Subject: [PATCH 2/2] chore: black lint --- ruterstop/__init__.py | 156 ++++++++++++++++++++---------- ruterstop/__main__.py | 2 +- ruterstop/tests/test_cli.py | 19 ++-- ruterstop/tests/test_ruterstop.py | 31 +++--- ruterstop/tests/test_utils.py | 29 +++--- ruterstop/tests/test_webapp.py | 6 +- ruterstop/utils.py | 6 +- setup.py | 17 ++-- 8 files changed, 164 insertions(+), 102 deletions(-) diff --git a/ruterstop/__init__.py b/ruterstop/__init__.py index dfcfa49..c86bed6 100644 --- a/ruterstop/__init__.py +++ b/ruterstop/__init__.py @@ -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" @@ -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: @@ -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() @@ -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() @@ -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"): @@ -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"]), @@ -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. """ @@ -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 @@ -250,18 +273,23 @@ 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 @@ -269,31 +297,55 @@ 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="", - help="search for a stop by name") - par.add_argument('--stop-id', metavar="", - 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="", - help="minimum ETA of departures to return") - par.add_argument('--long-eta', type=int, default=DEFAULTS["long_eta"], metavar="", - 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="", - help="HTTP server hostname") - par.add_argument('--port', type=int, default=4000, metavar="", - 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="", help="search for a stop by name" + ) + par.add_argument( + "--stop-id", + metavar="", + 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="", + help="minimum ETA of departures to return", + ) + par.add_argument( + "--long-eta", + type=int, + default=DEFAULTS["long_eta"], + metavar="", + 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="", + help="HTTP server hostname", + ) + par.add_argument( + "--port", + type=int, + default=4000, + metavar="", + 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:]) @@ -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) diff --git a/ruterstop/__main__.py b/ruterstop/__main__.py index 65711c5..6b8c89d 100644 --- a/ruterstop/__main__.py +++ b/ruterstop/__main__.py @@ -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 diff --git a/ruterstop/tests/test_cli.py b/ruterstop/tests/test_cli.py index b8d8536..44ac96b 100644 --- a/ruterstop/tests/test_cli.py +++ b/ruterstop/tests/test_cli.py @@ -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() @@ -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): @@ -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): @@ -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) diff --git a/ruterstop/tests/test_ruterstop.py b/ruterstop/tests/test_ruterstop.py index 976b579..e90339a 100644 --- a/ruterstop/tests/test_ruterstop.py +++ b/ruterstop/tests/test_ruterstop.py @@ -13,18 +13,22 @@ class DepartureClassTestCase(TestCase): def test_str_representation(self): - with patch('ruterstop.utils.datetime') as mock_date: + with patch("ruterstop.utils.datetime") as mock_date: ref = datetime.min mock_date.now.return_value = ref in_7_mins = ref + timedelta(minutes=7) in_77_mins = ref + timedelta(minutes=77) # Test valid representation - d = ruterstop.Departure(line=21, name="twentyone", eta=in_7_mins, direction="o") + d = ruterstop.Departure( + line=21, name="twentyone", eta=in_7_mins, direction="o" + ) self.assertEqual(str(d), "21 twentyone 7 min") # Test long name trimming - d = ruterstop.Departure(line=21, name="longname" * 3, eta=in_77_mins, direction="o") + d = ruterstop.Departure( + line=21, name="longname" * 3, eta=in_77_mins, direction="o" + ) self.assertEqual(str(d), "21 longnamelon 77 min") @@ -36,7 +40,7 @@ def setUp(self): self.raw_stop_data = json.load(fp) def test_get_stop_search_result(self): - with patch('requests.post') as mock: + with patch("requests.post") as mock: ruterstop.get_stop_search_result(name_search="foobar") self.assertEqual(mock.call_count, 1) _, kwargs = mock.call_args @@ -58,11 +62,11 @@ class RuterstopTestCase(TestCase): def setUp(self): # Load test data for the external API 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: self.raw_departure_data = json.load(fp) def test_get_realtime_stop(self): - with patch('requests.post') as mock: + with patch("requests.post") as mock: ruterstop.get_realtime_stop(stop_id=1337) self.assertEqual(mock.call_count, 1) _, kwargs = mock.call_args @@ -101,13 +105,13 @@ def test_does_not_hide_realtime_departures_after_eta(self, _): deps.append(d("03", "c", past3min, "inbound", realtime=True)) deps.append(d("51", "d", futr1min, "inbound", realtime=True)) - args = " --stop-id=2121 --direction=inbound --grouped".split(' ') + args = " --stop-id=2121 --direction=inbound --grouped".split(" ") # Use the fake departure list in this patch with patch("ruterstop.parse_departures", return_value=deps) as mock: output = StringIO() ruterstop.main(args, stdout=output) - lines = output.getvalue().split('\n') + lines = output.getvalue().split("\n") output.close() self.assertEqual(lines[0], "01, 02, 03 naa") self.assertEqual(lines[1], "51 d 1 min") @@ -135,9 +139,9 @@ def test_groups_output_when_grouped_enabled(self, _): # Use the fake departure list in this patch with patch("ruterstop.parse_departures", return_value=deps) as mock: output = StringIO() - args = " --stop-id=2121 --direction=inbound --grouped".split(' ') + args = " --stop-id=2121 --direction=inbound --grouped".split(" ") ruterstop.main(args, stdout=output) - lines = output.getvalue().split('\n') + lines = output.getvalue().split("\n") output.close() self.assertEqual(lines[0], "01, 10, 11, 12 naa") self.assertEqual(lines[1], "20, 21 2 min") @@ -148,24 +152,25 @@ def test_groups_output_when_grouped_enabled(self, _): def test_shows_timestamp_for_long_etas(self, _): seed = datetime(2020, 1, 1, 10, 0, 0) with freeze_time(seed): + def futr(minutes): return seed + timedelta(minutes=minutes) d = ruterstop.Departure deps = [ # Shouldn't matter if a departure is realtime or not - d("01", "a", futr(60), "inbound", realtime=True), + d("01", "a", futr(60), "inbound", realtime=True), d("02", "b", futr(120), "inbound", realtime=True), d("03", "c", futr(150), "inbound", realtime=False), ] - args = " --stop-id=2121 --direction=inbound --long-eta=59".split(' ') + args = " --stop-id=2121 --direction=inbound --long-eta=59".split(" ") # Use the fake departure list in this patch with patch("ruterstop.parse_departures", return_value=deps) as mock: with StringIO() as output: ruterstop.main(args, stdout=output) - lines = output.getvalue().split('\n') + lines = output.getvalue().split("\n") self.assertEqual(lines[0], "01 a 11:00") self.assertEqual(lines[1], "02 b 12:00") self.assertEqual(lines[2], "03 c 12:30") diff --git a/ruterstop/tests/test_utils.py b/ruterstop/tests/test_utils.py index 53a5786..bcfc2e0 100644 --- a/ruterstop/tests/test_utils.py +++ b/ruterstop/tests/test_utils.py @@ -9,13 +9,13 @@ class HumanDeltaTestCase(TestCase): def test_output(self): ref = datetime.now() testcases = [ - (ref - timedelta(seconds=20), " naa"), - (ref + timedelta(seconds=20), " naa"), - (ref + timedelta(minutes=1), " 1 min"), + (ref - timedelta(seconds=20), " naa"), + (ref + timedelta(seconds=20), " naa"), + (ref + timedelta(minutes=1), " 1 min"), (ref + timedelta(seconds=100), " 1 min"), - (ref + timedelta(minutes=2), " 2 min"), - (ref + timedelta(minutes=10), "10 min"), - (ref + timedelta(hours=100), "99 min") + (ref + timedelta(minutes=2), " 2 min"), + (ref + timedelta(minutes=10), "10 min"), + (ref + timedelta(hours=100), "99 min"), ] for i, case in enumerate(testcases): @@ -24,7 +24,7 @@ def test_output(self): self.assertEqual(res, expected, "test case #%d" % (i + 1)) def test_default_kwarg_value(self): - with patch('ruterstop.utils.datetime') as mock_date: + with patch("ruterstop.utils.datetime") as mock_date: mock_date.now.return_value = datetime.min ruterstop.human_delta(until=datetime.min + timedelta(seconds=120)) self.assertEqual(mock_date.now.call_count, 1) @@ -33,10 +33,10 @@ def test_default_kwarg_value(self): class NorwegianAsciiTestCase(TestCase): def test_norwegian_ascii(self): testcases = [ - ("Snarøya", "Snaroeya"), - ("Ås", "aas"), - ("Ærlig", "aerlig"), - ("Voçé não gosta Açaí?", "Vo no gosta Aa?") + ("Snarøya", "Snaroeya"), + ("Ås", "aas"), + ("Ærlig", "aerlig"), + ("Voçé não gosta Açaí?", "Vo no gosta Aa?"), ] for i, case in enumerate(testcases): @@ -48,12 +48,12 @@ def test_norwegian_ascii(self): class TimedCacheTestCase(TestCase): def test_timed_cache(self): now = MagicMock() - spy = Mock(return_value=1) # don't need return value + spy = Mock(return_value=1) # don't need return value @ruterstop.timed_cache(expires_sec=60, now=now) def func(a, b=None): - spy() # for counting calls - return [a, b] # return list to compare references + spy() # for counting calls + return [a, b] # return list to compare references def test_set(): res1 = func(1, 2) @@ -66,7 +66,6 @@ def test_set(): self.assertEqual(res3, [2, 2]) self.assertEqual(spy.call_count, 2) - # Test cache function now.return_value = datetime.min test_set() diff --git a/ruterstop/tests/test_webapp.py b/ruterstop/tests/test_webapp.py index dc65769..9efbfdf 100644 --- a/ruterstop/tests/test_webapp.py +++ b/ruterstop/tests/test_webapp.py @@ -49,6 +49,6 @@ def test_simple_500_error(self, mock): def test_calls_api_with_querystring_params(self, get_mock, format_mock): self.app.get("/1234?direction=inbound&min_eta=5&bogusargs=1337") get_mock.assert_called_once_with(stop_id=1234) - format_mock.assert_called_once_with(dict(a="foo"), - directions="inbound", - min_eta=5) + format_mock.assert_called_once_with( + dict(a="foo"), directions="inbound", min_eta=5 + ) diff --git a/ruterstop/utils.py b/ruterstop/utils.py index d128d00..ae86b61 100644 --- a/ruterstop/utils.py +++ b/ruterstop/utils.py @@ -28,9 +28,11 @@ def decorator(func): @wraps(func) def wrapper(*_args, **_kwargs): time = now() - key = _make_key(_args, _kwargs, False) # pylint: disable=protected-access + key = _make_key(_args, _kwargs, False) # pylint: disable=protected-access - if key not in cache or time > cache[key]["timestamp"] + timedelta(seconds=expires_sec): + if key not in cache or time > cache[key]["timestamp"] + timedelta( + seconds=expires_sec + ): cache[key] = dict(value=func(*_args, **_kwargs), timestamp=time) return cache[key]["value"] diff --git a/setup.py b/setup.py index e568e81..387bcfe 100644 --- a/setup.py +++ b/setup.py @@ -4,20 +4,22 @@ from setuptools import setup -_version_re = re.compile(r'__version__\s+=\s+(.*)') +_version_re = re.compile(r"__version__\s+=\s+(.*)") + +with open("ruterstop/__init__.py", "rb") as f: + version = str( + ast.literal_eval(_version_re.search(f.read().decode("utf-8")).group(1)) + ) -with open('ruterstop/__init__.py', 'rb') as f: - version = str(ast.literal_eval(_version_re.search( - f.read().decode('utf-8')).group(1))) def readme(): - with open('README.md') as f: + with open("README.md") as f: return f.read() + setup( name="ruterstop", - description="Et program som viser sanntidsinformasjon for stoppesteder i " + - "Oslo og deler av Viken.", + description="Et program som viser sanntidsinformasjon for stoppesteder i Oslo og deler av Viken.", version=version, long_description=readme(), long_description_content_type="text/markdown", @@ -34,7 +36,6 @@ def readme(): ], author="stigok", author_email="stig@stigok.com", - packages=["ruterstop"], entry_points={ "console_scripts": [