1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
import logging
import re
from collections.abc import Iterable
from io import BytesIO
import jmespath
import yaml
from lxml import objectify
logger = logging.getLogger("xsar.xml_parser")
logger.addHandler(logging.NullHandler())
# TODO: no variable caching is not while https://github.com/dask/distributed/issues/5610 is not solved
class XmlParser:
"""
Parameters
----------
xpath_mappings: dict
first level key is xml file type
second level key is variable name to be created
mappings may be 'xpath', or 'tuple(func,xpath)', or 'dict'
- xpath is an lxml xpath
- func is a decoder function fed by xpath
- dict is a nested dict with same structure, to create more hierarchy levels.
compounds_vars: dict
compounds variables are variables composed of several variables taken from xpath_mappings
the key is the variable name, and the value is a tuple or a dict.
if dict: (key, jpath), where key is the
sub variable name to create, and jpath is a jmespath in xpath_mappings.
if tuple: ( func, iterable ), where func(iterable) will be called to convert the iterable to another object.
iterable values are jpath. if iterable is a tuple, func(*iterable) will be called.
namespaces: dict
xml namespaces, passed to lxml.xpath.
namespaces are mutualised between all handled xml files.
"""
def __init__(self, mapper, xpath_mappings={}, compounds_vars={}, namespaces={}):
self._namespaces = namespaces
self._xpath_mappings = xpath_mappings
self._compounds_vars = compounds_vars
self._mapper = mapper
def __del__(self):
logger.debug("__del__ XmlParser")
def getroot(self, xml_file):
"""return xml root object from xml_file. (also update self._namespaces with fetched ones)"""
raw_data = self._mapper[xml_file]
# xml_root = etree.fromstring(raw_data)
xml_root = objectify.parse(BytesIO(raw_data)).getroot()
self._namespaces.update(xml_root.nsmap)
return xml_root
def xpath(self, xml_file, path):
"""
get path from xml_file. this is a simple wrapper for `objectify.parse(xml_file).getroot().xpath(path)`
"""
xml_root = self.getroot(xml_file)
result = [
getattr(e, "pyval", e)
for e in xml_root.xpath(path, namespaces=self._namespaces)
]
return result
def get_var(self, xml_file, jpath, describe=False):
"""
get simple variable in xml_file.
Parameters
----------
xml_file: str
xml filename
jpath: str
jmespath string reaching xpath in xpath_mappings
describe: bool
If True, describe the variable (ie return xpath used)
Returns
-------
object
xpath list, or decoded object, if a conversion function was specified in xpath_mappings
"""
func = None
xpath = jmespath.search(jpath, self._xpath_mappings)
if xpath is None:
raise KeyError('jmespath "%s" not found in xpath_mappings' % jpath)
if isinstance(xpath, tuple) and callable(xpath[0]):
func, xpath = xpath
if describe:
return xpath
if not isinstance(xpath, str):
raise NotImplementedError(
'Non leaf xpath of type "%s" instead of str' % type(xpath).__name__
)
result = self.xpath(xml_file, xpath)
if func is not None:
result = func(result)
return result
def get_compound_var(self, xml_file, var_name, describe=False):
"""
Parameters
----------
var_name: str
key in self._compounds_vars
xml_file: str
xml_file to use.
describe: bool
If True, only returns a string describing the variable (file, xpath, etc...)
Returns
-------
object
"""
if describe:
# keep only informative parts in filename
# sub SAFE path
minifile = re.sub(".*SAFE/", "", xml_file)
minifile = re.sub(r"-.*\.xml", ".xml", minifile)
var_object = self._compounds_vars[var_name]
func = None
if (
isinstance(var_object, dict)
and "func" in var_object
and callable(var_object["func"])
):
func = var_object["func"]
if isinstance(var_object["args"], tuple):
args = var_object["args"]
else:
raise ValueError("args must be a tuple when func is called")
else:
args = var_object
result = None
if isinstance(args, dict):
result = {}
for key, path in args.items():
result[key] = self.get_var(xml_file, path, describe=describe)
elif isinstance(args, Iterable):
result = [self.get_var(xml_file, p, describe=describe) for p in args]
if isinstance(args, tuple):
result = tuple(result)
if func is not None and not describe:
# apply converter
result = func(*result)
if describe:
if isinstance(result, dict):
result = result.values()
description = yaml.safe_dump({var_name: {minifile: result}})
return description
else:
return result
|
