Permalink
Browse files

An attempt to split into classes

+ Initial support towards #17
* Things may be broken
  • Loading branch information...
1 parent 28b8aac commit f79d895bfc66fce4ed17baa6d54c4886699fbfbf @mlt committed Jul 30, 2012
Showing with 561 additions and 227 deletions.
  1. +15 −0 .gitignore
  2. +7 −0 src/commands.py
  3. +4 −3 src/csv2tcx.py
  4. +152 −0 src/device.py
  5. +50 −224 src/download.py
  6. +12 −0 src/progress_text.py
  7. +22 −0 src/reader.py
  8. +88 −0 src/reader_cresta.py
  9. +103 −0 src/reader_schwinn.py
  10. +21 −0 src/utils.py
  11. +87 −0 src/writer.py
View
15 .gitignore
@@ -0,0 +1,15 @@
+*~
+*.pyc
+*.bin
+*.png
+*.rda
+*.htm
+*.log
+*.csv
+*.points
+*.laps
+*.track
+*.tcx
+*.kml
+*.gpx
+
View
7 src/commands.py
@@ -0,0 +1,7 @@
+""" Communication commands for Schwinn 810 """
+
+CONNECT = bytearray.fromhex("EEEE000000000000000000000000000000000000000000000000000000000000")
+DISCONNECT = bytearray.fromhex("FFFFFFFF00000000000000000000000000000000000000000000000000000000")
+READ = bytearray.fromhex("0808AAAA00000000000000000000000000000000000000000000000000000000")
+DELETE = bytearray.fromhex("000000000000000000000000000000000000000000000000000000000000FFFD")
+READ_SETTINGS = bytearray.fromhex("000000000000000000000000000000000000000000000000000000000000EEEE")
View
7 src/csv2tcx.py
@@ -83,12 +83,13 @@
<LatitudeDegrees>{:s}</LatitudeDegrees>
<LongitudeDegrees>{:s}</LongitudeDegrees>
</Position>
- <AltitudeMeters>{:s}</AltitudeMeters>
- <DistanceMeters>{:f}</DistanceMeters>""".format(time.astimezone(utc).strftime("%Y-%m-%dT%H:%M:%SZ"), \
+ <DistanceMeters>{:f}</DistanceMeters>""".format(time.astimezone(utc).strftime("%Y-%m-%dT%H:%M:%SZ"), \
thePoint["Latitude"], \
thePoint["Longitude"], \
- thePoint["Elevation"], \
dist))
+ if thePoint['Elevation']:
+ print("""<AltitudeMeters>{:s}</AltitudeMeters>""".format(
+ thePoint["Elevation"]))
heart = int(float(thePoint["Heart"]))
if heart > 0:
print(""" <HeartRateBpm><Value>{:d}</Value></HeartRateBpm>
View
152 src/device.py
@@ -0,0 +1,152 @@
+""" Module with device class to communicate with Schwinn 810 """
+
+from __future__ import print_function
+import os, re, stat, tempfile
+import struct
+from serial import Serial, SerialException
+from commands import *
+from utils import unpack_bcd
+import logging
+# from collections import namedtuple
+from datetime import datetime
+from reader_schwinn import SchwinnReader
+from reader_cresta import CrestaReader
+
+_log = logging.getLogger(__name__)
+
+class NotConnected(Exception): pass
+
+class Device:
+ """ Device class to communicate with Schwinn 810 """
+
+ def __init__(self, device=None):
+ self.device = device
+ self.dump = False
+ self.debug = False
+ self.connected = False
+ self.port = None
+ self.reader = None
+ self.backup = None
+ if self.device:
+ self.open()
+
+ def open(self, device=None):
+ """ Open port or existing dump """
+ if self.port:
+ return
+ if device:
+ self.device = device
+ if os.name != 'nt' or not re.match("com\\d", self.device, re.IGNORECASE):
+ try:
+ self.dump = stat.S_ISREG(os.stat(self.device).st_mode) # is regular file, i.e. previously dumped
+ except OSError as e:
+ pass
+ if not self.dump:
+ self.port = Serial(self.device, 115200, timeout=1, writeTimeout=1)
+ else:
+ _log.warn("Parsing existing dump in {:s}".format(self.device))
+ self.port = open(self.device, "rb")
+ self.connect()
+
+ def connect(self):
+ """ Issue connect command and set up a reader"""
+ if self.connected:
+ return
+ if not self.dump:
+ self.port.write(CONNECT)
+
+ raw = self.port.read(0x20)
+ if len(raw)==0:
+ _log.fatal("Did you plug your watches? Check the clip!")
+ raise NotConnected()
+ else:
+ self.connected = True
+
+ if self.debug:
+ self.backup = open(os.path.join(tempfile.gettempdir(), "schwinn810.bin"), mode="wb")
+ self.backup.write(raw)
+
+ (ee, e1, e2, e3, bcd1, bcd2, bcd3, serial, v1, v2, sign1) = struct.unpack("sBBBBBB6s6s7s2xI", raw)
+ if sign1: # 0x0130ff00
+ raw = self.port.read(4)
+ if self.debug:
+ self.backup.write(raw)
+ (sign2,) = struct.unpack("I", raw)
+ if sign1 != sign2:
+ print("0x{:X} != 0x{:X}".format(sign1, sign2))
+ # port.close()
+ # print("We are reading something wrong. Are you trying to communicate with another device?", file=sys.stderr)
+ # exit(-1)
+ self.reader = SchwinnReader(self.port, self.backup)
+ else: # Cresta/Mio
+ self.reader = CrestaReader(self.port, self.backup)
+ _log.warn("File a bug if you are not using Cresta or Mio watch!")
+ id = "{0:s} {1:s} {2:s} {3:02d} {4:02d} {5:02d} {6:02d} {7:02d} {8:02d}" \
+ .format(serial.decode('ascii'), v1.decode('ascii'), ee.decode('ascii'), \
+ e1, e2, e3, unpack_bcd(bcd1), unpack_bcd(bcd2), unpack_bcd(bcd3))
+ _log.info("Found %s" % id)
+
+ def _read_tracks(self, writer, count):
+ tracks_with_points = 0
+ for track in range(count):
+ track = self.reader.read_track()
+ tracks_with_points += (track['Points']>0)
+ _log.info("There are %d laps in %s" % (track['Laps'], track['Track']))
+ # _log.debug("Testing", extra = track)
+
+ writer.add_track(track)
+
+ for theLap in range(track['Laps']):
+ lap = self.reader.read_lap()
+ lap['Track'] = track['Track']
+ # _log.debug("Lap={:d}, Beats={:d}, Elevation={:f}".format(lap['Lap'], lap['Beats'], lap['Elevation']))
+ writer.add_lap(lap)
+ return tracks_with_points
+
+ def read(self, writer, progress=None):
+ if not self.dump:
+ self.port.write(READ)
+
+ (tracks, waypoints) = self.reader.read_summary()
+ _log.info("We've got %d tracks and %d waypoints to download" % (tracks, waypoints))
+
+ tracks_with_points = self._read_tracks(writer, tracks)
+
+ # now all track points
+ # for tracks containing them only!
+ for track in range(tracks_with_points):
+ if progress:
+ progress.track(track, tracks_with_points)
+ summary = self.reader.read_points_summary()
+ _log.info("Fetching %d points from %s" % (summary['Points'], summary['Track']))
+ writer.begin_points(summary['Track'])
+
+ for thePoint in range(summary['Points']):
+ if progress:
+ progress.point(thePoint, summary['Points'])
+ point = self.reader.read_point()
+ point['Track'] = summary['Track']
+ point['Time'] = datetime.combine(summary['Start'], point['Time'])
+ writer.add_point(point)
+ writer.commit()
+
+ for wp in range(waypoints):
+ wpt = self.reader.read_waypoint()
+ writer.add_waypoint(wpt)
+
+ def clear(self):
+ if not self.dump and self.debug:
+ self.port.write(DELETE)
+ else:
+ _log.info("Debug is required for deletion for now")
+
+ def close(self):
+ if self.debug:
+ self.backup.close()
+
+ if not self.dump:
+ self.port.write(DISCONNECT)
+ self.port.close()
+
+if __name__ == '__main__':
+ pass
View
274 src/download.py
@@ -1,230 +1,56 @@
#!/usr/bin/env python
from __future__ import print_function
-import sys,os
-import subprocess,stat,tempfile
-import argparse
-import struct
-import csv
-import serial
-import re
-from sys import exit
-#from yaml import load, dump
-
-parser = argparse.ArgumentParser(description='Download tracks from Schwinn 810 GPS sport watch with HRM.')
-parser.add_argument('--port', nargs=1, dest='port',
- default=[ {'posix': '/dev/ttyUSB0', 'nt': 'COM5'}[os.name] ],
-# default=['COM5'] if os.name=='nt' else ['/dev/ttyUSB0'],
- help='Virtual COM port created by cp201x for Schwinn 810 GPS watch')
-parser.add_argument('--hook', nargs=1, dest='hook',
- help='Callback upon track extraction')
-parser.add_argument('--dir', nargs=1, dest='dir',
- default=['.'],
- help='Where to store data')
-parser.add_argument('--debug', dest='debug', action='store_true',
- help='Dump all replies in a binary form into a single file schwinn810.bin in TEMP dir')
-parser.add_argument('--delete', dest='delete', action='store_true',
- help='Delete all data from watches after download?')
-# parser.add_argument('--add-year', dest='add_year', action='store_true',
-# help='Creates subfolder in dir named after the current year')
-# parser.add_argument('--add-id', dest='add_id', action='store_true',
-# help='Creates subfolder with device id inside dir but before year')
-
-args = parser.parse_args()
-
-open_extra = dict();
-if sys.version_info >= (3,0):
- open_extra["newline"] = ''
-
-connect = bytearray.fromhex("EEEE000000000000000000000000000000000000000000000000000000000000")
-disconnect = bytearray.fromhex("FFFFFFFF00000000000000000000000000000000000000000000000000000000")
-download = bytearray.fromhex("0808AAAA00000000000000000000000000000000000000000000000000000000")
-delete = bytearray.fromhex("000000000000000000000000000000000000000000000000000000000000FFFD")
-
-def unpackBCD(x0):
- x00 = x0
- x = 0
- m = 0
- while x0:
- digit = x0 & 0x000f
- if digit>9:
- raise Exception("Non-BCD argument {:X} in {:X}".format(digit, x00))
- x = x + digit * 10**m
- m = m + 1
- x0 = x0 >> 4
- return x
-
-def pack_coord(c0, hemi):
- c1 = struct.unpack(">HBHs", c0)
- c2=[unpackBCD(x) for x in c1[0:3]]
- c = c2[0] + (c2[1]+c2[2]/10000.)/60.
- if (c1[3] == hemi) : c = -c
- return c
-
-#serial.Serial("COM5", 115200, timeout=20, writeTimeout=20)
-reg = False
-if os.name != 'nt' or not re.match("com\\d", args.port[0], re.IGNORECASE):
- reg = stat.S_ISREG(os.stat(args.port[0]).st_mode) # is regular file, i.e. previously dumped
-
-if not reg:
+from device import Device, SerialException
+from writer import Writer
+from progress_text import TextProgress
+import argparse, os
+import logging
+
+_log = logging.getLogger(__name__)
+
+def main():
+ print("""schwinn810 Copyright (C) 2012 Mikhail Titov
+
+This program comes with ABSOLUTELY NO WARRANTY.
+This is free software, and you are welcome to redistribute it
+under the terms of GPL-3 or later version.
+""")
+
+ parser = argparse.ArgumentParser(description='Download tracks from Schwinn 810 GPS sport watch with HRM.')
+ parser.add_argument('--port', nargs=1, dest='port',
+ default=[ {'posix': '/dev/ttyUSB0', 'nt': 'COM5'}[os.name] ],
+ # default=['COM5'] if os.name=='nt' else ['/dev/ttyUSB0'],
+ help='Virtual COM port created by cp201x for Schwinn 810 GPS watch')
+ parser.add_argument('--hook', nargs=1, default = [ None ],
+ help='Callback upon track extraction')
+ parser.add_argument('--dir', nargs=1, dest='dir',
+ default=['.'],
+ help='Where to store data')
+ parser.add_argument('--debug', action='store_true',
+ help='Dump all replies in a binary form into a single file schwinn810.bin in TEMP dir')
+ parser.add_argument('--delete', dest='delete', action='store_true',
+ help='Delete all data from watches after download?')
+ # parser.add_argument('--add-year', dest='add_year', action='store_true',
+ # help='Creates subfolder in dir named after the current year')
+ # parser.add_argument('--add-id', dest='add_id', action='store_true',
+ # help='Creates subfolder with device id inside dir but before year')
+
+ args = parser.parse_args()
+
+ d = Device(args.port[0])
+ d.debug = args.debug
try:
- port = serial.Serial(args.port[0], 115200, timeout=1, writeTimeout=1)
- except serial.SerialException as e:
- print("Port can't be opened :(", file=sys.stderr)
+ w = Writer(args.dir[0], args.hook[0])
+ d.read(w, TextProgress())
+ d.close()
+ except SerialException as e:
+ _log.fatal("Port can't be opened :(")
exit(-1)
- port.write(connect)
-else:
- print("Parsing existing dump in {:s}".format(args.port[0]))
- port = open(args.port[0], "rb")
-
-raw = port.read(0x24)
-if len(raw)==0:
- print("Did you plug your watches? Check the clip!", file=sys.stderr)
- exit(-1)
-
-if args.debug:
- dump = open(os.path.join(tempfile.gettempdir(), "schwinn810.bin"), mode="wb")
- dump.write(raw)
-
-(ee, e1, e2, e3, bcd1, bcd2, bcd3, serial, v1, v2, check1, check2) = struct.unpack("sBBBBBB6s6s7s2x2I", raw)
-if check1 != check2:
- print("0x{:X} != 0x{:X}".format(check1, check2))
-# port.close()
-# print("We are reading something wrong. Are you trying to communicate with another device?", file=sys.stderr)
-# exit(-1)
-id = "{0:s} {1:s} {2:s} {3:02d} {4:02d} {5:02d} {6:02d} {7:02d} {8:02d}" \
- .format(serial.decode('ascii'), v1.decode('ascii'), ee.decode('ascii'), e1, e2, e3, unpackBCD(bcd1), unpackBCD(bcd2), unpackBCD(bcd3))
-print("Found %s" % id)
-if not reg: port.write(download)
-raw = port.read(0x24)
-
-if args.debug:
- dump.write(raw)
-
-tracks_with_points = 0
-
-(tracks, waypoints) = struct.unpack("=28xHH4x", raw)
-print("We've got %d tracks and %d waypoints to download" % (tracks, waypoints))
-for track in range(tracks):
- raw = port.read(0x24)
-
- if args.debug:
- dump.write(raw)
-
- (sec1,min1, hr1, dd1, mm1, yr1, laps, \
- speed_max, speed, hrm, x4, pts, \
- x5, name0, ahr, min2, hr2, dd2, mm2, yr2) = \
- struct.unpack(">6BH HHBBH H7sB5B5x", raw)
- tracks_with_points += (pts>0)
- start="20%02d-%02d-%02d %d:%02d:%02d" % (yr1, mm1,dd1, hr1, min1, sec1)
- end="20%02d-%02d-%02d %d:%02d" % (yr2, mm2, dd2, hr2, min2)
- track_name = name0.decode('ascii')
- print("There are %d laps in %s" % (laps-1, track_name))
- name = os.path.join(args.dir[0], track_name)
- trkFile = open('%s.track' % name, "wb", **open_extra)
- trkWriter = csv.writer(trkFile)
- trkWriter.writerow(["Start", "End", "Laps","MaxHeart","Heart","MaxSpeed","Speed","x4","x5","Points","Track"])
- trkWriter.writerow([start, end, laps-1, hrm, ahr, speed_max/10., speed/10., x4, x5, pts, track_name])
- lapFile = open('%s.laps' % name, "wb", **open_extra)
- lapWriter = csv.writer(lapFile)
- lapWriter.writerow(["Time", "Speed", "Lap", "Distance", "kcal", "MaxSpeed", \
- "autolap", "Beats", "sec", "MaxHeart", "MinHeart", \
- "InZone", "y4", "Elevation", "y8", "Track"])
- for theLap in range(1,laps):
- raw = port.read(0x24)
-
- if args.debug:
- dump.write(raw)
-
- (hh, mm, ss, sss, dist, \
- cal, speed_max, speed, \
- autolap, beats_high, beats_mid, beats_low, sec,heart_max,heart_min, \
- iz_h, iz_m, iz_s, y4, z_low, z_mid, z_high, y8, \
- lap, track, sign) = \
- struct.unpack(">4BI IxBxB 4BH2B 3BB3BB HBB", raw)
- z = z_low | z_mid << 8 | z_high << 16
- time=hh*3600 + mm*60 + ss + sss/100.
- iz = iz_h*3600 + iz_m*60 + iz_s
- lapWriter.writerow([time, speed/10., lap, dist/1e5, cal/1e4, speed_max/10., autolap, \
- beats_high*65536 + beats_mid*256 + beats_low, \
- sec, heart_max, heart_min, iz, y4, z/1e2, y8, track_name])
- lapFile.close()
- trkFile.close()
-
-# now all track points
-# for tracks containing them only!
-for track in range(tracks_with_points):
- raw = port.read(0x24)
-
- if args.debug:
- dump.write(raw)
-
- (min1, hr1, dd1, mm1, yr1, laps, hrm, pts, name0, ahr, min2, hr2, dd2, mm2, yr2) = \
- struct.unpack(">x5BH4xBxHxx7sB5B5x", raw)
- track_name = name0.decode('ascii')
- print("Fetching %d points from %s" % (pts, track_name))
- name = os.path.join(args.dir[0], track_name)
- ptsFile = open('%s.points' % name, "wb", **open_extra)
- ptsWriter = csv.writer(ptsFile)
- ptsWriter.writerow(["Distance", "Speed", "Time", "Heart","x1","InZone","Latitude","Longitude","kcal","Elevation","No","Track"])
-# while pts:
- for thePoint in range(pts):
- raw = port.read(0x24)
-
- if args.debug:
- dump.write(raw)
-
- (dist, speed0, sec, min, hr, x1, hrm0, izhh, izss, izmm, lat0, lon0, cal, \
- z_low, z_high, pt0, trk, sign) = \
- struct.unpack("=I2s3BB2s3B5s6sH HB3sBB", raw)
- time = "20{:02d}-{:d}-{:d} {:02d}:{:02d}:{:02d}".format(yr1, mm1, dd1, hr, min, sec)
- (hrm) = struct.unpack(">H",hrm0)[0]/5.
- (speed) = struct.unpack(">H",speed0)[0]#/150
- (pt_high, pt_low) = struct.unpack(">BH",pt0)
- pt = pt_low | pt_high << 16
-# (inzone,) = struct.unpack(">H",inzone0)
- lat = pack_coord(b"\x00" + lat0, b'S')
- lon = pack_coord(lon0, b'W')
- inzone = "{:d}:{:d}:{:d}".format(izhh, izmm, izss) # FIXME replace with plain seconds
- ptsWriter.writerow([dist/1e5, speed/160., time, hrm, x1, inzone, lat, lon, cal, \
- (z_low | z_high << 16)/100., \
- pt, track_name])
- if pt % 100 == 0:
- print("{:.0f}%".format(100.*pt/pts))
-# pts = pts - 1
- ptsFile.close()
- if args.hook:
- print("Calling {:s}\n\twith {:s}".format(args.hook[0], name))
- subprocess.Popen([args.hook[0], name])
-
-# Waypoints
-# TODO make use of waypoints count
-name = os.path.join(args.dir[0], "waypoints.csv")
-wptWriter = csv.writer(open(name, "wb", **open_extra))
-wptWriter.writerow(["Timestamp", "Name", "Latitude", "Longitude","x1","x2","Elevation","No"])
-while True:
- raw = port.read(0x24)
- if args.debug:
- dump.write(raw)
- if raw[0:2] == b"\xee\xee":
- break
- (yy, mm, dd, hh, mn, ss, name0, lat0, lon0, x1,x2, z, num, sign) = \
- struct.unpack("=6B6s3x5s6s2BH4xBB", raw)
- time = "{:2d}/{:2d}/{:2d} {:2d}:{:02d}:{:02d}".format(mm,dd,yy,hh,mn,ss)
- name = name0.decode('ascii')
- lat = pack_coord(b"\x00" + lat0, b'S')
- lon = pack_coord(lon0, b'W')
- wptWriter.writerow([time, name, lat, lon, x1, x2, z/100., num])
-
-if not reg and args.delete:
- if args.debug:
- port.write(delete)
- else:
- print("--debug is required for deletion for now", file=sys.stderr)
-
-if args.debug:
- dump.close()
-if not reg: port.write(disconnect)
-port.close()
+ print("Done")
-print("Done")
+if __name__ == '__main__':
+ # FORMAT = FORMAT = '%(asctime)-15s %(message)s'
+ # logging.basicConfig(level=logging.DEBUG, format=FORMAT)
+ logging.basicConfig(level=logging.DEBUG)
+ main()
View
12 src/progress_text.py
@@ -0,0 +1,12 @@
+""" Simple text based progress indicator """
+
+from __future__ import print_function
+
+class TextProgress:
+
+ def track(self, at, end):
+ print("Fetching track {:d}/{:d}".format(at, end))
+
+ def point(self, at, end):
+ if at % 100 == 0:
+ print("{:.0f}%".format(100.*at/end))
View
22 src/reader.py
@@ -0,0 +1,22 @@
+""" Base reader class for data decoders """
+
+# import logging
+# _log = logging.getLogger(__name__)
+
+class NotConnected(Exception): pass
+class BadSignature(Exception): pass
+
+class Reader(object):
+
+ def __init__(self, port, dump=None):
+ self.port = port
+ self.dump = dump
+
+ def read(self, amount):
+ raw = self.port.read(amount)
+ if self.dump:
+ self.dump.write(raw)
+ return raw
+
+if __name__ == '__main__':
+ pass
View
88 src/reader_cresta.py
@@ -0,0 +1,88 @@
+""" Device class to communicate with Schwinn 810 """
+
+from __future__ import print_function
+import struct
+from utils import *
+import logging
+from datetime import datetime, time, date
+from reader import *
+
+_log = logging.getLogger(__name__)
+
+class CrestaReader(Reader):
+
+ def __init__(self, port, dump=None):
+ super(CrestaReader, self).__init__(port, dump)
+
+ def read_summary(self):
+ raw = self.read(0x20)
+ (tracks,) = struct.unpack("=28xH2x", raw)
+ return (tracks, 0)
+
+ def read_track(self):
+ raw = self.read(0x20)
+ track = {}
+ (sec1,min1, hr1, dd1, mm1, yr1, lap_count, \
+ speed_max, speed, track['MaxHeart'], track['x4'], track['Points'], \
+ track['x5'], name0, track['Heart'], sign) = \
+ struct.unpack(">6BH HHBBH H7sB5xB", raw)
+ if 0xFD != sign:
+ raise BadSignature
+ track['Start'] = datetime(2000+yr1, mm1, dd1, hr1, min1, sec1)
+ track['Laps'] = lap_count - 1
+ track['Speed'] = speed/10.
+ track['MaxSpeed'] = speed_max/10.
+ track['Track'] = name0.decode('ascii')
+ return track
+
+ def read_lap(self):
+ raw = self.read(0x20)
+ lap = {}
+ (hh, mm, ss, sss, dist, \
+ cal, speed_max, speed, \
+ lap['autolap'], beats_high, beats_mid, beats_low, lap['sec'], lap['MaxHeart'], lap['MinHeart'], \
+ z0, \
+ lap['Lap'], track_number, sign) = \
+ struct.unpack(">4BI IxBxB 4BH2B 4s HBB", raw)
+ if 0xFD != sign:
+ raise BadSignature()
+ lap['Elevation'] = struct.unpack("<I", z0)[0]/1e2
+ lap['Time'] = hh*3600 + mm*60 + ss + sss/100.
+ lap['Speed'] = speed/10.
+ lap['Distance'] = dist/1e5
+ lap['kcal'] = cal/1e4
+ lap['MaxSpeed'] = speed_max/10.
+ lap['Beats'] = beats_high << 16 | beats_mid << 8 | beats_low
+ return lap
+
+ def read_points_summary(self):
+ raw = self.port.read(0x20)
+ (min1, hr1, dd1, mm1, yr1, lap_count, hrm, pts, name0, ahr, min2, hr2, dd2, mm2, yr2, sign) = \
+ struct.unpack(">x5BH4xBxHxx7sB5BB", raw)
+ if 0xFA != sign:
+ raise BadSignature
+ track_name = name0.decode('ascii')
+ return {'Track': track_name, 'Points': pts, 'Start': date(2000+yr1, mm1, dd1)}
+
+ def read_point(self):
+ raw = self.read(0x20)
+ point = {}
+ (dist0, speed, sec, min, hr, x1, hrm, izhh, izss, izmm, lat0, lon0, \
+ raw2, pt, trk, sign) = \
+ struct.unpack(">4sH3BBH3B5s6s 2sHBB", raw)
+ if 0xFA != sign:
+ raise BadSignature
+ point['kcal'] = struct.unpack("<H", raw2)[0]
+ point['Distance'] = struct.unpack("<I", dist0)[0] / 1e5
+ point['Speed'] =speed/160.9 # really?
+ point['Heart'] = hrm/5
+ point['No'] = pt
+ point['Time'] = time(hr, min, sec)
+ # time = "20{:02d}-{:d}-{:d} {:02d}:{:02d}:{:02d}".format(yr1, mm1, dd1, hr, min, sec)
+ point['Latitude'] = pack_coord(b"\x00" + lat0, b'S')
+ point['Longitude'] = pack_coord(lon0, b'W')
+ point['InZone'] = izhh*3600 + izmm*60 + izss
+ return point
+
+if __name__ == '__main__':
+ pass
View
103 src/reader_schwinn.py
@@ -0,0 +1,103 @@
+""" Device class to communicate with Schwinn 810 """
+
+from __future__ import print_function
+import struct
+from utils import *
+import logging
+from datetime import datetime, time, date
+from reader import *
+
+_log = logging.getLogger(__name__)
+
+class SchwinnReader(Reader):
+
+ def __init__(self, port, dump=None):
+ super(SchwinnReader, self).__init__(port, dump)
+
+ def read_summary(self):
+ raw = self.read(0x24)
+ return struct.unpack("=28xHH4x", raw)
+
+ def read_track(self):
+ raw = self.read(0x24)
+ track = {}
+ (sec1,min1, hr1, dd1, mm1, yr1, lap_count, \
+ speed_max, speed, track['MaxHeart'], track['x4'], track['Points'], \
+ track['x5'], name0, track['Heart'], min2, hr2, dd2, mm2, yr2, sign) = \
+ struct.unpack(">6BH HHBBH H7sB5B4xB", raw)
+ if 0xFD != sign:
+ raise BadSignature
+ track['End'] = datetime(2000+yr2, mm2, dd2, hr2, min2)
+ track['Start'] = datetime(2000+yr1, mm1, dd1, hr1, min1, sec1)
+ track['Laps'] = lap_count - 1
+ track['Speed'] = speed/10.
+ track['MaxSpeed'] = speed_max/10.
+ track['Track'] = name0.decode('ascii')
+ return track
+
+ def read_lap(self):
+ raw = self.read(0x24)
+ lap = {}
+ (hh, mm, ss, sss, dist, \
+ cal, speed_max, speed, \
+ lap['autolap'], beats_high, beats_mid, beats_low, lap['sec'], lap['MaxHeart'], lap['MinHeart'], \
+ iz_h, iz_m, iz_s, lap['y4'], z0, \
+ lap['Lap'], track_number, sign) = \
+ struct.unpack(">4BI IxBxB 4BH2B 3BB4s HBB", raw)
+ if 0xFD != sign:
+ raise BadSignature()
+ lap['InZone'] = iz_h*3600 + iz_m*60 + iz_s
+ lap['Elevation'] = struct.unpack("<I", z0)[0]/1e2
+ lap['Time'] = hh*3600 + mm*60 + ss + sss/100.
+ lap['Speed'] = speed/10.
+ lap['Distance'] = dist/1e5
+ lap['kcal'] = cal/1e4
+ lap['MaxSpeed'] = speed_max/10.
+ lap['Beats'] = beats_high << 16 | beats_mid << 8 | beats_low
+ return lap
+
+ def read_points_summary(self):
+ raw = self.read(0x24)
+ (min1, hr1, dd1, mm1, yr1, lap_count, hrm, pts, name0, ahr, min2, hr2, dd2, mm2, yr2, sign) = \
+ struct.unpack(">x5BH4xBxHxx7sB5B4xB", raw)
+ if 0xFA != sign:
+ raise BadSignature
+ track_name = name0.decode('ascii')
+ return {'Track': track_name, 'Points': pts, 'Start': date(2000+yr1, mm1, dd1)}
+
+ def read_point(self):
+ raw = self.read(0x24)
+ point = {}
+ (dist0, speed, sec, min, hr, x1, hrm, izhh, izss, izmm, lat0, lon0, \
+ raw2, pt, trk, sign) = \
+ struct.unpack(">4sH3BBH3B5s6s 6sHBB", raw)
+ if 0xFA != sign:
+ raise BadSignature
+ (cal, z) = struct.unpack("<HI", raw2)
+ point['kcal'] = cal
+ point['Elevation'] = z / 1e2
+ point['Distance'] = struct.unpack("<I", dist0)[0] / 1e5
+ point['Speed'] =speed/160.9 # really?
+ point['Heart'] = hrm/5
+ point['No'] = pt
+ point['Time'] = time(hr, min, sec)
+ # time = "20{:02d}-{:d}-{:d} {:02d}:{:02d}:{:02d}".format(yr1, mm1, dd1, hr, min, sec)
+ point['Latitude'] = pack_coord(b"\x00" + lat0, b'S')
+ point['Longitude'] = pack_coord(lon0, b'W')
+ point['InZone'] = izhh*3600 + izmm*60 + izss
+ return point
+
+ def read_waypoint(self):
+ raw = self.read(0x24)
+ wpt = {}
+ (yy, mm, dd, hh, mn, ss, name0, lat0, lon0, x1,x2, z, num, sign) = \
+ struct.unpack("=6B6s3x5s6s2BH4xBB", raw)
+ wpt['Time'] = datetime(2000+yy, mm, dd, hh, mn, ss)
+ wpt['Name'] = name0.decode('ascii')
+ wpt['Latitude'] = pack_coord(b"\x00" + lat0, b'S')
+ wpt['Longitude'] = pack_coord(lon0, b'W')
+ wpt['Elevation'] = z/1e2
+ return wpt
+
+if __name__ == '__main__':
+ pass
View
21 src/utils.py
@@ -0,0 +1,21 @@
+import struct
+
+def unpack_bcd(x0):
+ x00 = x0
+ x = 0
+ m = 0
+ while x0:
+ digit = x0 & 0x000f
+ if digit>9:
+ raise Exception("Non-BCD argument {:X} in {:X}".format(digit, x00))
+ x = x + digit * 10**m
+ m = m + 1
+ x0 = x0 >> 4
+ return x
+
+def pack_coord(c0, hemi):
+ c1 = struct.unpack(">HBHs", c0)
+ c2=[unpack_bcd(x) for x in c1[0:3]]
+ c = c2[0] + (c2[1]+c2[2]/10000.)/60.
+ if (c1[3] == hemi) : c = -c
+ return c
View
87 src/writer.py
@@ -0,0 +1,87 @@
+""" Simple CSV writer for Schwinn 810 """
+
+from __future__ import print_function
+import sys,os
+import subprocess
+import csv
+import logging
+
+_log = logging.getLogger(__name__)
+
+open_extra = dict();
+if sys.version_info >= (3,0):
+ open_extra["newline"] = ''
+
+class Writer:
+ """ Default files writer """
+
+ def __init__(self, dir, hook=None):
+ self.dir = dir
+ self.hook = hook
+
+ self.ptsWriters = {}
+ self.pts_count = {}
+
+ name = os.path.join(self.dir, "waypoints.csv")
+ wptFile = open(name, "wb", **open_extra)
+ waypoint_keys = ["Timestamp", "Name", "Latitude", "Longitude","x1","x2","Elevation","No"]
+ self.wptWriter = csv.DictWriter(wptFile, waypoint_keys)
+ self.wptWriter.writeheader()
+
+ def add_track(self, track):
+ """ Append track to database """
+ name = os.path.join(self.dir, track['Track'])
+ trkFile = open('%s.track' % name, "wb", **open_extra)
+ track_keys = ["Start", "End", "Laps", "MaxHeart", "Heart", "MaxSpeed", \
+ "Speed", "x4", "x5", "Points", "Track"]
+ trkWriter = csv.DictWriter(trkFile, track_keys)
+ trkWriter.writeheader()
+ trkWriter.writerow(track)
+ trkFile.close()
+
+ lapFile = open('%s.laps' % name, "wb", **open_extra)
+ lap_keys = ["Time", "Speed", "Lap", "Distance", "kcal", "MaxSpeed", \
+ "autolap", "Beats", "sec", "MaxHeart", "MinHeart", \
+ "InZone", "y4", "Elevation", "Track"]
+ self.lapWriter = csv.DictWriter(lapFile, lap_keys)
+ self.lapWriter.writeheader()
+
+ point_keys = ["Distance", "Speed", "Time", "Heart", "x1", "InZone", \
+ "Latitude", "Longitude", "kcal", "Elevation", "No", "Track"]
+ if track['Points']>0:
+ ptsFile = open('%s.points' % name, "wb", **open_extra)
+ ptsWriter = csv.DictWriter(ptsFile, point_keys)
+ ptsWriter.writeheader()
+ name = track['Track']
+ self.ptsWriters[name] = ptsWriter
+ self.pts_count[name] = track['Points']
+
+ def add_lap(self, lap):
+ """ Append lap to a database """
+ self.lapWriter.writerow(lap)
+
+ def begin_points(self, track):
+ """ Points for track will follow """
+ self.track = track
+
+ def add_point(self, point):
+ """ Append point to a database """
+ self.ptsWriters[self.track].writerow(point)
+
+ def add_waypoint(self, wp):
+ """ Append point to a database """
+ self.wptWriter.writerow(wp)
+
+ def commit(self):
+ """ Commit last track in a database """
+ if self.hook:
+ name = os.path.join(self.dir, self.track)
+ _log.info("Calling %s\n\twith %s", self.hook, name)
+ subprocess.Popen([self.hook, name])
+
+ def __del__(self):
+ # self.lapFile.close()
+ _log.info("Writer destroyed")
+
+if __name__ == '__main__':
+ pass

0 comments on commit f79d895

Please sign in to comment.