#!/usr/bin/env python # -*- coding: utf-8 -*- from lxml import etree from xml.sax.expatreader import AttributesImpl import sys, urllib, re, copy, traceback from astropy.io.votable import parse, parse_single_table from collections import OrderedDict from util_classes import * class Globals_Mapper: # # global resources # path_separator = re.compile("[\.:]+") def __init__(self, filename): self.name = "Globals" self.filename = filename self.globals_node = None self.value_map = OrderedDict() self.root_tuples = {} self.print_message("Init " + self.name + " on " + self.filename ) def __str__(self): retour = "Mapper " + self.name + " on " + self.filename return retour def __repr__(self): retour = self.__str__() + "\n" retour += "Root Tuple\n" for key, value in self.root_tuples.items(): retour += " " + key + "\n" for skey, svalue in value.items(): retour += " " + skey + " : " + svalue.__str__() return retour; def print_message(self, message): print("Mapper " + self.name + " " + message) def read_annotation(self): self.print_message( "Reading FILE " + self.filename) # # init parser # if self.filename.startswith("http://") or self.filename.startswith("https://") : tree = etree.ElementTree(file=urllib.request.urlopen(self.filename)) else : tree = etree.parse(self.filename) self.globals_node = tree.find("//GLOBALS") for node in self.globals_node.xpath("./TUPLE"): self.root_class_node = node sk = self.root_class_node.get("dmrole") id = self.root_class_node.get("ID") if sk == None: self.print_message("Root tuple without dmtype found in globals") sys.exit(1) self.print_message("Root class with dmtype=" + sk + " found") self.parse_tuple(self.root_class_node, "#" + sk) self.root_tuples["#" + sk] = copy.deepcopy(self.value_map) #self.root_tuples[id] = copy.deepcopy(self.value_map) self.value_map = {} def parse_tuple(self, node, key): role = node.get("dmrole") if role == None: self.print_message("tuple without dmrole found") sys.exit(1) # Avoid duplicate elements if role != None and key != ("#" + role): key = key + "#" + role if node.get("ref") != None: mapper = Mapper("../../data/globals.tuple.votable.xml", "//GLOBALS/TUPLE[@ID='" + node.get("ref") + "']") mapper.read_annotation() for refkey, value in mapper.value_map.items(): # ignore shortcuts if refkey.startswith("#"): # remove the root class dmtype self.value_map[key + '#' + re.sub("^#.*#", '', refkey)] = value else : for child in node.xpath("*"): if( child.tag == 'VALUE'): self.parse_value(child, key) elif( child.tag == 'TUPLE'): self.parse_tuple(child, key) elif( child.tag == 'COLLECTION'): self.parse_collection(child, key) def parse_value(self, node, key): source = node.get("source") val = None if( source.startswith('@') ): val = source elif( source == 'child'): val = node.text if val == None: print("Invalid VALUE in globals" ) sys.exit(1) value = ValueDescripor(val, "globals") #{"value": val, "table": "globals", "multiplicity": None, "field": {"name": None, "index": None}, "filter": None} ckey = key + "#" + node.get("dmrole") self.value_map[ckey ] = value def parse_collection(self, node, key): for child in node.xpath("*"): if( child.tag == 'VALUE'): self.parse_value(child, key) elif( child.tag == 'TUPLE'): self.parse_tuple(child, key) elif( child.tag == 'COLLECTION'): self.parse_collection(child, key)