Initial import of lefrecce - lefrecce - Retrieve information about next trains and stations via lefrecce.it (HTM) hg clone https://bitbucket.org/iamleot/lefrecce (DIR) Log (DIR) Files (DIR) Refs (DIR) README --- (DIR) changeset a76124023feb0e0db87e89dfca3aebb7edb20b7f (HTM) Author: Leonardo Taccari <iamleot@gmail.com> Date: Mon, 1 Jan 2018 01:34:26 Initial import of lefrecce Diffstat: lefrecce.py | 177 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 files changed, 177 insertions(+), 0 deletions(-) --- diff -r 000000000000 -r a76124023feb lefrecce.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/lefrecce.py Mon Jan 01 01:34:26 2018 +0100 @@ -0,0 +1,177 @@ +#!/usr/pkg/bin/python3.6 + +# +# Copyright (c) 2017 Leonardo Taccari +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions +# are met: +# +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED +# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS +# BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + + +from datetime import datetime +import json +from urllib import parse, request + + +def print_solution(solution): + print('{departure} ({departure_datetime:%H:%M %d/%m/%Y}) - ' + '{arrival} ({arrival_datetime:%H:%M %d/%m/%Y}) ' + '({duration})'.format(**solution)) + + for train in solution['trains']: + print(' [{acronym}] {id}:\t' + '{departure_station} ({departure_datetime:%H:%M}) - ' + '{arrival_station} ({arrival_datetime:%H:%M})'.format(**train)) + + for stop in train['stoplist']: + print(' . {station:14}\t' + '{arrival_datetime:%H:%M} - ' + '{departure_datetime:%H:%M}'.format(**stop)) + + +def solutions(origin, destination, adate, atime, verbose=True): + url = 'https://www.lefrecce.it/msite/api/solutions?' \ + + 'origin=' + parse.quote(origin) + '&' \ + + 'destination=' + parse.quote(destination) + '&' \ + + 'arflag=A' + '&' \ + + 'adate=' + parse.quote(adate) + '&' \ + + 'atime=' + parse.quote(atime) + '&' \ + + 'adultno=1&childno=0&direction=A&frecce=false&onlyRegional=false' + + res = '' + req = request.Request(url, headers={'Accept-Language': 'en-US'}) + with request.urlopen(req) as r: + for l in r: + res += l.decode() + j = json.loads(res) + + for solution in j: + s = {} + s['departure'] = solution['origin'] + s['departure_datetime'] = datetime.fromtimestamp(solution['departuretime'] / 1000) + s['arrival'] = solution['destination'] + s['arrival_datetime'] = datetime.fromtimestamp(solution['arrivaltime'] / 1000) + s['duration'] = solution['duration'] + s['trains'] = [] + + if verbose: + trains_url = 'https://www.lefrecce.it/msite/api/solutions/' + parse.quote(solution['idsolution']) + '/info' + res = '' + req = request.Request(trains_url, headers={'Accept-Language': 'en-US'}) + # XXX: Often the request trains information regarding the current + # XXX: solution just return an empty string... Recheck until we have + # XXX: something useful... + while res == '': + try: + r = request.urlopen(req) + for l in r: + res += l.decode() + except: + pass + + trains = json.loads(res) + + for train in trains: + stoplist = [] + if 'stoplist' in train: + for stop in train['stoplist']: + if stop['departuretime'] == None: + stop['departuretime'] = stop['arrivaltime'] + if stop['arrivaltime'] == None: + stop['arrivaltime'] = stop['departuretime'] + + stoplist.append({ + 'departure_datetime': datetime.fromtimestamp(stop['departuretime'] / 1000), + 'arrival_datetime': datetime.fromtimestamp(stop['arrivaltime'] / 1000), + 'station': stop['stationname'].lower().capitalize(), + }) + + s['trains'].append({ + 'acronym': train['trainacronym'], + 'id': train['trainidentifier'], + 'departure_station': train['departurestation'].lower().capitalize(), + 'departure_datetime': datetime.fromtimestamp(train['departuretime'] / 1000), + 'arrival_station': train['arrivalstation'].lower().capitalize(), + 'arrival_datetime': datetime.fromtimestamp(train['arrivaltime'] / 1000), + 'stoplist': stoplist, + }) + + yield s + + +def search_stations(name): + url = 'https://www.lefrecce.it/msite/api/geolocations/locations?name=' + \ + parse.quote(name) + + res = '' + req = request.Request(url, headers={'Accept-Language': 'en-US'}) + with request.urlopen(req) as r: + for l in r: + res += l.decode() + j = json.loads(res) + + stations = [] + for station in j: + stations.append(station['name']) + + return stations + + +if __name__ == '__main__': + import argparse + import sys + + ap = argparse.ArgumentParser(description='Retrieve information from lefrecce.it') + ap.add_argument('-s', metavar='search_station', type=str) + ap.add_argument('-o', metavar='origin', type=str, \ + help='origin station') + ap.add_argument('-d', metavar='destination', type=str, \ + help='destination station') + ap.add_argument('-a', metavar='adate', type=str, \ + help='arrival date') + ap.add_argument('-t', metavar='atime', type=str, \ + help='arrival time') + ap.add_argument('-q', action='store_true', help='silent output') + args = ap.parse_args() + + # Search stations and exit + if args.s: + for s in search_stations(args.s): + print(s) + sys.exit(0) + + # Set current date and time if the user does not provide them + if not args.a: + args.a = datetime.today().strftime('%d/%m/%Y') + if not args.t: + args.t = datetime.today().strftime('%H') + + # Check arguments + for a in args.o, args.d, args.a, args.t: + if not a: + ap.print_usage() + sys.exit(1) + + origin, destination, adate, atime = args.o, args.d, args.a, args.t + for s in solutions(origin, destination, adate, atime, verbose=(not args.q)): + print_solution(s)