lefrecce.py - 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
---
lefrecce.py
---
1 #!/usr/pkg/bin/python3.7
2
3 #
4 # Copyright (c) 2017-2019 Leonardo Taccari
5 # All rights reserved.
6 #
7 # Redistribution and use in source and binary forms, with or without
8 # modification, are permitted provided that the following conditions
9 # are met:
10 #
11 # 1. Redistributions of source code must retain the above copyright
12 # notice, this list of conditions and the following disclaimer.
13 # 2. Redistributions in binary form must reproduce the above copyright
14 # notice, this list of conditions and the following disclaimer in the
15 # documentation and/or other materials provided with the distribution.
16 #
17 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
18 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
19 # TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
20 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS
21 # BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
22 # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
23 # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
24 # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
25 # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
26 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27 # POSSIBILITY OF SUCH DAMAGE.
28 #
29
30
31 """
32 retrieve information about trains/stations via lefrecce.it
33
34 lefrecce is a Python script/module to retrieve information about next trains
35 and stations via lefrecce.it
36 """
37
38
39 from datetime import datetime
40 import json
41
42 import requests
43
44
45 LEFRECCE_API_URL = 'https://www.lefrecce.it/msite/api'
46 LEFRECCE_SOLUTIONS_URL = LEFRECCE_API_URL + '/solutions?' + \
47 'origin={origin}&' + \
48 'destination={destination}&' + \
49 'arflag=A&' + \
50 'adate={adate}&' + \
51 'atime={atime}&' + \
52 'adultno=1&' + \
53 'childno=0&' + \
54 'direction=A&' + \
55 'frecce=false&' + \
56 'onlyRegional=false'
57 LEFRECCE_SOLUTIONS_INFO_URL = LEFRECCE_API_URL + '/solutions/' + \
58 '{solution}/info'
59 LEFRECCE_LOCATIONS_URL = LEFRECCE_API_URL + '/geolocations/locations?' + \
60 'name={name}'
61
62
63 def print_solution(solution):
64 """Pretty print each solution returned by solutions()."""
65
66 print('{departure} ({departure_datetime:%H:%M %d/%m/%Y}) - '
67 '{arrival} ({arrival_datetime:%H:%M %d/%m/%Y}) '
68 '({duration})'.format(**solution))
69
70 for train in solution['trains']:
71 print(' [{acronym}] {id}:\t'
72 '{departure_station} ({departure_datetime:%H:%M}) - '
73 '{arrival_station} ({arrival_datetime:%H:%M})'.format(**train))
74
75 for stop in train['stoplist']:
76 print(' . {station:14}\t'
77 '{arrival_datetime:%H:%M} - '
78 '{departure_datetime:%H:%M}'.format(**stop))
79
80
81 def solutions(origin, destination, adate, atime, verbose=True):
82 """Given origin, destination, departure date and departure time return
83 next 5 solutions via a generator.
84
85 Needs origin station (origin), destination station (destination),
86 departure date (in `%d/%m/%Y' format) (adate),
87 departure time (in `%H' format) (atime) and an optional boolean verbose
88 flag.
89
90 Each solution is a dict with the following keys:
91 - departure: departure station
92 - departure_datetime: datetime.datetime() of the departure time
93 - arrival: arrival station
94 - arrival_datetime: datetime.datetime() of the arrival time
95 - trains: a list of dicts with the details of the train(s):
96 + acronym: acronym (REG, RV, ...)
97 + id: identifier of the train
98 + departure_station: departure station
99 + departure_datetime: datetime.datetime() of the departure time
100 + arrival_datetime: datetime.datetime() of the arrival time
101 + arrival_station: departure station
102
103 >>> sols = list(solutions(
104 ... 'milano centrale',
105 ... 'roma termini',
106 ... datetime.today().strftime('%d/%m/%Y'),
107 ... datetime.today().strftime('%H'),
108 ... verbose=True,
109 ... ))
110 >>> 0 < len(sols) <= 5
111 True
112 >>> sols[0]['departure']
113 'Milano Centrale'
114 >>> sols[0]['arrival']
115 'Roma Termini'
116 >>> sols[0]['trains'][0]['acronym'] != ''
117 True
118 >>> sols[0]['trains'][0]['id'] != ''
119 True
120 >>> sols[0]['trains'][0]['departure_station'] != ''
121 True
122 >>> sols[0]['trains'][0]['arrival_station'] != ''
123 True
124
125 >>> sols = list(solutions(
126 ... 'roma termini',
127 ... 'milano centrale',
128 ... datetime.today().strftime('%d/%m/%Y'),
129 ... datetime.today().strftime('%H'),
130 ... verbose=False,
131 ... ))
132 >>> sols[0]['trains'] == []
133 True
134 """
135 url = LEFRECCE_SOLUTIONS_URL.format(origin=origin,
136 destination=destination,
137 adate=adate,
138 atime=atime)
139
140 sess = requests.Session()
141 sess.headers['Accept-Language'] = 'en-US'
142 r = sess.get(url)
143 j = json.loads(r.text)
144
145 for solution in j:
146 s = {}
147 s['departure'] = solution['origin']
148 s['departure_datetime'] = \
149 datetime.fromtimestamp(solution['departuretime'] / 1000)
150 s['arrival'] = solution['destination']
151 s['arrival_datetime'] = \
152 datetime.fromtimestamp(solution['arrivaltime'] / 1000)
153 s['duration'] = solution['duration']
154 s['trains'] = []
155
156 if verbose:
157 trains_url = LEFRECCE_SOLUTIONS_INFO_URL.format(
158 solution=solution['idsolution'])
159 r = sess.get(trains_url)
160 trains = json.loads(r.text)
161
162 for train in trains:
163 stoplist = []
164 if 'stoplist' in train:
165 for stop in train['stoplist']:
166 if stop['departuretime'] is None:
167 stop['departuretime'] = stop['arrivaltime']
168 if stop['arrivaltime'] is None:
169 stop['arrivaltime'] = stop['departuretime']
170
171 stoplist.append({
172 'departure_datetime':
173 datetime.fromtimestamp(
174 stop['departuretime'] / 1000),
175 'arrival_datetime':
176 datetime.fromtimestamp(
177 stop['arrivaltime'] / 1000),
178 'station':
179 stop['stationname'].lower().capitalize(),
180 })
181
182 s['trains'].append({
183 'acronym': train['trainacronym'],
184 'id': train['trainidentifier'],
185 'departure_station':
186 train['departurestation'].lower().capitalize(),
187 'departure_datetime':
188 datetime.fromtimestamp(train['departuretime'] / 1000),
189 'arrival_station':
190 train['arrivalstation'].lower().capitalize(),
191 'arrival_datetime':
192 datetime.fromtimestamp(train['arrivaltime'] / 1000),
193 'stoplist': stoplist,
194 })
195
196 yield s
197
198
199 def search_stations(name):
200 """Given a station name return a list of stations matching the name.
201
202 Please note that the API is limited to a maximum of 5 matching stations.
203
204 >>> search_stations('macerata')
205 ['MACERATA', 'MACERATA FONTESCODELLA']
206 """
207 url = LEFRECCE_LOCATIONS_URL.format(name=name)
208
209 sess = requests.Session()
210 sess.headers['Accept-Language'] = 'en-US'
211 r = sess.get(url)
212 j = json.loads(r.text)
213
214 stations = []
215 for station in j:
216 stations.append(station['name'])
217
218 return stations
219
220
221 if __name__ == '__main__':
222 import argparse
223 import sys
224
225 ap = argparse.ArgumentParser(
226 description='Retrieve information from lefrecce.it')
227 ap.add_argument('-s', metavar='search_station', type=str)
228 ap.add_argument('-o', metavar='origin', type=str,
229 help='origin station')
230 ap.add_argument('-d', metavar='destination', type=str,
231 help='destination station')
232 ap.add_argument('-a', metavar='adate', type=str,
233 help='arrival date')
234 ap.add_argument('-t', metavar='atime', type=str,
235 help='arrival time')
236 ap.add_argument('-q', action='store_true', help='silent output')
237 args = ap.parse_args()
238
239 # Search stations and exit
240 if args.s:
241 for s in search_stations(args.s):
242 print(s)
243 sys.exit(0)
244
245 # Set current date and time if the user does not provide them
246 if not args.a:
247 args.a = datetime.today().strftime('%d/%m/%Y')
248 if not args.t:
249 args.t = datetime.today().strftime('%H')
250
251 # Check arguments
252 for a in args.o, args.d, args.a, args.t:
253 if not a:
254 ap.print_usage()
255 sys.exit(1)
256
257 for s in solutions(args.o, args.d, args.a, args.t, verbose=(not args.q)):
258 print_solution(s)