# -*- 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 ... 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('' + 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 + '') 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 + '') 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))