cyclograph 1.6.1-1 / usr / share / cyclograph / iofile.py

# -*- coding: utf-8 -*-
"""iofile.py"""

# Copyright (C) 2008, 2009, 2010, 2011, 2012, 2013 Federico Brega, Pierluigi Villani

# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

from __future__ import with_statement
import csv
import codecs
import logging
import os
import xml.dom.minidom
import zlib
from altitude_downloader import distance, haversine, point_conversion

CSV_VERSION = "1.1"
CGX_VERSION = "1.0"

STAT_VER = 2

def save_file(filename, slope):
    """Save .cgx file"""
    doc = xml.dom.minidom.Document()
    cg = doc.createElement(u"cyclographxml")
    cg.setAttribute(u"version", CGX_VERSION)
    doc.appendChild(cg)
    sl = doc.createElement(u"slope")
    cg.appendChild(sl)

    sl.appendChild(__createTextElement(doc, u"name", slope.name))
    sl.appendChild(__createTextElement(doc, u"country", slope.country))

    author = doc.createElement(u"author")
    author.appendChild(__createTextElement(doc, u"name", slope.author))
    author.appendChild(__createTextElement(doc, u"email", slope.email))
    sl.appendChild(author)

    sl.appendChild(__createTextElement(doc, u"info", slope.comment))

    cps = doc.createElement(u"cps")
    for cpt in slope.cps:
        cp = doc.createElement(u"cp")
        (dist, alt, desc) = cpt
        cp.setAttribute(u"alt", "%.3f" % alt)
        cp.setAttribute(u"dist", "%.3f" % dist)
        if desc != "":
            cp.setAttribute(u"desc", desc)
        cps.appendChild(cp)
    sl.appendChild(cps)

    coords = doc.createElement(u"coords")
    for coor in slope.coords:
        coord = doc.createElement(u"coord")
        (lat, lon) = coor
        coord.setAttribute(u"lon", "%.6f" % lon)
        coord.setAttribute(u"lat", "%.6f" % lat)
        coords.appendChild(coord)
    sl.appendChild(coords)

    with open(filename,"wb") as fid:
        doc.writexml(fid, encoding='utf-8', newl='\n')

def __createTextElement(doc, name, text):
    elem = doc.createElement(name)
    ptext = doc.createTextNode(text)
    elem.appendChild(ptext)
    return elem

#def save_file(filename, slope):
#    """Save .csv file"""
#    #Control if the right extension is given by user
#    if not filename.endswith('.csv'):
#        filename = filename + '.csv'
#    #fid = open(filename,'wb')
#    # Too bad there is no csv unicode writer so open in ASCII mode and convert utf-8 to ASCII
#    # Be carreful to also encode delimiter and quotechar if they are not ASCII
#    with open(filename, mode='wb') as fid:
#        #fid.write ("%s;%s;%s\n" % (slope.name, slope.country, slope.comment) )
#        csv_writer = csv.writer(fid, delimiter=';', quotechar='\"')
#        def to_unicode(item):
#            if isinstance(item, unicode):
#                item = item.encode('utf-8')
#            return item
#        csv_writer.writerow( map(to_unicode, (str(CSV_VERSION), slope.name, slope.country,
#                               slope.author, slope.email, slope.comment)) )
#        for row in slope.cps:
#            csv_writer.writerow(map(to_unicode, row))
#        #for cpt in slope.cps:
#        #    fid.write ("%.3f;%.3f;%s\n" % cpt)
#    #fid.close()

def open_file(filename, slopelist, slopenum, extension=''):
    """Open generic file"""
    statusok = True
    if not extension:
        if isinstance(filename, basestring):
            extension = filename.split('.')[-1].lower()
        else:
            extension = 'cgx' #when used stdin
    if extension == 'cgx':
        try:
            statusok = __open_cgx(filename, slopelist, slopenum)
        except IOError:
            statusok = False
    elif extension == 'csv':
        try:
            statusok = __open_csv(filename, slopelist, slopenum)
        except IOError:
            statusok = False
    elif extension == 'xml':
        try:
            statusok = __open_xml(filename, slopelist, slopenum)
        except IOError:
            statusok = False
    elif extension == 'gpx':
        try:
            gpxfile = load_gpxfile(filename)
            gpxfile.newSlope(slopelist, slopenum, -1)
            statusok = True
        except IOError:
            statusok = False
    elif extension == 'kml':
        try:
            kmlfile = load_kmlfile(filename)
            kmlfile.newSlope(slopelist, slopenum, -1)
            statusok = True
        except IOError:
            statusok = False
    elif extension == 'fitlog':
        try:
            fitlogfile = load_fitlogfile(filename)
            fitlogfile.newSlope(slopelist, slopenum, -1)
            statusok = True
        except IOError:
            statusok = False
    elif extension == 'tcx':
        try:
            tcxfile = load_tcxfile(filename)
            tcxfile.newSlope(slopelist, slopenum, -1)
            statusok = True
        except IOError:
            statusok = False
    elif extension == 'sal':
        try:
            __open_sal(filename, slopelist, slopenum)
        except IOError:
            statusok = False
    elif extension == 'crp':
        try:
            __open_crp(filename, slopelist, slopenum)
        except IOError:
            statusok = False
    elif extension == 'txt':
        try:
            __open_txt(filename, slopelist, slopenum)
        except IOError:
            statusok = False
    else:
        statusok = False
    return statusok

### Functions defining how to open a certain file format.

#csv unicode implementation, std library doesn't implements it yet :-(
#from http://docs.python.org/library/csv.html#csv-examples
def unicode_csv_reader(unicode_csv_data, dialect=csv.excel, **kwargs):
    # csv.py doesn't do Unicode; encode temporarily as UTF-8:
    csv_reader = csv.reader(utf_8_encoder(unicode_csv_data),
                            dialect=dialect, **kwargs)
    for row in csv_reader:
        # decode UTF-8 back to Unicode, cell by cell:
        yield [unicode(cell, 'utf-8') for cell in row]

def utf_8_encoder(unicode_csv_data):
    for line in unicode_csv_data:
        yield line.encode('utf-8')

def __open_csv(filename, slopelist, slopenum):
    """Open a Cyclograph CSV file (legacy)"""
    if isinstance(filename, basestring):
        fid = codecs.open(filename, encoding='utf-8', mode='rb')
    else:
        fid = filename
    #with open(filename, 'rb') as fid:
    csv_reader = unicode_csv_reader(fid, delimiter=';', quotechar='\"')
    line = csv_reader.next()
    #Make it more robust: if file has less than the required elements
    #it considers them as empty strings and doesn't consider exceding ones.
    if len(line) < 6:
        line.extend([""] * (6-len(line)) )
    (version, slopename, slopestate, slopeauthor, authoremail, slopecomment) = line[:6]
    status = True
    
    #This is a legacy format so always warn about that, regardless the version number
    status = STAT_VER

    slopelist.set_name(slopenum, slopename)
    slopelist.set_country(slopenum, slopestate)
    slopelist.set_author(slopenum, slopeauthor)
    slopelist.set_email(slopenum, authoremail)
    slopelist.set_comment(slopenum, slopecomment)

    for line in csv_reader:
        #(dist, alt, name)=line.strip().split(';')
        try:
            (dist, alt, name) = line
        except ValueError:
            return False
        slopelist.add_cp(slopenum, float(dist), float(alt), name)
    fid.close()
    return status

def __open_cgx(filename, slopelist, slopenum):
    """Open a Cyclograph cgx file"""
    fid = xml.dom.minidom.parse(filename)
    slop = fid.getElementsByTagName("slope")[0]
    slopename = slop.getElementsByTagName("name")[0]
    if slopename.hasChildNodes():
        slopelist.set_name(slopenum, slopename.childNodes[0].data)
    slopestate = slop.getElementsByTagName("country")[0]
    if slopestate.hasChildNodes():
        slopelist.set_country(slopenum, slopestate.childNodes[0].data)

    auth = slop.getElementsByTagName("author")[0]
    author = auth.getElementsByTagName("name")[0]
    if author.hasChildNodes():
        slopelist.set_author(slopenum, author.childNodes[0].data)
    mail = auth.getElementsByTagName("email")[0]
    if mail.hasChildNodes():
        slopelist.set_email(slopenum, mail.childNodes[0].data)

    slopecomment = slop.getElementsByTagName("info")[0]
    if slopecomment.hasChildNodes():
        slopelist.set_comment(slopenum, slopecomment.childNodes[0].data)

    for cpt in slop.getElementsByTagName('cp'):
        try:
            slopelist.add_cp(slopenum, float(cpt.attributes["dist"].value)
                                     , float(cpt.attributes["alt"].value)
                                     , cpt.getAttribute("desc"))
        except Exception, err:
            logging.exception(err)
            fid.unlink()
            return False
    for coord in slop.getElementsByTagName('coord'):
        try:
            slopelist.add_coord(slopenum,
                                float(coord.attributes["lat"].value),
                                float(coord.attributes["lon"].value))
        except Exception, err:
            logging.exception(err)
            fid.unlink()
            return False
    fid.unlink()
    return True

def __open_xml(filename, slopelist, slopenum):
    """Open a Cyclomaniac xml file"""
    fid = xml.dom.minidom.parse(filename)
    slopename = fid.getElementsByTagName("name")[0]
    slopename = slopename.childNodes
    slopelist.set_name(slopenum, slopename[0].data)
    samples = fid.getElementsByTagName("samples")[0]
    for sample in samples.getElementsByTagName('sample'):
        for ent in sample.childNodes:
            if ent.localName == "distance":
                dist = ent.childNodes[0].data.replace(',','.')
            elif ent.localName == "height":
                alt = ent.childNodes[0].data.replace(',','.')
            elif ent.localName == "label":
                name = ent.childNodes[0].data
        slopelist.add_cp(slopenum, float(dist), float(alt), name)
        name = u""
    fid.unlink()
    return True

def load_gpxfile(xmlfile):
    """Load a gpx file"""
    try:
        if isinstance(xmlfile, basestring) and xmlfile.strip().startswith('<'):
            fid = xml.dom.minidom.parseString(xmlfile)
        else:
            fid = xml.dom.minidom.parse(xmlfile)
        ptls = fid.getElementsByTagName('trkpt')
        coords = [(el.attributes["lon"].value, el.attributes["lat"].value) for el in ptls]
        cps = distance(coords)
        next_cp = 0
        max_dist = cps[-1][0]
        indices = []
        for i in range(len(cps)):
            dist = cps[i][0]
            if i == 0 or dist > next_cp or i == len(cps)-1:
                indices.append(i)
                next_cp = dist + max_dist/len(cps)
        #Use waypoints to determine the name of a cp.
        waypoints = {}
        for waypoint in fid.getElementsByTagName('wpt'):
            #find the nearest (filtered) point to this waypont
            (wpt_lon, wpt_lat) = point_conversion(waypoint.attributes["lon"].value,
                                                  waypoint.attributes["lat"].value)
            (min_i, min_distance) = (-1, float(3)) #use minimum accetable distance 3 km.
            #print cps
            for i in indices:
                #calculate distance to point i
                (lon, lat) = point_conversion(cps[i][1], cps[i][2])
                dist = haversine(lat, wpt_lat, lon, wpt_lon)
                #print (lon, lat, dist)
                if dist < min_distance:
                    (min_i, min_distance) = (i, dist)
            if min_i >= 0:
                #is found a point near enought to the waypoint: use it.
                waypoints[min_i] = waypoint.getElementsByTagName("name")[0].firstChild.data
        returncp = []
        is_with_alt = True
        alt_count = 0
        for i in indices:
            #get elevation
            (dist, lng, lat) = cps[i]
            sample = ptls[i]
            elelist = sample.getElementsByTagName('ele')
            if len(elelist) > 0:
                ele = elelist[0]
                node = ele.childNodes
                alt = float(node[0].data)
            else:
                is_with_alt = False
                alt = 0
            if (alt == 0):
                alt_count += 1
            if i in waypoints:
                name = waypoints[i]
            else:
                name = u""
            returncp.append((dist, alt, name, lat, lng))
        fid.unlink()
        retfile = SlopeFile(returncp)
        if len(indices) == alt_count:
            is_with_alt = False
        retfile.hasAltitudes = is_with_alt
        return retfile
    except Exception, err:
        logging.exception(err)
        return None

def load_fitlogfile(xmlfile):
    """Load a fitlog file"""
    try:
        if isinstance(xmlfile, basestring) and xmlfile.strip().startswith('<'):
            fid = xml.dom.minidom.parseString(xmlfile)
        else:
            fid = xml.dom.minidom.parse(xmlfile)
        ptls = fid.getElementsByTagName('pt')
        coords = [(el.attributes["lon"].value, el.attributes["lat"].value) for el in ptls]
        cps = distance(coords)
        next_cp = 0
        max_dist = cps[-1][0]
        indices = []
        for i in range(len(cps)):
            dist = cps[i][0]
            if i == 0 or dist > next_cp or i == len(cps)-1:
                indices.append(i)
                next_cp = dist + max_dist/len(cps)
        returncp = []
        is_with_alt = True
        alt_count = 0
        name = u""
        for i in indices:
            #get elevation
            (dist, lng, lat) = cps[i]
            sample = ptls[i]
            if sample.getAttribute('ele'):
                elev = sample.attributes["ele"].value
                if len(elev) > 0:
                    alt = float(elev)
                else:
                    is_with_alt = False
                    alt = 0
            else:
                is_with_alt = False
                alt = 0
            if (alt == 0):
                alt_count += 1
            returncp.append((dist, alt, name, lat, lng))
        fid.unlink()
        retfile = SlopeFile(returncp)
        if len(indices) == alt_count:
            is_with_alt = False
        retfile.hasAltitudes = is_with_alt
        return retfile
    except Exception, err:
        logging.exception(err)
        return None

def load_kmlfile(xmlfile):
    """Load a kml file"""
    try:
        if isinstance(xmlfile, basestring) and xmlfile.strip().startswith('<'):
            fid = xml.dom.minidom.parseString(xmlfile)
        else:
            fid = xml.dom.minidom.parse(xmlfile)
            
        coordlist = []
        altlist = []

        for linstr in fid.getElementsByTagName("LineString"):
            coords = linstr.getElementsByTagName("coordinates")[0]
            node = coords.childNodes[0]

            data = node.data.strip(' \n')
            coordlist += [el.split(',')[:2] for el in data.split(' ')]

        #if LineString is not present try gx:coord (KML 2.2 ext)
        if len(coordlist) == 0:
            for gxcoord in fid.getElementsByTagName("gx:coord"):
                node = gxcoord.childNodes[0]
                temp = node.data.split(' ')
                coordlist += [temp[:2]]
                altlist += [float(temp[2])]

            if len(coordlist) == 0:
                return SlopeFile([])

        cps = distance(coordlist)

        next_cp = 0
        max_dist = cps[-1][0]
        is_with_alt = len(altlist) >= len(cps)
        returncp = []
        for i in range(len(cps)):
            (dist, lng, lat) = cps[i]
            #create self.num_cps check points
            #print i
            if i == 0 or dist > next_cp or i == len(cps)-1:
                if is_with_alt:
                    # altitude is already in the kml file
                    alt = altlist[i]
                else:
                    # altitude has to be downloaded
                    alt = 0

                returncp.append((dist, alt, "", lat, lng))
                next_cp = dist + max_dist/len(cps)

        fid.unlink()
        retfile = SlopeFile(returncp)
        retfile.hasAltitudes = is_with_alt
        return retfile
    except Exception, err:
        logging.exception(err)
        return None

def load_tcxfile(xmlfile):
    """Load a tcx file"""
    try:
        fid = xml.dom.minidom.parse(xmlfile)
        ptls = fid.getElementsByTagName('Trackpoint')
        coords = []
        is_with_alt = False
        for sample in ptls:
            for ent in sample.childNodes:
                if ent.localName == "Position":
                    #Iterate on child nodes to get latitude and longitude
                    for pos_node in ent.childNodes:
                        if pos_node.localName == "LatitudeDegrees":
                            lat = pos_node.childNodes[0].data
                        elif pos_node.localName == "LongitudeDegrees":
                            lng = pos_node.childNodes[0].data
                elif ent.localName == "AltitudeMeters":
                    alt = float(ent.childNodes[0].data)
                    if (alt != 0):
                        is_with_alt = True
            coords.append((lng, lat, alt))
        dist = 0.0
        (lon_old, lat_old) = point_conversion(coords[0][0], coords[0][1])
        returncp = []
        for i in range(len(coords)):
            (lon_new, lat_new) = point_conversion(coords[i][0], coords[i][1])
            dist += haversine(lat_old, lat_new, lon_old, lon_new)
            lon_old = lon_new
            lat_old = lat_new
            returncp.append((dist, coords[i][2], u"", coords[i][1], coords[i][0]))
        retfile = SlopeFile(returncp)
        logging.info("is with alt " + str(is_with_alt))
        retfile.hasAltitudes = is_with_alt
        return retfile
    except Exception, err:
        logging.exception(err)
        return None

(UNSUPPORTED, CPS, AUTHOR, DATA) = range(4)
def __open_txt(file_, slopelist, slopenum):
    """Open a Cyclomaniac TXT file"""
    if isinstance(file_, basestring):
        fid = codecs.open(file_, encoding='utf-8', mode='r', errors='replace')
    else:
        fid = file_
    #with open(filename, 'rb') as fid:
    mode = CPS
    for line in fid:
        line = line.strip()
        #Find if switching to a different mode.
        if line.startswith('['):
            modename = line.strip('[]').lower()
            #print modename
            if modename == 'author':
                mode = AUTHOR
            elif modename == 'data':
                mode = DATA
            else:
                mode = UNSUPPORTED
            continue

        #mode cps import check point to slope.
        if mode == CPS:
            #format is dist, alt, name, other.
            (dist, alt, name) = line.split(',')[:3]
            (dist, alt) = (float(dist), float(alt))
            name = name.strip('\"')
            slopelist.add_cp(slopenum, dist, alt, name)

        #mode data imports extra information about the slope
        elif mode == DATA:
            try:
                (lhs, rhs) = line.split('=')[:2]
            except:
                continue
            #print lhs +" <- " + rhs
            if lhs == 'IDPercorso':
                slopelist.set_name(slopenum, rhs)
            elif lhs == 'Stato':
                slopelist.set_country(slopenum, rhs)
            elif lhs == 'Url':
                slopelist.set_url(slopenum, rhs)
    fid.close()

def __open_sal(filepathname, slopelist, slopenum):
    """Open a Salitaker SAL file"""
    if isinstance(filepathname, basestring):
        filename = os.path.split(filepathname)[-1]
        name = filename.split('.')[0]
        slopelist.set_name(slopenum, name)

        fid = codecs.open(filepathname, encoding='utf-8', mode='r')
    else:
        fid = filepathname
    #with open(filepathname, 'rb') as fid:
    ncp = fid.readline().strip().split(' ')[4]
    for _ in range(int(ncp)):
        name = fid.readline().strip()
        (dist, alt) = fid.readline().strip().split(' ')
        slopelist.add_cp(slopenum, float(dist), float(alt), name)
    fid.close()


def __open_crp(filepathname, slopelist, slopenum, num_cps=-1):
    """Open  Ciclotour crp
    The parameter num_cps is the number ok cp to import (at most).
    Particular values are:
    0: import every cp
    -1: import the default number of cp"""
    data = ""
    if isinstance(filepathname, basestring):
        fid = open(filepathname, mode='rb')#needed byte access not unicode string
    else:
        fid = filepathname
    #with open(filepathname, 'rb') as fid:
    byte = " "
    while byte:
        byte = fid.read(1)
        data += byte
    fid.close()
    if not data:
        #File not accesible or empty
        return -1

    if data[0] == '\x78' and data[1] == '\xda':
        plaincsv = zlib.decompress(data)
    elif data[0] == 'H' and data[1] == 'R':
        plaincsv = data
    else:
        #Invalid file format
        return -1
    #print plaincsv
    points = plaincsv.split('***')
    line = points[0].split('\n')   #plaincsv.split('\n')
    #print line[0]
    next_cp = 0
    max_dist = float(line[-2].split('\t')[2])/100
    lastline = len(line)-3
    if num_cps < 0:
        N = 20 #default value: often looks good.
    elif num_cps == 0:
        N = lastline
    else:
        N = num_cps
    for i in range(lastline):
        fields = line[i+2].split('\t')
        #print fields
        dist = float(fields[2])/100
        #create N check points
        if dist > next_cp or i == lastline or fields[8] != '':
            alt = float(fields[3])
            slopelist.add_cp(slopenum, dist, alt, unicode(fields[8], 'utf-8', errors='replace'))
            next_cp = dist + max_dist/N
    return True

### Function that return kml string from gpx
def gpxtokml(xmlfile):
    """Load a gpx file into kml"""
    if isinstance(xmlfile, basestring) and xmlfile.strip().startswith('<'):
        fid = xml.dom.minidom.parseString(xmlfile)
    else:
        fid = xml.dom.minidom.parse(xmlfile)
    ptls = fid.getElementsByTagName('trkpt')
    coords = [(el.attributes["lon"].value, el.attributes["lat"].value) for el in ptls]
    kmlstring = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
    kmlstring += "<kml xmlns=\"http://earth.google.com/kml/2.0\" xmlns:atom=\"http://www.w3.org/2005/Atom\">\n"
    kmlstring += "<Document>\n"
    kmlstring += "<LineString><coordinates>"

    for coord in coords:
        (lon, lat) = coord
        kmlstring += lon + "," + lat + " "
    kmlstring += "</coordinates></LineString>\n"
    kmlstring += "</Document>\n</kml>"
    return kmlstring
### class for various formats

class SlopeFile:
    def __init__(self, cps):
        """Load a track file"""
        self.cps = cps
        self.hasAltitudes = False
    def max_dist(self):
        """ Maximum distance of the track """
        return self.cps[-1][0]
    def __len__(self):
        """ Number of points of the track """
        return len(self.cps)
    def newSlope(self, slopelist, slopenum, num, seltype=0):
        """ Create a new slope in the slopelist"""
        if seltype == 0:#using the given number of check points in num
            max_dist = self.max_dist()
            next_cp = 0.0
            if num == -1:
                N = 30
            elif num == 0:
                N = len(self.cps)
            else:
                N = num
            for i in range(len(self.cps)):
                (dist, alt, name, lat, lng)= self.cps[i]
                slopelist.add_coord(slopenum, float(lat), float(lng))
                #create approximately N check points
                if i == 0 or dist > next_cp or i == len(self.cps)-1:
                    slopelist.add_cp(slopenum, dist, alt, name)
                    next_cp = dist + max_dist/N
            return True
        elif seltype == 1:#using the passed minimum distance in metres in num
            next_dist = num
            for i in range(len(self.cps)):
                (dist, alt, name, lat, lng)= self.cps[i]
                slopelist.add_coord(slopenum, float(lat), float(lng))
                if i == 0 or dist > next_dist or i == len(self.cps)-1:
                    slopelist.add_cp(slopenum, dist, alt, name)
                    next_dist = dist + num
            return True
        else:
            return False
    def getCoords(self, slopelist, slopenum):
        for i in range(len(self.cps)):
            (dist, alt, name, lat, lng)= self.cps[i]
            slopelist.add_coord(slopenum, float(lat), float(lng))

# vim:sw=4:softtabstop=4:expandtab