#!/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 # May 27, 2012 bar doxygen namespace # May 28, 2014 bar put thread id in threads # October 6, 2014 bar print total distances # --gpx_trk # January 20, 2015 bar able to handle minidom error from non-xml (nmea from gpsbabel) file # but reading nmea files doesn't work gpgga regex is nowhere near working - doesn't match the code, either # January 28, 2015 bar --new_speeds # figure out whether the gpx speed is kph or meters per second and fix them # use tzlib's find_argi # April 8, 2015 bar fix a None arith in GPX speed fixing # August 19, 2015 bar pyflakes # March 6, 2016 bar print the track duration and average speed # March 28, 2016 bar get_waypoints_from_tracks() # April 16, 2016 bar write zero speed to gpx files if the speed is not None # get rid of TINY_SPEED kludge # put enumerates in some places where it can go # April 22, 2016 bar nmea reading code cleanup # put out .gpx speeds in meters per second # put out mills in .gpx times # put out mph in .gpx speed also # May 11, 2016 bar make a_track_segmenter_settings cmparable # May 24, 2016 bar comment # July 21, 2016 bar speed up building big points for tracks that sit at one place for hours # July 22, 2016 bar fix that new code # July 23, 2016 bar tweak that new code # bikify up to 40kph from 35 because of a lake washington 23.6 sample - needs to be more resilient in the face of short bursts of slightly higher than allowed speeds - filter them or something # December 20, 2016 bar fix a spelling in comments # February 4, 2017 bar a_track.can_have_point() # February 5, 2017 bar time stamp the generically written file with the time of the last point # August 6, 2017 bar oh, golly, there are not 50 seconds in a minute # November 7, 2017 bar maxint->maxsize # July 3, 2018 bar fail to add a_track.__len__(). see note # raise bikify max_speed from 40 to 47 to include a downhill spot (need downhill_max_speed and trim_max_speed) # print mph on average speed printout # September 9, 2018 bar fix spaces in Duration printouts' seconds values # June 3, 2019 bar str of lat/lon circles # July 4, 2019 bar allow only file name extension for -o files # July 24, 2019 bar comments # July 26, 2019 bar find_spatially_separated_points() # July 28, 2019 bar a_point.set_when() # more on merge-prep # July 29, 2019 bar rename speed_to_kml() to something that makes sense (probably used for debugging - not by gps/tzpython/trodtrack # October 7, 2019 bar fix a crash in the new merge prep logic # December 5, 2019 bar max_max param to only_big_points_from_max_tracks() # January 12, 2020 bar track sort time now sub-sorts by dupeness, name, etc. # remove the dupes caused by all_hikes.kmz # January 13, 2020 bar .points, not .tracks in that sort # April 26, 2020 bar fix track color to BGR rather than GBR (apparently, that's how Google Earth likes 'em) # give the tracks that are half R and half G a lot of blue - Little_Si_geotrack effect is kinda kid-ugly, but the 5xx tracks can be seen, at least # April 27, 2020 bar a_geotrack track combination logic moved to here from thumbnail_htm (will this file ever be broken up?) # April 28, 2020 bar don't be a CPU hog when doing track combo logic # tune the point combiner and test it some on all hikes # April 29, 2020 bar handle all_hikes in the track combiner # May 12, 2020 bar a_track.nearest_point() and a_track.flat_nearest_point() # a_track don't do weird description setting from appended points (set the description to the appended descip if the track already had a description) # May 13, 2020 bar little cleanups # May 14, 2020 bar make_uncombined_geotracks() for thumbnail_htm.py and ancillary logic # May 28, 2020 bar a printout # August 20, 2020 bar --no_gpx_extensions option # --eodstamps-- ## \file # \namespace tzpython.tz_gps # # # 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 # http://aprs.gids.nl/nmea/ # # # 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. # # 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 glob import math import os import random import re import sys import threading import time import zipfile import latlon import replace_file import tzlib import tz_parse_time import tzspline INCLUDE_ALL_BIG_POINTS = False # -1 # True # set to True if we want big points to include all points - like if averaging a long time when the device is stable. Set to -1 for original, slow logic 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, sub_seconds = 0.0) : try : h = tm.tm_hour except AttributeError : tm = time.gmtime(tm) if sub_seconds : return( "%04u-%02u-%02uT%02u:%02u:%06.3fZ" % ( tm.tm_year, tm.tm_mon, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec + sub_seconds) ) 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 ) ) def latlon_str(lat, lon) : s = "" s += "lat=%.6f " % ( lat ) s += "lon=%.6f " % ( lon ) return(s) 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?) (lower the better. <1 is over-determined. 1.0 in 3D with 4 satellites is perfect) me.hdop = hdop # horizontal dilution of precision me.vdop = vdop # vertical dilution of precision 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_gpgga(object) : @staticmethod def from_string(nmea_gpgga_string) : if not nmea_gpgga_string : return(None) g = nmea_gpgga_re.search(nmea_gpgga_string) if not g : return(None) ( 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 return(a_gpgga(when, lat, lon, fix_typ, sat_cnt, hz_dispersion, altitude)) def __init__(me, when, lat, lon, fix_typ, sat_cnt, hz_dispersion, altitude) : me.when = when me.lat = lat me.lon = lon me.fix_typ = fix_typ me.sat_cnt = sat_cnt me.hz_dispersion = hz_dispersion me.altitude = altitude # a_gpgga class a_gprmc(object) : @staticmethod def from_string(nmea_gprmc_string) : if not nmea_gprmc_string : return(None) g = nmea_gprmc_re.search(nmea_gprmc_string) if not g : return(None) ( 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)) return(a_gprmc(when, lat, lon, speed, track_angle, month, day, year)) def __init__(me, when, lat, lon, speed, track_angle, month, day, year) : me.when = when me.lat = lat me.lon = lon me.speed = speed me.track_angle = track_angle me.month = month me.day = day me.year = year # a_gprmc class a_point(object) : def clear(me) : me.name = "" me.label = "" # from GPX - 'name' is sure used in many ways (waypointness, for instance) - and this label takes back seat to 'name' July 30, 2019 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.may_be_waypoint = True #: this point may be a waypoint (made false for lat/lon files) 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 set_when(me, when) : ov = me.when me.when = float(when) return(ov) def is_likely_waypoint(me) : """ Return whether this point is probably a waypoint. """ if me.duration or me.speed or (not me.may_be_waypoint) : 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 nautical 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 nautical 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 nautical 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.hypot(d, ad) return(d) def geo_speed_from(me, p) : """ Return None or the kph speed, given having moved flatly from 'p' to this point, or vice versa. """ td = abs((me.when - p.when) / 3600.0) if not td : return(None) d = me.flat_distance_from(p) * (latlon.metersPerNauticalMile / 1000.0) return(d / td) 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 += re.sub(r' +', ' ', latlon_str(me.lat, me.lon).strip()) 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 += latlon_str(me.lat, 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, gpx_trk = False, include_extensions = True) : 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_altitude_in_kml_string(me, is_waypoint = False, icon_image_url = None, description = None, gpx_trk = False) : """ Used for debugging? """ 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, gpx_trk = False, include_extensions = True) : s = "" if (me.lat != None) and (me.lon != None) : tag = (gpx_trk and "trkpt") or "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, me.when % 1.0) ) 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 include_extensions : # 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 : es += '\n %s%.2f' % ( ind, me.speed * (1000.0 / 3600.0)) es += '\n %s%.2f' % ( ind, me.speed * tzlib.KILOMETERS_TO_MILES) 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, gpx_trk = False) : 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, gpx_trk = False, include_extensions = True) : 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 = [ pp for pp 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 = [ pp.when for pp in points ] else : sa = [ float(i) for i in xrange(len(points)) ] ta = [ float(i) for i in xrange(len(points)) ] if True : ctc = [ pp.x_y_z() for pp 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 = [ pp.lat for pp in points ] aa = tzspline.a_smoother(ta, a)(sa) a = [ pp.lon for pp in points ] oa = tzspline.a_smoother(ta, a)(sa) a = [ pp.altitude for pp in points ] try : a.index(None) la = [ None ] * len(sa) except ValueError : la = tzspline.a_smoother(ta, a)(sa) a = [ pp.speed for pp in points ] try : a.index(None) pa = [ None ] * len(sa) except ValueError : pa = tzspline.a_smoother(ta, a)(sa) a = [ pp.heart_rate for pp 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. Note (May 24, 2016): Tried https://bost.ocks.org/mike/simplify/ a long time ago but found it didn't do quite as well at the corners for the hikify logic, if I recall. Visvalingam's algorithm and Douglas-Peucker algorithm. July 29, 2019 : Installed but did not try: https://pypi.org/project/visvalingamwyatt/ import visvalingamwyatt as vw Installed but did not try: https://github.com/urschrei/simplification import simplification.cutil """ 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 : # note: allow points with all zero speeds to be geo-speeded sc = None break pass for pi in xrange(1, len(points)) : p = points[pi] if (p.speed == sc) or (p.speed == None) or force : sp = p.geo_speed_from(points[pi - 1]) if sp != None : p.speed = sp 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) : """ Set values for points in lat/lon files and other places that don't have anything to say about the points other than lat/lon. """ 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.set_when(0.0) p.speed = 0.0 p.heart_rate = p.duration = None p.may_be_waypoint = False 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.set_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].set_when(when) when = points[0].when for p in points[1:] : if not p.when : if p.duration != None : if when : p.set_when(when + p.duration) pass else : retval = False pass when = p.when 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) def find_spatially_separated_points(points, how_many = 128) : """ Return an array of points that can be used to align two points lists in time. """ fix_points_speeds(points) how_many = max(1, how_many) pts = tzlib.find_representative_points_on_a_rubber_line(range(len(points)), how_many * 2, lambda i, j : points[i].flat_distance_from(points[j])) pts = [ points[pi] for pi in pts ] pts.sort(key = lambda p : p.speed) ln = len(pts) / 20 pts = pts[min(0, -(how_many + ln + ln)):] ln = len(pts) - how_many if ln > 0 : pts = pts[ln / 2 : -((ln + 1) / 2)] # take some off the start and end pts.sort(key = lambda p : p.when) return(pts) # # # 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, gpx_trk = False) : 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)) ) if gpx_trk : s = """ %s %s """ % ( tzlib.maybe_wrap_with_cdata(name), tzlib.maybe_wrap_with_cdata(desc + description), ) else : 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(gpx_trk = False) : if gpx_trk : return(" \n") 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") if not len(pts) : pts = r.getElementsByTagName("trkpt") 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() try : me.dom = xml.dom.minidom.parseString(file_or_str) except xml.parsers.expat.ExpatError : me.dom = None pass def get_tracks(me) : if not me.dom : return(None) 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) : # # # This should really extend a list - getting rid of .points. But, then, having .points is how other code detects a_track-ness, and who wants to find all the tz_gps-using modules anyway? # AND, if this extends list, then make_array_of_tracks() must be changed when it tries to sense arrays of arrays of tracks given to it! # # 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 me._mx_lat = None me._mx_lon = None # This can't be done unless make_array_of_tracks() is not so smart - or uses a better way to work, like looking for our type or attributes # def __len__(me) : # return(len(me.points)) def __init__(me, points = None, id_num = 0, when = None, name = None, description = None, file_name = None, color = None, opacity = None) : me.clear() me.append_track(points or []) 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 me._mx_lat = None me._mn_lat = None me._mx_lon = None me._mn_lon = None def append_track(me, t) : points = t if hasattr(t, 'points') : points = t.points me.file_name = me.file_name or t.file_name me.name = me.name or t.name me.description = me.description or t.description if len(points) : points[0].lat # type check the points points[0].lon # type check the points me._mx_lat = None me._mx_lon = None me.points += points return(len(points)) MAX_NEAR_LAT_LON = 0.001 def can_have_point(me, pnt, recalc = False, max_near_lat_or_lon = 0) : """ Return whether this track can have the given point, loosely speaking. This is used to speed up programs by ignoring tracks that the point can't be in. """ if recalc : me._mx_lat = None me._mx_lon = None if not len(me.points) : return(False) if pnt is None : return(False) max_near_lat_or_lon = max_near_lat_or_lon or me.MAX_NEAR_LAT_LON if me._mx_lat is None : me._mx_lat = max([ p.lat for p in me.points ]) me._mn_lat = min([ p.lat for p in me.points ]) if not (me._mn_lat - max_near_lat_or_lon <= pnt.lat <= me._mx_lat + max_near_lat_or_lon) : return(False) if me._mx_lat >= 89 : return(True) # be liberal in what we accept - longitude doesn't mean a whole lot at the poles if me._mx_lon is None : me._mx_lon = max([ p.lon for p in me.points ]) me._mn_lon = min([ p.lon for p in me.points ]) if me._mx_lon - me._mn_lon < 180 : if not (me._mn_lon - max_near_lat_or_lon <= pnt.lon <= me._mx_lon + max_near_lat_or_lon) : return(False) pass else : # note: it might be better to (maybe) sort the points and look at where the longest gap between adjacent points is to decide which direction the points go around the world - that is, consider the begin and end to be the two points furthest apart pass return(True) def nearest_point(me, pnt, metric_rtn = None) : if not me.points : return(None) metric_rtn = metric_rtn or a_point.distance_from da = [ metric_rtn(p, pnt) for p in me.points ] return(me.points[tzlib.index_of_min(da)]) def flat_nearest_point(me, pnt) : return(me.nearest_point(pnt, metric_rtn = a_point.flat_distance_from)) 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 = (b << 16) + (g << 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 blue = 255 - max(red, green) + 127 return(red, green, blue) 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 len(t.points) ] # simply get rid of empty tracks and throw an exception if the tracks don't have points (which points do, but we'll deal with them a couple of lines down) if len(rtracks) : t = rtracks[0] t.id_num # is the 1st track really a_track? t.when pass except AttributeError : try : tracks[0][0] # is tracks an array of arrays? except AttributeError : tracks = [ tracks ] # no, so make it one except TypeError : tracks = [ tracks ] # no, so make it one tracks = [ ta for ta in tracks if len(ta) ] if not tracks : return([]) if isinstance(tracks[0][0], a_track) : # is this an array of arrays of tracks rather than points? rtracks = tzlib.flatten_array(tracks) # flatten it and let them be recolored, too else : rtracks = [ a_track(pnts, i, pnts[0].when) for i, pnts in enumerate(tracks) ] # note that these tracks won't have a file_name 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 don't color (or recolored) his tracks return(rtracks) def sorted_tracks_by_when(tracks) : tracks = make_array_of_tracks(tracks) tracks = list(tracks) fns = {} for t in tracks : fn = t.file_name or "" fns[fn] = fns.get(fn, 0) + 1 # we'll sort the files with the most dupes to the end so all_hikes.kmz tracks will be whacked by remove_duplicate_tracks() tracks.sort(lambda a, b : cmp(a.when, b.when) or cmp(fns[a.file_name or ""], fns[b.file_name or ""]) or cmp(a.name or "", b.name or "") or cmp(a.file_name or "", b.file_name or "") or cmp(a.points[0].lon, b.points[0].lon) or cmp(a.points[0].lat, b.points[0].lat) or cmp(a.points[0].altitude, b.points[0].altitude)) 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) ] ) def find_nearest_tracks_point(tracks, point) : """ Return the closest-distance point to the given point in the given tracks. """ tracks = make_array_of_tracks(tracks) bp = None bd = sys.maxsize for t in tracks : if len(t.points) : if t.can_have_point(point) : p = t.nearest_point(point) d = point.distance_from(p) if bd >= d : bd = d bp = p pass pass pass return(bp) def find_most_contemporary_tracks_point(tracks, point) : """ Return the closest-time point to the given point in the given tracks. """ tracks = make_array_of_tracks(tracks) bp = None bt = sys.maxsize for t in tracks : if len(t.points) : if point.when <= t.points[ 0].when : bp = t.points[0] elif point.when >= t.points[-1].when : bp = t.points[-1] else : if t.points[0].when - bt <= point.when <= t.points[-1].when + bt : for p in t.points : d = abs(p.when - point.when) if bt >= d : bt = d bp = p pass pass pass pass pass return(bp) 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) def __str__(me) : return("%s radius:%.2f Miles" % ( latlon_str(me.lat, me.lon), me.radius * latlon.milesPerNauticalMile) ) 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, debug = False) : """ 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 gpgsa = None when = 0 month = 0 day = 0 year = 0 speed = None for ln in lns : ll = None gpgga = a_gpgga.from_string(ln) if gpgga : when = gpgga.when if p : if (int(p.when) % (60 * 60 * 24)) != int(gpgga.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 = gpgga.altitude p.fix_typ = gpgga.fix_typ p.sat_cnt = gpgga.sat_cnt p.hz_dispersion = gpgga.hz_dispersion pass if not p : if day : when = tz_parse_time.make_utime(month, day, year, 0, 0, gpgga.when) p = a_point(when = when, lat = gpgga.lat, lon = gpgga.lon, altitude = gpgga.altitude, speed = speed, gpgsa = gpgsa, fix_typ = gpgga.fix_typ, sat_cnt = gpgga.sat_cnt, hz_dispersion = gpgga.hz_dispersion) gpgsa = None ll = ln pass else : gprmc = a_gprmc.from_string(ln) if gprmc : when = gprmc.when speed = gprmc.speed month = gprmc.month or month day = gprmc.day or day year = gprmc.year or year 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.set_when(tz_parse_time.make_utime(month, day, year, 0, 0, when)) p.speed = gprmc.speed p.track_angle = gprmc.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 = gprmc.lat, lon = gprmc.lon, speed = gprmc.speed, gpgsa = gpgsa, track_angle = gprmc.track_angle) gpgsa = None ll = ln pass else : ngpgsa = a_gpgsa.from_string(ln) if ngpgsa : gpgsa = ngpgsa elif debug : print "@@@@", ln 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, debug = False) : if not isinstance(file_or_str, basestring) : file_or_str = file_or_str.read() tracks = [] lns = file_or_str.splitlines() points = extract_nmea_points_from_lines(lns, debug = debug) 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 fix_gpx_speeds(points) : """ GPX files apparently are supposed to have speeds in meters per second. gpsbabel puts them out that way, anyway. We used to put out kph, which we have the speed in internally. So, return a count of the number of mps speeds and set the speeds for the points to kph from mps if it seems like the best thing to do. """ mpscnt = 0 cnt = 0 kphs = [ p.speed for p in points ] pp = None for pi, p in enumerate(points) : if p.speed != None : kphs[pi] = p.speed * (3600.0 / 1000.0) # kph speed assuming p.speed is in mps if pp and p.speed : cnt += 1 gs = p.geo_speed_from(pp) or 0.0 if abs(gs - kphs[pi]) <= abs(gs - p.speed) : mpscnt += 1 pass pass pp = p if cnt and (mpscnt >= cnt * 0.5) : for p, kph in zip(points, kphs) : p.speed = kph return(mpscnt) return(0) 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 fix_gpx_speeds(points) 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 = [ [ pp[1], pp[0], pp[2] ] for pp in points ] # ugh. perhaps it's a kml file that we could not read points = [ a_point(lat = pp[0], lon = pp[1], altitude = pp[2]) for pp 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, filter_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. """ filter_rtn = filter_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 : if t.can_have_point(point) : for p in t.points : if filter_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 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 or '').lower(), (p2.point.name or '').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 each '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, filter_rtn = lambda p : p.when and True) # only find timed points - get_nearby_tracks_points() sorts the list by distance 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 = [] me.id_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) : if INCLUDE_ALL_BIG_POINTS or (len(me.points) < 3000) or (random.paretovariate(1/7.0) < len(me.points) + len(p.points)) : if (len(me.points) != len(me.id_points)) or (INCLUDE_ALL_BIG_POINTS == -1) : 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.id_points = pts # print "@@@@", len(me.points), len(me.id_points) else : pts = dict( [ ( id(pt), pt ) for pt in p.points ] ) me.id_points.update(pts) me.points = me.id_points.values() # print "@@@@", len(me.id_points), len(me.points), len(p.points), me, "|", len(me.points) and me.points[0], "|", len(p.points) and p.points[0] 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 pass 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) def set_when(me, when) : ov = me.when when = float(when) me.when_tot = when * me.cnt me.when = when return(ov) 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.maxsize 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) # make this a routine after calling count_points_in_tracks() because it wants the default (which is true for all points, but the code is optimized if the default is used) rcnt = tcnt for t in tracks : if len(t.points) : bpt = None bp = None for i, p in enumerate(t.points) : 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 pass 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.set_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 make_points_on_even_seconds(points) : """ Given an array of a_point()'s or a_big_point()'s, return an array of a_big_points()'s that all are on even seconds and cover all the seconds with (possibly interpolated) points. """ pts = [] if len(points) : ev = 1.0 pts = points stw = points[0].when w = ev for p in points[1:] : if p.when != stw + w : # is there a point not on the next even second? p = copy.deepcopy(points[0]) # well, then construct a new list of big points p.set_when(round(p.when)) # make the first point on an even second pts = [ a_big_point(p) ] # and include it in the output list (we could interpolate its location back from the 2nd point, but pffft) for pi, p in enumerate(points[1:]) : t = round(p.when) if t < pts[-1].when + 1 : # if the next point at, effectively, the same time as the latest one we're output? w = pts[-1].when pts[-1].include_point(p) # include it in the latest big point pts[-1].set_when(w) # but restore that big point's when else : pp = points[pi] d = p.when - pp.when # the duration between the previous input point and this input point - the X difference n = ev # make points at even seconds while pts[-1].when + ev < t : # until the lastest output point is just a second behind the input point f = n / d np = a_point(when = pts[-1].when + ev, lat = pp.lat + (f * (p.lat - pp.lat )), lon = pp.lon + (f * (p.lon - pp.lon )), altitude = pp.altitude + (f * (p.altitude - pp.altitude)) ) # make an interpolated point to fill in the seconds pts.append(a_big_point(np)) n += ev # and keep ticking the clock until we get to the input point's when pts.append(a_big_point(p)) # now add the input point to the output points pts[-1].set_when(round(pts[-1].when)) # and make sure its when is even pass set_points_duration(pts) # recalculate all the durations break # and, we're done w += ev pass return(pts) def points_location_difference(in_pts, find_pts, when_offset = 0.0, diff_func = None) : """ Given an array of points returned from make_points_on_even_seconds() and an array of points to find in them at a given time offset, return the average squared distance between the points. """ if diff_func is None : diff_func = a_point.distance_from cnt = 0 sm = 0.0 start = in_pts[0].when for p in find_pts : w = int(round(p.when + when_offset - start)) if 0 < len(in_pts) : cnt += 1 d = diff_func(in_pts[w], p) # print "@@@@", w, in_pts[w], p, d sm += (d ** 2) pass if not cnt : return(sys.float_info.max) # print "@@@@", sm, cnt, sm / cnt, when_offset return(sm / cnt) def find_points_time_offset(to_pts, pts, diff_func = None) : """ Find a time to offset the points in pts so that they seem to best match the to_pts points in some way (default: distance_from). """ to_pts = make_points_on_even_seconds(to_pts) bd0 = points_location_difference(to_pts, pts, 0.0, diff_func = diff_func) bp = bd0 pi = 0.0 while True : d = points_location_difference(to_pts, pts, pi + 1.0, diff_func = diff_func) if d > bp : break bp = d pi += 1.0 bn = bd0 ni = 0.0 while True : d = points_location_difference(to_pts, pts, ni - 1.0, diff_func = diff_func) if d > bn : break bn = d ni -= 1.0 if bp > bn : return(ni) return(pi) def prep_tracks_for_merge_to_points(tracks) : """ This routine changes the 'when' time stamps on the points so that the tracks match up in time to the second as best they can. """ tracks = [ t for t in tracks if len(t.points) ] for trk in tracks : trk.when = 0.0 trk.points.sort(lambda a, b : cmp(a.when, b.when)) trk.when = trk.points[0].when # make the track points rational - in order in time last_st = max([ t.when for t in tracks ]) frst_en = min([ t.points[-1].when for t in tracks ]) trks = [] for t in tracks : for bi, p in enumerate(t.points) : if p.when >= last_st : break pass for ei, p in enumerate(t.points[::-1]) : if p.when <= frst_en : break pass ei = max(0, len(t.points) - ei - 1) if bi < ei : trks.append(t.points[bi:ei]) # clip the ends off the tracks so they start and end at the same time pass tracks = make_array_of_tracks(trks) tracks.sort(lambda a, b : cmp(len(b.points), len(a.points))) # sync in to the track with the most points from where all tracks have points in roughly the same time period if len(tracks) : tracks[0].points = make_points_on_even_seconds(tracks[0].points) # make the first, most-point track have points at every second, so it's fast to access points by 'when' time stamp (these will be a_big_point()'s) for t in tracks[1:] : offset = find_points_time_offset(tracks[0].points, find_spatially_separated_points(t.points)) # print "@@@@", offset for p in t.points : p.set_when(p.when + offset) # change the 'when' times in the other tracks so that they match as well as possible with the 1st, most-point track pass pass return(tracks) 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 []. This logic all depends upon the track and points' time stamps. ' The object of this routine is to merge hiking/biking tracks recorded at the same time by the same person, but with multiple devices. Ideally, we want multiple tracks to blend to a more accurate track. """ # # Two tracks in parallel with each other may cause the # combined track to ping pong between the tracks. # Though the 'err' count may just be too high. # # Merging in a track to another and forgetting the 'when' # information from one of the tracks is fraught with peril, # too. That track's points would, presumably need to have # 'when' values interpolated from the two surrounding # points derived from the track whose 'when' values are # being used. Artificial 'when' values could affect # calculated speed. # # Another problem that comes up: What to do when one track # wanders off and the other stays put. Presumably, the # wandering track comes back to meet the "stay" track. And, # the "stay" tracks points should probably be forgotten. # But that does lead to trouble if the two merging tracks # are done at different times. Which is out-of-scope for # this logic. But still... # # Too, what happens when the two tracks fork and then join. # Two hikers taking slightly different routes. # # points = [] if len(tracks) : if True : tracks = prep_tracks_for_merge_to_points(tracks) # July 28, 2019 else : tracks = [ t for t in tracks if len(t.points) ] for trk in tracks : trk.when = 0.0 trk.points.sort(lambda a, b : cmp(a.when, b.when)) trk.when = trk.points[0].when 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) : # see "30 meters" comment below errs += 1 p.include_point(pp) points.append(p) # remember the big point for this time for p in pa : tia[p[1]] += 1 # bump the points[] index for the points we've merged pass if len(points) : bi = 0 while bi < len(points) - 1 : if points[bi].cnt > points[bi + 1].cnt : # strip big points at the start until there's one with more normal points going in to it than the next big point break bi += 1 psa = points[:bi] ei = len(points) - 1 while ei > bi : if points[ei].cnt > points[ei - 1].cnt : # strip big points from the end until there's one with more normal points going in to it than the previous big point 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 (this probably doesn't work with 3 tracks !!!! ) 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 pass return(points) def only_big_points_from_max_tracks(big_points, max_max = sys.maxsize) : mx_cnt = min(max_max, 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 airplane 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. Note: Use this: https://docs.scipy.org/doc/scipy-0.15.1/reference/generated/scipy.spatial.KDTree.query_ball_point.html#scipy.spatial.KDTree.query_ball_point using the points' XYZ positions And, if it doesn't make sense to pre-process the points being looked for, the cdist or faster method here may help: https://codereview.stackexchange.com/questions/28207/finding-the-closest-point-to-a-list-of-points """ def __init__(me, cutoff_distance = None) : me.cutoff_distance = cutoff_distance or (5.0 / latlon.metersPerNauticalMile) # print "@@@@", me.cutoff_distance, "naut miles" ( 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)) * 1.01 # oddly enough, 1.0 is faster than 2.0 - they both combine the same all hike/*.kmz tracks I have 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("%d:%d:%d" % ( 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 if me.points : 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 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) ############################################################ # # # GPS track inclusion # # Combine nearby tracks in to big a_geotrack()'s. # And make our 'trk_pts' for finding pictures' a_geotrack()'s. # And make our 'gts' lists of a_geotrack()'s for writing out as .kml/.kmz files, pictures and all. # # # class a_geotrack(object) : class a_geotrack_picture(object) : def __init__(me, name, kml = None) : me.name = name or "" # note: appears to be unused except as a key to the .pictures in a_geotrack - though if, say, a_geotrack has only one picture, it might make sense to name the geotrack by the picture name rather than the track name me.kml = kml or "" # KML text for the picture # a_geotrack_picture def __init__(me) : me.real_me = None # a pointer to the next in a chain ending with the combined geotrack that includes this one and maybe others (used by the logic to combine the tracks in to one, big track) me.url = "" # this will get some track name later me.pictures = {} # indexed by file name, the a_geotrack_picture()'s known to be near one of our tracks me.tracks = [] # array of actual tracks in this combined web of nearby tracks def copy(me) : """ Return a copy of us that keeps the tracks and pictures the same data. copy.copy(), would do the trick, too. """ nm = a_geotrack() nm.real_me = me.real_me nm.url = me.url nm.pictures.update(me.pictures) nm.tracks = list(me.tracks) return(nm) def point_count(me) : """ Return how many points we have in our tracks. """ return(sum([ len(t.points) for t in me.tracks ] + [ len(me.pictures) ])) def remove_dupe_tracks(me) : me.tracks = tzlib.without_dupes(me.tracks) def add_track(me, track) : """ Add a_track to this geotrack. """ me.tracks.append(track) def add_into(me, om) : """ Add the another other a_geotrack in to this one. """ me.tracks += om.tracks # use his tracks (watch out! multiple refs to these tracks!) me.pictures.update(om.pictures) def move_into(me, om) : """ Destructively merge another a_geotrack in to this one, moving the tracks and pictures over and pointing the other one to this one as the replacement of the other. """ if id(me) != id(om) : me.add_into(om) # take his tracks om.tracks = [] om.pictures = {} om.real_me = me # point him to us, so a geotrack points to a bigger, more inclusive geotrack (which becomes the parent of all the connected tracks) pass def add_kml_picture(me, name, pic_kml) : """ Add a picture's KML text to our 'name' keyed dictionary of pictures. The KML will be put in an output KML file made from this geotrack. """ me.pictures[name] = me.a_geotrack_picture(name, pic_kml) def me(me) : """ Return the geotrack that has this geotrack's tracks. """ while me.real_me : me = me.real_me # search up the tree to find the combined a_geotrack() that includes this one and maybe others pointing to it, too return(me) # a_geotrack def make_one_geotrack(gts, file_name) : """ Combine an array of a_geotrack()s in to a new one, setting the url in the new one. The resultant a_geotrack() may be empty if we are given no a_geotrack()s. Return the combined a_geotrack. """ ngt = a_geotrack() for gt in gts : gt = gt.me() for t in gt.tracks : ngt.add_track(t) ngt.pictures.update(gt.pictures) ngt.url = geotrack_name(os.path.splitext(os.path.basename(file_name))[0]) # note: the .ignore is replaced by .kmz return(ngt) MAX_TIME_DIST_FROM_TRACK = 1.0 # closest-time track points must be this close in nautical miles to the image point MAX_DIST_DIST_FROM_TRACK = 1.0 # closest-distance track points must be this close in nautical miles to the image point MAX_SECS_OVER_DIST_TRACK = 120.0 # if the closest-time track is this close in time, forget the closest-distance point MAX_REALLY_CLOSE_FROM_TRACK = 0.03 # if both found points are this close, forget the closest-distance one (note that the tracks have been optimized so the points are quite a distance from each other if the track is going in a straight line) def make_geotracks_from_point(gts, point) : """ Given a list of a_geotrack()s, return a list of any suitable a_geotrack()s for this (picture's) point. """ p_to_gt = {} # map points to a_geotrack that has the point p_to_gt[None] = None # the default, so to speak for gt in gts : for t in gt.tracks : for p in t.points : p_to_gt[p] = gt # so we can find the a_geotrack from the points we find pass pass dp = find_nearest_tracks_point( [ gt.tracks for gt in gts ], point) tp = find_most_contemporary_tracks_point([ gt.tracks for gt in gts ], point) if tp and (tp.distance_from(point) > MAX_TIME_DIST_FROM_TRACK) : # ignore close-time points if they are too far away to be reasonable tp = None if tp and (abs(tp.when - point.when) <= MAX_SECS_OVER_DIST_TRACK) : # ignore close-distance points if the time point is probably on the image's track dp = None if dp and (dp.distance_from(point) > MAX_DIST_DIST_FROM_TRACK) : # ignore close-distance points if they are too far away to be reasonable dp = None if tp and (tp.distance_from(point) <= MAX_REALLY_CLOSE_FROM_TRACK) : # if the closest-time point's distance is close, toss the closest-distance point (could be optimized) dp = None if dp and (dp.distance_from(point) <= MAX_REALLY_CLOSE_FROM_TRACK) : # if the closest-distance point's distance is close, toss the closest-time point tp = None if dp and tp and (id(p_to_gt[dp]) == id(p_to_gt[tp])) : tp = None # both are from the same track (or are both None), so forget one of 'em return([ t for t in [ p_to_gt[dp], p_to_gt[tp], ] if t ]) def _geotracks_from_gps_file(trk_pts, file_name_or_track) : """ Add a gps track file or a track to our a_geotrack's, returning a list of new a_geotrack()'s. This is where we combine nearby tracks in to big webs of tracks that can be written to one GPS (KML/KMZ) file. """ tracks = [ file_name_or_track ] if not hasattr(file_name_or_track, 'points') : tracks = tracks_from_file(file_name_or_track) gts = [] for t in tracks : if len(t.points) : gtd = {} # st = tzlib.elapsed_time() # print " @@@@", tzlib.elapsed_time(), "finding %u points" % len(t.points) cnt = 0 for p in t.points : p = trk_pts.find_nearest_point(p) if p : cnt += 1 gt = p._a_geotrack.me() # we've found a track with a point near one of ours, so add it to the ones we'll combine gtd[repr(gt)] = gt # note: we'll probaby assign the same a_geotrack to 'gts' for a lot of points in this track pass if gtd : gta = gtd.values() # get all the a_geotrack's we're near gt = gta.pop() # arbitrarily pick the first one to combine all of them in to for ogt in gta : gt.move_into(ogt) # move the tracks of all of the a_geotrack's we bridged to the first one pass else : gt = a_geotrack() # put this track in to a new a_geotrack because none of the ones we already have has a point near any point in the track gts.append(gt) # and now we know this new a_geotrack # print " @@@@", tzlib.elapsed_time(), "found %u trk_pts points %.3f per second, adding %u track points" % ( cnt, cnt / ((tzlib.elapsed_time() - st) or 0.0001), len(t.points), ) for p in t.points : cp = trk_pts.include_point(p) if cp : cp._a_geotrack = gt # reference the proper a_geotrack for each of the track's points so subsequent tracks' points can find this track's a_geotrack() pass gt.add_track(t) # put our track in the proper a_geotrack # print " @@@@", tzlib.elapsed_time(), "added track" pass return(gts) def _uncombined_geotracks_from_gps_file(trk_pts, file_name_or_track) : """ Add a gps track file or a track to our a_geotrack's, returning a list of new a_geotrack()'s. We don't combine the tracks. So, really, we're just creating a_geotracks() that are compatible, but uncombined. """ tracks = [ file_name_or_track ] if not hasattr(file_name_or_track, 'points') : tracks = tracks_from_file(file_name_or_track) gts = [] gt = None for t in tracks : if len(t.points) : if not gt : gt = a_geotrack() # create the a_geotrack() gts.append(gt) # we'll return the list of those we've created gt.add_track(t) # put this track in the a_geotrack pass return(gts) def kml_name(fn) : return(os.path.splitext(fn)[0] + ".kmz") def geotrack_name(n) : return(kml_name(tzlib.file_name_able(n).replace(" ", "_") + "_geotrack.ignore")) def _geotrack_name(t) : return(geotrack_name(t.name or t.file_name or "")) def _make_geotracks(fls_or_tracks, geotracks_from_gps_file_rtn, cutoff_distance = None) : """ Return an array of a_geotrack()'s and the a_point_combiner() used to combine the tracks in the a_geotrack()'s. """ trk_pts = a_point_combiner(cutoff_distance) gts = [] # all the a_geotrack's we make for fn_or_trk in fls_or_tracks : # print "@@@@", tzlib.elapsed_time(), fn, len(gts), len(trk_pts.points), sum([ len(tp) for tp in trk_pts.points ]) gts += geotracks_from_gps_file_rtn(trk_pts, fn_or_trk) gts = [ gt for gt in gts if gt.tracks ] # forget empty geotracks for gt in gts : gt.tracks = sorted_tracks_by_when(gt.tracks) gt.url = _geotrack_name(gt.tracks[0]) # give the combined a_geotrack the name of the earliest GPS track file's track name, or file name if the track(s) have no name return(gts, trk_pts) def make_geotracks(fls_or_tracks, cutoff_distance = None) : """ Return an array of a_geotrack()'s and the a_point_combiner() used to combine the tracks in the a_geotrack()'s. """ return(_make_geotracks(fls_or_tracks, _geotracks_from_gps_file, cutoff_distance = cutoff_distance)) def make_uncombined_geotracks(fls_or_tracks, cutoff_distance = None) : """ Return an array of a_geotrack()'s and the a_point_combiner() NOT used to combine the tracks in the a_geotrack()'s. """ return(_make_geotracks(fls_or_tracks, _uncombined_geotracks_from_gps_file, cutoff_distance = cutoff_distance)) def change_geotrack_lookup_distance(trk_pts, gts, cutoff_distance = None) : """ Given one that might be already, return a_point combiner() for the given distance for the given a_geotrack()'s. """ if (cutoff_distance is None) or (trk_pts.cutoff_distance != cutoff_distance) : # if the picture distance is different from the one we've already used, we must rebuild our trk_pts point finder to later find pictures using the picture distance ntrk_pts = a_point_combiner(cutoff_distance) # different distance - make a new one if trk_pts.points : # only include the tracks' points if they are already (presumably) in the trk_pts point-lookup thang - if no points, then trk_pts is a complicated mechanism to remember the cutoff_distance for gt in gts : for t in gt.tracks : for p in t.points : cp = ntrk_pts.include_point(p) if cp : cp._a_geotrack = gt # reference the a_geotrack that has the point pass pass pass pass trk_pts = ntrk_pts return(trk_pts) def find_nearest_geotrack(trk_pts, p) : """ Find a_geotrack that's nearest to the given point, if any. None if none found. """ cp = trk_pts.find_nearest_point(p) if cp : return(cp._a_geotrack.me()) # find a_geotrack near the point - that has the combined track(s) for the picture return(None) def write_geotrack_files(gts, to_dir, program_name = None, who = None, url_rtn = None) : for gt in gts : tracks = color_tracks(gt.tracks) s = kml_header(program_name, (url_rtn and url_rtn(gt.url)) or None, who) dt = 0 for fn, p in gt.pictures.items() : fn += ".jpg" if os.path.isfile(fn) : dt = max(dt, os.path.getmtime(fn)) s += p.kml ti = 0 for t in tracks : if t.points : dt = max(dt, t.points[-1].when) ti += 1 s += kml_timed_route(t.points, route_number = ti, number_of_routes = len(tracks), name = t.name) s += kml_trailer() fn = os.path.join(to_dir, gt.url) write_kml_or_kmz_string_to_file(fn, s) if dt : os.utime(fn, ( int(dt), int(dt) ) ) # set the file date/time to the newest picture's date/time pass pass def erase_geotrack_files(gts, to_dir) : for gt in gts : tzlib.whack_file(os.path.join(to_dir, gt.url)) pass # # # ############################################################ 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 = 47.0 me.min_distance = 1200.0 / latlon.metersPerNauticalMile return(me) def __cmp__(me, om) : if not om : return(1) return( cmp(me.max_speed, om.max_speed) or cmp(me.min_distance, om.min_distance) or cmp(me.ends_distance, om.ends_distance) or cmp(me.min_duration, om.min_duration) or cmp(me.max_jump, om.max_jump) or cmp(me.max_sleep, om.max_sleep) or cmp(me.req_alt_diff, om.req_alt_diff) ) pass def __str__(me) : return("MaxSpd:%.1fkph MinDist:%.1fmtrs MaxSlp:%.1fsecs MaxJmp:%.1fmtrs MinDur:%.1fsecs EndsDist:%.1fmtrs" % ( me.max_speed, me.min_distance * latlon.metersPerNauticalMile, me.max_sleep, ((me.max_jump is None) and -1.0) or me.max_jump * latlon.metersPerNauticalMile, me.min_duration, me.ends_distance * latlon.metersPerNauticalMile, ) ) pass 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)) return(True) return(False) 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(t.points) for t in ftracks ], "points" 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 si = -1 for pi, p in enumerate(fpoints) : click = True 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 : # print "@@@@", pi, p.speed pass if click : # maybe at the end of a track (if si >= 0)? push out the extracted track if _maybe_remember(otracks, fpoints, si, pi, settings, tname, tfn, tcolor) and show_info : print " ", len(otracks), len(fpoints), si, pi, (si >= 0) and fpoints[si], "|", fpoints[pi - 1], "---", otracks[-1].points[0], "|", otracks[-1].points[-1] si = -1 pass if _maybe_remember( otracks, fpoints, si, pi, settings, tname, tfn, tcolor) and show_info : # push out any extracted track at the end of the points print " ", len(otracks), len(fpoints), si, pi, (si >= 0) and fpoints[si], "|", fpoints[pi - 1], "---", otracks[-1].points[0], "|", otracks[-1].points[-1] 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) def get_waypoints_from_tracks(tracks) : """ Return an array of waypoints that are in the given tracks, and remove them from the tracks. """ 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() return(waypoints) # # # 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, gpx_trk = False, include_extensions = True) : 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, gpx_trk, include_extensions) : threading.current_thread().tid = tzlib.get_tid() sa[i] = "".join( [ rtn(p, gpx_trk = gpx_trk, include_extensions = include_extensions) 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, gpx_trk, include_extensions, ) ) 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, gpx_trk = False, include_extensions = True) : 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", gpx_trk = gpx_trk, include_extensions = include_extensions) 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, gpx_trk = gpx_trk) if sa : fs += sa[ti - 1] else : for p in t.points : fs += p.to_gpx(gpx_trk = gpx_trk, include_extensions = include_extensions) pass fs += gpx_route_trailer(gpx_trk = gpx_trk) 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, gpx_trk = False, include_extensions = True) : fs = gpx_file_string(file_name, program_name, creator_name, waypoints, tracks, label_points, gpx_trk = gpx_trk, include_extensions = include_extensions) _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, gpx_trk = False, date_the_file = True, include_extensions = True) : 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, gpx_trk = gpx_trk, include_extensions = include_extensions) 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, gpx_trk = gpx_trk, include_extensions = include_extensions) # default output file is .gpx as it contains the most information if date_the_file : bd = -sys.maxsize for trk in tracks : if len(trk.points) : d = trk.points[-1].when if bd < d : bd = d pass pass if bd > 0 : os.utime(file_name, ( os.path.getatime(file_name), int(round(bd)) ) ) pass pass def get_tracks_from_ambiguous_file_names(afns, start_time = None, end_time = None) : fns = [] while afns : afn = afns.pop(0) fna = glob.glob(afn) if not fna : print "No files found: %s!" % afn sys.exit(102) fns += fna fns = tzlib.remove_names_of_duplicate_files(fns) tracks = [] for fn in fns : tracks += tracks_from_file(fn) tracks = time_strip_tracks(tracks, start_time, end_time) tracks = remove_duplicate_tracks(tracks) # this also sorts them in time order - but it's here to get rid of all_hikes.kmz dupes return(tracks) 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. --new_speeds Bump the 'new_speed' count. 1->Fix speeds. 2->New speeds from lat/lon. --combine_zero_speed_points Contiguous points with zero speed values are all combined. --combine_near_points Make 'big points' from contiguous, nearby points. --combine_near_tracks out_dir Combine nearby tracks to output KMZ GPS files to the given directory. --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/_tracks --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). --gpx_trk Make output .gpx files contain trkpt's rather than rtept's. (as the world expects) --no_gpx_extensions Do not include our extensions in the output GPX files. """ # # # Main program # # if __name__ == '__main__' : import TZCommandLineAtFile program_name = sys.argv.pop(0) TZCommandLineAtFile.expand_at_sign_command_line_files(sys.argv) strip_spaces = False new_speeds = 0 zero = False rev = False merge = False sparsify = False near = 0 cutoff_distance = 5.0 / latlon.metersPerNauticalMile cutoff_time = None comb_trk_dir = None ofile_name = None start_time = None end_time = None gpx_trk = False include_extensions = True 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.find_argi(sys.argv, [ "--strip_spaces", "-s"] ) if oi < 0 : break del sys.argv[oi] strip_spaces = True while True : oi = tzlib.find_argi(sys.argv, [ "--new_speeds", "--ns"] ) if oi < 0 : break del sys.argv[oi] new_speeds += 1 while True : oi = tzlib.find_argi(sys.argv, [ "--combine_zero_speed_points", "-z"] ) if oi < 0 : break del sys.argv[oi] zero = True while True : oi = tzlib.find_argi(sys.argv, [ "--combine_near_points", "-n"] ) if oi < 0 : break del sys.argv[oi] near += 1 while True : oi = tzlib.find_argi(sys.argv, [ "--combine_near_tracks", ] ) if oi < 0 : break del sys.argv[oi] comb_trk_dir = sys.argv.pop(oi) if not os.path.isdir(comb_trk_dir) : print "%s is not a directory!" % comb_trk_dir sys.exit(102) pass while True : oi = tzlib.find_argi(sys.argv, [ "--merge", "-m"] ) if oi < 0 : break del sys.argv[oi] merge = True while True : oi = tzlib.find_argi(sys.argv, [ "--sparsify", "-r"] ) if oi < 0 : break del sys.argv[oi] sparsify = True while True : oi = tzlib.find_argi(sys.argv, [ "--reverse", "-r"] ) if oi < 0 : break del sys.argv[oi] rev = True while True : oi = tzlib.find_argi(sys.argv, [ "--cutoff_distance", "-d"] ) if oi < 0 : break del sys.argv[oi] v = sys.argv.pop(oi).strip().lower() if re.search(r'n\s*m$', v) : v = re.sub(r'n\s*m$', '', v) v = float(v) elif v.endswith('m') : v = v[:-1] v = float(v) / latlon.milesPerNauticalMile else : v = float(v) / latlon.metersPerNauticalMile cutoff_distance = v while True : oi = tzlib.find_argi(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.find_argi(sys.argv, [ "--start_time", ] ) 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.find_argi(sys.argv, [ "--end_time", ] ) 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.find_argi(sys.argv, [ "--out_put", "--out_file", "-o"] ) if oi < 0 : break del sys.argv[oi] ofile_name = sys.argv.pop(oi) while True : oi = tzlib.find_argi(sys.argv, [ "--gpx_trk", ] ) if oi < 0 : break del sys.argv[oi] gpx_trk = True while True : oi = tzlib.find_argi(sys.argv, [ "--no_gpx_extensions", ] ) if oi < 0 : break del sys.argv[oi] include_extensions = False fnd = False while True : oi = tzlib.find_argi_and_del(sys.argv, [ "--test_these_files", ]) if oi < 0 : break fnd = True afn = sys.argv.pop(oi) fns = glob.glob(afn) if not len(fns) : print "No files found for %s!" % afn sys.exit(102) trks = [] for fn in fns : tracks = tracks_from_file(fn) trks += tracks for t in tracks : pts = find_spatially_separated_points(t.points) print fn print time.asctime(time.localtime(t.when)) print "\n".join([ str(p) for p in pts ]) pass prep_tracks_for_merge_to_points(trks) # aaaand, that's it. I guess they've been tested. if fnd : sys.exit(0) if zero or near or sparsify or merge or ofile_name : mfc = 2 if ofile_name or comb_trk_dir : mfc = 1 if len(sys.argv) < mfc : print "Must have input .gpx/.nmea/.loc/.txt file(s) and output file given to do -z or -n ... or a -o output_file_name!" sys.exit(101) tracks = get_tracks_from_ambiguous_file_names(sys.argv[:len(sys.argv) - (mfc - 1)]) if new_speeds : tracks = fix_tracks_speeds(tracks, force = (new_speeds > 1)) 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) n, e = os.path.splitext(os.path.basename(ofile_name)) if (not e) and n.startswith('.') : bfn = os.path.splitext(os.path.basename(fn))[0] # pick up the last base file name we read in, extensionless for t in tracks : fn = getattr(t, 'file_name', None) if fn : fn = os.path.splitext(os.path.basename(fn))[0] if fn : bfn = fn break pass pass ofile_name = os.path.join(os.path.dirname(ofile_name), bfn + n) waypoints = get_waypoints_from_tracks(tracks) write_gps_file(ofile_name, "tz_gps.py", "tz_gps", waypoints, tracks, strip_spaces, gpx_trk = gpx_trk, date_the_file = True, include_extensions = include_extensions) sys.exit(0) tracks = get_tracks_from_ambiguous_file_names(sys.argv) print "%-5u track%s" % ( len(tracks), tzlib.s_except_1(tracks) ) if comb_trk_dir : import tz_os_priority tz_os_priority.set_proc_to_idle_priority() gts, trk_pts = make_geotracks(tracks, cutoff_distance = cutoff_distance) gts = [ gt for gt in gts if gt.tracks ] print "Writing %-5u combined tracks." % len(gts) write_geotrack_files(gts, comb_trk_dir, program_name = os.path.basename(program_name)) sys.exit(0) # # # Print out the average and median of the points in the gpx files # # alat = 0.0 alon = 0.0 aalt = 0.0 lats = [] lons = [] alts = [] fdistot = 0.0 distot = 0.0 durtot = 0.0 # # This should use X Y Z logic !!!! # for t in tracks : dur = None s = '%5u point%s in "%s" from %s' % ( len(t.points), tzlib.s_except_1(t.points), t.name or "", t.file_name or "" ) print s.rstrip() s = " %s %s %s" % ( t.points[ 0], t.points[ 0].time_description(), t.points[ 0].description[:64] or "", ) print s.rstrip() if len(t.points) > 1 : dur = max(0, t.points[-1].when - t.points[0].when) durtot += dur dis = points_distance(t.points) ds = "" if dur : ds = " Duration: %u:%02u:%02u Average speed: %.2f mph" % ( int(dur / 3600), int((dur / 60) % 60), int(dur % 60), (dis * latlon.milesPerNauticalMile) / (dur / 3600.0), ) s = " %s %s %s%s" % ( t.points[-1], t.points[-1].time_description(), t.points[-1].description[:64] or "", ds, ) print s aalt = 0.0 for p in t.points : alat += p.lat alon += p.lon aalt += p.altitude or aalt 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) fdistot += dis print "Flat Distance: %.2f meters %.2f miles" % ( dis * latlon.metersPerNauticalMile, dis * latlon.milesPerNauticalMile ) dis = points_distance(t.points) distot += dis 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 : pcnt = sum([ len(t.points) for t in tracks ]) dis = fdistot print "Flat Distance: %.2f meters %.2f miles Average: %.2f meters %.2f miles over %u tracks, %u points." % ( dis * latlon.metersPerNauticalMile, dis * latlon.milesPerNauticalMile, dis * latlon.metersPerNauticalMile / len(tracks), dis * latlon.milesPerNauticalMile / len(tracks), len(tracks), pcnt, ), if durtot : dur = durtot print "Duration: %u:%02u:%02u Average speed: %.2f mph" % ( int(dur / 3600), int((dur / 60) % 60), int(dur % 60), (dis * latlon.milesPerNauticalMile) / (dur / 3600.0), ) print dis = distot print " Distance: %.2f meters %.2f miles Average: %.2f meters %.2f miles over %u tracks, %u points." % ( dis * latlon.metersPerNauticalMile, dis * latlon.milesPerNauticalMile, dis * latlon.metersPerNauticalMile / len(tracks), dis * latlon.milesPerNauticalMile / len(tracks), len(tracks), pcnt, ), if durtot : dur = durtot print "Duration: %u:%02u:%02u Average speed: %.2f mph" % ( int(dur / 3600), int((dur / 60) % 60), int(dur % 60), (dis * latlon.milesPerNauticalMile) / (dur / 3600.0), ) print 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