#!/usr/bin/python # tz_gps.py # --copyright-- Copyright 2007 (C) Tranzoa, Co. All rights reserved. Warranty: You're free and on your own here. This code is not necessarily up-to-date or of public quality. # --url-- http://www.tranzoa.net/tzpython/ # --email-- pycode is the name to send to. tranzoa.com is the place to send to. # --bodstamps-- # July 1, 2007 bar # July 2, 2007 bar 4 satellites # add who to headers # July 3, 2007 bar kml pid of route number, not number of routes # July 5, 2007 bar allow None for altitude (put out zero - thumbnail_htm.py might have pictures without altitudes) # fix a CDATA in GPX points # July 22, 2007 bar find_point_index_by_when # fix time parsing from .gpx file # July 23, 2007 bar faster gpx parsing (by regx) # July 24, 2007 bar able to put put time stamps in kml files # July 25, 2007 bar points_from_all_tracks # distance_from # July 28, 2007 bar tessellate the waypoints so they aren't underground # July 29, 2007 bar comment # August 5, 2007 bar todo comment # August 7, 2007 bar space after a time in a comment # August 16, 2007 bar subtract if the binary search is too high # August 18, 2007 bar change gpx indentation # August 21, 2007 bar time_description() # where_description() # October 12, 2007 bar move the degree/minute/second routines to latlon.py # October 20, 2007 bar allow time.time() to go in to gpx_kml_date_time_string() # zero collapsation # move file writing to here from tz_gh615 # October 22, 2007 bar fool with big points # October 25, 2007 bar average speed in big points # smooth points # elevation/altitude can be negative # October 27, 2007 bar put the source track names in the output tracks # October 28, 2007 bar a_track._clear() for tz_gh615 use # make_all_big_points() from hikes.py # November 1, 2007 bar use a_smoother instead of a_spliner # November 18, 2007 bar turn on doxygen # November 27, 2007 bar insert boilerplate copyright # December 12, 2007 bar a_track.color # December 17, 2007 bar heart rate in big points # don't assume points are include_point()'ed in time order # December 18, 2007 bar more big point fixes # December 19, 2007 bar fix big point, hikify logic # gpx crc optional on per-point basis # smooth keeps speed and heart rate in the picture # December 20, 2007 bar comment # December 21, 2007 bar read waypoints from gpx files in regex reader # detect all-waypoint tracks in gpx writer # get_points_near_untimed_points_in_tracks() # December 22, 2007 bar no, read each waypoint from a gpx file in to a separate track # December 23, 2007 bar use latlon x/y/z conversion routines to average the lat/lon in big points # December 29, 2007 bar set_tracks_points_to_know_track_point_idxes() # January 4, 2008 bar set the track pointer and index in points copied by points_from_all_tracks() # make_interpolated_time_point() # change the meaning of find_point_index_by_when() in boundary cases # January 5, 2008 bar don't override gpx file when values with durations. only use duration if there is no time # distance and altitude sum/info routines # January 8, 2008 bar dink with fix_points_speeds() # January 9, 2008 bar flesh out first_raw_point() and last_raw_point() for a_point and a_big_point # remember all the raw points that go in to a big point # January 12, 2008 bar allow altitude to be None # properly make an interpolated point with the right time when it's before or after all of the points # March 14, 2008 bar file string routines # March 15, 2008 bar fix goofed param # May 11, 2008 bar a_rectangle and a_circle # May 12, 2008 bar objectify the classes # May 13, 2008 bar get combine logic working to smooth tracks, if not to create map/nets # May 14, 2008 bar don't output tiny speeds to gpx points # May 16, 2008 bar read nmea files (ours, anyway) # May 17, 2008 bar email adr # May 18, 2008 bar more nmea (speed is apparently in knots, not kph, so output has changed) # May 20, 2008 bar combine option of map_them - and check 'em off by pairs, for gosh sakes # June 8, 2008 bar cmd line read nmea files, too # and generally get the cmd line up to date # June 16, 2008 bar cmd line, show distance unsmoothed # June 26, 2008 bar read raw lat/lon text files # remove_redundant_points # June 28, 2008 bar bike # June 29, 2008 bar move the xyz stuff in to a_point routines # July 2, 2008 bar promote a_nearby_point to global level # July 3, 2008 bar comment # take spaces out from inside the numbers in kml files # July 5, 2008 bar write label_points to gpx files # July 8, 2008 bar remove_single_label_tracks() # July 19, 2008 bar gpx trk elements (which I should be using) # able to smooth untimed points # August 28, 2008 bar NMEA GPGSA parse and output # August 29, 2008 bar basestring instead of StringType because of unicode strings and others # September 2, 2008 bar todo comment # September 4, 2008 bar fool with write file speed (doesn't help to "".join( [ ps ... ] ) the point strings together) # ended up tossing point string creation on per-track threads (kludge alert!) # September 7, 2008 bar set the unix atributes to doc.kml in kmz files # September 9, 2008 bar when sorting nearby points, handle equally nearby points # October 21, 2008 bar get rid of extra comma in altitude in kml # take antialias out of kml # write_kml_or_kmz_string_to_file() # October 27, 2008 bar just output the base file name, not the path, for gpx and km? files as file names # the summary/average main program can read multiple ambiguously named files, now # first cut at reading kml/kmz files # October 28, 2008 bar allow folder-less kmlz # and many other kmlz fixes # do altitude right in text file reading # able to read our own nmea files # allow a big altitude difference to satisfy hikify's minimum distance-from-start requirement ('cause i didn't notice the poo poo point tracks are already included. sheesh) # don't put out gpx labels if they are blank # October 30, 2008 bar read the nmea stuff from gpx files # accept and put out gpx floating altitude # November 5, 2008 bar put track description stuff out in addition to our boilerplate # cut off altitude gain/loss at 50 meters rather than at 5 # November 12, 2008 bar remove_untimed_points_from_tracks() # November 13, 2008 bar typo in name # November 14, 2008 bar clearer logic in combining points # November 15, 2008 bar egad! I've been making default params as [] and {} # find_hold_still_point_pairs() # November 16, 2008 bar kmlz : don't strip the last coord from a previous-location-point # November 24, 2008 bar named parms to write_gps_file # allow default distance to a_point_combiner # December 20, 2008 bar cdata kml folder name and description if needed # December 21, 2008 bar a_track has opacity value # December 27, 2008 bar make bikify min_distance be more than hikify # April 7, 2009 bar comment # April 30, 2009 bar flat option to get_nearby_tracks_points() # January 9, 2010 bar try to carry over track file names when possible # allow empty in gpx to not cause a file naming problem (allows empty file names) # May 14, 2010 bar only_nearby_points() # May 16, 2010 bar radian_angle_to() # July 10, 2010 bar print non-flat distance # print altitude information # August 10, 2010 bar note about direction from one point to another # September 12, 2010 bar .loc files # default point altitude to None # March 20, 2011 bar typos in old, combine code # merge_tracks_to_points() # April 17, 2011 bar fix altitude of zero problem of going off to cm.altitude # April 25, 2011 bar merge logic includes the ends from 1 track # May 21, 2011 bar set the durations for merged points # --start_time --end_time # July 11, 2011 bar append_track # July 14, 2011 bar don't do anything in merge_tracks_to_points if the merged tracks look like they are way out of sync # November 29, 2011 bar pyflake cleanup # --eodstamps-- ## \file # # # GPS stuff. # # # # Online GPX validator: http://www.fahrradspass.de/ # Online KML validator: http://www.kmlvalidator.org/ not http://feedvalidator.org/ # # # # GPX rtept's can have 'extensions', apparently. # http://www.topografix.com/GPX/1/1/#type_extensionsType # # NMEA format: # http://gpsinformation.org/dale/nmea.htm # # # Todo: # # --merge doesn't do sailing tracks well - straight lines for coming about # gps units find roughly the same track, but are shifted in time. When driving, it's dramatic. # # gpx output should probably be ...... # not ...... # or this should at least be optional. # # take a hint about so much of this file being out of control # massive re-org needed # # kml/kmz files: # detect waypoints and mark 'em # # bikify: # stop biking when the guy is going fast uphill # in fact, if there are hills, whack the whole track if the downhill speed is not much faster than the uphill # at > 20 mph, there really should not be a lot of twisty turnies? # # remove_redundant_points' 'ed' should compensate for back and forth rather than accumulating errors of absolute magnitude # yet another thought (as opposed to the "alternates" below): # do remove_redundant_points for cutoff / 8, cutoff / 4, cutoff / 2, cutoff # alternative to fix the problem with corners being rounded or cut off: # insert the intervening point that's the furthest from a line drawn between the two output points # try this: # don't assume that the previous line continues. Start anew at each output point. # alternative to fix the problem with corners being rounded or cut off: # insert the intervening points that are furthest from the new point and the current point whenever a new point is added # include the two points in the prospective "furthest" points, but don't dupe them, of course # use an_xyz_point.closest_point_on_line() # and an adaptive filter would be good. when the path is curving sharply, put more points in it for a smoother output curve # and time might be thrown in, too. like if the speed is real slow, then make the cutoff lower # # a_point.label and a_point.name and a_point._label (geonames cache logic) are out of control # # find "further afield" # # get the waypointness of points under control # consider a point a waypoint if it's in a 1-point track alone because untimed tracks can describe trails and streets and such. # and fix any code that puts a set of waypoints in one track or one array of points (tz_gh615.py ?) # # while working on map_them combination points, have map_tracks build tracks with extra point(s) on each end if possible (to try to discourage having a sub-track disconnect from the main track) # # add nmea data to a_big_point or do something with it # # do nmea $GPGSV # # bring these in to control a_point and ilk: # ._prev_p (ref to previous point - used in trodtrack) # .track (ref to the track the point came from) # .track_i (which is the same thing as _point_idx ? ) # ._track_idx (index in to the array of tracks of the track) # ._point_idx (index in to track.points of the point) # # points_flat_distance gets 4466 nautical miles for a smoothed great circle trip from mv to hiroshima, which latlon.calcDistance calculates to be 4830 apart (maybe we fly through the earth) # # %.6f causes trailing zeros in lat/lon values that are not inherently 6 digits of resolution (how do we know the resolution?) # # point description logic is a bit out of control (to_kml takes a description, but i added .description to a_point so a description could be added to gpx points) # # Timed KML/Z files come up in GE with the time cursors both on the left end, so the route doesn't show. # Apparently, that's GE and can't be fixed. # # The "a_point" has gotten a bit out of hand. # In fact this whole file is over the iterative-design edge. # Note the routines with 4 or more parameters. # Especially the ones that create km? files. # # smooth - when interpolating, lie to the spline logic by telling it all the points are equally spaced in X (time) # then interpolate the interpolation # this might help the ringing problem when two points are the same in lat or lon or altitude # Or, just do a different spline logic without the problem # # import copy import math import os import re import sys import threading import time import zipfile import latlon import replace_file import tzlib import tz_parse_time import tzspline TINY_SPEED = 0.000000000001 # a speed that stops automatic waypoint detection, but is effectively zero def fix_unsigned_int(i) : if i == None : return(None) if 60000 <= i <= 0x10000 : i = i - 0x10000 if 0x10000 < i : i = i - 0x100000000L # October 30, 2008 What is this fix? if i > 0 : i = 0 pass return(i) def _blank_or_float(f) : if f == None : return("") return("%.2f" % f) def gpx_kml_date_time_string(tm) : try : h = tm.tm_hour except AttributeError : tm = time.gmtime(tm) return("%04u-%02u-%02uT%02u:%02u:%02uZ" % ( tm.tm_year, tm.tm_mon, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec ) ) # this should put out fractional seconds !!!! nmea_flt_num_s = tzlib.float_regx_str nmea_hms_s = r"(\d?\d)(\d\d)(\d\d(?:\.\d*)?)" nmea_lat_lon_s = r"(\d+?)(\d\d(?:\.\d*)?)" nmea_dmy_s = r"(\d\d)(\d\d)(\d\d)" nmea_gpgga_re = re.compile(r"\$GPGGA\s*,\s*" + nmea_hms_s + r"\s*,\s*" + nmea_lat_lon_s + r"\s*,\s*(N|S)\s*,\s*" + nmea_lat_lon_s + r"\s*,\s*(E|W)\s*,\s*" + r"(\d+)" + r"\s*,\s*" + r"(\d+)" + r"\s*,\s*(" + nmea_flt_num_s + r"|\s*)\s*,\s*(" + nmea_flt_num_s + r"|\s*)\s*,\s*(.?)", re.DOTALL | re.IGNORECASE) nmea_gprmc_re = re.compile(r"\$GPRMC\s*,\s*" + nmea_hms_s + r"\s*,\s*[^V],\s*" + nmea_lat_lon_s + r"\s*,\s*(N|S)\s*,\s*" + nmea_lat_lon_s + r"\s*,\s*(E|W)\s*,\s*(" + nmea_flt_num_s + r")\s*,\s*(" + nmea_flt_num_s + r"|\s*)\s*,\s*" + nmea_dmy_s + r"\s*,", re.DOTALL | re.IGNORECASE) class a_gpgsa(object) : nmea_gpgsa_re = re.compile(r"\$GPGSA\s*,\s*([AM])\s*,\s*(\d+)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(\d*)\s*,\s*(" + nmea_flt_num_s + r")\s*,\s*(" + nmea_flt_num_s + r")\s*,\s*(" + nmea_flt_num_s + r")", re.DOTALL | re.IGNORECASE) @staticmethod def from_string(nmea_gpgsa_string) : if not nmea_gpgsa_string : return(None) g = a_gpgsa.nmea_gpgsa_re.search(nmea_gpgsa_string) if not g : return(None) sats = [ int(g.group(si) or "0") or None for si in xrange(3, 15) ] return(a_gpgsa(g.group(1), int(g.group(2)), sats, float(g.group(15)), float(g.group(16)), float(g.group(17)))) def __init__(me, am, fix, sats, pdop, hdop, vdop) : me.am = am # A or M (auto or manual) me.fix = fix # 1=none, 2=2D, 3=3D me.sats = sats # array[12] of integer satellite numbers or None me.pdop = pdop # dilution of precision (combo of hdop and vdop?) me.hdop = hdop me.vdop = vdop def __str__(me) : ss = "" for s in (me.sats + ([ None ] * 12))[0:12] : if s != None : ss += "%02d" % s ss += "," return("GPGSA,%s,%u,%s%s,%s,%s" % ( me.am or "A", me.fix or 0, ss, _blank_or_float(me.pdop), _blank_or_float(me.hdop), _blank_or_float(me.vdop) ) ) pass # a_gpgsa class a_point(object) : def clear(me) : me.name = "" me.label = "" me.description = "" me.when = 0.0 me.lat = 0.0 me.lon = 0.0 me.altitude = None me.gpgsa = None # information from NMEA $GPGSA me.fix_typ = 1 # information from NMEA $GPGGA me.sat_cnt = 0 me.hz_dispersion = None me.track_angle = None # information from NMEA $GPRMC me.typ = None me.duration = None # duration since the previous point me.speed = None # speed from previous point to this one me.heart_rate = None me.output_gpx_crc = True me.points = [ me ] me.label_point = None # a_nearby_point that has a (presumably) nearby location and a name (or label). me.info = {} def __init__(me, name = "", when = 0.0, lat = 0.0, lon = 0.0, altitude = None, typ = None, duration = None, speed = None, heart_rate = None, description = None, label = None, info = None, copied_point = None, gpgsa = None, fix_typ = 1, sat_cnt = 0, hz_dispersion = None, track_angle = None) : me.clear() cm = copied_point or me name = name or getattr(cm, 'name', "") or "" me.name = name + "" label = label or getattr(cm, 'label', "") or "" me.label = label + "" # a label of, perhaps, some point nearby, or this *is* a label point description = description or getattr(cm, 'description', "") or "" me.description = description + "" me.when = when + cm.when me.lat = lat + cm.lat me.lon = lon + cm.lon if altitude == None : altitude = cm.altitude me.altitude = altitude if typ != None : me.typ = typ or cm.typ else : me.typ = cm.typ if duration != None : me.duration = duration + 0.0 else : me.duration = cm.duration if speed != None : me.speed = speed + 0.0 else : me.speed = cm.speed if heart_rate != None : me.heart_rate = heart_rate + 0 else : me.heart_rate = cm.heart_rate me.gpgsa = gpgsa or None # !!!! make this a_gpgga along with the last 3 fields (hite above geoid, time in seconds since last DGPS fix, DGPS station number) me.fix_typ = fix_typ or 1 # assume GPS (( 0=invalid, 1=GPS fix (SPS), 2=DGPS fix, 3=PPS fix, 4=Real Time Kinematic, 5=Float RTK, 6=estimated (dead reckoning) (2.3 feature), 7=Manual input mode, 8=Simulation mode )) me.sat_cnt = sat_cnt or 0 me.hz_dispersion = hz_dispersion or None # !!!! make this a_gprmc me.track_angle = track_angle or None if info == None : info = cm.info if info == None : info = {} me.info = info def is_likely_waypoint(me) : """ Return whether this point is probably a waypoint. """ if me.duration or me.speed : return(False) return((me.typ != None) or me.name or (not me.when) or me.description) def flat_distance_from(me, p) : """ Return the distance from this point to the other point in natical miles without regard to altitude. """ return(latlon.calcDistance(me.lat, me.lon, p.lat, p.lon)) def flat_distance_from_lat_lon(me, lat, lon) : """ Return the distance from this point to the given locaion in natical miles without regard to altitude. """ latlon.calcDistance(me.lat, me.lon, lat, lon) def distance_from(me, p) : """ Return the distance from this point to the other point in natical miles. """ d = me.flat_distance_from(p) if (d < 10.0) and (me.altitude != None) and (p.altitude != None) : ad = (me.altitude - p.altitude) / latlon.metersPerNauticalMile d = math.sqrt((d * d) + (ad * ad)) return(d) def radian_angle_to(me, p) : """ Find the radian angle to the given point. Convert to degrees by dividing by latlon.rad (or multiplying by 180/math.pi). """ # Note: # From: http://instruct1.cit.cornell.edu/courses/ee476/FinalProjects/s2010/gp244_nva2_mrk99/gp244_nva2_mrk99/index.html # atan2(sin(dlon)*cos(lat2), (cos(lat1)*sin(lat2))-(sin(lat1)*cos(lat2)*cos(dlon))); # where lat1=current latitude, lat2=target latitude, and dlon=difference in current and target longitude. f = a_point(when = 1.0, lat = me.lat, lon = me.lon, altitude = me.altitude) t = a_point(when = 100001.0, lat = p.lat, lon = p.lon, altitude = p.altitude ) if 90.0 - f.lat < 0.5 : f.lat -= 0.5 t.lat -= 0.5 if 90.0 - t.lat < 0.5 : f.lat -= 0.5 t.lat -= 0.5 if 90.0 + f.lat < 0.5 : f.lat += 0.5 t.lat += 0.5 if 90.0 + t.lat < 0.5 : f.lat += 0.5 t.lat += 0.5 if 180.0 - f.lon < 0.5 : f.lon -= 0.5 t.lon -= 0.5 if 180.0 - t.lon < 0.5 : f.lon -= 0.5 t.lon -= 0.5 if 180.0 + f.lon < 0.5 : f.lon += 0.5 t.lon += 0.5 if 180.0 + t.lon < 0.5 : f.lon += 0.5 t.lon += 0.5 f.lat = latlon.lat_in_world_modulo(f.lat) t.lat = latlon.lat_in_world_modulo(t.lat) f.lon = latlon.lon_in_world_modulo(f.lon) t.lon = latlon.lon_in_world_modulo(t.lon) p = make_interpolated_time_point(f, t, 2.0) return(math.atan2(p.lat - f.lat, p.lon - f.lon)) def flat_between_points(me, p1, p2) : """ Return whether this point is probably somewhere between the two given points without regard to altitude. """ p1p2d = p1.flat_distance_from(p2) p1d = me.flat_distance_from(p1) if p1d < p1p2d : p2d = me.flat_distance_from(p2) if p2d < p1p2d : return(True) pass return(False) def between_points(me, p1, p2) : """ Return whether this point is probably somewhere between the two given points taking altitude in to effect. """ p1p2d = p1.distance_from(p2) p1d = me.distance_from(p1) if p1d < p1p2d : p2d = me.distance_from(p2) if p2d < p1p2d : return(True) pass return(False) def interpolate_point_value_by_when(me, val_func, p, w) : if (val_func(me) == None) or (val_func(p) == None) : return(None) m = float(w - me.when) / float(p.when - me.when) v = val_func(me) + ((val_func(p) - val_func(me)) * m) return(v) def xyz_point(me) : """ Return an XYZ point for this point. """ return(latlon.convert_lat_lon_to_xyz_point(me.lat, me.lon)) def x_y_z(me) : """ Return the X Y Z location of this point. """ return(latlon.convert_lat_lon_to_x_y_z(me.lat, me.lon)) def set_x_y_z(me, x, y, z) : """ Set this point's lat/lon from the given XYZ values. """ ( me.lat, me.lon ) = latlon.convert_x_y_z_to_lat_lon(x, y, z) def set_xyz_point(me, xyz_point) : """ Set this point's lat/lon from the given xyz point. (untested) """ me.set_x_y_z(xyz_point.x, xyz_point.y, xyz_point.z) def time_description(me) : s = "" if me.when : s = time.asctime(time.gmtime(me.when)) + " GMT" return(s) start_when = property(lambda me : me.when) # cheesy way to let points be used in places big points can be used end_when = property(lambda me : me.when) def where_description(me) : s = "" s += "%.6f" % ( me.lat ) s += " " s += "%.6f" % ( me.lon ) s += " --- " l = me.lat d = "N" if l < 0.0 : l = -l d = "S" dms = latlon.lat_lon_in_degrees_minutes_seconds(l) s += "%3u %u\' %6.3f\" %s" % ( dms[0], dms[1], dms[2], d ) s += " " l = me.lon d = "E" if l < 0.0 : l = -l d = "W" dms = latlon.lat_lon_in_degrees_minutes_seconds(l) s += "%3u %u\' %6.3f\" %s" % ( dms[0], dms[1], dms[2], d ) if me.altitude != None : s += " --- " s += "%i meters : %i feet" % ( me.altitude, me.altitude * latlon.feetPerMeter ) return(s) def __str__(me) : s = "" # # Note that the order the items are printed out affects the __cmp__ function! # if me.when : s += time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(me.when)) + " " s += "lat=%.6f " % ( me.lat ) s += "lon=%.6f " % ( me.lon ) if me.altitude != None : s += "alt=%i " % ( int(round(me.altitude)) ) if me.name : s += me.name + " " kys = me.info.keys() kys.sort() for k in kys : s += "%s=%s " % ( str(me.info[k]) ) if me.typ != None : s += "type=%s" % ( str(me.typ) ) if me.speed != None : s += "speed=%.2f " % ( me.speed ) if me.heart_rate != None : s += "heart=%u " % ( me.heart_rate ) s = s.strip() return(s) def __cmp__(me, other) : ms = str(me) so = str(other) return(cmp(ms, so)) kml_placemark = """ %s %s%s%s 1 %.6f,%6f%s """ kml_style = """ """ undoc = """ 0 """ def to_kml(me, is_waypoint = False, icon_image_url = None, description = None) : s = "" if (me.lat != None) and (me.lon != None) : alts = "" if me.altitude != None : alts = ",%u" % me.altitude if is_waypoint : s = "" crcs = "%08x" % ( tzlib.blkcrc32(0, str(me)) ) nam = tzlib.printable(me.name) ts = "" if me.when : ts = "\n %s" % ( gpx_kml_date_time_string(time.gmtime(me.when)) ) styleUrl = "" icon_image_url = icon_image_url or getattr(me, 'icon_image_url', None) if icon_image_url : icrc = tzlib.blkcrc32(0, icon_image_url) s += a_point.kml_style % ( icrc, icon_image_url ) styleUrl = "\n #%u" % ( icrc ) description = description or me.description or ("Waypoint " + nam) nam = nam or description description = tzlib.maybe_wrap_with_cdata(description) nam = tzlib.maybe_wrap_with_cdata(nam) s += a_point.kml_placemark % ( crcs, nam, description, ts, styleUrl, me.lon, me.lat, alts ) else : s = " %.6f,%.6f%s\n" % ( me.lon, me.lat, alts ) pass return(s) def speed_to_kml(me, is_waypoint = False, icon_image_url = None, description = None) : a = me.altitude me.altitude = int(me.speed * 100.0) ps = me.to_kml(is_waypoint = is_waypoint, icon_image_url = icon_image_url, description = description) me.altitude = a return(ps) def to_gpx(me, is_waypoint = False, icon_image_url = None, description = None) : s = "" if (me.lat != None) and (me.lon != None) : tag = "rtept" ind = " " if is_waypoint : tag = "wpt" ind = "" tms = "" if me.when : tm = time.gmtime(me.when) tms = "\n %s" % ( ind, gpx_kml_date_time_string(tm) ) ns = "" if me.name : ns = "\n %s%s%s%s" % ( ind, ind, ind, tzlib.maybe_wrap_with_cdata(tzlib.printable(str(me.name))) ) ds = "" description = description or me.description if description : ds = "\n %s%s%s%s" % ( ind, ind, ind, tzlib.maybe_wrap_with_cdata(str(description)) ) ts = "" if me.typ != None : ts = "\n %s%s%s%s" % ( ind, ind, ind, tzlib.maybe_wrap_with_cdata(str(me.typ)) ) es = "" if True : # create the extensions' string if not is_waypoint : es += ns + ds + ts ns = ds = ts = "" if me.output_gpx_crc : es += '\n %s%08x' % ( ind, tzlib.blkcrc32(0, str(me)) ) kys = me.info.keys() kys.sort() for k in kys : ks = re.sub(r"[^a-zA-Z0-9]", "", k) ks = re.sub(r"^[^a-zA-Z]", "", k) if not ks : ks = "oddinfotag" es += '\n %s<%s>%s' % ( ind, ks, tzlib.maybe_wrap_with_cdata(str(me.info[k])), ks ) if (me.speed != None) and (abs(me.speed) > TINY_SPEED) : es += '\n %s%.2f' % ( ind, me.speed ) if me.heart_rate : es += '\n %s%u' % ( ind, me.heart_rate ) if me.duration != None : es += '\n %s%.1f' % ( ind, me.duration ) info = me.make_nmea_info() nmeas = me.gpgga_string(info).strip() if nmeas and ((me.fix_typ != 1) or me.sat_cnt or (me.hz_dispersion != None)) : es += '\n %s%s' % ( ind, tzlib.maybe_wrap_with_cdata(nmeas) ) nmeas = me.gpgsa_string().strip() if nmeas : es += '\n %s%s' % ( ind, tzlib.maybe_wrap_with_cdata(nmeas) ) nmeas = me.gprmc_string(info).strip() if nmeas and (me.track_angle != None) : es += '\n %s%s' % ( ind, tzlib.maybe_wrap_with_cdata(nmeas) ) if me.label != None : try : ls = str(me.label).strip() if ls : # October 28, 2008 why was this not here? jeez, the label/name/etc are out of control. es += '\n %s' % ( ind, ls ) pass except UnicodeEncodeError : pass pass pass if es : sx = es es = '\n %s' % ( ind ) es += '\n %s' % ( ind ) es += sx es += '\n %s' % ( ind ) es += '\n %s' % ( ind ) pass eles = "%i" % ( me.altitude or 0 ) if me.altitude and (me.altitude != round(me.altitude)) : eles = "%.2f" % ( me.altitude ) s = """ %s<%s lat="%.6f" lon="%.6f"> %s%s%s%s%s%s%s %s """ % ( ind, tag, me.lat, me.lon, ind, eles, tms, ns, ds, ts, es, ind, tag ) pass return(s) def to_loc(me, is_waypoint = True, url = None, description = None) : s = "" if (me.lat != None) and (me.lon != None) : tag = "point" if is_waypoint : tag = "waypoint" ns = "" if me.name : ns = ' id="%s"' % tzlib.printable(str(me.name)) ds = tzlib.maybe_wrap_with_cdata(str(description or me.description or "")) or tzlib.maybe_wrap_with_cdata(tzlib.printable(str(me.name or ""))) als = "" if me.altitude != None : als = ' ele="%i"' % int(round(me.altitude)) # not in geocaching.com's loc files ts = "" if me.typ != None : ts = "\n %s" % ( tzlib.maybe_wrap_with_cdata(str(me.typ)) ) us = "" url = url or getattr(me, 'url', None) if url != None : us = '\n %s' % ( getattr(me, 'link_text', 'Details'), tzlib.maybe_wrap_with_cdata(str(url)) ) tms = "" if me.when : tm = time.gmtime(me.when) tms = "\n " % ( gpx_kml_date_time_string(tm) ) # not in geocaching.com's loc files s = """ %s %s%s%s """ % ( ns, ds, me.lat, me.lon, als, ts, us, tms, tag ) pass return(s) def gpgga_string(me, info) : """ Get the altitude out. """ if (not info) or (me.altitude == None) : return("") # $GPGGA,081821.000,4724.1826,N,12158.7843,W,1,04,4.2,235.8,M,-17.3,M,,0000*65 outs = "GPGGA,%02u%02u%02u.%02u,%02u%02u.%04u,%s,%03u%02u.%04u,%s,%u,%02u,%s,%.2f,M,,," % ( info.tm.tm_hour, info.tm.tm_min, info.tm.tm_sec, int(me.when * 100) % 100, info.lata[0], info.lata[1], info.lata[2], info.ns, info.lona[0], info.lona[1], info.lona[2], info.ew, me.fix_typ or 1, me.sat_cnt or 0, _blank_or_float(me.hz_dispersion), me.altitude or 0 ) return("$%s*%02X\n" % ( outs, tzlib.xorsum(outs) & 0xff )) def gpgsa_string(me) : # $GPGSA,A,3,05,04,12,02,,,,,,,,,4.3,4.2,1.0*32 if not me.gpgsa : return("") outs = str(me.gpgsa) return("$%s*%02X\n" % ( outs, tzlib.xorsum(outs) & 0xff )) def gprmc_string(me, info) : """ This gets the lat/lon, speed (knots, apparently) and GMT time. """ if not info : return(s) # $GPRMC,081818.000,A,4724.1837,N,12158.7845,W,0.26,335.82,250607,,*1D ss = "" if me.speed != None : ss = "%.2f" % ( ((me.speed or 0.0) * 1000.0) / latlon.metersPerNauticalMile ) outs = "GPRMC,%02u%02u%02u.%02u,A,%02u%02u.%04u,%s,%03u%02u.%04u,%s,%s,%s,%02u%02u%02u,," % ( info.tm.tm_hour, info.tm.tm_min, info.tm.tm_sec, int(me.when * 100) % 100, info.lata[0], info.lata[1], info.lata[2], info.ns, info.lona[0], info.lona[1], info.lona[2], info.ew, ss, _blank_or_float(me.track_angle), info.tm.tm_mday, info.tm.tm_mon, info.tm.tm_year % 100 ) return("$%s*%02X\n" % ( outs, tzlib.xorsum(outs) & 0xff )) class a_nmea_info(object) : def __init__(me, p) : lat = p.lat me.ns = 'N' if lat < 0.0 : lat = -lat me.ns = 'S' lon = p.lon me.ew = 'E' if lon < 0.0 : lon = -lon me.ew = 'W' me.lata = latlon.lat_lon_in_degrees_minutes_10_thousandth_minutes(lat) me.lona = latlon.lat_lon_in_degrees_minutes_10_thousandth_minutes(lon) w = p.when if w : w += 0.0005 else : w = 0.0 me.tm = time.gmtime(w) pass # a_nmea_info def make_nmea_info(me) : if (me.lat == None) or (me.lon == None) or (me.when == None) : return(None) return(me.a_nmea_info(me)) def to_nmea(me, is_waypoint = False, icon_image_url = None, description = None) : s = "" info = me.make_nmea_info() s += me.gpgga_string(info) s += me.gpgsa_string() s += me.gprmc_string(info) return(s) def first_raw_point(me) : return(me) def last_raw_point(me) : return(me) pass # a_point def cmp_points_by_when(a, b) : return(cmp(a.when, b.when)) def do_points_distance(points, rtn) : """ Return the sum of all the nautical mile distance values for an array of points. """ if not points : return(0.0) sm = 0.0 pp = points[0] for pi in xrange(1, len(points)) : p = points[pi] sm += rtn(p, pp) pp = p return(sm) def points_distance(points) : """ Return the total points' nautical mile a_point.distance_from()'s. """ return(do_points_distance(points, a_point.distance_from)) def points_flat_distance(points) : """ Return the total points' nautical mile a_point.flat_distance_from()'s. """ return(do_points_distance(points, a_point.flat_distance_from)) def points_sparse_flat_distance(points) : """ Return the total points' nautical mile a_point.flat_distance_from()'s that you really want. """ return(points_flat_distance(remove_redundant_points(list(points)))) REAL_ALTITUDE_DIFF = 50.0 # this many meters to really count for an altitude change def points_altitude_info(points, min_distance = None) : """ Return None or information about the points' altitude meters lost, gained, high and low extremes. """ if not points : return(None) min_distance = min_distance or REAL_ALTITUDE_DIFF class an_altitude_info(object) : def __init__(me) : me.lowest = 10000000000000000000000000.0 me.highest = -me.lowest me.lost = 0.0 me.gained = 0.0 def merge_in(me, om) : if om : me.lowest = min(me.lowest, om.lowest) me.highest = max(me.highest, om.highest) me.lost += om.lost me.gained += om.gained pass def __str__(me) : return("low=%.0f high=%.0f lost=%.0f gain=%.0f" % ( me.lowest, me.highest, me.lost, me.gained ) ) pass info = an_altitude_info() pp = None for p in points : alt = p.altitude if alt != None : info.lowest = min(info.lowest, p.altitude) info.highest = max(info.highest, p.altitude) if not pp : pp = p elif abs(alt - pp.altitude) >= REAL_ALTITUDE_DIFF : if alt >= pp.altitude : info.gained += ( alt - pp.altitude) else : info.lost += (-alt + pp.altitude) pp = p pass else : pp = None pass if info.lowest >= info.highest : return(None) return(info) def probable_max_distance_between_routes(points1, points2) : """ Find the maximum distance from each other the two tracks are. Sort of. """ if (not points1) or (not points2) : return(0.0) points1 = list(points1) points1 = remove_redundant_points(points1) d = points_flat_distance(points1) / 2.0 hd = 0.0 pp = fp = points1[0] for p in points1[1:] : hd += p.flat_distance_from(pp) if hd >= d : fp = p break # find the half-way point in distance pp = p return(find_closest_point(fp, points2).flat_distance_from(fp)) def make_interpolated_time_point(fp, tp, when) : """ Create a point interpolated by time between the two given points. If the time is before 'fp' or after 'tp', then the returned point will be a dupe of 'fp' or 'tp', respectively. """ if fp.when > tp.when : ( fp, tp ) = ( tp, fp ) if when >= tp.when : return(a_point(when = when, copied_point = tp)) if when <= fp.when : return(a_point(when = when, copied_point = fp)) d = float((when - fp.when) / (tp.when - fp.when)) ( fx, fy, fz ) = fp.x_y_z() ( tx, ty, tz ) = tp.x_y_z() fx += ((tx - fx) * d) fy += ((ty - fy) * d) fz += ((tz - fz) * d) ( lat, lon ) = latlon.convert_x_y_z_to_lat_lon(fx, fy, fz) if (fp.altitude != None) and (tp.altitude != None) : alt = fp.altitude + ((tp.altitude - fp.altitude) * d) else : alt = None p = a_point(lat = lat, lon = lon, altitude = alt, when = when) return(p) def find_point_index_by_when(points_sorted_by_when, t) : """ Find the index of the point whose time is less than or equal to 't'. If the first point in 'points_sorted_by_when' is after 't', then return -1. """ if not points_sorted_by_when : return(None) # # Binary search 'em # i = 0 lo = 0 hi = len(points_sorted_by_when) while lo < hi : i = (lo + hi) / 2 if t > points_sorted_by_when[i].when : i += 1 lo = i else : hi = i pass i = min(i, len(points_sorted_by_when) - 1) if points_sorted_by_when[i].when > t : i -= 1 return(i) def find_closest_point(p, points, rtn = None) : """ Find the point that's closest to the given point. """ rtn = rtn or a_point.flat_distance_from bp = None bd = 100000000000000.0 for pp in points : d = rtn(pp, p) if bd > d : bd = d bp = pp pass return(bp) DEAD_POINT_MULT = 5.0 def smooth_points(points) : """ Given an array of points, smooth them out. """ if not points : return( [] ) lt = 0.0 dt = points[-1].when - points[0].when have_whens = False if dt > 0.0 : have_whens = True for pi in xrange(1, len(points) - 1) : if points[pi].when - points[pi - 1].when <= 0.0 : have_whens = False # es = "ouch %u %s %s %s %s %u %u" % ( pi, time.asctime(time.gmtime(points[pi - 1].when)), time.asctime(time.gmtime(points[pi].when)), str(points[pi - 1]), str(points[pi]), len(points), pi ) break d = points[pi + 1].when - points[pi - 1].when dt = min(dt, d) lt = max(lt, d) pass dt = max(dt, 1.0) """ The problem here is that a_spliner overshoots when two values are the same (or near the same) and their neighbors are much different. So, when each little step is spline-computed, a simple, one way walk that goes through some locations that change a lot, then a little, then a lot, causes the spline interpolation to overshoot the "little" changes. So the guy walks backward, even though he's always going forward. Which says that the spline logic is not the right thing to use here. Or that it must be limited somehow. For instance, if the points surrounding two points that are being interpolated between are both above or both below their neighbors (of the two interpolation points) below: keep the interpolation values within min(v[1], v[2]) and min(v[1] + (v[1] - v[0]) / 2, v[2] + (v[2] - v[3]) / 2) above: max(v[1], v[2]) and max(v[1] + (v[1] - v[0]) / 2, v[2] + (v[2] - v[3]) / 2) and if the shoulder points v[0] and v[3] are on opposite sides, limit the interpolation points to min(v[1], v[2]) and max(v[1], v[2]) So a_smoother is a hack at that. I'd guess that the official, major league way to do it is with http://en.wikipedia.org/wiki/Monotone_cubic_interpolation logic. """ # # # Construct points at each end of big points # # if False : pi = len(points) while pi > 0 : pi -= 1 p = points[pi] if p.start_when != p.end_when : if p.end_when != p.when : d = p.end_when - p.when np = a_point(when = p.end_when, lat = p.lat, lon = p.lon, altitude = p.altitude, typ = p.typ, speed = p.speed, duration = d, heart_rate = p.heart_rate, name = p.name, description = p.description) if (pi < len(points) - 1) and points[pi + 1].duration : points[pi + 1].duration -= d points.insert(pi + 1, np) if p.start_when != p.when : d = 0.0 if pi : d = p.start_when - points[pi - 1].end_when np = a_point(when = p.start_when, lat = p.lat, lon = p.lon, altitude = p.altitude, typ = p.typ, speed = p.speed, duration = d, heart_rate = p.heart_rate, name = p.name, description = p.description) if p.duration : p.duration -= p.end_when - np.when points.insert(pi, np) pass pass pass # # # Since some of the points can be long, long duration (big points) and they can be bordered by short duration points # to keep the spline logic from taking all that long, long time to zoom all the way to the moon and back, # we may need to lock in the location during those long, long points' times. # We'll add shoulder points around the long, long duration points. # The shoulder points will be at the same location as the long, long duration point. # # Another way of saying that restriction: # The slope of the interpolation line 'tween two points can't be of two signs if the shoulders are on opposite sides of their neighbors. # And the limit of the interpolation is the least distance on the other side of athe point its shoulder # # if False : if lt >= dt * DEAD_POINT_MULT : points = [ p for p in points ] pi = len(points) - 1 while pi > 0 : pi -= 1 while True : p = points[pi] pn = points[pi + 1] d = pn.when - p.when if d <= dt * DEAD_POINT_MULT : break w = pn.when - 2.0 # dt * 2.0 d = w - p.when np = a_point(when = w, lat = p.interpolate_point_value_by_when(lambda p : p.lat, pn, w), lon = p.interpolate_point_value_by_when(lambda p : p.lon, pn, w), altitude = p.interpolate_point_value_by_when(lambda p : p.altitude, pn, w), typ = p.typ, speed = p.speed, duration = d, heart_rate = p.heart_rate, name = p.name, description = p.description ) if pn.duration : pn.duration -= d points.insert(pi + 1, np) pass pass pass # print len(points) # points = points[6950:7020] # return( [ copy.copy(p) for p in points ] ) d = int(max(dt / 8.0, 1.0)) if have_whens : sa = [ int(t) for t in xrange(int(points[0].when), int(points[-1].when) + d, d) ] # make array of 1 second ticks or 1/8th of the shortest point's duration, whichever is longest ta = [ p.when for p in points ] else : sa = [ float(i) for i in xrange(len(points)) ] ta = [ float(i) for i in xrange(len(points)) ] if True : ctc = [ p.x_y_z() for p in points ] a = [ xyz[0] for xyz in ctc ] ax = tzspline.a_smoother(ta, a)(sa) a = [ xyz[1] for xyz in ctc ] ay = tzspline.a_smoother(ta, a)(sa) a = [ xyz[2] for xyz in ctc ] az = tzspline.a_smoother(ta, a)(sa) lls = [ latlon.convert_x_y_z_to_lat_lon(ax[i], ay[i], az[i]) for i in xrange(len(ax)) ] aa = [ ll[0] for ll in lls ] oa = [ ll[1] for ll in lls ] else : # !!!! this does not work going over the poles or around 180 degrees a = [ p.lat for p in points ] aa = tzspline.a_smoother(ta, a)(sa) a = [ p.lon for p in points ] oa = tzspline.a_smoother(ta, a)(sa) a = [ p.altitude for p in points ] try : a.index(None) la = [ None ] * len(sa) except ValueError : la = tzspline.a_smoother(ta, a)(sa) a = [ p.speed for p in points ] try : a.index(None) pa = [ None ] * len(sa) except ValueError : pa = tzspline.a_smoother(ta, a)(sa) a = [ p.heart_rate for p in points ] try : a.index(None) ha = [ None ] * len(sa) except ValueError : ha = tzspline.a_smoother(ta, a)(sa) sm = [ a_point(when = sa[i], lat = aa[i], lon = oa[i], altitude = la[i], speed = pa[i], heart_rate = ha[i]) for i in xrange(len(sa)) ] sm[0] = a_point(copied_point = points[ 0]) # note: probably unnecessary sm[-1] = a_point(copied_point = points[-1]) # this gets the starting and ending point to be the original start/end points so that they will match up with other start/ends, even if they don't match up with other points in smoothed tracks return(sm) def remove_redundant_points(points, cutoff_distance = None) : """ Given an array of points, remove the points that are on a path between the two points before and after the removed point. If the middle point is nearer to each of its two neighbors than the two neighbors are to each other and if it's within cutoff_distance from the xyz line between the neighbors, delete the middle point. """ if not cutoff_distance : cutoff_distance = 5.0 / latlon.metersPerNauticalMile # default to 5 meter smoothing if he didn't spec any cutoff if points and (len(points) > 2) : ed = 0.0 for pi in xrange(len(points) - 3, -1, -1) : dl = False pp = points[pi + 2] p = points[pi + 1] np = points[pi] if p.flat_between_points(np, pp) : # only delete points that are between the two neighbors ppxyz = pp.xyz_point() pxyz = p.xyz_point() npxyz = np.xyz_point() ed += pxyz.distance_from_line(ppxyz, npxyz) * latlon.nauticalEarthRadius if ed < cutoff_distance : dl = True pass if dl : if (pp.duration != None) and (p.duration != None) : pp.duration += p.duration del(points[pi + 1]) else : ed = 0.0 if (not pp.name) and p.name : pp.name = p.name if (not pp.label) and p.label : pp.label = p.label pass pass return(points) def fix_points_speeds(points, force = False) : """ If all these points have zero speeds or if any of the points have speed == None, set them to values computed from the locations as best we can without getting too fancy. Return whether the speeds are all ok. """ retval = True sc = 0.0 for p in points : if p.speed : sc = None break pass for pi in xrange(1, len(points)) : p = points[pi] if (p.speed == sc) or (p.speed == None) or force : pp = points[pi - 1] d = pp.flat_distance_from(p) d *= (latlon.metersPerNauticalMile / 1000.0) td = (p.when - pp.when) / 3600.0 if td : p.speed = d / td else : retval = False pass pass if points : p = points[0] if (p.speed == sc) or (p.speed == None) or force : p.speed = 0.0 pass return(retval) def set_no_gpx_crcs(points) : for p in points : p.output_gpx_crc = False pass def set_no_whens(points) : for p in points : try : p.start_when = 0.0 except AttributeError : pass try : p.end_when = 0.0 except AttributeError : pass p.when_tot = 0.0 p.when = 0.0 p.speed = TINY_SPEED # non-zero so that the automatic waypoint detection doesn't see the point as a waypoint ( !!!! set a flag or something in the point) p.heart_rate = p.duration = None pass def set_equal_durations(points, when = 1.0, duration = 1.0) : for p in points : try : p.start_when = when except AttributeError : pass try : p.end_when = when except AttributeError : pass try : p.when_tot = when / p.cnt except AttributeError : pass except ZeroDivisionError : pass p.when = when p.duration = duration when += duration pass def fix_points_whens(points, when = None) : """ Try to give the points a reasonable 'when' value using duration information if there is any. """ retval = True if points : if not when : when = points[0].when if not when : retval = False if not points[0].when : points[0].when = when when = points[0].when pi = 1 while pi < len(points) : p = points[pi] if not p.when : if p.duration != None : if when : p.when = when + p.duration pass else : retval = False pass when = p.when pi += 1 pass return(retval) def fix_points_duration(points) : """ Fix any ungiven durations from the points' 'when' values if there are 'when' values. """ for pi in xrange(1, len(points)) : p = points[pi] if p.duration == None : pp = points[pi - 1] if p.when or pp.when : p.duration = p.when - pp.when if p.duration < 0 : p.duration = None pass pass pass pass def set_points_duration(points) : """ Set durations from the points' 'when' values if there are 'when' values. """ for p in points : p.duration = None fix_points_duration(points) def make_no_duplicate_points(points, ids = None) : ids = ids or {} for pi in xrange(len(points)) : p = points[pi] if ids.has_key(id(p)) : p = copy.copy(p) # if a point points back to tracks or to other points, which happens, then this must be 'copy', not 'deepcopy'. And, combined points' prev's and next's are broken try : del(p.prevs) except AttributeError : pass try : del(p.nexts) except AttributeError : pass points[pi] = p ids[id(p)] = True pass def are_all_points_waypoints(points) : if not points : return(False) for p in points : if not p.is_likely_waypoint() : return(False) return(True) # # # This page has stuff about atom: in kml files (for putting author and link stuff in. # # http://code.google.com/apis/kml/documentation/kmlSearch.html # # def kml_header(program_name = "", file_name = "", who = None, when = None, url = None) : program_name = program_name or "tz_gps.py" file_name = file_name or "" who = who or "" if who : who = " for " + who when = when or time.time() if url : name = url else : name = tzlib.safe_html(os.path.split(file_name)[1]) s = """ %s 0 %s KML created by %s %s%s """ % ( name, tzlib.safe_html(file_name), program_name, time.asctime(time.localtime(when)), who ) return(s) # # If this is set, user must reset it thru properties to see/control the inside stuff # # # KML_COLOR_GREEN = 0xff0000 KML_COLOR_BLUE = 0x00ff00 KML_COLOR_RED = 0x0000ff KML_ALPHA_BLEND = 0x7f kml_linestring = """ 1 """ kml_hover_linestring = """ 0 relativeToGround """ def kml_when_strings(when, until) : ts = "" tos = "" if when : tw = time.gmtime(when) ts = "\n %s" % ( gpx_kml_date_time_string(tw) ) else : when = time.time() tw = time.gmtime(when) if until and (until >= when) : tu = time.gmtime(until) ts = "\n %s%s" % ( gpx_kml_date_time_string(tw), gpx_kml_date_time_string(tu) ) tos = " to %s" % ( time.asctime(time.localtime(until)) ) return( ( ts, tos ) ) def kml_folder_placemark_header(route_number = None, when = None, until = None, hover = False) : if route_number == None : route_number = 1 ( ts, tos ) = kml_when_strings(when, until) hs = kml_linestring if hover : hs = kml_hover_linestring s = """ %s #ls%u%s """ % ( ts, route_number, hs ) return(s) def kml_route_header(route_number = None, number_of_routes = None, name = None, when = None, until = None, color = None, opacity = None, in_folder = False, description = None) : description = description or "" if description : description = description.strip(" ") + " " if route_number == None : route_number = 1 if number_of_routes == None : number_of_routes = route_number if name == None : name = "%u" % ( route_number ) name = str(name) if color == None : ( red, blue, green ) = nice_track_color(number_of_routes, route_number - 1) else : green = (int(color) >> 16) & 0xff blue = (int(color) >> 8) & 0xff red = int(color) & 0xff if opacity == None : alpha = KML_ALPHA_BLEND else : alpha = int(round((min(255, max(0, opacity)) * 255.0) / a_track.OPAQUE)) # # # Note: # absolute will shove things underground often, so stick to # # ( ts, tos ) = kml_when_strings(when, until) if not in_folder : tag = "Placemark" ifs = kml_linestring else : tag = "Folder" ifs = "" d = "%s%s%s" % ( description, time.asctime(time.localtime(when)), tos ) s = """ <%s id="pid%u"> %s %s%s %s """ % ( tag, route_number, tzlib.maybe_wrap_with_cdata(name), tzlib.maybe_wrap_with_cdata(d), ts, route_number, alpha, green, blue, red, ifs ) # root://styleMaps#default+nicon=0x467+hicon=0x477 return(s) def kml_route_trailer() : return(" \n \n \n") def kml_folder_trailer() : return(" \n") def kml_trailer() : return(" \n\n") def kml_timed_route(points, from_point_idx = 0, until_point_idx = None, route_number = None, number_of_routes = None, name = None, color = None, opacity = None, description = None) : description = description or "" if description : description = " " + description.strip(" ") if until_point_idx == None : until_point_idx = len(points) s = "" if from_point_idx + 1 < until_point_idx : sa = [] s = kml_route_header(route_number = route_number, number_of_routes = number_of_routes, name = name, when = points[from_point_idx].when, until = points[until_point_idx - 1].when, color = color, opacity = opacity, in_folder = True, description = description, ) ps = points[from_point_idx].to_kml() for pi in xrange(from_point_idx, until_point_idx - 1) : ss = kml_folder_placemark_header(route_number = route_number, when = points[pi].when, until = points[pi + 1].when) ss += ps ps = points[pi + 1].to_kml() ss += ps ss += kml_route_trailer() sa.append(ss) s += "".join(sa) s += kml_folder_trailer() return(s) def gpx_header(program_name = "", file_name = "", who = None, when = None, description = None) : description = description or "" if description : description = " " + description.strip(" ") program_name = program_name or "tz_gps.py" file_name = file_name or "" who = who or "unknown" when = when or time.time() tm = time.gmtime(when) desc = tzlib.maybe_wrap_with_cdata("%s GPX created by %s %s for %s" % ( file_name, program_name, time.asctime(time.localtime(when)), who ) ) who = tzlib.maybe_wrap_with_cdata(who) program_name = tzlib.printable(program_name).replace('"', "_").replace("'", " ") file_name = tzlib.maybe_wrap_with_cdata(file_name) s = """ %s %s %s """ % ( program_name, os.path.split(file_name)[1], desc, who, gpx_kml_date_time_string(tm) ) # return(s) def gpx_route_header(route_number = None, number_of_routes = None, name = None, when = None, description = None) : description = description or "" if description : description = " " + description.strip(" ") if route_number == None : route_number = 1 if number_of_routes == None : number_of_routes = route_number if name == None : name = "%u" % ( route_number ) name = str(name) when = when or time.time() desc = "GPX path %s created %s" % ( name, time.asctime(time.localtime(when)) ) s = """ %s %s %u """ % ( tzlib.maybe_wrap_with_cdata(name), tzlib.maybe_wrap_with_cdata(desc + description), route_number ) return(s) def gpx_route_trailer() : return(" \n") def gpx_trailer() : return("\n") import xml.dom.minidom def inner_text(n) : t = "" if (n.nodeType == xml.dom.minidom.Node.TEXT_NODE) or (n.nodeType == xml.dom.minidom.Node.CDATA_SECTION_NODE) or (n.nodeType == xml.dom.minidom.Node.ENTITY_NODE) : t += n.data elif (n.nodeType == xml.dom.minidom.Node.ELEMENT_NODE) : n = n.firstChild while n : t += inner_text(n) n = n.nextSibling pass return(t) class a_gpx(object) : def __init__(me, file_or_str) : if not isinstance(file_or_str, basestring) : file_or_str = file_or_str.read() me.dom = xml.dom.minidom.parseString(file_or_str) def get_tracks(me) : rtes = me.dom.getElementsByTagName("rte") tracks = [] for r in rtes : pts = r.getElementsByTagName("rtept") pa = [] for p in pts : lat = float(p.getAttribute('lat')) lon = float(p.getAttribute('lon')) alt = None e = p.getElementsByTagName('ele') if e.length == 1 : alt = fix_unsigned_int(float(inner_text(e.item(0)))) t = 0.0 e = p.getElementsByTagName('time') if e.length == 1 : t = tz_parse_time.parse_time(inner_text(e.item(0))) if not t : t = 0.0 pass typ = None e = p.getElementsByTagName('type') if e.length == 1 : typ = inner_text(e.item(0)).strip() nam = None e = p.getElementsByTagName('name') if e.length == 1 : nam = inner_text(e.item(0)).strip() des = None e = p.getElementsByTagName('desc') if e.length == 1 : des = inner_text(e.item(0)).strip() spd = None e = p.getElementsByTagName('speed') if e.length == 1 : spd = float(inner_text(e.item(0))) hrt = None e = p.getElementsByTagName('heart_rate') if e.length == 1 : hrt = float(inner_text(e.item(0))) dur = None e = p.getElementsByTagName('duration') if e.length == 1 : dur = float(inner_text(e.item(0))) gpgga = None e = p.getElementsByTagName('gpgga') if e.length == 1 : gpgga = inner_text(e.item(0)).strip() gpgsa = None e = p.getElementsByTagName('gpgsa') if e.length == 1 : gpgsa = inner_text(e.item(0)).strip() gprmc = None e = p.getElementsByTagName('gprmc') if e.length == 1 : gprmc = inner_text(e.item(0)).strip() ( fix_typ, sat_cnt, hz_dispersion, gpgsa, track_angle ) = nmea_parse_extras(gpgga = gpgga, gpgsa = gpgsa, gprmc = gprmc) pa.append(a_point(name = nam, when = t, lat = lat, lon = lon, altitude = alt, typ = typ, speed = spd, duration = dur, heart_rate = hrt, description = des, gpgsa = gpgsa, fix_typ = fix_typ, sat_cnt = sat_cnt, hz_dispersion = hz_dispersion, track_angle = track_angle)) pa.sort(cmp_points_by_when) tracks.append(pa) return(tracks) def to_string(me) : return(xml.dom.minidom.toprettyxml(me.dom)) pass # a_gpx class a_loc(object) : def __init__(me, file_or_str) : if not isinstance(file_or_str, basestring) : file_or_str = file_or_str.read() me.dom = xml.dom.minidom.parseString(file_or_str) def get_tracks(me) : pa = [] wps = me.dom.getElementsByTagName("waypoint") for p in wps : e = p.getElementsByTagName('coord') if e.length == 1 : e = e.item(0) lat = float(e.getAttribute('lat')) lon = float(e.getAttribute('lon')) ast = e.getAttribute('ele') alt = (ast and fix_unsigned_int(float(ast))) or None t = 0.0 e = p.getElementsByTagName('time') if e.length == 1 : t = tz_parse_time.parse_time(inner_text(e.item(0))) if not t : t = 0.0 pass typ = None e = p.getElementsByTagName('type') if e.length == 1 : typ = inner_text(e.item(0)).strip() nam = None des = None e = p.getElementsByTagName('name') if e.length == 1 : des = inner_text(e.item(0)).strip() nam = e.item(0).getAttribute('id') url = None lt = None e = p.getElementsByTagName('link') if e.length == 1 : url = inner_text(e.item(0)).strip() lt = e.item(0).getAttribute('text') pnt = a_point(name = nam, when = t, lat = lat, lon = lon, altitude = alt, typ = typ, description = des) pnt.url = url pnt.link_text = lt pa.append(pnt) pass pa.sort(cmp_points_by_when) tracks = [ [ p, ] for p in pa ] return(tracks) def to_string(me) : return(xml.dom.minidom.toprettyxml(me.dom)) pass # a_loc def loc_header(program_name = "", file_name = "", who = None, when = None, description = None) : description = description or "" if description : description = " " + description.strip(" ") program_name = program_name or "tz_gps.py" file_name = file_name or "" who = who or "unknown" when = when or time.time() tm = time.gmtime(when) desc = tzlib.maybe_wrap_with_cdata("%s GPX created by %s %s for %s" % ( file_name, program_name, time.asctime(time.localtime(when)), who ) ) who = tzlib.maybe_wrap_with_cdata(who) program_name = tzlib.printable(program_name).replace('"', "_").replace("'", " ") file_name = tzlib.maybe_wrap_with_cdata(file_name) s = """ %s %s %s """ % ( program_name, os.path.split(file_name)[1], desc, who, gpx_kml_date_time_string(tm) ) # return(s) def loc_trailer() : return("\n") import xml.parsers.expat class a_kml(object) : kml_num_re_str = tzlib.float_regx_str kml_all_coords_re = re.compile(r"(\s*" + kml_num_re_str + " *, *" + kml_num_re_str + " *(?:, *" + kml_num_re_str + ")?)", re.DOTALL) # yes, the altitude could be negative kml_coords_re = re.compile(r"\s*(" + kml_num_re_str + ") *, *(" + kml_num_re_str + ") *(?:, *(" + kml_num_re_str + "))?", re.DOTALL) class a_folder(object) : def __init__(me, parent) : if parent : me.level = parent.level + 1 else : me.level = 0 me.parent = parent me.sub_folders = [] me.placemarks = [] me.points = [] me.name = None me.description = None def append(me, f) : me.sub_folders.append(f) pass class a_placemark(object) : def __init__(me) : me.name = None me.description = None me.coords = [] me.swhen = None me.ewhen = None def append(me, c) : me.coords.append(c) pass def _fix_placemarks(me, f) : i = 0 ppm = None for pm in f.placemarks : if ppm and ppm.coords and (len(ppm.coords) > 1) and (str(pm.coords[0]) == str(ppm.coords[-1])) and (pm.swhen == ppm.ewhen) : ppm.coords.pop() ppm.ewhen = None # get rid of the last coord on the previous placemark, as it's duped in this one ppm = pm for pm in f.placemarks : if pm.coords : f.points.append( a_point(name = pm.name, when = pm.swhen or 0.0, lat = pm.coords[ 0][1], lon = pm.coords[ 0][0], altitude = pm.coords[ 0][2], description = pm.description)) if len(pm.coords) > 1 : for i in xrange(1, len(pm.coords) - 1) : f.points.append(a_point(name = pm.name, lat = pm.coords[ i][1], lon = pm.coords[ i][0], altitude = pm.coords[ i][2], description = pm.description)) f.points.append( a_point(name = pm.name, when = pm.ewhen or 0.0, lat = pm.coords[-1][1], lon = pm.coords[-1][0], altitude = pm.coords[-1][2], description = pm.description)) pass pass def start_element(me, name, attrs) : if name == "Folder" : f = me.a_folder(me.pf) me.pf.append(f) me.pf = f me.pc.append('f') elif name == "Placemark" : me.pm = me.a_placemark() me.pc.append('p') elif name == "name" : me.pc.append('n') me.ps = "" elif name == "description" : me.pc.append('d') me.ps = "" elif name == "TimeSpan" : me.pc.append('t') elif name == "begin" : me.pc.append('b') me.ps = "" elif name == "end" : me.pc.append('e') me.ps = "" elif name == "Point" : me.pc.append('o') elif name == "LineString" : me.pc.append('l') elif name == "coordinates" : me.pc.append('c') me.ps = "" else : me.pc.append('x') me.ps = "" pass def char_data(me, data) : if me.pc[-1] in "cbend" : # if one of these elements is inside the other, then there would be trouble, but we'll just live with it me.ps += data pass def end_element(me, name) : me.pc.pop() if name == "Folder" : me._fix_placemarks(me.pf) me.pf = me.pf.parent elif name == "Placemark" : if me.pf and me.pm.coords : me.pf.placemarks.append(me.pm) me.pm = None elif name == "name" : if me.pc[-1] == 'p' : me.pm.name = me.ps elif me.pc[-1] == 'f' : me.pf.name = me.ps pass elif name == "description" : if me.pc[-1] == 'p' : me.pm.description = me.ps elif me.pc[-1] == 'f' : me.pf.description = me.ps pass elif name == "begin" : if (me.pc[-2] == 'p') and (me.pc[-1] == 't') and me.ps : me.pm.swhen = tz_parse_time.parse_time(me.ps) pass elif name == "end" : if (me.pc[-2] == 'p') and (me.pc[-1] == 't') and me.ps : me.pm.ewhen = tz_parse_time.parse_time(me.ps) pass elif name == "coordinates" : if me.pm and (me.pc[-2] == 'p') and ((me.pc[-1] == 'l') or (me.pc[-1] == 'o')) : csa = me.kml_all_coords_re.findall(me.ps) if csa : for cs in csa : g = me.kml_coords_re.match(cs) lla = [ float(g.group(1)), float(g.group(2)) ] if g.lastindex > 2 : lla.append(float(g.group(3))) else : lla.append(None) me.pm.coords.append(lla) pass pass pass else : me.ps = "" pass def __init__(me, s, file_name = None) : me.file_name = file_name or None me.folder = me.a_folder(None) me.pf = me.folder # current folder me.pm = None # current placemark me.pc = [] # context array me.ps = "" # character string we're getting me.p = xml.parsers.expat.ParserCreate() me.p.buffer_text = True # speed it up by glomming text together if possible me.p.StartElementHandler = me.start_element me.p.CharacterDataHandler = me.char_data me.p.EndElementHandler = me.end_element try : me.p.Parse(s) except xml.parsers.expat.ExpatError : # print "expat", me.p.CurrentLineNumber, me.p.CurrentColumnNumber # print "expat", me.p.code, me.p.lineno, me.p.offset raise ValueError except ValueError : # print "value", me.p.CurrentLineNumber, me.p.CurrentColumnNumber raise ValueError pass def _get_tracks(me, tracks, folders) : for f in folders : if f.placemarks : t = a_track(points = f.points, id_num = len(tracks), when = f.points[0].when, name = f.name, description = f.description, file_name = me.file_name) tracks.append(t) me._get_tracks(tracks, f.sub_folders) pass def get_tracks(me) : tracks = [] me._fix_placemarks(me.folder) me._get_tracks(tracks, [ me.folder ] ) tracks = color_tracks(tracks) return(tracks) pass # a_kml class a_track(object) : def clear(me) : me.points = [] me.id_num = 0 me.when = 0 me.name = None me.file_name = None me.color = None me._opacity = None me.description = None def __init__(me, points = None, id_num = 0, when = None, name = None, description = None, file_name = None, color = None, opacity = None) : me.clear() me.points = points or [] # array of points me.id_num = id_num # needed for some file output if (when == None) and points : when = points[0].when me.when = when # needed for some file output me.name = name me.description = description me.file_name = file_name me.color = color # set GGBBRR numeric value (should take "#RRGGBB" and such, too) me._opacity = opacity def append_track(me, t) : me.points += t.points if not me.file_name : me.file_name = t.file_name if not me.name : me.name = t.name if me.description : me.description = t.description pass def full_name(me) : return(((me.file_name or "") + " " + (me.name or "")).strip() or None) def set_color(me, r_or_rgb, g = None, b = None) : ov = me.color if g != None : if b == None : raise ValueError("Green but no blue!") if not (0 <= r_or_rgb <= 255) : raise ValueError("Bad red [%s]!" % ( str(r_or_rgb) ) ) if not (0 <= g <= 255) : raise ValueError("Bad green [%s]!" % ( str(g) ) ) if not (0 <= b <= 255) : raise ValueError("Bad blue [%s]!" % ( str(b) ) ) me.color = (g << 16) + (b << 8) + r_or_rgb elif not (0 <= r_or_rgb <= 0xffffff) : raise ValueError("Bad rgb [%x]!" % ( r_or_rgb ) ) else : me.color = r_or_rgb return(ov) def color_rgb(me) : if me.color == None : return( ( None, None, None ) ) return( ( me.color & 0xff, (me.color >> 16) & 0xff, (me.color >> 8) & 0xff ) ) def color_hash_hex_str(me) : if me.color == None : return("") ( r, g, b ) = me.color_rgb() return("#%02x%02x%02x" % ( r, g, b ) ) def color_gbr_val(me) : return(me.color) OPAQUE = 100 def set_opacity(me, v) : ov = me._opacity me._opacity = v if me._opacity != None : me._opacity = min(me.OPAQUE, max(0, v)) return(ov) def get_opacity(me) : if me._opacity == None : return(None) return(int(min(me.OPAQUE, max(0, me._opacity)))) pass # a_track def nice_track_color(num_tracks, zero_based_track_num) : green = 0 if num_tracks > 1 : num_tracks -= 1 green = int(((num_tracks - max(0, min(zero_based_track_num, num_tracks))) * 255.0) / num_tracks) red = 255 - green return(red, green, 0) def _color_tracks(tracks) : for i in xrange(len(tracks)) : ( red, green, blue ) = nice_track_color(len(tracks), i) tracks[i].set_color(red, green, blue) return(tracks) def make_array_of_tracks(tracks) : """ Make an array of a_tracks, whether 'tracks' is an array of a_tracks or and array of points or an array of arrays of points. """ if not tracks : return([]) try : rtracks = [ t for t in tracks if t.points ] if len(rtracks) : t = rtracks[0] x = t.id_num x = t.when pass except AttributeError : try : t = tracks[0][0] except AttributeError : tracks = [ tracks ] except TypeError : tracks = [ tracks ] rtracks = [] id_num = 0 i = 0 while i < len(tracks) : if len(tracks[i]) : rtracks.append(a_track(tracks[i], id_num, tracks[i][0].when)) id_num += 1 i += 1 rtracks = _color_tracks(rtracks) # just be nice, but if he passes us valid tracks, not array or points or array of array of points, then we will not have colored (or recolored) his tracks return(rtracks) def sorted_tracks_by_when(tracks) : tracks = make_array_of_tracks(tracks) tracks = list(tracks) tracks.sort(lambda a, b : cmp(a.when, b.when)) return(tracks) def reverse_tracks(tracks) : """ Reverse, in place, all the points in the tracks and the tracks, themselves. """ tracks.reverse() for t in tracks : t.points.reverse() pass def count_points_in_tracks(tracks, filter_rtn = None) : tracks = make_array_of_tracks(tracks) cnt = 0 for t in tracks : if filter_rtn : cnt += len( [ p for p in t.points if filter_rtn(p) ] ) else : cnt += len(t.points) pass return(cnt) def smooth_tracks(tracks) : tracks = make_array_of_tracks(tracks) for t in tracks : t.points = smooth_points(t.points) t.when = t.points[0].when return(tracks) def remove_redundant_points_from_tracks(tracks, cutoff_distance = None) : tracks = make_array_of_tracks(tracks) for t in tracks : t.points = remove_redundant_points(t.points, cutoff_distance) return(tracks) def fix_tracks_speeds(tracks, remove_bad_tracks = True, force = False) : tracks = make_array_of_tracks(tracks) ntracks = [] for t in tracks : if fix_points_speeds(t.points, force = force) or not remove_bad_tracks : ntracks.append(t) pass return(ntracks) def fix_tracks_whens(tracks, remove_bad_tracks = True) : tracks = make_array_of_tracks(tracks) ntracks = [] for t in tracks : if fix_points_whens(t.points, t.when) or (not remove_bad_tracks) : ntracks.append(t) pass return(ntracks) def remove_untimed_points_from_tracks(tracks) : """ Remove all points with zero or None 'when' values or fix the 'when' values using a 'duration' value and a previous point. """ tracks = make_array_of_tracks(tracks) ti = len(tracks) while ti > 0 : ti -= 1 t = tracks[ti] points = t.points pi = len(points) while pi > 0 : pi -= 1 if not points[pi].when : del(t[pi]) pass if not t.points : del(tracks[ti]) else : t.points.sort(cmp_points_by_when) t.when = t.points[0].when pass tracks = sorted_tracks_by_when(tracks) return(tracks) def remove_duplicate_tracks(tracks) : tracks = sorted_tracks_by_when(tracks) ti = len(tracks) while ti > 1 : ti -= 1 t = tracks[ti] tp = tracks[ti - 1] if t.points == tp.points : del(tracks[ti]) elif (len(t.points) == len(tp.points)) : if not t.points : del(tracks[ti]) else : same = True for pi in xrange(len(t.points)) : if (t.points[pi].when != tp.points[pi].when) or (t.points[pi].lat != tp.points[pi].lat) : same = False break if str(t.points[pi]) != str(tp.points[pi]) : same = False break pass if same : del(tracks[ti]) pass pass pass return(tracks) def color_tracks(tracks) : tracks = make_array_of_tracks(tracks) return(_color_tracks(tracks)) def set_no_gpx_crcs_in_tracks(tracks) : tracks = make_array_of_tracks(tracks) for t in tracks : set_no_gpx_crcs(t.points) return(tracks) def set_no_whens_in_tracks(tracks) : tracks = make_array_of_tracks(tracks) for t in tracks : t.when = 0.0 set_no_whens(t.points) return(tracks) def set_equal_durations_in_tracks(tracks, duration = 1.0) : tracks = make_array_of_tracks(tracks) for t in tracks : t.when = 1.0 set_equal_durations(t.points, 1.0, duration) return(tracks) def make_no_duplicate_points_in_tracks(tracks, ids = None) : ids = ids or {} tracks = make_array_of_tracks(tracks) for t in tracks : make_no_duplicate_points(t.points, ids) return(tracks) def remove_single_label_tracks(tracks) : """ Return an array of tracks that doesn't have any tracks that are empty or that have a single labeled, unnamed point. """ tracks = make_array_of_tracks(tracks) return( [ t for t in tracks if (len(t.points) > 1) or ((len(t.points) == 1) and t.points[0].label and not t.points[0].name) ] ) class a_rectangle(object) : """ A rectangle with NW and SE corners. Note: This thing does not handle areas draped over the poles. And things get very weird when the rectangle spans the globe, east and west. But we do our best in such cases. """ def __init__(me, lat, lon) : """ Create a_rectangle starting with a given latitude/longitude. """ me.north = lat me.south = lat me.east = lon me.west = lon def include_latlon(me, lat, lon) : """ Expand a_rectangle to include the given latitude/longitude. """ if not me.is_inside_latlon(me.south, lon) : if latlon.calcDistance(lat, lon, lat, me.west) < latlon.calcDistance(lat, lon, lat, me.east) : me.west = lon else : me.east = lon pass me.north = latlon.lat_in_world(max(me.north, lat)) me.south = latlon.lat_in_world(min(me.south, lat)) def include_point(me, p) : """ Expand a_rectangle to include the given point (which has p.lat, p.lon). """ if p : me.include_latlon(p.lat, p.lon) pass def is_inside_latlon(me, lat, lon) : """ Is the given latitude/longitude inside us? Note that a point on the boundary is 'inside' us. """ if me.south <= lat <= me.north : if me.east < me.west : return((lon >= me.west) or (lon <= me.east)) if me.west <= lon <= me.east : return(True) pass return(False) def is_inside(me, p) : """ Is the given point (p.lat p.lon) inside us? """ if p : return(me.is_inside_latlon(p.lat, p.lon)) return(False) def is_overlapping(me, ome) : """ Does the given a_rectangle overlap with us? """ if ome : if ((me.north < ome.south) and (me.south >= ome.north)) or (me.north == ome.north) : if ((me.west < ome.east ) and (me.east >= ome.west )) or (me.west == ome.west ) : return(True) pass pass return(False) pass # a_rectangle class a_circle(object) : """ A circle with a center and radius. """ def __init__(me, lat, lon, radius) : """ Create a_circle at the given latitude/longitude with the given radius in nautical miles . """ me.lat = lat me.lon = lon me.radius = radius def is_inside_latlon(me, lat, lon) : """ Is the given latitude/longitude inside us? Note that a point on the boundary is 'inside' us. """ return(latlon.calcDistance(me.lat, me.lon, lat, lon) < me.radius) def is_inside(me, p) : """ Is the given point (p.lat p.lon) inside us? """ if p : return(me.is_inside_latlon(p.lat, p.lon)) return(False) pass # a_circle def nmea_parse_extras(gpgga = None, gpgsa = None, gprmc = None) : fix_typ = 1 sat_cnt = 0 hz_dispersion = None if gpgga : g = nmea_gpgga_re.search(gpgga) if g : fix_typ = int(g.group(10)) sat_cnt = int(g.group(11)) hz_dispersion = None gs = g.group(12).strip() if gs : hz_dispersion = float(gs) pass pass gpgsa = a_gpgsa.from_string(gpgsa) or None track_angle = None if gprmc : g = nmea_gprmc_re.search(gprmc) if g : track_angle = None gs = g.group(11).strip() if gs : track_angle = float(gs) pass pass return( fix_typ, sat_cnt, hz_dispersion, gpgsa, track_angle ) def _decode_nmea_g_wll(g) : when = (int(g.group(1)) * 3600.0) + (int(g.group(2)) * 60.0) + float(g.group(3)) lat = int(g.group(4)) + (float(g.group(5)) / 60.0) if g.group(6).upper() == 'S' : lat = -lat lon = int(g.group(7)) + (float(g.group(8)) / 60.0) if g.group(9).upper() == 'W' : lon = -lon return( ( when, lat, lon ) ) def extract_nmea_points_from_lines(lns) : """ Get as many nmea points as we can, getting rid of all the ascii lines of text from the lns array except those that apply to the point's lines. Complete information about the last point is not guaranteed. """ points = [] p = None ll = None fix_typ = 1 sat_cnt = 0 hz_dispersion = None gpgsa = None month = 0 day = 0 year = 0 speed = None for ln in lns : ll = None g = nmea_gpgga_re.search(ln) if g : ( when, lat, lon ) = _decode_nmea_g_wll(g) fix_typ = int(g.group(10)) sat_cnt = int(g.group(11)) hz_dispersion = None gs = g.group(12).strip() if gs : hz_dispersion = float(gs) # print "fsz", fix_typ, sat_cnt, hz_dispersion altitude = None if g.group(14).upper() == 'M' : # we only understand altitude in meters, for now gs = g.group(13).strip() if gs : altitude = float(gs) pass if p : if (int(p.when) % (60 * 60 * 24)) != int(when) : if p.when > 3600 * 101 : if gpgsa and not p.gpgsa : p.gpgsa = gpgsa gpgsa = None points.append(p) p = None pass else : p.altitude = altitude p.fix_typ = fix_typ p.sat_cnt = sat_cnt p.hz_dispersion = hz_dispersion pass if not p : if day : when = tz_parse_time.make_utime(month, day, year, 0, 0, when) p = a_point(when = when, lat = lat, lon = lon, altitude = altitude, speed = speed, gpgsa = gpgsa, fix_typ = fix_typ, sat_cnt = sat_cnt, hz_dispersion = hz_dispersion) gpgsa = None ll = ln pass else : g = nmea_gprmc_re.search(ln) if g : ( when, lat, lon ) = _decode_nmea_g_wll(g) speed = (float(g.group(10)) * latlon.metersPerNauticalMile) / 1000.0 track_angle = None gs = g.group(11).strip() if gs : track_angle = float(gs) # print "ta", track_angle month = int(g.group(13)) day = int(g.group(12)) year = int(g.group(14)) if p : if (int(p.when) % (60 * 60 * 24)) != int(when) : if p.when > 3600 * 101 : if gpgsa and not p.gpgsa : p.gpgsa = gpgsa gpgsa = None points.append(p) p = None pass else : p.when = tz_parse_time.make_utime(month, day, year, 0, 0, when) p.speed = speed p.track_angle = track_angle pass if not p : if day : when = tz_parse_time.make_utime(month, day, year, 0, 0, when) p = a_point(when = when, lat = lat, lon = lon, speed = speed, gpgsa = gpgsa, track_angle = track_angle) gpgsa = None ll = ln pass else : ngpgsa = a_gpgsa.from_string(ln) if ngpgsa : gpgsa = ngpgsa pass pass pass if p and (p.when > 3600 * 101) : if gpgsa and not p.gpgsa : p.gpgsa = gpgsa gpgsa = None points.append(p) ll = None if ll : lns[0] = ll del(lns[1:]) else : del(lns[0:]) # print "\n".join([ str(p) for p in points ]) return(points) def tracks_from_nmea_file_or_string(file_or_str, file_name = None, name = None, desc = None) : if not isinstance(file_or_str, basestring) : file_or_str = file_or_str.read() tracks = [] lns = re.split("(\r?\n|\r)", file_or_str) points = extract_nmea_points_from_lines(lns) if points : tracks.append(a_track(points, 0, name = name, description = desc, file_name = file_name)) tracks = color_tracks(tracks) return(tracks) def tracks_from_nmea_file(file_name) : fi = open(file_name, "r") s = fi.read() fi.close() return(tracks_from_nmea_file_or_string(s, file_name)) # # # This is a *very* much faster way to get the points from a .gpx file. # # gpx_metadata_re = re.compile(r"(.*?)", re.DOTALL) gpx_wpt_re = re.compile(r"]+)>\s*(.*?)", re.DOTALL) gpx_rte_re = re.compile(r".*?", re.DOTALL) gpx_rtept_re = re.compile(r"]+)>\s*(.*?)", re.DOTALL) gpx_trk_re = re.compile(r".*?", re.DOTALL) gpx_trkseg_re = re.compile(r".*?", re.DOTALL) gpx_trkpt_re = re.compile(r"]+)>\s*(.*?)", re.DOTALL) gpx_name_re = re.compile(r"\s*(?:|([^<]*?))\s*", re.DOTALL) gpx_desc_re = re.compile(r"\s*(?:|([^<]+?))\s*", re.DOTALL) gpx_type_re = re.compile(r"\s*(?:|([^<]+?))\s*", re.DOTALL) gpx_label_re = re.compile(r"\s*(?:|([^<]+?))\s*", re.DOTALL) gpx_gpgga_re = re.compile(r"\s*(?:|([^<]+?))\s*", re.DOTALL) gpx_gpgsa_re = re.compile(r"\s*(?:|([^<]+?))\s*", re.DOTALL) gpx_gprmc_re = re.compile(r"\s*(?:|([^<]+?))\s*", re.DOTALL) gpx_lat_re = re.compile(r"lat\s*=\s*[\"']\s*(" + tzlib.float_regx_str + r")\s*[\"']", re.DOTALL) gpx_lon_re = re.compile(r"lon\s*=\s*[\"']\s*(" + tzlib.float_regx_str + r")\s*[\"']", re.DOTALL) gpx_ele_re = re.compile(r"\s*(" + tzlib.float_regx_str + r")\s*", re.DOTALL) gpx_time_re = re.compile(r"\s*([^<]+)", re.DOTALL) gpx_speed_re = re.compile(r"\s*([^<]+)", re.DOTALL) gpx_duration_re = re.compile(r"\s*([^<]+)", re.DOTALL) gpx_heart_rate_re = re.compile(r"\s*([^<]+)", re.DOTALL) def re_num(regx, s) : g = regx.search(s) if g : try : return(float(g.group(1).strip()) + 0.0) except ValueError : pass except TypeError : pass except OverflowError : pass pass return(None) def gpx_string_stripped(s, regx) : name = None g = regx.search(s) if g : name = g.group(1) or g.group(2) name = name.strip() return(name) def gpx_string_name(s) : return(gpx_string_stripped(s, gpx_name_re)) def gpx_string_desc(s) : return(gpx_string_stripped(s, gpx_desc_re)) def gpx_string_type(s) : return(gpx_string_stripped(s, gpx_type_re)) def gpx_string_label(s) : return(gpx_string_stripped(s, gpx_label_re)) def points_from_gpx_point_string(all_pts_strs) : points = [] pwhen = None for p in all_pts_strs : lat = re_num(gpx_lat_re, p[0]) lon = re_num(gpx_lon_re, p[0]) alt = fix_unsigned_int(re_num(gpx_ele_re, p[1])) speed = re_num(gpx_speed_re, p[1]) duration = re_num(gpx_duration_re, p[1]) heart_rate = re_num(gpx_heart_rate_re, p[1]) label = gpx_string_label(p[1]) or None typ = gpx_string_type(p[1]) name = gpx_string_name(p[1]) or None description = gpx_string_desc(p[1]) ( fix_typ, sat_cnt, hz_dispersion, gpgsa, track_angle ) = nmea_parse_extras(gpgga = gpx_string_stripped(p[1], gpx_gpgga_re) or None, gpgsa = gpx_string_stripped(p[1], gpx_gpgsa_re) or "", gprmc = gpx_string_stripped(p[1], gpx_gprmc_re) or None) when = gpx_time_re.search(p[1]) if when : when = tz_parse_time.parse_time(when.group(1)) if not when : when = None pass if duration and pwhen : w = pwhen + duration if (when == None) or (abs(when - w) <= 1.0) : # if either the point doesn't have an explicit 'when' or if the previous point's 'when' plus the duration is within a second of this point's 'when' when = w # use durations, if we have them - they can go down to less than 1 second resolution pass when = when or 0.0 pwhen = when if lat and lon : try : points.append(a_point(name = name, when = when, lat = lat, lon = lon, altitude = alt, typ = typ, speed = speed, duration = duration, heart_rate = heart_rate, description = description, label = label, gpgsa = gpgsa, fix_typ = fix_typ, sat_cnt = sat_cnt, hz_dispersion = hz_dispersion, track_angle = track_angle)) except TypeError : pass except ValueError : pass except OverflowError : pass pass pass points.sort(cmp_points_by_when) return(points) def points_from_gpx_rte_string(s) : rtepts = gpx_rtept_re.findall(s) return(points_from_gpx_point_string(rtepts)) def points_from_gpx_trkseg_string(s) : trkpts = gpx_trkpt_re.findall(s) return(points_from_gpx_point_string(trkpts)) def points_from_gpx_wpts(s) : pts = gpx_wpt_re.findall(s) return(points_from_gpx_point_string(pts)) def tracks_from_gpx_file_or_string(file_or_str, file_name = None) : if not isinstance(file_or_str, basestring) : file_or_str = file_or_str.read() fname = file_name md = gpx_metadata_re.search(file_or_str) if md : fname = gpx_string_name(md.group(1)) or fname tracks = [] points = points_from_gpx_wpts(file_or_str) for p in points : name = "waypoint " + re.sub(r"^(?i)waypoint\s*", "", p.name or "") name = name.strip() tracks.append(a_track([p], len(tracks), name = name, file_name = fname)) rtes = gpx_rte_re.findall(file_or_str) for r in rtes : rhdr = r[0 : r.find(" lonloncnt : points = [ [ p[1], p[0], p[2] ] for p in points ] # ugh. perhaps it's a kml file that we could not read points = [ a_point(lat = p[0], lon = p[1], altitude = p[2]) for p in points ] if ll : lns[0] = ll del(lns[1:]) else : del(lns[0:]) return(points) def tracks_from_lat_lon_text_file_or_string(file_or_str, file_name = None, name = None, desc = None) : if not isinstance(file_or_str, basestring) : file_or_str = file_or_str.read() tracks = [] lns = re.split("(\r?\n|\r|\s*\|\s*|\s*\/\s*)", file_or_str) points = extract_lat_lon_text(lns) if points : tracks.append(a_track(points, 0, name = name, description = desc, file_name = file_name)) tracks = set_no_whens_in_tracks(tracks) tracks = color_tracks(tracks) return(tracks) def tracks_from_lat_lon_text_file(file_name) : fi = open(file_name, "r") s = fi.read() fi.close() return(tracks_from_lat_lon_text_file_or_string(s, file_name)) def tracks_from_file(file_name) : tracks = tracks_from_kmlz_file(file_name) if tracks : return(tracks) fd = tzlib.read_whole_text_file(file_name) tracks = tracks_from_gpx_file_or_string( fd, file_name = file_name) if not tracks : tracks = tracks_from_loc_file_or_string( fd, file_name = file_name) if not tracks : tracks = tracks_from_nmea_file_or_string(fd, file_name = file_name) if not tracks : tracks = tracks_from_lat_lon_text_file_or_string( fd, file_name = file_name); return(tracks) def points_from_all_tracks(tracks) : """ Return one, sorted, flat array of a copy of all points in all tracks. """ tracks = make_array_of_tracks(tracks) points = [] for t in tracks : pts = [ copy.copy(p) for p in t.points ] pi = 0 for p in pts : if not hasattr(p, 'track') : p.track = t p.track_i = pi pi += 1 points += pts points.sort(cmp_points_by_when) return(points) def _valid_from_to_i(points, from_i = 0, to_i = None) : from_i = from_i or 0 from_i = min(len(points), max(0, from_i)) to_i = to_i or len(points) to_i = min(len(points), max(0, to_i)) return( ( from_i, to_i ) ) def _point_values_sum(func, points, from_i = 0, to_i = None) : ( from_i, to_i ) = _valid_from_to_i(points, from_i, to_i) sum = 0 while from_i < to_i : v = func(points[from_i]) if v == None : return(None) sum += v from_i += 1 return(sum) def make_points_their_average_location(points, from_i = 0, to_i = None) : """ Move all the given points to the same, average location. """ ( from_i, to_i ) = _valid_from_to_i(points, from_i, to_i) # This needs to use X Y Z logic !!!! lat_tot = _point_values_sum(lambda p : p.lat, points, from_i, to_i) lon_tot = _point_values_sum(lambda p : p.lon, points, from_i, to_i) alt_tot = _point_values_sum(lambda p : p.altitude, points, from_i, to_i) if (lat_tot != None) and (lon_tot != None) : cnt = to_i - from_i lat_tot /= cnt lon_tot /= cnt if alt_tot != None : alt_tot /= cnt for ti in xrange(from_i, to_i) : points[ti].lat = lat_tot points[ti].lon = lon_tot if alt_tot != None : points[ti].altitude = alt_tot pass pass pass def combine_zero_speed_points(tracks) : """ For any group of consecutive points with zero speed, set them all to their average location (lat/lon/altitude). """ if len(tracks) : tracks = make_array_of_tracks(tracks) for t in tracks : t = t.points gi = -1 i = 1 while i < len(t) : sp = t[i].speed if sp == None : if gi >= 0 : make_points_their_average_location(t, gi, i) gi = -1 i += 2 else : if sp : if gi >= 0 : make_points_their_average_location(t, gi, i) gi = -1 pass elif gi < 0 : gi = i - 1 i += 1 pass if gi >= 0 : # print "end", gi, i make_points_their_average_location(t, gi, len(t) - 1) pass pass return(tracks) class a_nearby_point(object) : def __init__(me, point, distance) : me.point = point me.distance = distance pass def only_nearby_points(nearby_points, how_many = None, near_dist = None, far_dist = None) : """ Bogus routine: Given an array of sorted, close-to-far a_nearby_points, return a good place to lop off the nearby_points, nearby_points[:return_value]. """ how_many = how_many or min(15, max(1, len(nearby_points))) bd = -1 bdi = 0 for di, p in enumerate(nearby_points[how_many:]) : d = p.distance if ((not near_dist) or (d >= near_dist)) and ((not far_dist) or (d <= far_dist)) : d = abs(d - nearby_points[di - 1].distance) if bdi <= d : bdi = d # find the index at the far side of the largest distance gap bd = di pass pass return(bdi or len(nearby_points)) def get_nearby_tracks_points(tracks, point, max_distance = None, flt_rtn = None, flat = False) : """ Get all the points in the tracks that are near the given point. Returns an array of a_nearby_point's, sorted by ascending distance to the given point. """ flt_rtn = flt_rtn or (lambda p : True) tracks = make_array_of_tracks(tracks) max_distance = max_distance or (7.0 / latlon.metersPerNauticalMile) # this default multiplier might be changed if we used flat_distance_from() instead of using the altitude, too (which may be too flakey) nearby = [] lat = point.lat lon = point.lon ( latd, lond ) = latlon.distance_in_lat_lon_here(lat, lon, max_distance) # find lat and lon values differences, if a point is outside them, the point is more than our max_distance away for t in tracks : for p in t.points : if flt_rtn(p) : if (abs(p.lat - lat) <= latd) and ((abs(p.lon - lon) <= lond) or (abs(p.lon + lon) <= lond)) : # quick test to avoid needing to do the big arithmetic (the longitude check is a bit inclusive, but handles 180/-180 and -180/180 in one test) if flat : d = point.flat_distance_from(p) else : d = point.distance_from(p) if d <= max_distance : nearby.append(a_nearby_point(p, d)) pass pass pass pass def _cmp_nbp(p1, p2) : d = cmp(p1.distance, p2.distance) if d : return(d) if p1.point.name or p2.point.name : return(-cmp(p1.point.name.lower(), p2.point.name.lower())) # take the higher one in the alphabet, so that the unnamed one won't be chosen and so that the more detailed name will be chosen return(0) nearby.sort(_cmp_nbp) return(nearby) def get_points_near_untimed_points_in_tracks(tracks, max_distance = None) : """ Run through all the points and, for points without 'when' values, find nearby points that can tell the untimed point's 'when'. Return an array of an_untimed_point's with 'nearby' sorted by ascending distance from 'point'. """ class an_untimed_point(object) : def __init__(me, point) : me.point = point me.nearby = [] pass tracks = list(make_array_of_tracks(tracks)) # since we take a while, we'll use our own copy of the array - in case the caller changes it on another thread points = [] for t in tracks : for p in t.points : if not p.when : points.append(an_untimed_point(p)) pass pass for utp in points : utp.nearby = get_nearby_tracks_points(tracks, utp.point, max_distance = max_distance, flt_rtn = lambda p : p.when and True) return(points) class a_big_point(a_point) : """ A big point is a point that has the location and time (lat, lon, altitude, when) of multiple points. """ start_when = None end_when = None def __init__(me, point = None) : super(a_big_point, me).__init__() me.speed = me.speed or 0.0 me.heart_rate = me.heart_rate or 0.0 me.x_tot = 0.0 me.y_tot = 0.0 me.z_tot = 0.0 me.alt_tot = 0.0 me.when_tot = 0.0 me.speed_tot = None me.heart_rate_tot = None me.start_when = 0.0 me.end_when = 0.0 me.cnt = 0 me.points = [] if point : me.include_point(point) pass def include_point(me, p) : me.start_when = me.start_when or p.start_when me.start_when = min(me.start_when, p.start_when) me.end_when = max(me.end_when, p.end_when) if isinstance( p, a_big_point) : pts = dict( [ ( id(pt), pt ) for pt in me.points ] ) pts.update( dict( [ ( id(pt), pt ) for pt in p.points ] )) me.points += pts.values() me.x_tot += p.x_tot me.y_tot += p.y_tot me.z_tot += p.z_tot me.when_tot += p.when_tot if not me.cnt : me.alt_tot = p.alt_tot me.speed_tot = p.speed_tot me.heart_rate_tot = p.heart_rate_tot else : if p.alt_tot == None : me.alt_tot = None elif me.alt_tot != None : me.alt_tot += p.alt_tot if p.speed_tot == None : me.speed_tot = None elif me.speed_tot != None : me.speed_tot += p.speed_tot if p.heart_rate_tot == None : me.heart_rate_tot = None elif me.heart_rate_tot != None : me.heart_rate_tot += p.heart_rate_tot pass me.cnt += p.cnt else : me.points.append(p) ( x, y, z ) = p.x_y_z() me.x_tot += x me.y_tot += y me.z_tot += z me.when_tot += p.when if not me.cnt : me.alt_tot = p.altitude me.speed_tot = p.speed me.heart_rate_tot = p.heart_rate else : if p.altitude == None : me.alt_tot = None elif me.alt_tot != None : me.alt_tot += p.altitude if p.speed == None : me.speed_tot = None elif me.speed_tot != None : me.speed_tot += p.speed if p.heart_rate == None : me.heart_rate_tot = None elif me.heart_rate_tot != None : me.heart_rate_tot += p.heart_rate pass pass me.cnt += 1 if p.duration != None : me.duration = (me.duration or 0.0) + p.duration me.set_x_y_z(me.x_tot / me.cnt, me.y_tot / me.cnt, me.z_tot / me.cnt) me.when = me.when_tot / me.cnt if me.alt_tot != None : me.altitude = me.alt_tot / me.cnt else : me.altitude = None if me.speed_tot != None : me.speed = me.speed_tot / me.cnt else : me.speed = None if me.heart_rate_tot != None : me.heart_rate = me.heart_rate_tot / me.cnt else : me.heart_rate = None pass def first_raw_point(me) : if me.points : return(me.points[0]) return(me) def last_raw_point(me) : if me.points : return(me.points[-1]) return(me) pass # a_big_point class a_track_big_point(a_big_point) : """ A track big point is a big point that knows what track and where in the track the big point's points are. """ def __init__(me, track, i = -1) : super(a_track_big_point, me).__init__() me.track = track me.track_i = i if me.track_i >= 0 : p = track[i] if isinstance( p, a_track_big_point) : me.track = p.track me.track_i = p.track_i me.include_point(p) pass def first_raw_point(me) : return(me.track[me.track_i]) def last_raw_point(me) : return(me.track[me.track_i + me.cnt - 1]) pass # a_track_big_point def make_big_points(tracks, cutoff_distance = None, cutoff_time = None, cutoff_count = None, wanted_count = None, filter_rtn = None) : """ Given an array of tracks, convert them to an array of tracks of "big" points - points that are combo/averages of all consecutive points within the constraints given. Warning: If there is no filter routine and a wanted_count greater than or equal to the number of points in 'tracks', then we simply return 'tracks' - not even a copy of 'em! Otherwise, this routine returns an array of completely new tracks. """ if not (cutoff_distance or cutoff_time or cutoff_count or filter_rtn or wanted_count) : cutoff_distance = 5.0 / latlon.metersPerNauticalMile # default to 5 meter smoothing if he didn't spec any cutoffs or filter cutoff_distance = cutoff_distance or None cutoff_time = cutoff_time or 100000000000000000000000.0 cutoff_count = cutoff_count or sys.maxint wanted_count = wanted_count or 0 big_points = [] tracks = make_array_of_tracks(tracks) tcnt = count_points_in_tracks(tracks, filter_rtn) if (not filter_rtn) and (wanted_count >= tcnt) : return(tracks) # note: it might be better to be slow and return a copy here, so that the caller doesn't figure that he can mess with the returned tracks' values and not affect his input to this routine filter_rtn = filter_rtn or (lambda(p) : True) rcnt = tcnt for t in tracks : if len(t.points) : bpt = None bp = None i = 0 while i < len(t.points) : p = t.points[i] f = filter_rtn(p) if f : rcnt -= 1 if not bpt : bpt = a_track([], len(big_points), p.when, name = t.name, file_name = t.file_name, color = t.color) tot = wanted_count bp = a_track_big_point(t.points, i) elif not bp : bp = a_track_big_point(t.points, i) else : ntot = tot + wanted_count if (ntot < tcnt) and (p.when - bp.when < cutoff_time) and (bp.cnt < cutoff_count) and f and ((cutoff_distance == None) or (bp.flat_distance_from(p) < cutoff_distance)) : tot = ntot bp.include_point(p) else : tot -= tcnt wanted_count = max(0, wanted_count - 1) # to compensate for other things causing points to be appended tcnt = rcnt bpt.points.append(bp) bp = a_track_big_point(t.points, i) pass pass elif bp : # if the point is filtered out, kick out whatever big point we're working on tot -= tcnt wanted_count = max(0, wanted_count - 1) # to compensate for other things causing points to be appended tcnt = rcnt bpt.points.append(bp) bp = None i += 1 if bpt != None : if bp : tot -= tcnt wanted_count = max(0, wanted_count - 1) # to compensate for other things causing points to be appended tcnt = rcnt bpt.points.append(bp) big_points.append(bpt) pass pass return(make_array_of_tracks(big_points)) def make_all_big_points(tracks, max_loops = 1, cutoff_distance = 5.0 / latlon.metersPerNauticalMile, cutoff_time = None) : """ make_big_points() until they don't get any bigger, up to a given number of iterations. """ max_loops = max(max_loops, 1) ncnt = count_points_in_tracks(tracks) while max_loops > 0 : cnt = ncnt tracks = make_big_points(tracks, cutoff_distance, cutoff_time) ncnt = count_points_in_tracks(tracks) if cnt == ncnt : break max_loops -= 1 return(tracks) def make_new_points_in_tracks(tracks) : tracks = make_array_of_tracks(tracks) for t in tracks : t.points = [ a_point(copied_point = p) for p in t.points ] return(tracks) def big_point_tracks_to_original_tracks(tracks, btracks) : """ Given original tracks and equivalent tracks of big_points, make a set of tracks with the points being the a_point values from big points, but with the 'when'set to the original point's 'when'. """ btracks = make_array_of_tracks(btracks) bps = {} for bt in btracks : for bp in bt.points : for tp in bp.points : p = tracks[tp._track_idx].points[tp._point_idx] bps[id(p)] = bp # now we can find all the big_points for each of the original points pass pass rtracks = [] for t in tracks : points = [] for p in t.points : if id(p) in bps : np = a_point(copied_point = bps[id(p)]) np.when = p.when points.append(np) pass rtracks.append(a_track(points = points, id_num = t.id_num, when = t.when, name = t.name, description = t.description, file_name = t.file_name, color = t.color)) return(rtracks) def merge_tracks_to_points(tracks) : """ Merge points recorded at the same time in to new tracks. If it doesn't look like these tracks are really the same place, return []. """ points = [] if len(tracks) : for trk in tracks : trk.when = 0.0 if len(trk.points) : trk.points.sort(lambda a, b : cmp(a.when, b.when)) trk.when = trk.points[0].when pass tracks = sorted_tracks_by_when(tracks) errs = 0 tia = [ 0 ] * len(tracks) # make the merge-indices in to each track's points while True : pa = [ [ tracks[ti].points[tia[ti]], ti ] for ti in xrange(len(tia)) if tia[ti] < len(tracks[ti].points) ] # create array of the [ current point to merge and the point's points[] index ] for each track if not len(pa) : break # kick out if we've gathered all the points from all the tracks t = min([ int(round(p[0].when)) for p in pa ]) pa = [ p for p in pa if int(round(p[0].when)) == t ] # winnow the list down to those at the earliest time p = a_big_point() for pp in [ tp[0] for tp in pa ] : if p.cnt and (p.flat_distance_from(pp) > 30.0 / latlon.metersPerNauticalMile) : errs += 1 p.include_point(pp) points.append(p) for p in pa : tia[p[1]] += 1 # bump the points[] index for the points we've merged pass bi = 0 while bi < len(points) - 1 : if points[bi].cnt > points[bi + 1].cnt : break bi += 1 psa = points[:bi] ei = len(points) - 1 while ei > bi : if points[ei].cnt > points[ei - 1].cnt : break ei -= 1 ei += 1 pea = points[ei:] points = [ points[bi] ] + points[bi:ei] + [ points[ei - 1] ] for pi in xrange(len(points) - 2, 0, -1) : cnt = points[pi].cnt if (cnt == 1) and ((points[pi - 1].cnt > cnt) or (points[pi + 1] > cnt)) : # take out 1-track points in the middle - they will inject too much noise in the track del(points[pi]) # note: I've found that on the highway, the two GPS watches can differ by seconds in location! pass points = psa + points[1:-1] + pea # note: this stitch-together generates glitches set_points_duration(points) # print "@@@@", len(points), errs, errs / float(len(points)) if errs >= len(points) * 0.6 : # this logic and the 30 meters logic isn't very good !!!! return([]) pass return(points) def only_big_points_from_max_tracks(big_points) : mx_cnt = max([ p.cnt for p in big_points ]) return([ p for p in big_points if p.cnt == mx_cnt ]) class a_combined_point(a_big_point) : def __init__(me, point = None) : super(a_combined_point, me).__init__(point) me.nexts = {} # dictionary of points after this one, keyed by the points' ids me.prevs = {} def link_to_next_point(me, p) : if p and not tzlib.same_object(me, p): me.nexts[id(p)] = p p.prevs[id(me)] = me pass def link_to_prev_point(me, p) : if p and not tzlib.same_object(me, p): me.prevs[id(p)] = p p.nexts[id(me)] = me pass pass # a_combined_point class a_point_combiner(object) : """ This is a dictionary keyed by xyz point location values that have been scaled so we can look for nearby points. In particular, we know that the closest point to any location is within a 3x3 array of dicts of points. We do that by taking the location's xyz and adding and subtracting a quantum that assures us that we'll see all the combined points within the cutoff_distance from our location. Big points are stashed in our database as they develope. If the big point moves (because of points added to it) the big point might end up in more than 9 of our lists. But no biggy. Just slows down lookup a hair. We keep the points in 9 (or more) dictionaries keyed by ids so that we can quickly update a point's references. Note: Altitude isn't taken in to consideration because the big point xyz doesn't factor it in. It really should be so that airplace flights don't get combined with strolls down the sidewalk. But the same problem exists for driving and walking. Driving in the road, walking along side the road. """ def __init__(me, cutoff_distance = None) : me.cutoff_distance = cutoff_distance or (5.0 / latlon.metersPerNauticalMile) ( lat, lon ) = latlon.distance_in_lat_lon_here(0.0, 0.0, me.cutoff_distance) ( xd, yd, zd ) = latlon.convert_lat_lon_to_x_y_z(lat, lon) ( xz, yz, zz ) = latlon.convert_lat_lon_to_x_y_z(0.0, 0.0) me.scale_div = max(abs(xd - xz), abs(yd - yd), abs(zd - zz)) * 2.0 # the 2x is so i can experiment with other things, though indications are that it also affects the logic, which it should not !!!! me.points = {} # dictionary of lists of a_combined_point's keyed by scaled xyz values. me.done_cnt = 1 def point_xyz(me, p) : try : return( ( int((p.x_tot / p.cnt) / me.scale_div), int((p.y_tot / p.cnt) / me.scale_div), int((p.z_tot / p.cnt) / me.scale_div) ) ) # if it's a big point we have the xyz already except AttributeError : pass except ZeroDivisionError : pass ( x, y, z ) = p.x_y_z() return( ( int(x / me.scale_div), int(y / me.scale_div), int(z / me.scale_div) ) ) def xyz_key(me, x, y, z) : return("%u-%u-%u" % ( x, y, z ) ) def include_point(me, p) : if p : try : x = p.prevs x = p.nexts except AttributeError : p = a_combined_point(p) xyz = me.point_xyz(p) for x in xrange(-1, 2) : for y in xrange(-1, 2) : for z in xrange(-1, 2) : k = me.xyz_key(xyz[0] + x, xyz[1] + y, xyz[2] + z) pts = me.points.get(k, {}) pts[id(p)] = p me.points[k] = pts pass pass pass return(p) # tell him the combined point so he can add info to it def find_nearest_point(me, p) : bp = None bd = me.cutoff_distance ids = {} xyz = me.point_xyz(p) for x in xrange(-1, 2) : for y in xrange(-1, 2) : for z in xrange(-1, 2) : k = me.xyz_key(xyz[0] + x, xyz[1] + y, xyz[2] + z) pts = me.points.get(k, None) if pts : for i in pts.keys() : if not ids.has_key(i) : # avoid doing flat_distance_from by ignoring points we've already looked at ids[i] = True pp = pts[i] d = pp.flat_distance_from(p) if bd > d : bd = d bp = pp pass pass pass pass pass pass return(bp) def combine_near_points(me, points) : pp = None for p in points : np = me.find_nearest_point(p) if not np : np = a_combined_point(p) else : np.include_point(p) # put the original point in to the combined point me.include_point(np) # make sure we know about the combined point np.link_to_prev_point(pp) pp = np pass def _add_tracks_from_point(me, ids, track) : tracks = [] p = track[-1] pid = id(p) ppt = tracks[-2:] cutoff = 1 nps = p.nexts for nid in nps.keys() : np = nps[nid] if nid > pid : i = str(pid) + " " + str(nid) else : i = str(nid) + " " + str(pid) if not ids.has_key(i) : ids[i] = True track.append(np) tracks += me._add_tracks_from_point(ids, track) # do the current track depth first track = [ p ] # copy.copy(ppt) # now try for any others starting from this point - THIS HAS JUST THE opposite effect as we want cutoff = len(track) pass if len(track) > cutoff : # we don't include stand-alone points if nps : # track.append(nps.values()[0]) # this hurts. why? pass tracks.append(track) return(tracks) def convert_tracks_to_tracks_of_combined_points(me, tracks) : """ Given the tracks, return the tracks with all the points being combined points or raw, singleton big points that match to the tracks' points. """ rtrks = [] tracks = make_array_of_tracks(tracks) for t in tracks : t = copy.copy(t) if t.points : points = [ me.find_nearest_point(p) or a_big_point(p) for p in t.points ] lp = points[-1] points = [ points[pi] for pi in xrange(len(points) - 1) if not tzlib.same_object(points[pi], points[pi + 1]) ] if (len(t.points) >= 2) : if not tzlib.same_object(points[-1], lp) : points.append(lp) pass t.points = points rtrks.append(t) pass return(rtrks) def map_tracks(me) : tracks = [] ids = {} for k in me.points.keys() : pts = me.points[k] for pid in pts.keys() : if not ids.has_key(pid) : p = pts[pid] if not len(p.prevs) : # do one way tracks first by starting with only points that don't have predecessors ids[pid] = True tracks += me._add_tracks_from_point(ids, [ p ]) pass pass pass for k in me.points.keys() : pts = me.points[k] for pid in pts.keys() : if not ids.has_key(pid) : ids[pid] = True p = pts[pid] tracks += me._add_tracks_from_point(ids, [ p ]) # do circular tracks pass pass tracks = make_no_duplicate_points_in_tracks(tracks) return(tracks) def find_hold_still_point_pairs(me, tracks, hold_still_time = None) : hold_still_time = hold_still_time or (1.0 * 60.0) tracks = make_array_of_tracks(tracks) stops = [] for t in tracks : fp = None pp = None pbp = None for p in t.points : bp = me.find_nearest_point(p) if not tzlib.same_object(bp, pbp) : if fp and pp and (pp.when - fp.when >= hold_still_time) : stops.append( [ fp, pp ] ) fp = p if not bp : fp = None pp = p pbp = bp if fp and pp and (pp.when - fp.when >= hold_still_time) : stops.append( [ fp, pp ] ) pass return(stops) # return an array of point-pairs marking when we stopped and were last stopped pass # a_point_combiner if False : def make_ends_combiner_for_tracks(tracks, cutoff_distance) : tracks = make_array_of_tracks(tracks) cutoff_distance = cutoff_distance or None if cutoff_distance : cutoff_distance *= 4 ends = a_point_combiner(cutoff_distance) for t in tracks : if t.points and (len(t.points) >= 2) : ends.include_point(a_combined_point(t.points[ 0])) ends.include_point(a_combined_point(t.points[-1])) pass return( ( tracks, ends ) ) def attach_ends_to_tracks(tracks, ends) : tracks = make_array_of_tracks(tracks) for t in tracks : if t.points and (len(t.points) >= 2) : p = ends.find_nearest_point(t.points[0]) if p : t.points.insert(0, copy.copy(p)) # print "beg FOUND", t.points[-1] else : # print "beg not found", t.points[0] pass p = ends.find_nearest_point(t.points[-1]) if p : t.points.append(copy.copy(p)) # print "end FOUND", t.points[-1] else : # print "end not found", t.points[-1] pass pass pass return(tracks) pass def _combine_near_points_in_tracks(tracks, cutoff_distance, end_points = None, show_info = False) : end_points = end_points or [] c = a_point_combiner(cutoff_distance) for t in tracks : c.combine_near_points(t.points) map_them = False if end_points : map_them = True for be in end_points : np = c.find_nearest_point(be[0]) # !!!! we need to look further afield for the nearest point since it may have drifted to far to find using this combiner if np : p = a_combined_point(be[0]) c.include_point(p) p.link_to_next_point(np) # !!!! this is a guess. the found point may be the new ending point, for instance pp = c.find_nearest_point(be[1]) if pp : p = a_combined_point(be[1]) c.include_point(p) p.link_to_prev_point(pp) pass pass if tracks : tracks = c.convert_tracks_to_tracks_of_combined_points(tracks) if (not tracks) or map_them : tracks = c.map_tracks() if show_info : print count_points_in_tracks(tracks), "points in", len(tracks), "tracks" return(tracks) def combine_near_points_in_tracks(tracks, cutoff_distance, map_them = False, show_info = False) : tracks = make_array_of_tracks(tracks) cnt = count_points_in_tracks(tracks) if show_info : print cnt, "points in", len(tracks), "tracks" if True : tracks = _combine_near_points_in_tracks(tracks, cutoff_distance / 8.0, show_info = show_info) tracks = _combine_near_points_in_tracks(tracks, cutoff_distance / 4.0, show_info = show_info) tracks = _combine_near_points_in_tracks(tracks, cutoff_distance / 2.0, show_info = show_info) end_points = [] if map_them : for t in tracks : if t.points : end_points.append((t.points[ 0], t.points[-1])) pass pass tracks = _combine_near_points_in_tracks(tracks, cutoff_distance, end_points, show_info = show_info) while map_them : ncnt = count_points_in_tracks(tracks) spp = 2.0 if ncnt : spp = max(spp, float(cnt) / float(ncnt)) tracks = set_equal_durations_in_tracks(tracks, spp) tracks = smooth_tracks(tracks) tracks = _combine_near_points_in_tracks(tracks, cutoff_distance, end_points, show_info = show_info) ccnt = count_points_in_tracks(tracks) if ccnt >= ncnt : break pass return(tracks) class a_track_segmenter_settings(object) : def __init__(me) : me.hike() def hike(me) : me.ends_distance = 40.0 / latlon.metersPerNauticalMile # first big points are extracted to this level of spacial resolution me.min_distance = 450.0 / latlon.metersPerNauticalMile # segment must go at least this far from the first point (poorly named !!!! min_span? min_extent_from_start? req_extent_from_start? min_distance should be req_distance and refer to farthest-from-each-other two points in tracks) me.req_alt_diff = 1000 / latlon.feetPerMeter # sement required at least this difference in altitude from highest to lowest points (or min_distance) ( !!!! might consider not using flat_distance ) me.min_duration = 15.0 * 60.0 # segment must take this much time, at least me.max_speed = 15.0 # all segment points' speeds must be under this kph me.max_jump = None # all segment points cannot move more than this far from the previous point me.max_sleep = 3.0 * 60.0 # all segment points cannot be move more than this time after the previous point return(me) def bike(me) : me.hike() me.max_speed = 35 me.min_distance = 1200.0 / latlon.metersPerNauticalMile return(me) pass # a_track_segment_settings def extract_track_segments(tracks, settings, show_info = False) : """ Extract track segments from the given tracks. The segments are recognized 'cause they satisfy the settings' values. """ def _maybe_found(fpoints, from_i, to_i, settings) : if from_i >= 0 : ffp = fpoints[from_i] lfp = fpoints[to_i - 1] fsi = ffp.track_i fei = lfp.track_i + lfp.cnt # # # strip too-fast speeds from the beginning and end # speed is so glitchy that this won't for sure get rid of driving to a trailhead or driving away, but it will help some # # trk = ffp.track for i in xrange(ffp.cnt) : if trk[fsi].speed <= settings.max_speed : break fsi += 1 trk = lfp.track for i in xrange(lfp.cnt) : if trk[fei - 1].speed <= settings.max_speed : break fei -= 1 # # Note that if both points are completely whacked off, something is very wrong. The big point had a low speed, but all the little points don't! # # # # !!!! Now, what should be done on each end is to look for the minimum speed points that are followed by at least one point at 1.3 times the minimum speed found so far. # We're looking for standing around, then walking. And, we're looking to get rid of the final slow-down driving in, and the initial ramp-up in speed as the parking lot is traversed. # # if lfp.when - ffp.when >= settings.min_duration : # the segment must have been for a minimum amount of time mx_d = 0 for i in xrange(from_i + 1, to_i) : d = ffp.flat_distance_from(fpoints[i]) if d >= settings.min_distance : # the guy must have gone a minimum distance from the starting point # print "%s is %.3f from %s : %u:%u" % ( str(ffp), ffp.flat_distance_from(fpoints[i]), str(fpoints[i]), ffp.track_i, lfp.track_i + lfp.cnt ) return( ( lfp.track, fsi, fei ) ) mx_d = max(mx_d, d) if mx_d >= settings.min_distance / 2.0 : # check whether this was a half-min_distance-from-start-in-flat-distance climb or drop hi_alt = -10000000 lo_alt = 10000000 for i in xrange(from_i + 1, to_i) : alt = fpoints[i].altitude or 0 hi_alt = max(hi_alt, alt) lo_alt = min(lo_alt, alt) if hi_alt - lo_alt >= settings.req_alt_diff : # allow a flat distance to be low if the altitude difference from high to low is a lot return( ( lfp.track, fsi, fei ) ) pass pass pass return( ( None, 0, 0 ) ) def _maybe_remember(otracks, fpoints, from_i, to_i, setttings, name, file_name, color) : ( fnd_t, fsi, fei ) = _maybe_found(fpoints, si, pi, settings) # get the track and beginning and ending offset of the points to include if fnd_t : otracks.append(a_track(fnd_t[fsi:fei], len(otracks), name = name, file_name = file_name, color = color)) pass tracks = fix_tracks_speeds(tracks) # the points must have valid speeds to be used because we cut segments when the speed exceeds settings.max_speed ftracks = make_all_big_points(tracks, 10, settings.ends_distance, 0) # note if ends_distance is zero then we just get a copy of everything in a_big_point format - that's ok, 'cause we return the original, raw points if show_info : print "Computed filtered tracks:", len(ftracks), if len(ftracks) : print "with", len(ftracks[0].points), "points in first track" print otracks = [] for t in ftracks : olen = len(otracks) fpoints = t.points tname = t.full_name() tcolor = t.color_gbr_val() tfn = t.file_name pi = 0 si = -1 while pi < len(fpoints) : click = True p = fpoints[pi] if p.speed <= settings.max_speed : if si < 0 : si = pi click = False elif fpoints[pi - 1].flat_distance_from(p) >= settings.max_jump : # end on hyperspace jump pass elif p.first_raw_point().when - fpoints[pi - 1].last_raw_point().when >= settings.max_sleep : # end if the point is too far apart in time beyond the previous point pass else : click = False pass else : pass if click : _maybe_remember(otracks, fpoints, si, pi, settings, tname, tfn, tcolor) si = -1 pi += 1 _maybe_remember( otracks, fpoints, si, pi, settings, tname, tfn, tcolor) if show_info and (len(otracks) != olen) : print len(otracks) - olen, "tracks found from", tname pass return(otracks) def set_tracks_points_to_known_track_point_idxes(tracks) : """ Set each point._track_idx and point._point_idx to values appropriate for the point. """ tracks = make_array_of_tracks(tracks) ti = 0 for t in tracks : pi = 0 for p in t.points : p._track_idx = ti p._point_idx = pi pi += 1 ti += 1 return(tracks) def do_tracks_distance(tracks, rtn) : """ Return the sum of all the nautical mile distances for an array of tracks. """ tracks = make_array_of_tracks(tracks) sm = 0.0 for t in tracks : sm += do_points_distance(t.points, rtn) return(sm) def tracks_distance(tracks) : """ Sum up all the p.distance_from() values for an array of tracks. Return the tracks' nautical mile distance. """ return(do_tracks_distance(tracks, a_point.distance_from)) def tracks_flat_distance(tracks) : """ Sum up all the a_point.flat_distance_from() values for an array of tracks. Return the tracks' nautical mile distance. """ return(do_tracks_distance(tracks, a_point.flat_distance_from)) def tracks_altitude_info(tracks) : """ Return None or information about the tracks' altitude meters lost, gained, high and low extremes. """ tracks = make_array_of_tracks(tracks) info = None for t in tracks : i = points_altitude_info(t.points) if i : if info : info.merge_in(i) else : info = i pass pass return(info) def time_strip_tracks(tracks, start_time = None, end_time = None) : """ Return the array of tracks stripped of any points outside the start/end times [start end). """ if (start_time != None) or (end_time != None) : if start_time == None : start_time = -10000000 * 365 * 24 * 60 * 60 if end_time == None : end_time = 10000000 * 365 * 24 * 60 * 60 tracks = make_array_of_tracks(tracks) for t in tracks : t.points = [ p for p in t.points if (p.when != None) and (start_time <= p.when < end_time) ] tracks = [ t for t in tracks if len(t.points) ] return(tracks) # # # File writing stuff... # # leading_spaces_re = re.compile(r"^\s+", re.MULTILINE) def _write_file(file_name, fs, strip_spaces = False) : if strip_spaces : fs = leading_spaces_re.sub("", fs) tname = file_name + ".tmp" fo = open(tname, "w") fo.write(fs) fo.close() replace_file.replace_file(file_name, tname, file_name + ".bak") def write_label_points(points, to_rtn) : fs = "" for p in points : fs += to_rtn(p, is_waypoint = True) return(fs) def write_waypoints_and_waypoint_tracks(waypoints, tracks, to_rtn) : fs = "" # # First output the real waypoints, if any # waypoints = waypoints or [] for p in waypoints : fs += to_rtn(p, is_waypoint = True) tracks = tracks or [] tracks = make_array_of_tracks(tracks) # # Then output any intuited waypoints' tracks # wpt_tracks = [] for ti in xrange(len(tracks) - 1, -1, -1) : if are_all_points_waypoints(tracks[ti].points) : wpt_tracks.append(tracks.pop(ti)) pass wpt_tracks.reverse() # we did the tracks in reverse so we could whack 'em if they were waypoint tracks for t in wpt_tracks : for p in t.points : fs += to_rtn(p, is_waypoint = True) pass return( ( tracks, fs ) ) def _do_threads_to_save(tracks, rtn, name) : if threading.activeCount() > 1 : """ OK. This needs an explanation. 1st: It does no great harm, apparently, from the command line even if there is more than one thread - or it does this logic whether the tread count is 1 or not. 2nd: Under py2.4:XP and py2.5:Ubuntu, it makes the file save routine drastically faster in a (TrodTrack) multitasking situation. (presumably by hogging the cpu, but i have no idea why. probably not a gc.collect() thing, though. ) """ def _do_one(rtn, sa, i, t) : sa[i] = "".join( [ rtn(p) for p in t.points ] ) sa = [ "" for t in tracks ] ta = [ None for t in tracks ] ti = 0 for t in tracks : ta[ti] = threading.Thread(target = _do_one, name = "%s_%u" % ( name, ti ), args = ( rtn, sa, ti, t) ) ta[ti].setDaemon(True) ta[ti].start() ti += 1 while ti >= 0 : ti -= 1 ta[ti].join() pass return(sa) return(None) def gpx_file_string(file_name, program_name, creator_name, waypoints = None, tracks = None, label_points = None) : waypoints = waypoints or [] tracks = tracks or [] label_points = label_points or [] fs = gpx_header(program_name, file_name, creator_name) fs += write_label_points(label_points, a_point.to_gpx) ( tracks, s ) = write_waypoints_and_waypoint_tracks(waypoints, tracks, a_point.to_gpx) fs += s sa = _do_threads_to_save(tracks, a_point.to_gpx, "gpx") ti = 0 for t in tracks : ti += 1 fs += gpx_route_header(ti, len(tracks), t.name or t.id_num, t.when, description = t.description) if sa : fs += sa[ti - 1] else : for p in t.points : fs += p.to_gpx() pass fs += gpx_route_trailer() fs += gpx_trailer() return(fs) def write_gpx_file(file_name, program_name, creator_name, waypoints = None, tracks = None, strip_spaces = False, timed = True, label_points = None) : fs = gpx_file_string(file_name, program_name, creator_name, waypoints, tracks, label_points) _write_file(file_name, fs, strip_spaces) def loc_file_string(file_name, program_name, creator_name, waypoints = None) : waypoints = waypoints or [] fs = loc_header(program_name, file_name, creator_name) ( tracks, s ) = write_waypoints_and_waypoint_tracks(waypoints, [], a_point.to_loc) fs += s fs += loc_trailer() return(fs) def write_loc_file(file_name, program_name, creator_name, waypoints = None, tracks = None, strip_spaces = False, timed = True, label_points = None) : fs = loc_file_string(file_name, program_name, creator_name, waypoints) _write_file(file_name, fs, strip_spaces) def kml_file_string(file_name, program_name, creator_name, waypoints = None, tracks = None, timed = True) : waypoints = waypoints or [] tracks = tracks or [] fs = kml_header(program_name, file_name, creator_name) ( tracks, s ) = write_waypoints_and_waypoint_tracks(waypoints, tracks, a_point.to_kml) fs += s sa = _do_threads_to_save(tracks, a_point.to_kml, "kml") ti = 0 for t in tracks : ti += 1 if not timed : fs += kml_route_header(ti, len(tracks), t.name or t.id_num, t.when, name = t.name, color = t.color_gbr_val(), opacity = t.get_opacity(), description = t.description) if sa : fs += sa[ti - 1] else : for p in t.points : fs += p.to_kml() pass fs += kml_route_trailer() else : fs += kml_timed_route(t.points, route_number = ti, number_of_routes = len(tracks), name = t.name, color = t.color_gbr_val(), opacity = t.get_opacity(), description = t.description) pass fs += kml_trailer() return(fs) def write_kml_or_kmz_string_to_file(file_name, s, strip_spaces = False, when = None) : if os.path.splitext(file_name)[1].lower() == ".kmz" : if strip_spaces : s = leading_spaces_re.sub("", s) tname = file_name + ".tmp" zf = zipfile.ZipFile(tname, "w", zipfile.ZIP_DEFLATED) when = when or time.time() i = zipfile.ZipInfo(filename = "doc.kml", date_time = time.localtime(when)) i.compress_type = zipfile.ZIP_DEFLATED i.internal_attr = 0 i.external_attr = (0x8000 | 0774) << 16 zf.writestr(i, s) zf.close() replace_file.replace_file(file_name, tname, file_name + ".bak") else : _write_file(file_name, s, strip_spaces) pass def write_kml_file(file_name, program_name, creator_name, waypoints = None, tracks = None, strip_spaces = False, timed = True) : s = kml_file_string(file_name, program_name, creator_name, waypoints, tracks, timed) write_kml_or_kmz_string_to_file( file_name, s, strip_spaces = strip_spaces) def nmea_file_string(waypoints = None, tracks = None) : waypoints = waypoints or [] tracks = tracks or [] fs = "" ( tracks, s ) = write_waypoints_and_waypoint_tracks(waypoints, tracks, a_point.to_nmea) fs += s sa = _do_threads_to_save(tracks, a_point.to_nmea, "nmea") ti = 0 for t in tracks : ti += 1 if sa : fs += sa[ti - 1] else : for p in t.points : fs += p.to_nmea() pass pass return(fs) def write_nmea_file(file_name, program_name, creator_name, waypoints = None, tracks = None, strip_spaces = False, timed = True) : fs = nmea_file_string(waypoints = waypoints, tracks = tracks) _write_file(file_name, fs, strip_spaces) def write_gps_file(file_name, program_name = None, creator_name = None, waypoints = None, tracks = None, strip_spaces = False, timed = True, label_points = None) : waypoints = waypoints or [] tracks = tracks or [] label_points = label_points or [] tracks = make_array_of_tracks(tracks) if os.path.splitext( file_name)[1].lower() == ".kmz" : write_kml_file( file_name, program_name, creator_name, waypoints, tracks, strip_spaces, timed) elif os.path.splitext(file_name)[1].lower() == ".kml" : write_kml_file( file_name, program_name, creator_name, waypoints, tracks, strip_spaces, timed) elif os.path.splitext(file_name)[1].lower() == ".nmea" : write_nmea_file( file_name, program_name, creator_name, waypoints, tracks, strip_spaces, timed) elif os.path.splitext(file_name)[1].lower() == ".gpx" : write_gpx_file( file_name, program_name, creator_name, waypoints, tracks, strip_spaces, timed, label_points) elif os.path.splitext(file_name)[1].lower() == ".loc" : write_loc_file( file_name, program_name, creator_name, waypoints, tracks, strip_spaces, timed, label_points) else : write_gpx_file( file_name, program_name, creator_name, waypoints, tracks, strip_spaces, timed, label_points) # default output file is .gpx as it contains the most information pass help_str = """ %s (options) gpx_kmz_kml_nmea_file(s) Print information about the given GPS track files. Or convert file(s) to a new, output file. Options: --output file_name Convert to the given file name. Type will be file name extension's. All input files are combined. --strip_spaces Strip leading spaces in output file. --combine_zero_speed_points Contiguous points with zero speed values are all combined. --combine_near_points Make 'big points' from contiguous, nearby points. --merge Merge point recorded at same time/place. --sparsify Eliminate points in direct lines. --reverse Reverse the points in each track. --cutoff_distance meters Set a value used by --combine_near_points --sparsify --merge (default: %f) --cutoff_time hh:mm:ss Set a value used by --combine_near_points. (default: forever) --start_time when Strip points before this time (or with unknown times). --end_time when Strip points at this time or after (or with unknown times). """ # # # Main program # # if __name__ == '__main__' : import glob import TZCommandLineAtFile program_name = sys.argv.pop(0) TZCommandLineAtFile.expand_at_sign_command_line_files(sys.argv) strip_spaces = False zero = False rev = False merge = False sparsify = False near = 0 cutoff_distance = 5.0 / latlon.metersPerNauticalMile cutoff_time = None ofile_name = None start_time = None end_time = None while True : oi = tzlib.array_find(sys.argv, [ "--help", "-h", "-?", "/h", "/?", ] ) if oi < 0 : break del sys.argv[oi] print >> sys.stderr, help_str % ( program_name, cutoff_distance * latlon.metersPerNauticalMile, ) sys.exit(254) while True : oi = tzlib.array_find(sys.argv, [ "--strip_spaces", "-s"] ) if oi < 0 : break del sys.argv[oi] strip_spaces = True while True : oi = tzlib.array_find(sys.argv, [ "--combine_zero_speed_points", "-z"] ) if oi < 0 : break del sys.argv[oi] zero = True while True : oi = tzlib.array_find(sys.argv, [ "--combine_near_points", "-n"] ) if oi < 0 : break del sys.argv[oi] near += 1 while True : oi = tzlib.array_find(sys.argv, [ "--merge", "-m"] ) if oi < 0 : break del sys.argv[oi] merge = True while True : oi = tzlib.array_find(sys.argv, [ "--sparsify", "-r"] ) if oi < 0 : break del sys.argv[oi] sparsify = True while True : oi = tzlib.array_find(sys.argv, [ "--reverse", "-r"] ) if oi < 0 : break del sys.argv[oi] rev = True while True : oi = tzlib.array_find(sys.argv, [ "--cutoff_distance", "-d"] ) if oi < 0 : break del sys.argv[oi] cutoff_distance = float(sys.argv.pop(oi)) / latlon.metersPerNauticalMile while True : oi = tzlib.array_find(sys.argv, [ "--cutoff_time", "-t"] ) if oi < 0 : break del sys.argv[oi] cutoff_time = tz_parse_time.parse_time_zone(sys.argv.pop(oi)) if cutoff_time == None : print "I cannot understand the cutoff time!" sys.exit(102) pass while True : oi = tzlib.array_find(sys.argv, [ "--start_time", "--start-time", "--starttime", ] ) if oi < 0 : break del sys.argv[oi] start_time = tz_parse_time.parse_time(sys.argv.pop(oi)) if start_time == None : print "I cannot understand the start time!" sys.exit(102) pass while True : oi = tzlib.array_find(sys.argv, [ "--end_time", "--end-time", "--endtime", ] ) if oi < 0 : break del sys.argv[oi] end_time = tz_parse_time.parse_time(sys.argv.pop(oi)) if end_time == None : print "I cannot understand the end time!" sys.exit(102) pass while True : oi = tzlib.array_find(sys.argv, [ "--output", "--outfile", "-o"] ) if oi < 0 : break del sys.argv[oi] ofile_name = sys.argv.pop(oi) if zero or near or sparsify or merge or ofile_name : mfc = 2 if ofile_name : mfc = 1 if len(sys.argv) < mfc : print "Must have input .gpx file and output file given to do -z or -n ... or a -o output_file_name!" sys.exit(101) tracks = [] while len(sys.argv) >= mfc : fns = sys.argv.pop(0) fna = glob.glob(fns) if not fna : print "No files found: %s!" % fns sys.exit(102) for fn in fna : tracks += tracks_from_file(fn) pass tracks = time_strip_tracks(tracks, start_time, end_time) if merge : tracks = make_array_of_tracks(only_big_points_from_max_tracks(merge_tracks_to_points(tracks))) if zero : combine_zero_speed_points(tracks) if rev : rev = copy.deepcopy(tracks) reverse_tracks(rev) tracks = rev if near : tracks = make_all_big_points(tracks, near, cutoff_distance, cutoff_time) if sparsify : tracks = remove_redundant_points_from_tracks(tracks, cutoff_distance) if rev : reverse_tracks(tracks) if not ofile_name : ofile_name = sys.argv.pop(0) waypoints = [] for ti in xrange(len(tracks) - 1, -1, -1) : if len(tracks[ti].points) == 1 : waypoints += tracks.pop(ti).points else : wp = True for p in tracks[ti].points : if not p.is_likely_waypoint() : wp = False break pass if wp : waypoints += tracks.pop(ti).points pass pass waypoints.reverse() write_gps_file(ofile_name, "tz_gps.py", "tz_gps", waypoints, tracks, strip_spaces) sys.exit(0) # # # Print out the average and median of the points in the gpx files # # tracks = [] while len(sys.argv) : fns = sys.argv.pop(0) fna = glob.glob(fns) if not fna : print "No files found: %s!" % fns sys.exit(102) for fn in fna : tracks += tracks_from_file(fn) pass print "%-5u track%s" % ( len(tracks), tzlib.s_except_1(tracks) ) alat = 0.0 alon = 0.0 aalt = 0.0 lats = [] lons = [] alts = [] tracks = time_strip_tracks(tracks, start_time, end_time) # # This should use X Y Z logic !!!! # for t in tracks : print "%5u point%s in %s from %s" % ( len(t.points), tzlib.s_except_1(t.points), t.name or "", t.file_name or "" ) print " ", t.points[ 0], t.points[ 0].time_description(), t.points[ 0].description[:64] or "" if len(t.points) > 1 : print " ", t.points[-1], t.points[-1].time_description(), t.points[-1].description[:64] or "" for p in t.points : alat += p.lat alon += p.lon aalt += p.altitude if False and ((p.lat == 0.0) or (p.lon == 0.0) or (p.altitude == 0)) : print "z", p.lat, p.lon, p.altitude if p.lat : lats.append(p.lat) if p.lon : lons.append(p.lon) if p.altitude : alts.append(p.altitude) pass dis = points_flat_distance(t.points) print "Flat Distance: %.2f meters %.2f miles" % ( dis * latlon.metersPerNauticalMile, dis * latlon.milesPerNauticalMile ) dis = points_distance(t.points) print " Distance: %.2f meters %.2f miles" % ( dis * latlon.metersPerNauticalMile, dis * latlon.milesPerNauticalMile ) ai = points_altitude_info(t.points) if ai : print "Altitude: Low/High/Ediff: %u:%u:%u feet Up/Down %u:%u" % ( ai.lowest * latlon.feetPerMeter, ai.highest * latlon.feetPerMeter, (ai.highest - ai.lowest) * latlon.feetPerMeter, ai.gained * latlon.feetPerMeter, ai.lost * latlon.feetPerMeter ) pass print if lats : alat /= (len(lats) or 1) alon /= (len(lons) or 1) print print "Averages:" print " Latitude = %13.8f" % ( alat ) ( d, m, s, neg ) = latlon.lat_lon_in_degrees_minutes_seconds(alat) print " %4d.%02u.%.6f" % ( d, m, s ) ( d, m, s, neg ) = latlon.lat_lon_in_degrees_minutes_seconds(alon) print " Longitude = %13.8f" % ( alon ) print " %4d.%02u.%.6f" % ( d, m, s ) if len(alts) : aalt /= len(alts) print " Altitude = %13.8f meters" % ( aalt ) print " Altitude = %13.8f feet" % ( aalt * latlon.feetPerMeter ) lats.sort() lons.sort() mlat = lats[len(lats) / 2] mlon = lons[len(lons) / 2] print print "Medians:" print " Latitude = %13.8f" % ( mlat ) ( d, m, s, neg ) = latlon.lat_lon_in_degrees_minutes_seconds(mlat) print " %4d.%02u.%.6f" % ( d, m, s ) print " Longitude = %13.8f" % ( mlon ) ( d, m, s, neg ) = latlon.lat_lon_in_degrees_minutes_seconds(mlon) print " %4d.%02u.%.6f" % ( d, m, s ) if alts : alts.sort() malt = alts[len(alts) / 2] print " Altitude = %13.8f meters" % ( malt ) print " Altitude = %13.8f feet" % ( malt * latlon.feetPerMeter ) pass pass # # # eof