Skip to content

Instantly share code, notes, and snippets.

@slackorama

slackorama/app.py

Last active Jan 8, 2019
Embed
What would you like to do?
pull info from strava
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import cStringIO
import codecs
import csv
import logging
import sys
import webbrowser
from stravalib import unithelper
from stravalib.client import Client
CLIENT_ID = <YOUR CLIENT ID>
ACCESS_TOKEN = '<YOUR CODE HERE>'
LOG = logging.basicConfig(level=logging.ERROR)
class UTF8Recoder:
"""
Iterator that reads an encoded stream and encodes the input to UTF-8
"""
def __init__(self, f, encoding):
self.reader = codecs.getreader(encoding)(f)
def __iter__(self):
return self
def next(self):
return self.reader.next().encode('utf-8')
class UnicodeReader:
"""
A CSV reader which will iterate over lines in the CSV file "f",
which is encoded in the given encoding.
"""
def __init__(self, f, dialect=csv.excel, encoding='utf-8', **kwds):
f = UTF8Recoder(f, encoding)
self.reader = csv.reader(f, dialect=dialect, **kwds)
def next(self):
row = self.reader.next()
return [unicode(s, 'utf-8') for s in row]
def __iter__(self):
return self
class UnicodeWriter:
"""
A CSV writer which will write rows to CSV file "f",
which is encoded in the given encoding.
"""
def __init__(self, f, dialect=csv.excel, encoding='utf-8', **kwds):
# Redirect output to a queue
self.queue = cStringIO.StringIO()
self.writer = csv.writer(self.queue, dialect=dialect, **kwds)
self.stream = f
self.encoder = codecs.getincrementalencoder(encoding)()
def writerow(self, row):
therow =[]
for s in row:
if isinstance(s, basestring):
therow.append(s.encode('utf-8'))
else:
therow.append(s)
self.writer.writerow(therow)
# Fetch UTF-8 output from the queue ...
data = self.queue.getvalue()
data = data.decode('utf-8')
# ... and reencode it into the target encoding
data = self.encoder.encode(data)
# write to the target stream
self.stream.write(data)
# empty queue
self.queue.truncate(0)
def writerows(self, rows):
for row in rows:
self.writerow(row)
def get_auth_code(client):
auth_url = 'http://localhost:8000/'
authorize_url = client.authorization_url(client_id=CLIENT_ID,
redirect_uri=auth_url)
webbrowser.open(authorize_url, True, True)
def get_bikes(athlete):
d = dict()
if hasattr(athlete, 'bikes'):
for b in athlete.bikes:
d[b.id] = b
return d
def main():
client = Client()
client.access_token = ACCESS_TOKEN
athlete = client.get_athlete()
bikes = get_bikes(athlete)
act_cache = dict()
seg_cache = dict()
if len(sys.argv) > 1:
activities = [client.get_activity(int(sys.argv[1]),
include_all_efforts=True)]
else:
activities = client.get_activities(limit=1)
output = cStringIO.StringIO()
csv_out = UnicodeWriter(output)
csv_out.writerow(['id', 'segment', 'bike', 'distance'])
for a in activities:
if a.id not in act_cache and a.segment_efforts is None:
a = client.get_activity(a.id, include_all_efforts=True)
act_cache[a.id] = a
efforts = a.segment_efforts
for e in efforts:
segment = seg_cache.get(e.segment.id, None) or \
client.get_segment(e.segment.id)
seg_cache[e.segment.id] = segment
top_efforts = client.get_segment_efforts(e.segment.id,
athlete_id=athlete.id,
limit=1)
for tope in top_efforts:
topa = act_cache.get(tope.activity.id, None)
if not topa:
topa = act_cache.get(tope.activity.id, None) or \
client.get_activity(tope.activity.id)
act_cache[topa.id] = topa
csv_out.writerow([segment.id,
segment.name,
bikes.get(topa.gear_id).name,
unithelper.miles(segment.distance).num])
print(output.getvalue())
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment