import os import json import getpass from datetime import datetime import itertools import numpy import data class NumpyEncoder(json.JSONEncoder): def default(self, o): if type(o).__module__ == numpy.__name__: return o.item() super(NumpyEncoder, self).default(o) class EGJ(object): def save(self, path=getpass.getuser(), append=False): path = os.path.join(data.path, 'visualizer', path) if append: if not os.path.isdir(path): raise ValueError("Can't append to the given directory") name = str(1+max(map(int, filter(str.isdigit, os.listdir(path)))+[-1])) path = os.path.join(path, name) else: while os.path.isdir(path): path = os.path.join(path, '0') with open(path, 'w') as f: self.write(f) def write(self, file): file.write(json.dumps(self.object(), cls=NumpyEncoder)) def type(self): return 'raw' def options(self): return [] def object(self): return { 'type': self.type(), 'data': { 'type': 'FeatureCollection', 'crs': { 'type': 'name', 'properties': { 'name': 'urn:ogc:def:crs:OGC:1.3:CRS84' } }, 'features': self.features() } } class Point(EGJ): def __init__(self, latitude, longitude, info=None): self.latitude = latitude self.longitude = longitude self.info = info def features(self): d = { 'type': 'Feature', 'geometry': { 'type': 'Point', 'coordinates': [self.longitude, self.latitude] } } if self.info is not None: d['properties'] = { 'info': self.info } return [d] class Path(EGJ): def __init__(self, path, info=''): self.path = path self.info = info def features(self): info = self.info + '''trip_id: %(trip_id)s
call_type: %(call_type_f)s
origin_call: %(origin_call)d
origin_stand: %(origin_stand)d
taxi_id: %(taxi_id)d
timestamp: %(timestamp_f)s
day_type: %(day_type_f)s
missing_data: %(missing_data)d
''' \ % dict(self.path, call_type_f = ['central', 'stand', 'street'][self.path['call_type']], timestamp_f = datetime.fromtimestamp(self.path['timestamp']).strftime('%c'), day_type_f = ['normal', 'holiday', 'holiday eve'][self.path['day_type']]) return [{ 'type': 'Feature', 'properties': { 'info': info, 'display': 'path', 'timestamp': self.path['timestamp'] }, 'geometry': { 'type': 'LineString', 'coordinates': [[lon, lat] for (lat, lon) in zip(self.path['latitude'], self.path['longitude'])] } }] class Vlist(EGJ, list): def __init__(self, cluster=False, heatmap=False, *args): list.__init__(self, *args) self.cluster = cluster self.heatmap = heatmap def type(self): if self.cluster or self.heatmap: if all(isinstance(c, Point) for c in self): if self.cluster: return 'cluster' elif self.heatmap: return 'heatmap' else: raise ValueError('Building a %s with something that is not a Point' % ('cluster' if self.cluster else 'heatmap')) else: return 'raw' def features(self): return list(itertools.chain.from_iterable(p.features() for p in self))