"""File containing Infoscreen HSL data fetcher classes.""" import urllib.request import json import logging from datetime import datetime, timedelta from django.conf import settings from infoscreen.models import HSLDataModel class HSLFetcher: """Main class of Infoscreen HSL fetcher.""" last_fetched = datetime.fromtimestamp(0) # epoch INTERVAL = 1 # minutes logging.info( "Set up scheduled HSL API fetch every {} minutes".format(INTERVAL)) def fetch_if_needed(self): """Check if new fetch from HSL API is needed.""" if (datetime.now() - HSLFetcher.last_fetched > timedelta(minutes=HSLFetcher.INTERVAL)): self.fetch() def fetch(self): """Fetch data from HSL API.""" location_coords = (2545565, 6675319) src = urllib.request.urlopen( ("https://api.reittiopas.fi/hsl/prod/?userhash={}" "&request=stops_area¢er_coordinate={},{}") .format(settings.HSL_USERHASH, location_coords[0], location_coords[1]))\ .read().decode("utf-8") data = json.loads(src) arr = [] time = (datetime.now() + timedelta(minutes=settings.HSL_DEPARTURE_THRESHOLD)) time = "{0:02d}{0:02d}".format(time.hour, time.minute) for element in data: src = urllib.request.urlopen( ("https://api.reittiopas.fi/hsl/prod/?userhash={}" "&request=stop&code={}&dep_limit=20&time={}") .format(settings.HSL_USERHASH, element['code'], time) ).read().decode("utf-8") parsed = json.loads(src)[0] arr.append({ "name": parsed['name_fi'], "lines": parsed['lines'], "dist": element['dist'], "departures": parsed['departures']}) model_arr = HSLDataModel.objects.all() count = len(model_arr) json_dump = json.dumps(arr) if count == 0: HSLDataModel.objects.create(data=json_dump) else: obj = model_arr[count - 1] obj.data = json_dump obj.save() now = datetime.now() HSLFetcher.last_fetched = now logging.info( "Fetched HSL timetable data with size {} bytes.".format(len(src)))