· 9 years ago · Oct 22, 2016, 01:36 AM
1import csv
2from datetime import date, datetime, time, timedelta
3import glob
4import logging
5import os
6from flask.ext.sqlalchemy import SQLAlchemy
7from fixtures.days import DAYS
8from mana_app import app
9from mana_app.models import (
10 PublicHoliday,
11 Days,
12 Route,
13 Stop,
14 StopMap,
15 Timetable,
16 TimetableEntry,
17 Measurement,
18)
19from mana_app.dbtools import get_or_create, get_day_type
20from mana_app.constants import STOP_MAPPING
21from settings import (
22 MEASUREMENTS,
23 SIMPLE_MEASUREMENTS,
24 TIMETABLE_DIR,
25 PUBLIC_HOLIDAYS,
26)
27logging.basicConfig(
28 filename='info.log',
29 level=logging.INFO,
30 format='%(levelname)s:%(message)s',
31)
32db = SQLAlchemy(app)
33def load_public_holidays(filepath):
34 """Table of public holidays
35 Used to check if a day is a national or Wellington holiday. Also includes
36 all other regional holidays in case this is used for other regions.
37 """
38 public_holiday_dates = []
39 try:
40 with open(filepath, 'r') as csvfile:
41 reader = csv.DictReader(csvfile)
42 for row in reader:
43 if row['NationalHoliday'] == 'Y' or row['HolidayName'] == 'Wellington Anniversary':
44 observed_date = row['ObservedDate']
45 public_holiday, created = get_or_create(
46 db.session,
47 PublicHoliday,
48 date=observed_date,
49 name=row['HolidayName'],
50 )
51 if created:
52 logging.info('Added: {}'.format(public_holiday))
53 else:
54 logging.info('Duplicate: {}'.format(public_holiday))
55 observed_date = datetime.strptime(observed_date, '%m/%d/%Y')
56 public_holiday_dates.append(observed_date)
57 db.session.flush()
58 else:
59 logging.info(
60 'Not National or Wellington holiday - {}'.format(
61 row['HolidayName'],
62 )
63 )
64 db.session.commit()
65 except:
66 db.session.rollback()
67 raise
68 return public_holiday_dates
69def load_days():
70 """Days for which a timetable is valid"""
71 all_days = {}
72 try:
73 for day in DAYS:
74 days, created = get_or_create(
75 db.session,
76 Days,
77 name=day[0],
78 includes_monday=day[1],
79 includes_tuesday=day[2],
80 includes_wednesday=day[3],
81 includes_thursday=day[4],
82 includes_friday=day[5],
83 includes_saturday=day[6],
84 includes_sunday=day[7],
85 includes_public_holiday=day[8],
86 )
87 all_days[day[0]] = days
88 if created:
89 logging.info('Added: Days {}'.format(days))
90 else:
91 logging.info('Duplicate: Days {}'.format(days))
92 db.session.flush()
93 db.session.commit()
94 except:
95 db.session.rollback()
96 raise
97 return all_days
98def load_stop_map():
99 try:
100 for timetable_stopname in STOP_MAPPING.keys():
101 stop_map, created = get_or_create(
102 db.session,
103 StopMap,
104 timetable_stopname=timetable_stopname,
105 measurement_stopname=STOP_MAPPING[timetable_stopname],
106 )
107 if created:
108 logging.info('Added: Stop Mapping: {} maps to {}'.format(
109 STOP_MAPPING[timetable_stopname],
110 timetable_stopname,
111 ))
112 db.session.commit()
113 except:
114 db.session.rollback()
115 raise
116def load_routes(filepath):
117 """..."""
118 all_routes = {}
119 try:
120 with open(filepath, 'r') as csvfile:
121 reader = csv.reader(csvfile)
122 full_route_name = next(reader)[0].strip()
123 number = full_route_name.split()[-1].strip()
124 direction = next(reader)[0].strip()
125 variants = set()
126 for row in reader:
127 if 'Variant' in row:
128 for col in row:
129 if col.isdigit():
130 variants.add(int(col.strip()))
131 break
132 if number and direction:
133 for variant in variants:
134 route, created = get_or_create(
135 db.session,
136 Route,
137 number=number,
138 direction=direction,
139 variant=variant,
140 )
141 if created:
142 logging.info('Filepath: {} Added: Route {}'.format(
143 filepath,
144 route
145 ))
146 else:
147 logging.info(
148 'Filepath: {} Duplicate: Route {} not added'.format(
149 filepath,
150 route
151 ))
152 db.session.flush()
153 all_routes[(number, direction, variant)] = route
154 else:
155 logging.info('Filepath: {} Invalid: Route needs both number and direction'.format(
156 filepath,
157 ))
158 db.session.commit()
159 except:
160 db.session.rollback()
161 raise
162 return all_routes
163def load_stops(filepath):
164 """ Stops loaded according the new proposed timetables
165 This means using all the new stop names.
166 """
167 all_stops = {}
168 try:
169 with open(filepath, 'r') as csvfile:
170 reader = csv.reader(csvfile)
171 timing_points = []
172 row = next(reader)
173 while row[0].strip().lower().startswith('route bus stop'):
174 # skip unnecessary rows
175 row = next(reader)
176 next(reader)
177 next(reader)
178 next(reader)
179 for row in reader:
180 name = row[0]
181 number = row[1]
182 if name and number:
183 if (name, number) in timing_points:
184 stop, created = get_or_create(
185 db.session,
186 Stop,
187 name=name,
188 number=number,
189 timing_point=True,
190 )
191 else:
192 stop, created = get_or_create(
193 db.session,
194 Stop,
195 name=name,
196 number=number,
197 )
198 if created:
199 logging.info('Filepath: {} Added: Stop {}'.format(
200 filepath,
201 stop
202 ))
203 else:
204 logging.info('Filepath: {} Duplicate: Stop {}'.format(
205 filepath,
206 stop
207 ))
208 all_stops[(name, number)] = stop
209 db.session.flush()
210 else:
211 logging.info('Filepath: {} Invalid: Stop needs both name and number'.format(
212 filepath,
213 ))
214 db.session.commit()
215 except:
216 db.session.rollback()
217 raise
218 return all_stops
219def load_timetable(filepath):
220 print(filepath)
221 all_days = load_days()
222 all_routes = load_routes(filepath)
223 all_timetables = {}
224 try:
225 with open(filepath, 'r') as csvfile:
226 reader = csv.reader(csvfile)
227 full_route_name = next(reader)[0]
228 number = full_route_name.split()[-1]
229 direction = next(reader)[0]
230 days_name = next(reader)[0]
231 row = next(reader)
232 while not row[0].strip().lower().startswith('route bus stop'):
233 row = next(reader)
234 # List of variant numbers, will become list of
235 # tuples with second element being the start time for that timetable
236 # (same route variant can have multiple busses running at different times)
237 row = next(reader)
238 variants = [int(variant) for variant in row[2:] if variant != '']
239 row = next(reader)
240 while not row[1].strip().lower().startswith('valid day'):
241 # Sometimes there are extra rows after 'Variant' and before
242 # the required rows
243 row = next(reader)
244 departure_times = [None for variant in variants]
245 for row in reader:
246 # Insert departure times from first stop until we have all
247 stop_times = row[2:]
248 for index, stop_time in enumerate(stop_times):
249 if stop_time and departure_times[index] is None:
250 departure_times[index] = stop_time
251 if None not in departure_times:
252 break
253 for variant, departure_time in zip(variants, departure_times):
254 if number and direction and variant and days_name:
255 days = all_days[days_name.strip()]
256 route = all_routes[number.strip(), direction.strip(), variant]
257 if ':' not in departure_time:
258 departure_time = float(departure_time) - 1
259 seconds = round(float(departure_time) * 24 * 3600)
260 departure_time = time(
261 int(seconds // 3600),
262 int((seconds % 3600) // 60),
263 int(seconds % 60)
264 )
265 departure_time = departure_time.strftime('%H:%M:%S')
266 hour, minute, second = departure_time.split(':')
267 departure_time = time(int(hour), int(minute), int(second))
268 timetable, created = get_or_create(
269 db.session,
270 Timetable,
271 days=days,
272 route=route,
273 departure_time=departure_time,
274 )
275 all_timetables[days, route, departure_time] = timetable
276 if created:
277 logging.info('Filepath: {} Added: Timetable {}'.format(
278 filepath,
279 timetable
280 ))
281 else:
282 logging.info('Filepath: {} Duplicate: Timetable {}'.format(
283 filepath,
284 timetable,
285 ))
286 else:
287 logging.info('Filepath: {} Timetable needs number, direction, variant and days'.format(
288 filepath,
289 ))
290 db.session.flush()
291 print('committing transaction')
292 db.session.commit()
293 print('transaction committed')
294 except:
295 db.session.rollback()
296 print('rolling back transaction')
297 raise
298 return all_timetables
299def load_timetable_entries(filepath, get_or_create=True):
300 """Load timetable entries into the database
301 get_or_create slows down as the the data grows. For initial loading
302 it makes sense to ignore the get_or_create to improve the
303 efficiency of the loading process. However, once this is done we
304 don't want to be adding duplicates.
305 Some of the files are very different. I have separated these files
306 out and will deal with them manually, or write better code to do
307 them automatically.
308 """
309 try:
310 with open(filepath, 'r') as csvfile:
311 reader = csv.reader(csvfile)
312 full_route_name = next(reader)[0]
313 number = full_route_name.split()[-1]
314 direction = next(reader)[0].strip()
315 days_name = next(reader)[0].strip()
316 timing_points = []
317 row = next(reader)
318 while row[0].lower().strip() != 'timing points':
319 row = next(reader)
320 next(reader)
321 next(reader)
322 next(reader)
323 next(reader)
324 row = next(reader)
325 while row[0] != '':
326 if 'timetable' in row[0].strip().lower():
327 break
328 timing_points.append((row[0], row[1]))
329 row = next(reader)
330 while not row[0].lower().strip().startswith('route bus stop'):
331 row = next(reader)
332 variants = next(reader)
333 next(reader)
334 next(reader)
335 trips = []
336 for col in zip(variants, *reader):
337 trips.append(col)
338 for journey in range(len(trips)):
339 if journey < 2:
340 continue
341 orders = {}
342 variant = None # set the variant for each trip
343 departure_time = None
344 for stop_name, stop_number, stop_time in zip(
345 trips[0],
346 trips[1],
347 trips[journey]
348 ):
349 # stop numbers are a 4 digit string e.g. '1001'
350 if len(stop_number) == 4:
351 stop_number = stop_number
352 # variants are a 1 digit string e.g. '1'
353 if len(stop_time) == 1:
354 variant = int(stop_time)
355 if stop_time and (stop_name, stop_number) in all_stops.keys() \
356 and number and direction and variant and days_name:
357 orders[variant] = orders.get(variant, 0)
358 if ':' not in stop_time:
359 stop_time = float(stop_time) - 1
360 seconds = round(float(stop_time) * 24 * 3600)
361 stop_time = time(
362 int(seconds // 3600),
363 int((seconds % 3600) // 60),
364 int(seconds % 60)
365 )
366 stop_time = stop_time.strftime('%H:%M:%S')
367 hour, minute, second= stop_time.split(':')
368 stop_time = time(int(hour), int(minute), int(second))
369 if departure_time is None:
370 departure_time = stop_time
371 days = all_days[days_name]
372 route = all_routes[(number, direction, variant)]
373 stop = all_stops[(stop_name, stop_number)]
374 timetable = all_timetables[(days, route, departure_time)]
375 if get_or_create:
376 if (stop_name, stop_number) in timing_points:
377 timetable_entry, created = get_or_create(
378 db.session,
379 TimetableEntry,
380 order=orders[variant],
381 stop=stop,
382 stop_time=stop_time,
383 timetable=timetable,
384 is_timing_point=True,
385 )
386 else:
387 timetable_entry, created = get_or_create(
388 db.session,
389 TimetableEntry,
390 order=orders[variant],
391 stop=stop,
392 stop_time=stop_time,
393 timetable=timetable,
394 )
395 else:
396 if (stop_name, stop_name) in timing_points:
397 timetable_entry = TimetableEntry(
398 order=orders[variant],
399 stop=stop,
400 stop_time=stop_time,
401 timetable=timetable,
402 is_timing_point=True,
403 )
404 else:
405 timetable_entry = TimetableEntry(
406 order=orders[variant],
407 stop=stop,
408 stop_time=stop_time,
409 timetable=timetable,
410 )
411 db.session.add(timetable_entry)
412 created = True
413 if created:
414 logging.info('Filepath: {} Added: Timetable Entry - {}'.format(filepath, timetable_entry))
415 else:
416 logging.info('Filepath: {} Duplicate: Timetable Entry - {}'.format(filepath, timetable_entry))
417 db.session.flush
418 orders[variant] += 1
419 db.session.commit()
420 except:
421 db.session.rollback()
422 raise
423def load_measurements(
424 measurements_filepath,
425 all_days,
426 all_stops,
427 all_routes,
428 all_timetables,
429 get_or_create=True
430):
431 """Load measurements into database
432 For csvs with many rows as the get_or_create function has to do a query to
433 check whether it should get or create. As the table gets larger, the query
434 takes longer and longer. Using the argument get_or_create allows the choice
435 between simply adding or using the get_or_create function.
436 """
437 today = date.today()
438 try:
439 with open(measurements_filepath, 'r') as csvfile:
440 reader = csv.reader(csvfile)
441 fieldnames = next(reader)
442 measurement_stopnames = fieldnames[4:]
443 unknown_stopnames = set()
444 stop_by_measurement_stopname = {}
445 for measurement_stopname in measurement_stopnames:
446 # take the raw_stopname, see if its in the mapping table,
447 # if it is in the mapping table, use the timetable_stopname
448 # that corresponds.
449 mapped_raw_stopname = db.session.query(
450 StopMap.timetable_stopname
451 ).filter(
452 StopMap.measurement_stopname == measurement_stopname
453 ).first()
454 if mapped_raw_stopname:
455 raw_stopname = mapped_raw_stopname[0]
456 else:
457 raw_stopname = measurement_stopname
458 stopname = ' '.join(i for i in raw_stopname.split()[:-1]).strip()
459 stopnumber = raw_stopname.split()[-1].strip('()')
460 # Try matching on both stopname and stop number
461 stopkey = (stopname, stopnumber)
462 stop = all_stops.get(stopkey, None)
463 # Then try matching on just the stop number
464 if stop is None:
465 matching_stops = [
466 stop for stop in all_stops if stop[1] == stopnumber
467 ]
468 if matching_stops:
469 stop = all_stops[matching_stops[0]]
470 if stop is None:
471 unknown_stopnames.add(measurement_stopname)
472 continue
473 else:
474 # Regardless of how the stop was found, map it to
475 # the stop name as used in the measurement file
476 stop_by_measurement_stopname[measurement_stopname] = stop
477 # Some timetables have a different number of 'header' lines
478 # Skip those lines until the first 'date' exists in the first column
479 row = next(reader)
480 while '/' not in row[0]:
481 row = next(reader)
482 for row_number, raw_row in enumerate(reader, 1):
483 # Prepare row dict like a DictReader but let duplicate headers
484 # turn the value into a list holding the multiple values
485 row = {}
486 for key, value in zip(fieldnames, raw_row):
487 if not value.strip():
488 value = None
489 if key in row:
490 # duplicate row header: value to be a list of all
491 # values
492 if not isinstance(row[key], list):
493 row[key] = [row[key]]
494 row[key].append(value)
495 else:
496 # first time row header, as per DictReader
497 row[key] = value
498 if row['Date'] and row['Scheduled Departure Time']:
499 # only process rows that contain measurements
500 hour, minute = row['Scheduled Departure Time'].split(':')
501 departure_time = time(int(hour), int(minute))
502 measurement_date = datetime.strptime(row['Date'], '%d/%m/%Y')
503 day, public_holiday = get_day_type(
504 measurement_date,
505 public_holidays=public_holidays
506 )
507 previous_stop = None
508 previous_time = None
509 stopnames_visited = []
510 for stopname in measurement_stopnames:
511 stop = stop_by_measurement_stopname.get(stopname, None)
512 if stop is None:
513 # Unknown stop - ignore measurement
514 continue
515 if row[stopname] is None:
516 # No value: route not yet started or gap in route
517 continue
518 if isinstance(row[stopname], list):
519 # Stop appears more than once, so the measurements
520 # are a list - grab the one depending on how often
521 # we've already grabbed a value for the stop before
522 num_visited = stopnames_visited.count(stopname)
523 if row[stopname][num_visited] is None:
524 continue
525 elapsed_to = float(
526 row[stopname][num_visited]
527 )
528 else:
529 elapsed_to = float(row[stopname])
530 stopnames_visited.append(stopname)
531 if previous_stop:
532 # no origin, no measurement
533 to_datetime = datetime.combine(today, departure_time) + timedelta(minutes=elapsed_to)
534 to_time = to_datetime.time()
535 elapsed_time = datetime.combine(today, to_time) - datetime.combine(today, previous_time)
536 if get_or_create:
537 new_measurement, created = get_or_create(
538 db.session,
539 Measurement,
540 measurement_date=measurement_date,
541 day_of_week=day,
542 public_holiday=public_holiday,
543 from_stop_id=previous_stop.id,
544 to_stop_id=stop.id,
545 from_time=previous_time,
546 elapsed_time=elapsed_time,
547 )
548 else:
549 new_measurement = Measurement(
550 measurement_date=measurement_date,
551 day_of_week=day,
552 public_holiday=public_holiday,
553 from_stop_id=previous_stop.id,
554 to_stop_id=stop.id,
555 from_time=previous_time,
556 elapsed_time=elapsed_time,
557 )
558 db.session.add(new_measurement)
559 created = True
560 if created:
561 logging.info(
562 'Filepath: {} Row {} - Added: {}'.format(
563 measurements_filepath,
564 row_number,
565 new_measurement,
566 )
567 )
568 else:
569 'Filepath: {} Row {} - Added: {}'.format(
570 measurements_filepath,
571 row_number,
572 new_measurement,
573 )
574 if row_number % 50 == 0:
575 db.session.flush()
576 if row_number % 500 == 0:
577 db.session.commit()
578 previous_stop = stop
579 # First stop may not have started at zero
580 previous_datetime = datetime.combine(
581 today,
582 departure_time
583 ) + timedelta(minutes=elapsed_to)
584 previous_time = previous_datetime.time()
585 db.session.commit()
586 except:
587 db.session.rollback()
588 raise
589 if unknown_stopnames:
590 logging.warning(
591 'Unknown stop names: {}'.format(', '.join(unknown_stopnames))
592 )
593 print('Success')
594if __name__ == '__main__':
595 all_days = load_days()
596 public_holidays = load_public_holidays(PUBLIC_HOLIDAYS)
597 load_stop_map()
598 all_stops = {}
599 all_routes = {}
600 all_timetables = {}
601 timetable_filepaths = glob.glob('data/Timetables/*.csv')
602 for timetable_filepath in timetable_filepaths:
603 all_stops.update(load_stops(timetable_filepath))
604 all_routes.update(load_routes(timetable_filepath))
605 for timetable_filepath in timetable_filepaths:
606 all_timetables.update(load_timetable(timetable_filepath))
607 # load_timetable_entries(timetable_filepath, get_or_create=False)
608 # move completed csv files into the 'Done' folder
609 # timetable_filepath_split = timetable_filepath.split('/')
610 # timetable_filepath_split.insert(2, 'Done')
611 # done_timetable_filepath = '/'.join([i for i in timetable_filepath_split])
612 # os.rename(timetable_filepath, done_timetable_filepath)
613 measurement_filepaths = glob.glob('data/Measurements/*.csv')
614 for measurement_filepath in measurement_filepaths:
615 print(measurement_filepath)
616 load_measurements(
617 measurement_filepath,
618 all_days,
619 all_stops,
620 all_routes,
621 all_timetables,
622 get_or_create=False
623 )
624 # move completed csvs files to the 'Done' folder
625 measurement_filepath_split = measurement_filepath.split('/')
626 measurement_filepath_split.insert(2, 'Done')
627 done_measurement_filepath = '/'.join([i for i in measurement_filepath_split])
628 os.rename(measurement_filepath, done_measurement_filepath)