mirror of
https://github.com/elisspace/autopsy.git
synced 2026-10-04 00:09:53 +00:00
Merge pull request #5627 from markmckinnon/5966-open-source-GPX-module-and-update-how-it-makes-artifacts
5966-open-source-GPX-module-and-update-how-it-makes-artifacts
This commit is contained in:
Richard Cordovano
@@ -0,0 +1,283 @@
|
||||
"""
|
||||
Autopsy Forensic Browser
|
||||
|
||||
Copyright 2019-2020 Basis Technology Corp.
|
||||
Contact: carrier <at> sleuthkit <dot> org
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
"""
|
||||
|
||||
import os
|
||||
import jarray
|
||||
import inspect
|
||||
import time
|
||||
import calendar
|
||||
from datetime import datetime
|
||||
|
||||
from java.lang import System
|
||||
from java.util.logging import Level
|
||||
from java.io import File
|
||||
from java.util import ArrayList
|
||||
|
||||
from org.sleuthkit.datamodel import SleuthkitCase
|
||||
from org.sleuthkit.datamodel import AbstractFile
|
||||
from org.sleuthkit.datamodel import ReadContentInputStream
|
||||
from org.sleuthkit.datamodel import Blackboard
|
||||
from org.sleuthkit.datamodel import BlackboardArtifact
|
||||
from org.sleuthkit.datamodel import BlackboardAttribute
|
||||
from org.sleuthkit.datamodel import TskCoreException
|
||||
from org.sleuthkit.datamodel.blackboardutils import GeoArtifactsHelper
|
||||
from org.sleuthkit.datamodel.blackboardutils.attributes import GeoWaypoint
|
||||
from org.sleuthkit.datamodel.blackboardutils.attributes import GeoTrackPoints
|
||||
from org.sleuthkit.autopsy.datamodel import ContentUtils
|
||||
from org.sleuthkit.autopsy.ingest import IngestModule
|
||||
from org.sleuthkit.autopsy.ingest.IngestModule import IngestModuleException
|
||||
from org.sleuthkit.autopsy.ingest import DataSourceIngestModule
|
||||
from org.sleuthkit.autopsy.ingest import FileIngestModule
|
||||
from org.sleuthkit.autopsy.ingest import IngestModuleFactoryAdapter
|
||||
from org.sleuthkit.autopsy.ingest import IngestMessage
|
||||
from org.sleuthkit.autopsy.ingest import IngestServices
|
||||
from org.sleuthkit.autopsy.coreutils import Logger
|
||||
from org.sleuthkit.autopsy.casemodule import Case
|
||||
from org.sleuthkit.autopsy.casemodule.services import Services
|
||||
from org.sleuthkit.autopsy.casemodule.services import FileManager
|
||||
from org.sleuthkit.autopsy.ingest import ModuleDataEvent
|
||||
|
||||
# Based on gpxpy module: https://github.com/tkrajina/gpxpy
|
||||
import gpxpy
|
||||
import gpxpy.gpx
|
||||
import gpxpy.parser
|
||||
|
||||
# Factory that defines the name and details of the module and allows Autopsy
|
||||
# to create instances of the modules that will do the analysis.
|
||||
class GPXParserDataSourceIngestModuleFactory(IngestModuleFactoryAdapter):
|
||||
|
||||
moduleName = "GPX Parser Module"
|
||||
|
||||
# True - Verbose debugging messages sent to log file.
|
||||
# False - Verbose debugging turned off.
|
||||
debuglevel = False
|
||||
|
||||
def getModuleDisplayName(self):
|
||||
return self.moduleName
|
||||
|
||||
# TODO: Give it a description
|
||||
def getModuleDescription(self):
|
||||
return "Module that extracts GEO data from GPX files."
|
||||
|
||||
def getModuleVersionNumber(self):
|
||||
return "1.1"
|
||||
|
||||
def isDataSourceIngestModuleFactory(self):
|
||||
return True
|
||||
|
||||
def createDataSourceIngestModule(self, ingestOptions):
|
||||
return GPXParserDataSourceIngestModule()
|
||||
|
||||
|
||||
# Data Source-level ingest module. One gets created per data source.
|
||||
class GPXParserDataSourceIngestModule(DataSourceIngestModule):
|
||||
|
||||
_logger = Logger.getLogger(GPXParserDataSourceIngestModuleFactory.moduleName)
|
||||
|
||||
def log(self, level, msg):
|
||||
self._logger.logp(level, self.__class__.__name__, inspect.stack()[1][3], msg)
|
||||
|
||||
def __init__(self):
|
||||
self.context = None
|
||||
|
||||
# Where any setup and configuration is done.
|
||||
def startUp(self, context):
|
||||
self.context = context
|
||||
|
||||
# Where the analysis is done.
|
||||
def process(self, dataSource, progressBar):
|
||||
|
||||
# We don't know how much work there is yet.
|
||||
progressBar.switchToIndeterminate()
|
||||
|
||||
# This will work in 4.0.1 and beyond.
|
||||
# Use blackboard class to index blackboard artifacts for keyword search.
|
||||
blackboard = Case.getCurrentCase().getServices().getBlackboard()
|
||||
|
||||
# Get the sleuthkitcase
|
||||
skCase = Case.getCurrentCase().getSleuthkitCase()
|
||||
|
||||
# In the name and then count and read them.
|
||||
fileManager = Case.getCurrentCase().getServices().getFileManager()
|
||||
|
||||
files = fileManager.findFiles(dataSource, "%.gpx")
|
||||
# TODO: Would like to change this to find files based on mimetype rather than extension.
|
||||
#files = findFiles(dataSource, "text/xml")
|
||||
#if (file.isMimeType('text/xml') == False):
|
||||
|
||||
numFiles = len(files)
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "found " + str(numFiles) + " files")
|
||||
progressBar.switchToDeterminate(numFiles)
|
||||
fileCount = 0;
|
||||
|
||||
# Get module name for adding attributes
|
||||
moduleName = GPXParserDataSourceIngestModuleFactory.moduleName
|
||||
|
||||
for file in files:
|
||||
|
||||
# Get the GeoArtifactsHelper
|
||||
geoArtifactHelper = GeoArtifactsHelper(skCase, moduleName, file)
|
||||
|
||||
# Check if the user pressed cancel while we were busy.
|
||||
if self.context.isJobCancelled():
|
||||
return IngestModule.ProcessResult.OK
|
||||
|
||||
#self.log(Level.INFO, "GPX: Processing file: " + file.getName())
|
||||
fileCount += 1
|
||||
|
||||
# Check if module folder is present. If not, create it.
|
||||
dirName = os.path.join(Case.getCurrentCase().getTempDirectory(), "GPX_Parser_Module")
|
||||
try:
|
||||
os.stat(dirName)
|
||||
except:
|
||||
os.mkdir(dirName)
|
||||
fileName = os.path.join(dirName, "tmp.gpx")
|
||||
|
||||
# Check to see if temporary file exists. If it does, remove it.
|
||||
if os.path.exists(fileName):
|
||||
try:
|
||||
os.remove(fileName)
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "GPX:\t" + "FILE DELETED " + fileName )
|
||||
except:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "GPX:\t" + "FILE NOT DELETED " + fileName)
|
||||
|
||||
# This writes the file to the local file system.
|
||||
localFile = File(fileName)
|
||||
ContentUtils.writeToFile(file, localFile)
|
||||
|
||||
# Send to gpxpy for parsing.
|
||||
gpxfile = open(fileName)
|
||||
try:
|
||||
gpx = gpxpy.parse(gpxfile)
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "GPX:\t" + "FILE PARSED")
|
||||
except:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.SEVERE, "GPX:\t" + file.getName() + " - FILE NOT PARSED")
|
||||
continue
|
||||
|
||||
if gpx:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "GPX: TRACKS")
|
||||
for track in gpx.tracks:
|
||||
for segment in track.segments:
|
||||
geoPointList = ArrayList()
|
||||
for point in segment.points:
|
||||
|
||||
elevation = 0
|
||||
if point.elevation != None:
|
||||
elevation = point.elevation
|
||||
|
||||
dateTime = 0
|
||||
try:
|
||||
if (point.time != None):
|
||||
datetime = long(time.mktime(point.time.timetuple()))
|
||||
except:
|
||||
pass
|
||||
|
||||
geoPointList.add(GeoWaypoint.GeoTrackPoint(point.latitude, point.longitude, elevation, 0, 0, 0, dateTime))
|
||||
|
||||
try:
|
||||
# Add the trackpoint using the helper class
|
||||
geoartifact = geoArtifactHelper.addTrack("Trackpoint", geoPointList)
|
||||
except Blackboard.BlackboardException as e:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.SEVERE, "GPX: Error using geo artifact helper with blackboard " )
|
||||
except TskCoreException as e:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.SEVERE, "GPX: Error using geo artifact helper tskcoreexception" )
|
||||
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "GPX: WAYPOINTS")
|
||||
for waypoint in gpx.waypoints:
|
||||
attributes = ArrayList()
|
||||
art = file.newArtifact(BlackboardArtifact.ARTIFACT_TYPE.TSK_GPS_BOOKMARK)
|
||||
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LATITUDE.getTypeID(), moduleName, waypoint.latitude))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LONGITUDE.getTypeID(), moduleName, waypoint.longitude))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_FLAG.getTypeID(), moduleName, "Waypoint"))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_NAME.getTypeID(), moduleName, waypoint.name))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_PROG_NAME.getTypeID(), moduleName, "GPXParser"))
|
||||
|
||||
art.addAttributes(attributes)
|
||||
|
||||
try:
|
||||
# Post the artifact to blackboard
|
||||
skCase.getBlackboard().postArtifact(art, moduleName)
|
||||
except Blackboard.BlackboardException as e:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.SEVERE, "GPX: Error using geo artifact helper with blackboard for waypoints" )
|
||||
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "GPX: ROUTES")
|
||||
for route in gpx.routes:
|
||||
firstTimeThru = 0
|
||||
startingPoint = list()
|
||||
endingPoint = list()
|
||||
for point in route.points:
|
||||
# If first time in loop only populate starting point
|
||||
if (firstTimeThru == 0):
|
||||
startingPoint.append((point.latitude, point.longitude))
|
||||
firstTimeThru = 1
|
||||
else:
|
||||
startingPoint.append((point.latitude, point.longitude))
|
||||
endingPoint.append((point.latitude, point.longitude))
|
||||
|
||||
if (len(endingPoint) > 0):
|
||||
# get length of ending point as this ensures that we have equal points to process.
|
||||
for i in range(0,len(endingPoint) -1):
|
||||
attributes = ArrayList()
|
||||
art = file.newArtifact(BlackboardArtifact.ARTIFACT_TYPE.TSK_GPS_ROUTE)
|
||||
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LATITUDE_START.getTypeID(), moduleName, startingPoint[i][0]))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LONGITUDE_START.getTypeID(), moduleName, startingPoint[i][1]))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LATITUDE_END.getTypeID(), moduleName, endingPoint[i][0]))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LONGITUDE_END.getTypeID(), moduleName, endingPoint[i][1]))
|
||||
|
||||
art.addAttributes(attributes)
|
||||
|
||||
try:
|
||||
# Post the artifact to blackboard
|
||||
skCase.getBlackboard().postArtifact(art, moduleName)
|
||||
except Blackboard.BlackboardException as e:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.SEVERE, "GPX: Error using geo artifact helper with blackboard for waypoints" )
|
||||
else:
|
||||
if (len(startingPoint) > 0):
|
||||
attributes = ArrayList()
|
||||
art = file.newArtifact(BlackboardArtifact.ARTIFACT_TYPE.TSK_GPS_ROUTE)
|
||||
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LATITUDE_START.getTypeID(), moduleName, startingPoint[0][0]))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LONGITUDE_START.getTypeID(), moduleName, startingPoint[0][1]))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LATITUDE_END.getTypeID(), moduleName, startingPoint[0][0]))
|
||||
attributes.add(BlackboardAttribute(BlackboardAttribute.ATTRIBUTE_TYPE.TSK_GEO_LONGITUDE_END.getTypeID(), moduleName, startingPoint[0][1]))
|
||||
|
||||
art.addAttributes(attributes)
|
||||
|
||||
try:
|
||||
# Post the artifact to blackboard
|
||||
skCase.getBlackboard().postArtifact(art, moduleName)
|
||||
except Blackboard.BlackboardException as e:
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.SEVERE, "GPX: Error using geo artifact helper with blackboard for waypoints" )
|
||||
|
||||
|
||||
# Update the progress bar.
|
||||
progressBar.progress(fileCount)
|
||||
if os.path.exists(fileName):
|
||||
try:
|
||||
os.remove(fileName)
|
||||
if GPXParserDataSourceIngestModuleFactory.debuglevel: self.log(Level.INFO, "GPX:\t" + "FILE DELETED")
|
||||
except:
|
||||
self.log(Level.SEVERE, "GPX:\t" + "FILE NOT DELETED")
|
||||
|
||||
# Post a message to the ingest messages inbox.
|
||||
message = IngestMessage.createMessage(IngestMessage.MessageType.DATA, "GPX Parser Data Source Ingest Module", "Found %d files" % fileCount)
|
||||
IngestServices.getInstance().postMessage(message)
|
||||
return IngestModule.ProcessResult.OK;
|
||||
@@ -0,0 +1,13 @@
|
||||
Metadata-Version: 1.1
|
||||
Name: gpxpy
|
||||
Version: 0.8.8
|
||||
Summary: GPX file parser and GPS track manipulation library
|
||||
Home-page: http://www.trackprofiler.com/gpxpy/index.html
|
||||
Author: Tomo Krajina
|
||||
Author-email: tkrajina@gmail.com
|
||||
License: Apache License, Version 2.0
|
||||
Description: UNKNOWN
|
||||
Platform: UNKNOWN
|
||||
Classifier: Programming Language :: Python
|
||||
Classifier: Programming Language :: Python :: 2
|
||||
Classifier: Programming Language :: Python :: 3
|
||||
@@ -0,0 +1,38 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2011 Tomo Krajina
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
|
||||
__version__ = '1.3.5'
|
||||
|
||||
def parse(xml_or_file, version = None):
|
||||
"""
|
||||
Parse xml (string) or file object. This is just an wrapper for
|
||||
GPXParser.parse() function.
|
||||
|
||||
parser may be 'lxml', 'minidom' or None (then it will be automatically
|
||||
detected, lxml if possible).
|
||||
|
||||
xml_or_file must be the xml to parse or a file-object with the XML.
|
||||
|
||||
version may be '1.0', '1.1' or None (then it will be read from the gpx
|
||||
xml node if possible, if not then version 1.0 will be used).
|
||||
"""
|
||||
|
||||
from . import parser as mod_parser
|
||||
|
||||
parser = mod_parser.GPXParser(xml_or_file)
|
||||
|
||||
return parser.parse(version)
|
||||
@@ -0,0 +1,390 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2011 Tomo Krajina
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import logging as mod_logging
|
||||
import math as mod_math
|
||||
|
||||
from . import utils as mod_utils
|
||||
|
||||
log = mod_logging.getLogger(__name__)
|
||||
|
||||
# Generic geo related function and class(es)
|
||||
|
||||
# latitude/longitude in GPX files is always in WGS84 datum
|
||||
# WGS84 defined the Earth semi-major axis with 6378.137 km
|
||||
EARTH_RADIUS = 6378.137 * 1000
|
||||
|
||||
# One degree in meters:
|
||||
ONE_DEGREE = (2*mod_math.pi*EARTH_RADIUS) / 360 # ==> 111.319 km
|
||||
|
||||
|
||||
def to_rad(x):
|
||||
return x / 180. * mod_math.pi
|
||||
|
||||
|
||||
def haversine_distance(latitude_1, longitude_1, latitude_2, longitude_2):
|
||||
"""
|
||||
Haversine distance between two points, expressed in meters.
|
||||
|
||||
Implemented from http://www.movable-type.co.uk/scripts/latlong.html
|
||||
"""
|
||||
d_lat = to_rad(latitude_1 - latitude_2)
|
||||
d_lon = to_rad(longitude_1 - longitude_2)
|
||||
lat1 = to_rad(latitude_1)
|
||||
lat2 = to_rad(latitude_2)
|
||||
|
||||
a = mod_math.sin(d_lat/2) * mod_math.sin(d_lat/2) + \
|
||||
mod_math.sin(d_lon/2) * mod_math.sin(d_lon/2) * mod_math.cos(lat1) * mod_math.cos(lat2)
|
||||
c = 2 * mod_math.atan2(mod_math.sqrt(a), mod_math.sqrt(1-a))
|
||||
d = EARTH_RADIUS * c
|
||||
|
||||
return d
|
||||
|
||||
|
||||
def length(locations=None, _3d=None):
|
||||
locations = locations or []
|
||||
if not locations:
|
||||
return 0
|
||||
length = 0
|
||||
for i in range(len(locations)):
|
||||
if i > 0:
|
||||
previous_location = locations[i - 1]
|
||||
location = locations[i]
|
||||
|
||||
if _3d:
|
||||
d = location.distance_3d(previous_location)
|
||||
else:
|
||||
d = location.distance_2d(previous_location)
|
||||
if d:
|
||||
length += d
|
||||
return length
|
||||
|
||||
|
||||
def length_2d(locations=None):
|
||||
""" 2-dimensional length (meters) of locations (only latitude and longitude, no elevation). """
|
||||
locations = locations or []
|
||||
return length(locations, False)
|
||||
|
||||
|
||||
def length_3d(locations=None):
|
||||
""" 3-dimensional length (meters) of locations (it uses latitude, longitude, and elevation). """
|
||||
locations = locations or []
|
||||
return length(locations, True)
|
||||
|
||||
|
||||
def calculate_max_speed(speeds_and_distances):
|
||||
"""
|
||||
Compute average distance and standard deviation for distance. Extremes
|
||||
in distances are usually extremes in speeds, so we will ignore them,
|
||||
here.
|
||||
|
||||
speeds_and_distances must be a list containing pairs of (speed, distance)
|
||||
for every point in a track segment.
|
||||
"""
|
||||
assert speeds_and_distances
|
||||
if len(speeds_and_distances) > 0:
|
||||
assert len(speeds_and_distances[0]) == 2
|
||||
# ...
|
||||
assert len(speeds_and_distances[-1]) == 2
|
||||
|
||||
size = len(speeds_and_distances)
|
||||
|
||||
if size < 20:
|
||||
log.debug('Segment too small to compute speed, size=%s', size)
|
||||
return None
|
||||
|
||||
distances = list(map(lambda x: x[1], speeds_and_distances))
|
||||
average_distance = sum(distances) / float(size)
|
||||
standard_distance_deviation = mod_math.sqrt(sum(map(lambda distance: (distance-average_distance)**2, distances))/float(size))
|
||||
|
||||
# Ignore items where the distance is too big:
|
||||
filtered_speeds_and_distances = filter(lambda speed_and_distance: abs(speed_and_distance[1] - average_distance) <= standard_distance_deviation * 1.5, speeds_and_distances)
|
||||
|
||||
# sort by speed:
|
||||
speeds = list(map(lambda speed_and_distance: speed_and_distance[0], filtered_speeds_and_distances))
|
||||
if not isinstance(speeds, list): # python3
|
||||
speeds = list(speeds)
|
||||
if not speeds:
|
||||
return None
|
||||
speeds.sort()
|
||||
|
||||
# Even here there may be some extremes => ignore the last 5%:
|
||||
index = int(len(speeds) * 0.95)
|
||||
if index >= len(speeds):
|
||||
index = -1
|
||||
|
||||
return speeds[index]
|
||||
|
||||
|
||||
def calculate_uphill_downhill(elevations):
|
||||
if not elevations:
|
||||
return 0, 0
|
||||
|
||||
size = len(elevations)
|
||||
|
||||
def __filter(n):
|
||||
current_ele = elevations[n]
|
||||
if current_ele is None:
|
||||
return False
|
||||
if 0 < n < size - 1:
|
||||
previous_ele = elevations[n-1]
|
||||
next_ele = elevations[n+1]
|
||||
if previous_ele is not None and current_ele is not None and next_ele is not None:
|
||||
return previous_ele*.3 + current_ele*.4 + next_ele*.3
|
||||
return current_ele
|
||||
|
||||
smoothed_elevations = list(map(__filter, range(size)))
|
||||
|
||||
uphill, downhill = 0., 0.
|
||||
|
||||
for n, elevation in enumerate(smoothed_elevations):
|
||||
if n > 0 and elevation is not None and smoothed_elevations is not None:
|
||||
d = elevation - smoothed_elevations[n-1]
|
||||
if d > 0:
|
||||
uphill += d
|
||||
else:
|
||||
downhill -= d
|
||||
|
||||
return uphill, downhill
|
||||
|
||||
|
||||
def distance(latitude_1, longitude_1, elevation_1, latitude_2, longitude_2, elevation_2,
|
||||
haversine=None):
|
||||
"""
|
||||
Distance between two points. If elevation is None compute a 2d distance
|
||||
|
||||
if haversine==True -- haversine will be used for every computations,
|
||||
otherwise...
|
||||
|
||||
Haversine distance will be used for distant points where elevation makes a
|
||||
small difference, so it is ignored. That's because haversine is 5-6 times
|
||||
slower than the dummy distance algorithm (which is OK for most GPS tracks).
|
||||
"""
|
||||
|
||||
# If points too distant -- compute haversine distance:
|
||||
if haversine or (abs(latitude_1 - latitude_2) > .2 or abs(longitude_1 - longitude_2) > .2):
|
||||
return haversine_distance(latitude_1, longitude_1, latitude_2, longitude_2)
|
||||
|
||||
coef = mod_math.cos(latitude_1 / 180. * mod_math.pi)
|
||||
x = latitude_1 - latitude_2
|
||||
y = (longitude_1 - longitude_2) * coef
|
||||
|
||||
distance_2d = mod_math.sqrt(x * x + y * y) * ONE_DEGREE
|
||||
|
||||
if elevation_1 is None or elevation_2 is None or elevation_1 == elevation_2:
|
||||
return distance_2d
|
||||
|
||||
return mod_math.sqrt(distance_2d ** 2 + (elevation_1 - elevation_2) ** 2)
|
||||
|
||||
|
||||
def elevation_angle(location1, location2, radians=False):
|
||||
""" Uphill/downhill angle between two locations. """
|
||||
if location1.elevation is None or location2.elevation is None:
|
||||
return None
|
||||
|
||||
b = float(location2.elevation - location1.elevation)
|
||||
a = location2.distance_2d(location1)
|
||||
|
||||
if a == 0:
|
||||
return 0
|
||||
|
||||
angle = mod_math.atan(b / a)
|
||||
|
||||
if radians:
|
||||
return angle
|
||||
|
||||
return 180 * angle / mod_math.pi
|
||||
|
||||
|
||||
def distance_from_line(point, line_point_1, line_point_2):
|
||||
""" Distance of point from a line given with two points. """
|
||||
assert point, point
|
||||
assert line_point_1, line_point_1
|
||||
assert line_point_2, line_point_2
|
||||
|
||||
a = line_point_1.distance_2d(line_point_2)
|
||||
|
||||
if a == 0:
|
||||
return line_point_1.distance_2d(point)
|
||||
|
||||
b = line_point_1.distance_2d(point)
|
||||
c = line_point_2.distance_2d(point)
|
||||
|
||||
s = (a + b + c) / 2.
|
||||
|
||||
return 2. * mod_math.sqrt(abs(s * (s - a) * (s - b) * (s - c))) / a
|
||||
|
||||
|
||||
def get_line_equation_coefficients(location1, location2):
|
||||
"""
|
||||
Get line equation coefficients for:
|
||||
latitude * a + longitude * b + c = 0
|
||||
|
||||
This is a normal cartesian line (not spherical!)
|
||||
"""
|
||||
if location1.longitude == location2.longitude:
|
||||
# Vertical line:
|
||||
return float(0), float(1), float(-location1.longitude)
|
||||
else:
|
||||
a = float(location1.latitude - location2.latitude) / (location1.longitude - location2.longitude)
|
||||
b = location1.latitude - location1.longitude * a
|
||||
return float(1), float(-a), float(-b)
|
||||
|
||||
|
||||
def simplify_polyline(points, max_distance):
|
||||
"""Does Ramer-Douglas-Peucker algorithm for simplification of polyline """
|
||||
|
||||
if len(points) < 3:
|
||||
return points
|
||||
|
||||
begin, end = points[0], points[-1]
|
||||
|
||||
# Use a "normal" line just to detect the most distant point (not its real distance)
|
||||
# this is because this is faster to compute than calling distance_from_line() for
|
||||
# every point.
|
||||
#
|
||||
# This is an approximation and may have some errors near the poles and if
|
||||
# the points are too distant, but it should be good enough for most use
|
||||
# cases...
|
||||
a, b, c = get_line_equation_coefficients(begin, end)
|
||||
|
||||
# Initialize to safe values
|
||||
tmp_max_distance = 0
|
||||
tmp_max_distance_position = 1
|
||||
|
||||
# Check distance of all points between begin and end, exclusive
|
||||
for point_no in range(1,len(points)-1):
|
||||
point = points[point_no]
|
||||
d = abs(a * point.latitude + b * point.longitude + c)
|
||||
if d > tmp_max_distance:
|
||||
tmp_max_distance = d
|
||||
tmp_max_distance_position = point_no
|
||||
|
||||
# Now that we have the most distance point, compute its real distance:
|
||||
real_max_distance = distance_from_line(points[tmp_max_distance_position], begin, end)
|
||||
|
||||
# If furthest point is less than max_distance, remove all points between begin and end
|
||||
if real_max_distance < max_distance:
|
||||
return [begin, end]
|
||||
|
||||
# If furthest point is more than max_distance, use it as anchor and run
|
||||
# function again using (begin to anchor) and (anchor to end), remove extra anchor
|
||||
return (simplify_polyline(points[:tmp_max_distance_position + 1], max_distance) +
|
||||
simplify_polyline(points[tmp_max_distance_position:], max_distance)[1:])
|
||||
|
||||
|
||||
class Location:
|
||||
""" Generic geographical location """
|
||||
|
||||
latitude = None
|
||||
longitude = None
|
||||
elevation = None
|
||||
|
||||
def __init__(self, latitude, longitude, elevation=None):
|
||||
self.latitude = latitude
|
||||
self.longitude = longitude
|
||||
self.elevation = elevation
|
||||
|
||||
def has_elevation(self):
|
||||
return self.elevation or self.elevation == 0
|
||||
|
||||
def remove_elevation(self):
|
||||
self.elevation = None
|
||||
|
||||
def distance_2d(self, location):
|
||||
if not location:
|
||||
return None
|
||||
|
||||
return distance(self.latitude, self.longitude, None, location.latitude, location.longitude, None)
|
||||
|
||||
def distance_3d(self, location):
|
||||
if not location:
|
||||
return None
|
||||
|
||||
return distance(self.latitude, self.longitude, self.elevation, location.latitude, location.longitude, location.elevation)
|
||||
|
||||
def elevation_angle(self, location, radians=False):
|
||||
return elevation_angle(self, location, radians)
|
||||
|
||||
def move(self, location_delta):
|
||||
self.latitude, self.longitude = location_delta.move(self)
|
||||
|
||||
def __add__(self, location_delta):
|
||||
latitude, longitude = location_delta.move(self)
|
||||
return Location(latitude, longitude)
|
||||
|
||||
def __str__(self):
|
||||
return '[loc:%s,%s@%s]' % (self.latitude, self.longitude, self.elevation)
|
||||
|
||||
def __repr__(self):
|
||||
if self.elevation is None:
|
||||
return 'Location(%s, %s)' % (self.latitude, self.longitude)
|
||||
else:
|
||||
return 'Location(%s, %s, %s)' % (self.latitude, self.longitude, self.elevation)
|
||||
|
||||
def __hash__(self):
|
||||
return mod_utils.hash_object(self, ('latitude', 'longitude', 'elevation'))
|
||||
|
||||
|
||||
class LocationDelta:
|
||||
"""
|
||||
Intended to use similar to timestamp.timedelta, but for Locations.
|
||||
"""
|
||||
|
||||
NORTH = 0
|
||||
EAST = 90
|
||||
SOUTH = 180
|
||||
WEST = 270
|
||||
|
||||
def __init__(self, distance=None, angle=None, latitude_diff=None, longitude_diff=None):
|
||||
"""
|
||||
Version 1:
|
||||
Distance (in meters).
|
||||
angle_from_north *clockwise*.
|
||||
...must be given
|
||||
Version 2:
|
||||
latitude_diff and longitude_diff
|
||||
...must be given
|
||||
"""
|
||||
if (distance is not None) and (angle is not None):
|
||||
if (latitude_diff is not None) or (longitude_diff is not None):
|
||||
raise Exception('No lat/lon diff if using distance and angle!')
|
||||
self.distance = distance
|
||||
self.angle_from_north = angle
|
||||
self.move_function = self.move_by_angle_and_distance
|
||||
elif (latitude_diff is not None) and (longitude_diff is not None):
|
||||
if (distance is not None) or (angle is not None):
|
||||
raise Exception('No distance/angle if using lat/lon diff!')
|
||||
self.latitude_diff = latitude_diff
|
||||
self.longitude_diff = longitude_diff
|
||||
self.move_function = self.move_by_lat_lon_diff
|
||||
|
||||
def move(self, location):
|
||||
"""
|
||||
Move location by this timedelta.
|
||||
"""
|
||||
return self.move_function(location)
|
||||
|
||||
def move_by_angle_and_distance(self, location):
|
||||
coef = mod_math.cos(location.latitude / 180. * mod_math.pi)
|
||||
vertical_distance_diff = mod_math.sin((90 - self.angle_from_north) / 180. * mod_math.pi) / ONE_DEGREE
|
||||
horizontal_distance_diff = mod_math.cos((90 - self.angle_from_north) / 180. * mod_math.pi) / ONE_DEGREE
|
||||
lat_diff = self.distance * vertical_distance_diff
|
||||
lon_diff = self.distance * horizontal_distance_diff / coef
|
||||
return location.latitude + lat_diff, location.longitude + lon_diff
|
||||
|
||||
def move_by_lat_lon_diff(self, location):
|
||||
return location.latitude + self.latitude_diff, location.longitude + self.longitude_diff
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,640 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2014 Tomo Krajina
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import inspect as mod_inspect
|
||||
import datetime as mod_datetime
|
||||
import re as mod_re
|
||||
import copy as mod_copy
|
||||
|
||||
from . import utils as mod_utils
|
||||
|
||||
|
||||
class GPXFieldTypeConverter:
|
||||
def __init__(self, from_string, to_string):
|
||||
self.from_string = from_string
|
||||
self.to_string = to_string
|
||||
|
||||
|
||||
RE_TIMESTAMP = mod_re.compile(
|
||||
r'^([0-9]{4})-([0-9]{1,2})-([0-9]{1,2})[T ]([0-9]{1,2}):([0-9]{1,2}):([0-9]{1,2})'
|
||||
r'(\.[0-9]{1,8})?(Z|[+-−][0-9]{2}:?(?:[0-9]{2})?)?$')
|
||||
|
||||
|
||||
class SimpleTZ(mod_datetime.tzinfo):
|
||||
__slots__ = ('offset',)
|
||||
|
||||
def __init__(self, s=None):
|
||||
self.offset = 0
|
||||
if s and len(s) >= 2:
|
||||
if s[0] in ('−', '-'):
|
||||
mult = -1
|
||||
s = s[1:]
|
||||
else:
|
||||
if s[0] == '+':
|
||||
s = s[1:]
|
||||
mult = 1
|
||||
hour = int(s[:2]) if s[:2].isdigit() else 0
|
||||
if len(s) >= 4:
|
||||
minute = int(s[-2:]) if s[-2:].isdigit() else 0
|
||||
else:
|
||||
minute = 0
|
||||
self.offset = mult * (hour * 60 + minute)
|
||||
|
||||
def utcoffset(self, dt):
|
||||
return mod_datetime.timedelta(minutes=self.offset)
|
||||
|
||||
def dst(self, dt):
|
||||
return mod_datetime.timedelta(0)
|
||||
|
||||
def tzname(self, dt):
|
||||
if self.offset == 0:
|
||||
return 'Z'
|
||||
return '{:02}:{:02}'.format(self.offset // 60, self.offset % 60)
|
||||
|
||||
def __repr__(self):
|
||||
return 'SimpleTZ("{}")'.format(self.tzname(None))
|
||||
|
||||
def __eq__(self, other):
|
||||
return self.offset == other.offset
|
||||
|
||||
|
||||
def parse_time(string):
|
||||
from . import gpx as mod_gpx
|
||||
if not string:
|
||||
return None
|
||||
m = RE_TIMESTAMP.match(string)
|
||||
if m:
|
||||
dt = [int(m.group(i)) for i in range(1, 7)]
|
||||
if m.group(7):
|
||||
f = m.group(7)[1:7]
|
||||
dt.append(int(f + "0" * (6 - len(f))))
|
||||
else:
|
||||
dt.append(0)
|
||||
dt.append(SimpleTZ(m.group(8)))
|
||||
return mod_datetime.datetime(*dt)
|
||||
raise mod_gpx.GPXException('Invalid time: {0}'.format(string))
|
||||
|
||||
|
||||
def format_time(time):
|
||||
offset = time.utcoffset()
|
||||
if not offset or offset == 0:
|
||||
tz = 'Z'
|
||||
else:
|
||||
tz = time.strftime('%z')
|
||||
if time.microsecond:
|
||||
ms = time.strftime('.%f')
|
||||
else:
|
||||
ms = ''
|
||||
return ''.join((time.strftime('%Y-%m-%dT%H:%M:%S'), ms, tz))
|
||||
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------------------------------
|
||||
# Type converters used to convert from/to the string in the XML:
|
||||
# ----------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FloatConverter:
|
||||
def __init__(self):
|
||||
self.from_string = lambda string : None if string is None else float(string.strip())
|
||||
self.to_string = lambda flt : mod_utils.make_str(flt)
|
||||
|
||||
|
||||
class IntConverter:
|
||||
def __init__(self):
|
||||
self.from_string = lambda string: None if string is None else int(string.strip())
|
||||
self.to_string = lambda flt: str(flt)
|
||||
|
||||
|
||||
class TimeConverter:
|
||||
def from_string(self, string):
|
||||
try:
|
||||
return parse_time(string)
|
||||
except:
|
||||
return None
|
||||
|
||||
def to_string(self, time):
|
||||
return format_time(time) if time else None
|
||||
|
||||
|
||||
INT_TYPE = IntConverter()
|
||||
FLOAT_TYPE = FloatConverter()
|
||||
TIME_TYPE = TimeConverter()
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------------------------------
|
||||
# Field converters:
|
||||
# ----------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class AbstractGPXField:
|
||||
def __init__(self, attribute_field=None, is_list=None):
|
||||
self.attribute_field = attribute_field
|
||||
self.is_list = is_list
|
||||
self.attribute = False
|
||||
|
||||
def from_xml(self, node, version):
|
||||
raise Exception('Not implemented')
|
||||
|
||||
def to_xml(self, value, version, nsmap):
|
||||
raise Exception('Not implemented')
|
||||
|
||||
|
||||
class GPXField(AbstractGPXField):
|
||||
"""
|
||||
Used for to (de)serialize fields with simple field<->xml_tag mapping.
|
||||
"""
|
||||
def __init__(self, name, tag=None, attribute=None, type=None,
|
||||
possible=None, mandatory=None):
|
||||
AbstractGPXField.__init__(self)
|
||||
self.name = name
|
||||
if tag and attribute:
|
||||
from . import gpx as mod_gpx
|
||||
raise mod_gpx.GPXException('Only tag *or* attribute may be given!')
|
||||
if attribute:
|
||||
self.tag = None
|
||||
self.attribute = name if attribute is True else attribute
|
||||
elif tag:
|
||||
self.tag = name if tag is True else tag
|
||||
self.attribute = None
|
||||
else:
|
||||
self.tag = name
|
||||
self.attribute = None
|
||||
self.type_converter = type
|
||||
self.possible = possible
|
||||
self.mandatory = mandatory
|
||||
|
||||
def from_xml(self, node, version):
|
||||
if self.attribute:
|
||||
if node is not None:
|
||||
result = node.get(self.attribute)
|
||||
else:
|
||||
__node = node.find(self.tag)
|
||||
if __node is not None:
|
||||
result = __node.text
|
||||
else:
|
||||
result = None
|
||||
if result is None:
|
||||
if self.mandatory:
|
||||
from . import gpx as mod_gpx
|
||||
raise mod_gpx.GPXException('{0} is mandatory in {1} (got {2})'.format(self.name, self.tag, result))
|
||||
return None
|
||||
|
||||
if self.type_converter:
|
||||
try:
|
||||
result = self.type_converter.from_string(result)
|
||||
except Exception as e:
|
||||
from . import gpx as mod_gpx
|
||||
raise mod_gpx.GPXException('Invalid value for <{0}>... {1} ({2})'.format(self.tag, result, e))
|
||||
|
||||
if self.possible:
|
||||
if not (result in self.possible):
|
||||
from . import gpx as mod_gpx
|
||||
raise mod_gpx.GPXException('Invalid value "{0}", possible: {1}'.format(result, self.possible))
|
||||
|
||||
return result
|
||||
|
||||
def to_xml(self, value, version, nsmap=None, prettyprint=True, indent=''):
|
||||
if value is None:
|
||||
return ''
|
||||
if not prettyprint:
|
||||
indent = ''
|
||||
if self.attribute:
|
||||
return '{0}="{1}"'.format(self.attribute, mod_utils.make_str(value))
|
||||
elif self.type_converter:
|
||||
value = self.type_converter.to_string(value)
|
||||
return mod_utils.to_xml(self.tag, content=value, escape=True,
|
||||
prettyprint=prettyprint, indent=indent)
|
||||
|
||||
|
||||
class GPXComplexField(AbstractGPXField):
|
||||
def __init__(self, name, classs, tag=None, is_list=None):
|
||||
AbstractGPXField.__init__(self, is_list=is_list)
|
||||
self.name = name
|
||||
self.tag = tag or name
|
||||
self.classs = classs
|
||||
|
||||
def from_xml(self, node, version):
|
||||
if self.is_list:
|
||||
result = []
|
||||
for child in node:
|
||||
if child.tag == self.tag:
|
||||
result.append(gpx_fields_from_xml(self.classs, child,
|
||||
version))
|
||||
return result
|
||||
else:
|
||||
field_node = node.find(self.tag)
|
||||
if field_node is None:
|
||||
return None
|
||||
return gpx_fields_from_xml(self.classs, field_node, version)
|
||||
|
||||
def to_xml(self, value, version, nsmap=None, prettyprint=True, indent=''):
|
||||
if not prettyprint:
|
||||
indent = ''
|
||||
if self.is_list:
|
||||
result = []
|
||||
for obj in value:
|
||||
result.append(gpx_fields_to_xml(obj, self.tag, version,
|
||||
nsmap=nsmap,
|
||||
prettyprint=prettyprint,
|
||||
indent=indent))
|
||||
return ''.join(result)
|
||||
else:
|
||||
return gpx_fields_to_xml(value, self.tag, version,
|
||||
prettyprint=prettyprint, indent=indent)
|
||||
|
||||
|
||||
class GPXEmailField(AbstractGPXField):
|
||||
"""
|
||||
Converts GPX1.1 email tag group from/to string.
|
||||
"""
|
||||
def __init__(self, name, tag=None):
|
||||
AbstractGPXField.__init__(self, is_list=False)
|
||||
self.name = name
|
||||
self.tag = tag or name
|
||||
|
||||
def from_xml(self, node, version):
|
||||
"""
|
||||
Extract email address.
|
||||
|
||||
Args:
|
||||
node: ETree node with child node containing self.tag
|
||||
version: str of the gpx output version "1.0" or "1.1"
|
||||
|
||||
Returns:
|
||||
A string containing the email address.
|
||||
"""
|
||||
email_node = node.find(self.tag)
|
||||
if email_node is None:
|
||||
return ''
|
||||
|
||||
email_id = email_node.get('id')
|
||||
email_domain = email_node.get('domain')
|
||||
return '{0}@{1}'.format(email_id, email_domain)
|
||||
|
||||
def to_xml(self, value, version, nsmap=None, prettyprint=True, indent=''):
|
||||
"""
|
||||
Write email address to XML
|
||||
|
||||
Args:
|
||||
value: str representing an email address
|
||||
version: str of the gpx output version "1.0" or "1.1"
|
||||
|
||||
Returns:
|
||||
None if value is empty or str of XML representation of the
|
||||
address. Representation starts with a \n.
|
||||
"""
|
||||
if not value:
|
||||
return ''
|
||||
|
||||
if not prettyprint:
|
||||
indent = ''
|
||||
|
||||
if '@' in value:
|
||||
pos = value.find('@')
|
||||
email_id = value[:pos]
|
||||
email_domain = value[pos+1:]
|
||||
else:
|
||||
email_id = value
|
||||
email_domain = 'unknown'
|
||||
|
||||
return ('\n' + indent +
|
||||
'<{0} id="{1}" domain="{2}" />'.format(self.tag,
|
||||
email_id, email_domain))
|
||||
|
||||
|
||||
class GPXExtensionsField(AbstractGPXField):
|
||||
"""
|
||||
GPX1.1 extensions <extensions>...</extensions> key-value type.
|
||||
"""
|
||||
def __init__(self, name, tag=None, is_list=True):
|
||||
AbstractGPXField.__init__(self, is_list=is_list)
|
||||
self.name = name
|
||||
self.tag = tag or 'extensions'
|
||||
|
||||
def from_xml(self, node, version):
|
||||
"""
|
||||
Build a list of extension Elements.
|
||||
|
||||
Args:
|
||||
node: Element at the root of the extensions
|
||||
version: unused, only 1.1 supports extensions
|
||||
|
||||
Returns:
|
||||
a list of Element objects
|
||||
"""
|
||||
result = []
|
||||
extensions_node = node.find(self.tag)
|
||||
if extensions_node is None:
|
||||
return result
|
||||
for child in extensions_node:
|
||||
result.append(mod_copy.deepcopy(child))
|
||||
return result
|
||||
|
||||
def _resolve_prefix(self, qname, nsmap):
|
||||
"""
|
||||
Convert a tag from Clark notation into prefix notation.
|
||||
|
||||
Convert a tag from Clark notation using the nsmap into a
|
||||
prefixed tag. If the tag isn't in Clark notation, return the
|
||||
qname back. Converts {namespace}tag -> prefix:tag
|
||||
|
||||
Args:
|
||||
qname: string with the fully qualified name in Clark notation
|
||||
nsmap: a dict of prefix, namespace pairs
|
||||
|
||||
Returns:
|
||||
string of the tag ready to be serialized.
|
||||
"""
|
||||
if nsmap is not None and '}' in qname:
|
||||
uri, _, localname = qname.partition("}")
|
||||
uri = uri.lstrip("{")
|
||||
qname = uri + ':' + localname
|
||||
for prefix, namespace in nsmap.items():
|
||||
if uri == namespace:
|
||||
qname = prefix + ':' + localname
|
||||
break
|
||||
return qname
|
||||
|
||||
def _ETree_to_xml(self, node, nsmap=None, prettyprint=True, indent=''):
|
||||
"""
|
||||
Serialize ETree element and all subelements.
|
||||
|
||||
Creates a string of the ETree and all children. The prefixes are
|
||||
resolved through the nsmap for easier to read XML.
|
||||
|
||||
Args:
|
||||
node: ETree with the extension data
|
||||
version: string of GPX version, must be 1.1
|
||||
nsmap: dict of prefixes and URIs
|
||||
prettyprint: boolean, when true, indent line
|
||||
indent: string prepended to tag, usually 2 spaces per level
|
||||
|
||||
Returns:
|
||||
string with all the prefixed tags and data for the node
|
||||
and its children as XML.
|
||||
|
||||
"""
|
||||
if not prettyprint:
|
||||
indent = ''
|
||||
|
||||
# Build element tag and text
|
||||
result = []
|
||||
prefixedname = self._resolve_prefix(node.tag, nsmap)
|
||||
result.append('\n' + indent + '<' + prefixedname)
|
||||
for attrib, value in node.attrib.items():
|
||||
attrib = self._resolve_prefix(attrib, nsmap)
|
||||
result.append(' {0}="{1}"'.format(attrib, value))
|
||||
result.append('>')
|
||||
if node.text is not None:
|
||||
result.append(node.text.strip())
|
||||
|
||||
|
||||
# Build subelement nodes
|
||||
for child in node:
|
||||
result.append(self._ETree_to_xml(child, nsmap,
|
||||
prettyprint=prettyprint,
|
||||
indent=indent+' '))
|
||||
|
||||
# Add tail and close tag
|
||||
tail = node.tail
|
||||
if tail is not None:
|
||||
tail = tail.strip()
|
||||
else:
|
||||
tail = ''
|
||||
if len(node) > 0:
|
||||
result.append('\n' + indent)
|
||||
result.append('</' + prefixedname + '>' + tail)
|
||||
|
||||
return ''.join(result)
|
||||
|
||||
def to_xml(self, value, version, nsmap=None, prettyprint=True, indent=''):
|
||||
"""
|
||||
Serialize list of ETree.
|
||||
|
||||
Creates a string of all the ETrees in the list. The prefixes are
|
||||
resolved through the nsmap for easier to read XML.
|
||||
|
||||
Args:
|
||||
value: list of ETrees with the extension data
|
||||
version: string of GPX version, must be 1.1
|
||||
nsmap: dict of prefixes and URIs
|
||||
prettyprint: boolean, when true, indent line
|
||||
indent: string prepended to tag, usually 2 spaces per level
|
||||
|
||||
Returns:
|
||||
string with all the prefixed tags and data for each node
|
||||
as XML.
|
||||
|
||||
"""
|
||||
if not prettyprint:
|
||||
indent = ''
|
||||
if not value or version != "1.1":
|
||||
return ''
|
||||
result = []
|
||||
result.append('\n' + indent + '<' + self.tag + '>')
|
||||
for extension in value:
|
||||
result.append(self._ETree_to_xml(extension, nsmap,
|
||||
prettyprint=prettyprint,
|
||||
indent=indent+' '))
|
||||
result.append('\n' + indent + '</' + self.tag + '>')
|
||||
return ''.join(result)
|
||||
|
||||
# ----------------------------------------------------------------------------------------------------
|
||||
# Utility methods:
|
||||
# ----------------------------------------------------------------------------------------------------
|
||||
|
||||
def _check_dependents(gpx_object, fieldname):
|
||||
"""
|
||||
Check for data in subelements.
|
||||
|
||||
Fieldname takes the form of 'tag:dep1:dep2:dep3' for an arbitrary
|
||||
number of dependents. If all the gpx_object.dep attributes are
|
||||
empty, return a sentinel value to suppress serialization of all
|
||||
subelements.
|
||||
|
||||
Args:
|
||||
gpx_object: GPXField object to check for data
|
||||
fieldname: string with tag and dependents delimited with ':'
|
||||
|
||||
Returns:
|
||||
Two strings. The first is a sentinel value, '/' + tag, if all
|
||||
the subelements are empty and an empty string otherwise. The
|
||||
second is the bare tag name.
|
||||
"""
|
||||
if ':' in fieldname:
|
||||
children = fieldname.split(':')
|
||||
field = children.pop(0)
|
||||
for child in children:
|
||||
if getattr(gpx_object, child.lstrip('@')):
|
||||
return '', field # Child has data
|
||||
return '/' + field, field # No child has data
|
||||
return '', fieldname # No children
|
||||
|
||||
def gpx_fields_to_xml(instance, tag, version, custom_attributes=None,
|
||||
nsmap=None, prettyprint=True, indent=''):
|
||||
if not prettyprint:
|
||||
indent = ''
|
||||
fields = instance.gpx_10_fields
|
||||
if version == '1.1':
|
||||
fields = instance.gpx_11_fields
|
||||
|
||||
tag_open = bool(tag)
|
||||
body = []
|
||||
if tag:
|
||||
body.append('\n' + indent + '<' + tag)
|
||||
if tag == 'gpx': # write nsmap in root node
|
||||
body.append(' xmlns="{0}"'.format(nsmap['defaultns']))
|
||||
namespaces = set(nsmap.keys())
|
||||
namespaces.remove('defaultns')
|
||||
for prefix in sorted(namespaces):
|
||||
body.append(
|
||||
' xmlns:{0}="{1}"'.format(prefix, nsmap[prefix])
|
||||
)
|
||||
if custom_attributes:
|
||||
# Make sure to_xml() always return attributes in the same order:
|
||||
for key in sorted(custom_attributes.keys()):
|
||||
body.append(' {0}="{1}"'.format(key, mod_utils.make_str(custom_attributes[key])))
|
||||
suppressuntil = ''
|
||||
for gpx_field in fields:
|
||||
# strings indicate non-data container tags with subelements
|
||||
if isinstance(gpx_field, str):
|
||||
# Suppress empty tags
|
||||
if suppressuntil:
|
||||
if suppressuntil == gpx_field:
|
||||
suppressuntil = ''
|
||||
else:
|
||||
suppressuntil, gpx_field = _check_dependents(instance,
|
||||
gpx_field)
|
||||
if not suppressuntil:
|
||||
if tag_open:
|
||||
body.append('>')
|
||||
tag_open = False
|
||||
if gpx_field[0] == '/':
|
||||
body.append('\n' + indent + '<{0}>'.format(gpx_field))
|
||||
if prettyprint and len(indent) > 1:
|
||||
indent = indent[:-2]
|
||||
else:
|
||||
if prettyprint:
|
||||
indent += ' '
|
||||
body.append('\n' + indent + '<{0}'.format(gpx_field))
|
||||
tag_open = True
|
||||
elif not suppressuntil:
|
||||
value = getattr(instance, gpx_field.name)
|
||||
if gpx_field.attribute:
|
||||
body.append(' ' + gpx_field.to_xml(value, version, nsmap,
|
||||
prettyprint=prettyprint,
|
||||
indent=indent + ' '))
|
||||
elif value is not None:
|
||||
if tag_open:
|
||||
body.append('>')
|
||||
tag_open = False
|
||||
xml_value = gpx_field.to_xml(value, version, nsmap,
|
||||
prettyprint=prettyprint,
|
||||
indent=indent + ' ')
|
||||
if xml_value:
|
||||
body.append(xml_value)
|
||||
|
||||
if tag:
|
||||
if tag_open:
|
||||
body.append('>')
|
||||
body.append('\n' + indent + '</' + tag + '>')
|
||||
|
||||
return ''.join(body)
|
||||
|
||||
|
||||
def gpx_fields_from_xml(class_or_instance, node, version):
|
||||
if mod_inspect.isclass(class_or_instance):
|
||||
result = class_or_instance()
|
||||
else:
|
||||
result = class_or_instance
|
||||
|
||||
fields = result.gpx_10_fields
|
||||
if version == '1.1':
|
||||
fields = result.gpx_11_fields
|
||||
|
||||
node_path = [node]
|
||||
|
||||
for gpx_field in fields:
|
||||
current_node = node_path[-1]
|
||||
if isinstance(gpx_field, str):
|
||||
gpx_field = gpx_field.partition(':')[0]
|
||||
if gpx_field.startswith('/'):
|
||||
node_path.pop()
|
||||
else:
|
||||
if current_node is None:
|
||||
node_path.append(None)
|
||||
else:
|
||||
node_path.append(current_node.find(gpx_field))
|
||||
else:
|
||||
if current_node is not None:
|
||||
value = gpx_field.from_xml(current_node, version)
|
||||
setattr(result, gpx_field.name, value)
|
||||
elif gpx_field.attribute:
|
||||
value = gpx_field.from_xml(node, version)
|
||||
setattr(result, gpx_field.name, value)
|
||||
|
||||
return result
|
||||
|
||||
def gpx_check_slots_and_default_values(classs):
|
||||
"""
|
||||
Will fill the default values for this class. Instances will inherit those
|
||||
values so we don't need to fill default values for every instance.
|
||||
This method will also fill the attribute gpx_field_names with a list of
|
||||
gpx field names. This can be used
|
||||
"""
|
||||
fields = classs.gpx_10_fields + classs.gpx_11_fields
|
||||
|
||||
gpx_field_names = []
|
||||
|
||||
instance = classs()
|
||||
|
||||
try:
|
||||
attributes = list(filter(lambda x : x[0] != '_', dir(instance)))
|
||||
attributes = list(filter(lambda x : not callable(getattr(instance, x)), attributes))
|
||||
attributes = list(filter(lambda x : not x.startswith('gpx_'), attributes))
|
||||
except Exception as e:
|
||||
raise Exception('Error reading attributes for %s: %s' % (classs.__name__, e))
|
||||
|
||||
attributes.sort()
|
||||
slots = list(classs.__slots__)
|
||||
slots.sort()
|
||||
|
||||
if attributes != slots:
|
||||
raise Exception('Attributes for %s is\n%s but should be\n%s' % (classs.__name__, attributes, slots))
|
||||
|
||||
for field in fields:
|
||||
if not isinstance(field, str):
|
||||
if field.is_list:
|
||||
value = []
|
||||
else:
|
||||
value = None
|
||||
try:
|
||||
actual_value = getattr(instance, field.name)
|
||||
except:
|
||||
raise Exception('%s has no attribute %s' % (classs.__name__, field.name))
|
||||
if value != actual_value:
|
||||
raise Exception('Invalid default value %s.%s is %s but should be %s'
|
||||
% (classs.__name__, field.name, actual_value, value))
|
||||
#print('%s.%s -> %s' % (classs, field.name, value))
|
||||
if not field.name in gpx_field_names:
|
||||
gpx_field_names.append(field.name)
|
||||
|
||||
gpx_field_names = tuple(gpx_field_names)
|
||||
## if not hasattr(classs, '__slots__') or not classs.__slots__ or classs.__slots__ != gpx_field_names:
|
||||
## try: slots = classs.__slots__
|
||||
## except Exception as e: slots = '[Unknown:%s]' % e
|
||||
## raise Exception('%s __slots__ invalid, found %s, but should be %s' % (classs, slots, gpx_field_names))
|
||||
@@ -0,0 +1,66 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import xml.dom.minidom as mod_minidom
|
||||
|
||||
def split_gpxs(xml):
|
||||
"""
|
||||
Split single tracks from this one, without parsing with gpxpy
|
||||
"""
|
||||
dom = mod_minidom.parseString(xml)
|
||||
gpx_node = _find_gpx_node(dom)
|
||||
gpx_track_nodes = []
|
||||
if gpx_node:
|
||||
for child_node in gpx_node.childNodes:
|
||||
if child_node.nodeName == 'trk':
|
||||
gpx_track_nodes.append(child_node)
|
||||
gpx_node.removeChild(child_node)
|
||||
|
||||
for gpx_track_node in gpx_track_nodes:
|
||||
gpx_node.appendChild(gpx_track_node)
|
||||
yield dom.toxml()
|
||||
gpx_node.removeChild(gpx_track_node)
|
||||
|
||||
def join_gpxs(xmls):
|
||||
"""
|
||||
Utility to join GPX files without parsing them with gpxpy
|
||||
"""
|
||||
result = None
|
||||
|
||||
wpt_elements = []
|
||||
rte_elements = []
|
||||
trk_elements = []
|
||||
|
||||
for xml in xmls:
|
||||
dom = mod_minidom.parseString(xml)
|
||||
if not result:
|
||||
result = dom
|
||||
|
||||
gpx_node = _find_gpx_node(dom)
|
||||
if gpx_node:
|
||||
for child_node in gpx_node.childNodes:
|
||||
if child_node.nodeName == 'wpt':
|
||||
wpt_elements.append(child_node)
|
||||
gpx_node.removeChild(child_node)
|
||||
elif child_node.nodeName == 'rte':
|
||||
rte_elements.append(child_node)
|
||||
gpx_node.removeChild(child_node)
|
||||
elif child_node.nodeName == 'trk':
|
||||
trk_elements.append(child_node)
|
||||
gpx_node.removeChild(child_node)
|
||||
|
||||
gpx_node = _find_gpx_node(result)
|
||||
if gpx_node:
|
||||
for wpt_element in wpt_elements:
|
||||
gpx_node.appendChild(wpt_element)
|
||||
for rte_element in rte_elements:
|
||||
gpx_node.appendChild(rte_element)
|
||||
for trk_element in trk_elements:
|
||||
gpx_node.appendChild(trk_element)
|
||||
|
||||
return result.toxml()
|
||||
|
||||
def _find_gpx_node(dom):
|
||||
for gpx_candidate_node in dom.childNodes:
|
||||
if gpx_candidate_node.nodeName == 'gpx':
|
||||
return gpx_candidate_node
|
||||
return None
|
||||
@@ -0,0 +1,155 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2011 Tomo Krajina
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import logging as mod_logging
|
||||
import re as mod_re
|
||||
|
||||
try:
|
||||
import lxml.etree as mod_etree # Load LXML or fallback to cET or ET
|
||||
except ImportError:
|
||||
try:
|
||||
import xml.etree.cElementTree as mod_etree
|
||||
except ImportError:
|
||||
import xml.etree.ElementTree as mod_etree
|
||||
|
||||
from . import gpx as mod_gpx
|
||||
from . import utils as mod_utils
|
||||
from . import gpxfield as mod_gpxfield
|
||||
|
||||
log = mod_logging.getLogger(__name__)
|
||||
|
||||
class GPXParser:
|
||||
"""
|
||||
Parse the XML and provide new GPX instance.
|
||||
|
||||
Methods:
|
||||
__init__: initialize new instance
|
||||
init: format XML
|
||||
parse: parse XML, build tree, build GPX
|
||||
|
||||
Attributes:
|
||||
gpx: GPX instance of the most recently parsed XML
|
||||
xml: string containing the XML text
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, xml_or_file=None):
|
||||
"""
|
||||
Initialize new GPXParser instance.
|
||||
|
||||
Arguments:
|
||||
xml_or_file: string or file object containing the gpx
|
||||
formatted xml
|
||||
|
||||
"""
|
||||
self.init(xml_or_file)
|
||||
self.gpx = mod_gpx.GPX()
|
||||
|
||||
def init(self, xml_or_file):
|
||||
"""
|
||||
Store the XML and remove utf-8 Byte Order Mark if present.
|
||||
|
||||
Args:
|
||||
xml_or_file: string or file object containing the gpx
|
||||
formatted xml
|
||||
|
||||
"""
|
||||
text = xml_or_file.read() if hasattr(xml_or_file, 'read') else xml_or_file
|
||||
self.xml = mod_utils.make_str(text)
|
||||
|
||||
def parse(self, version=None):
|
||||
"""
|
||||
Parse the XML and return a GPX object.
|
||||
|
||||
Args:
|
||||
version: str or None indicating the GPX Schema to use.
|
||||
Options are '1.0', '1.1' and None. When version is None
|
||||
the version is read from the file or falls back on 1.0.
|
||||
|
||||
Returns:
|
||||
A GPX object loaded from the xml
|
||||
|
||||
Raises:
|
||||
GPXXMLSyntaxException: XML file is invalid
|
||||
GPXException: XML is valid but GPX data contains errors
|
||||
|
||||
"""
|
||||
# Build prefix map for reserialization and extension handlings
|
||||
for namespace in mod_re.findall(r'\sxmlns:?[^=]*="[^"]+"', self.xml):
|
||||
prefix, _, URI = namespace[6:].partition('=')
|
||||
prefix = prefix.lstrip(':')
|
||||
if prefix == '':
|
||||
prefix = 'defaultns' # alias default for easier handling
|
||||
else:
|
||||
if prefix.startswith("ns"):
|
||||
mod_etree.register_namespace("noglobal_" + prefix, URI.strip('"'))
|
||||
else:
|
||||
mod_etree.register_namespace(prefix, URI.strip('"'))
|
||||
self.gpx.nsmap[prefix] = URI.strip('"')
|
||||
|
||||
schema_loc = mod_re.search(r'\sxsi:schemaLocation="[^"]+"', self.xml)
|
||||
if schema_loc:
|
||||
_, _, value = schema_loc.group(0).partition('=')
|
||||
self.gpx.schema_locations = value.strip('"').split()
|
||||
|
||||
# Remove default namespace to simplify processing later
|
||||
self.xml = mod_re.sub(r"""\sxmlns=(['"])[^'"]+\1""", '', self.xml, count=1)
|
||||
|
||||
# Build tree
|
||||
try:
|
||||
if GPXParser.__library() == "LXML":
|
||||
# lxml does not like unicode strings when it's expecting
|
||||
# UTF-8. Also, XML comments result in a callable .tag().
|
||||
# Strip them out to avoid handling them later.
|
||||
if mod_utils.PYTHON_VERSION[0] >= '3':
|
||||
self.xml = self.xml.encode('utf-8')
|
||||
root = mod_etree.XML(self.xml,
|
||||
mod_etree.XMLParser(remove_comments=True))
|
||||
else:
|
||||
root = mod_etree.XML(self.xml)
|
||||
|
||||
except Exception as e:
|
||||
# The exception here can be a lxml or ElementTree exception.
|
||||
log.debug('Error in:\n%s\n-----------\n' % self.xml, exc_info=True)
|
||||
|
||||
# The library should work in the same way regardless of the
|
||||
# underlying XML parser that's why the exception thrown
|
||||
# here is GPXXMLSyntaxException (instead of simply throwing the
|
||||
# original ElementTree or lxml exception e).
|
||||
#
|
||||
# But, if the user needs the original exception (lxml or ElementTree)
|
||||
# it is available with GPXXMLSyntaxException.original_exception:
|
||||
raise mod_gpx.GPXXMLSyntaxException('Error parsing XML: %s' % str(e), e)
|
||||
|
||||
if root is None:
|
||||
raise mod_gpx.GPXException('Document must have a `gpx` root node.')
|
||||
|
||||
if version is None:
|
||||
version = root.get('version')
|
||||
|
||||
mod_gpxfield.gpx_fields_from_xml(self.gpx, root, version)
|
||||
return self.gpx
|
||||
|
||||
@staticmethod
|
||||
def __library():
|
||||
"""
|
||||
Return the underlying ETree.
|
||||
|
||||
Provided for convenient unittests.
|
||||
"""
|
||||
if "lxml" in str(mod_etree):
|
||||
return "LXML"
|
||||
return "STDLIB"
|
||||
@@ -0,0 +1,92 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2011 Tomo Krajina
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import sys as mod_sys
|
||||
import math as mod_math
|
||||
import xml.sax.saxutils as mod_saxutils
|
||||
|
||||
PYTHON_VERSION = mod_sys.version.split(' ')[0]
|
||||
|
||||
|
||||
def to_xml(tag, attributes=None, content=None, default=None, escape=False, prettyprint=True, indent=''):
|
||||
if not prettyprint:
|
||||
indent = ''
|
||||
attributes = attributes or {}
|
||||
result = []
|
||||
result.append('\n' + indent + '<{0}'.format(tag))
|
||||
|
||||
if content is None and default:
|
||||
content = default
|
||||
|
||||
if attributes:
|
||||
for attribute in attributes.keys():
|
||||
result.append(make_str(' %s="%s"' % (attribute, attributes[attribute])))
|
||||
|
||||
if content is None:
|
||||
result.append('/>')
|
||||
else:
|
||||
if escape:
|
||||
result.append(make_str('>%s</%s>' % (mod_saxutils.escape(content), tag)))
|
||||
else:
|
||||
result.append(make_str('>%s</%s>' % (content, tag)))
|
||||
|
||||
result = make_str(''.join(result))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def is_numeric(object):
|
||||
try:
|
||||
float(object)
|
||||
return True
|
||||
except TypeError:
|
||||
return False
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def to_number(s, default=0, nan_value=None):
|
||||
try:
|
||||
result = float(s)
|
||||
if mod_math.isnan(result):
|
||||
return nan_value
|
||||
return result
|
||||
except TypeError:
|
||||
pass
|
||||
except ValueError:
|
||||
pass
|
||||
return default
|
||||
|
||||
|
||||
def total_seconds(timedelta):
|
||||
""" Some versions of python don't have the timedelta.total_seconds() method. """
|
||||
if timedelta is None:
|
||||
return None
|
||||
return (timedelta.days * 86400) + timedelta.seconds
|
||||
|
||||
|
||||
def make_str(s):
|
||||
""" Convert a str or unicode or float object into a str type. """
|
||||
if isinstance(s, float):
|
||||
result = str(s)
|
||||
if not 'e' in result:
|
||||
return result
|
||||
# scientific notation is illegal in GPX 1/1
|
||||
return format(s, '.10f').rstrip('0.')
|
||||
if PYTHON_VERSION[0] == '2':
|
||||
if isinstance(s, unicode):
|
||||
return s.encode("utf-8")
|
||||
return str(s)
|
||||
Reference in New Issue
Block a user