14916511482

Committed 08 May 2025 08:55PM UTC coverage: 80.239% (+0.05%) from 80.194%

Build # 14916511482

Build Type

push

github

Committed by

jharwell

Commit Message

feature(#326): Arrow storage

- Start updating docs/code to say "output files" instead of "csv"

- Move flattening to be a platform callback so it can be done before scaffolding
  a batch exp.

- Start hacking at statistics generation to support arrow and CSV. Things seem
  to work with arrow, but need to re-run some imagizing/csv tests to verify
  things aren't broken in other ways.

- Add a placeholder for fleshing out SIERRA's dataflow model, which is a really
  important aspect of usage which currently isn't documented.

- Remove excessive class usage in DataFrame{Reader,Writer}

- Overhaul collation and fix nasty bug where data was only being gathered from 1
  run per sim; no idea how long that has been in there. Added an assert so that
  can't happen again.

Run Details

349 of 385 new or added lines in 28 files covered. (90.65%)

3 existing lines in 3 files now uncovered.

5441 of 6781 relevant lines covered (80.24%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.0

/sierra/plugins/expdef/xml/plugin.py

# Copyright 2024 John Harwell, All rights reserved.
#
#  SPDX-License-Identifier: MIT
"""Plugin for parsing and manipulating template input files in XML format.

"""

# Core packages
import pathlib
import logging
import xml.etree.ElementTree as ET
import typing as tp

# 3rd party packages
import implements

# Project packages
from sierra.core.experiment import definition
from sierra.core import types


class Writer():
    """Write the XML experiment to the filesystem according to configuration.

    More than one file may be written, as specified.
    """

    def __init__(self, tree: ET.ElementTree) -> None:
        self.tree = tree
        self.root = tree.getroot()
        self.logger = logging.getLogger(__name__)

    def __call__(self,
                 write_config: definition.WriterConfig,
                 base_opath: pathlib.Path) -> None:
        for config in write_config.values:
            self._write_with_config(base_opath, config)

    def _write_with_config(self,
                           base_opath: tp.Union[pathlib.Path, str],
                           config: dict) -> None:
        tree, src_root, opath = self._prepare_tree(pathlib.Path(base_opath),
                                                   config)

        if tree is None:
            self.logger.warning("Cannot write non-existent tree@'%s' to '%s'",
                                src_root,
                                opath)
            return

        self.logger.trace("Write tree@%s to %s",  # type: ignore
                          src_root,
                          opath)

        # Renaming tree root is not required
        if 'rename_to' in config and config['rename_to'] is not None:
            tree.tag = config['rename_to']
            self.logger.trace("Rename tree root -> %s",  # type: ignore
                              config['rename_to'])

        # Adding new children not required
        if all(k in config and config[k] is not None for k in ['new_children_parent',
                                                               'new_children']):
            self._add_new_children(config, tree)

        # Grafts are not required
        if all(k in config and config[k] is not None for k in ['child_grafts_parent',
                                                               'child_grafts']):
            self._add_grafts(config, tree)

        to_write = ET.ElementTree(tree)

        ET.indent(to_write.getroot(), space="\t", level=0)
        ET.indent(to_write, space="\t", level=0)
        to_write.write(opath, encoding='utf-8')

    def _add_grafts(self,
                    config: dict,
                    tree: ET.Element) -> None:

        graft_parent = tree.find(config['child_grafts_parent'])
        assert graft_parent is not None, \
            f"Bad parent '{graft_parent}' for grafting"

        for g in config['child_grafts']:
            self.logger.trace("Graft tree@'%s' as child under '%s'",  # type: ignore
                              g,
                              graft_parent)
            elt = self.root.find(g)
            graft_parent.append(elt)

    def _add_new_children(self,
                          config: dict,
                          tree: ET.ElementTree) -> None:
        """Given the experiment definition, add new children as configured.

        We operate on the whole definition in-situ, rather than creating a new
        subtree with all the children because that is less error prone in terms
        of grafting the new subtree back into the experiment definition.
        """

        parent = tree.find(config['new_children_parent'])

        for spec in config['new_children']:
            if spec.as_root_elt:
                # Special case: Adding children to an empty tree
                tree = ET.Element(spec.path, spec.attr)
                continue

            elt = parent.find(spec.path)

            assert elt is not None, \
                (f"Could not find parent '{spec.path}' of new child element '{spec.tag}' "
                 "to add")

            ET.SubElement(elt, spec.tag, spec.attr)

            self.logger.trace("Create child element '%s' under '%s'",  # type: ignore
                              spec.tag,
                              spec.path)

    def _prepare_tree(self,
                      base_opath: pathlib.Path,
                      config: dict) -> tp.Tuple[tp.Optional[ET.Element],
                                                str,
                                                pathlib.Path]:
        assert 'src_parent' in config, "'src_parent' key is required"
        assert 'src_tag' in config and config['src_tag'] is not None, \
            "'src_tag' key is required"

        if config['src_parent'] is None:
            src_root = config['src_tag']
        else:
            src_root = "{0}/{1}".format(config['src_parent'],
                                        config['src_tag'])

        tree_out = self.tree.getroot().find(src_root)

        # Customizing the output write path is not required
        opath = base_opath
        if 'opath_leaf' in config and config['opath_leaf'] is not None:
            opath = base_opath.with_name(base_opath.name + str(config['opath_leaf']))

        self.logger.trace("Preparing subtree write of '%s' to '%s', root='%s'",
                          tree_out,
                          opath,
                          tree_out)

        return (tree_out, src_root, opath)


def root_querypath() -> str:
    return "."


@implements.implements(definition.BaseExpDef)
class ExpDef:
    """Read, write, and modify parsed XML files into experiment definitions.
    """

    def __init__(self,
                 input_fpath: pathlib.Path,
                 write_config: tp.Optional[definition.WriterConfig] = None) -> None:

        self.write_config = write_config
        self.input_fpath = input_fpath
        self.tree = ET.parse(self.input_fpath)
        self.root = self.tree.getroot()
        self.element_adds = definition.ElementAddList()
        self.attr_chgs = definition.AttrChangeSet()

        self.logger = logging.getLogger(__name__)

    def write_config_set(self, config: definition.WriterConfig) -> None:
        """Set the write config for the object.

        Provided for cases in which the configuration is dependent on whether or
        not certain tags/element are present in the input file.

        """
        self.write_config = config

    def write(self, base_opath: pathlib.Path) -> None:
        assert self.write_config is not None, \
            "Can't write without write config"

        writer = Writer(self.tree)
        writer(self.write_config, base_opath)

    def flatten(self, keys: tp.List[str]) -> None:
        raise NotImplementedError(
            "The XML expdef plugin does not support flattening")

    def attr_get(self, path: str, attr: str) -> tp.Optional[tp.Union[str, int, float]]:
        el = self.root.find(path)
        if el is not None and attr in el.attrib:
            return el.attrib[attr]
        return None

    def attr_change(self,
                    path: str,
                    attr: str,
                    value: tp.Union[str, int, float],
                    noprint: bool = False) -> bool:
        el = self.root.find(path)
        if el is None:
            if not noprint:
                self.logger.warning("Parent element '%s' not found", path)
            return False

        if attr not in el.attrib:
            if not noprint:
                self.logger.warning("Attribute '%s' not found in path '%s'",
                                    attr,
                                    path)
            return False

        el.attrib[attr] = value
        self.logger.trace("Modify attr: '%s/%s' = '%s'",  # type: ignore
                          path,
                          attr,
                          value)

        self.attr_chgs.add(definition.AttrChange(path, attr, str(value)))
        return True

    def attr_add(self,
                 path: str,
                 attr: str,
                 value: tp.Union[str, int, float],
                 noprint: bool = False) -> bool:
        el = self.root.find(path)
        if el is None:
            if not noprint:
                self.logger.warning("Parent element '%s' not found", path)
            return False

        if attr in el.attrib:
            if not noprint:
                self.logger.warning("Attribute '%s' already in path '%s'",
                                    attr,
                                    path)
            return False

        el.set(attr, value)
        self.logger.trace("Add new attribute: '%s/%s' = '%s'",  # type: ignore
                          path,
                          attr,
                          value)
        self.attr_chgs.add(definition.AttrChange(path, attr, str(value)))
        return True

    def has_element(self, path: str) -> bool:
        return self.root.find(path) is not None

    def has_attr(self, path: str, attr: str) -> bool:
        el = self.root.find(path)
        if el is None:
            return False
        return attr in el.attrib

    def element_change(self, path: str, tag: str, value: str) -> bool:
        el = self.root.find(path)
        if el is None:
            self.logger.warning("Parent element '%s' not found", path)
            return False

        for child in el:
            if child.tag == tag:
                child.tag = value
                self.logger.trace("Modify element: '%s/%s' = '%s'",  # type: ignore
                                  path,
                                  tag,
                                  value)
                return True

        self.logger.warning("No such element '%s' found in '%s'", tag, path)
        return False

    def element_remove(self, path: str, tag: str, noprint: bool = False) -> bool:
        parent = self.root.find(path)

        if parent is None:
            if not noprint:
                self.logger.warning("Parent node '%s' not found", path)
            return False

        victim = parent.find(tag)
        if victim is None:
            if not noprint:
                self.logger.warning("No victim '%s' found in parent '%s'",
                                    tag,
                                    path)
            return False

        parent.remove(victim)
        return True

    def element_remove_all(self,
                           path: str,
                           tag: str,
                           noprint: bool = False) -> bool:

        parent = self.root.find(path)

        if parent is None:
            if not noprint:
                self.logger.warning("Parent element '%s' not found", path)
            return False

        victims = parent.findall(tag)
        if not victims:
            if not noprint:
                self.logger.warning("No victims matching '%s' found in parent '%s'",
                                    tag,
                                    path)
            return False

        for victim in victims:
            parent.remove(victim)
            self.logger.trace("Remove matching element: '%s/%s'",  # type: ignore
                              path,
                              tag)

        return True

    def element_add(self,
                    path: str,
                    tag: str,
                    attr: tp.Optional[types.StrDict] = None,
                    allow_dup: bool = True,
                    noprint: bool = False) -> bool:
        """
        Add tag name as a child element of enclosing parent.
        """
        parent = self.root.find(path)

        if parent is None:
            if not noprint:
                self.logger.warning("Parent element '%s' not found", path)
            return False

        if not allow_dup:
            if parent.find(tag) is not None:
                if not noprint:
                    self.logger.warning("Child element '%s' already in parent '%s'",
                                        tag,
                                        path)
                return False

            ET.SubElement(parent, tag, attrib=attr if attr else {})
            self.logger.trace("Add new unique element: '%s/%s' = '%s'",  # type: ignore
                              path,
                              tag,
                              str(attr))
        else:
            # Use ET.Element instead of ET.SubElement so that child nodes with
            # the same 'tag' don't overwrite each other.
            child = ET.Element(tag, attrib=attr if attr else {})
            parent.append(child)
            self.logger.trace("Add new element: '%s/%s' = '%s'",  # type: ignore
                              path,
                              tag,
                              str(attr))

        self.element_adds.append(definition.ElementAdd(path, tag, attr, allow_dup))
        return True


def unpickle(fpath: pathlib.Path) -> tp.Optional[tp.Union[definition.AttrChangeSet,
                                                          definition.ElementAddList]]:
    """Unickle all XML modifications from the pickle file at the path.

    You don't know how many there are, so go until you get an exception.

    """
    try:
        return definition.AttrChangeSet.unpickle(fpath)
    except EOFError:
        pass

    try:
        return definition.ElementAddList.unpickle(fpath)
    except EOFError:
        pass

    raise NotImplementedError


__all__ = [
    'ExpDef',
    'unpickle'
]

1	# Copyright 2024 John Harwell, All rights reserved.
2	#
3	# SPDX-License-Identifier: MIT
4	"""Plugin for parsing and manipulating template input files in XML format.
5
6	"""
7
8	# Core packages
9	import pathlib	1✔
10	import logging	1✔
11	import xml.etree.ElementTree as ET	1✔
12	import typing as tp	1✔
13
14	# 3rd party packages
15	import implements	1✔
16
17	# Project packages
18	from sierra.core.experiment import definition	1✔
19	from sierra.core import types	1✔
20
21
22	class Writer():	1✔
23	"""Write the XML experiment to the filesystem according to configuration.
24
25	More than one file may be written, as specified.
26	"""
27
28	def __init__(self, tree: ET.ElementTree) -> None:	1✔
29	self.tree = tree	1✔
30	self.root = tree.getroot()	1✔
31	self.logger = logging.getLogger(__name__)	1✔
32
33	def __call__(self,	1✔
34	write_config: definition.WriterConfig,
35	base_opath: pathlib.Path) -> None:
36	for config in write_config.values:	1✔
37	self._write_with_config(base_opath, config)	1✔
38
39	def _write_with_config(self,	1✔
40	base_opath: tp.Union[pathlib.Path, str],
41	config: dict) -> None:
42	tree, src_root, opath = self._prepare_tree(pathlib.Path(base_opath),	1✔
43	config)
44
45	if tree is None:	1✔
46	self.logger.warning("Cannot write non-existent tree@'%s' to '%s'",	×
47	src_root,
48	opath)
49	return	×
50
51	self.logger.trace("Write tree@%s to %s", # type: ignore	1✔
52	src_root,
53	opath)
54
55	# Renaming tree root is not required
56	if 'rename_to' in config and config['rename_to'] is not None:	1✔
57	tree.tag = config['rename_to']	1✔
58	self.logger.trace("Rename tree root -> %s", # type: ignore	1✔
59	config['rename_to'])
60
61	# Adding new children not required
62	if all(k in config and config[k] is not None for k in ['new_children_parent',	1✔
63	'new_children']):
64	self._add_new_children(config, tree)	1✔
65
66	# Grafts are not required
67	if all(k in config and config[k] is not None for k in ['child_grafts_parent',	1✔
68	'child_grafts']):
69	self._add_grafts(config, tree)	1✔
70
71	to_write = ET.ElementTree(tree)	1✔
72
73	ET.indent(to_write.getroot(), space="\t", level=0)	1✔
74	ET.indent(to_write, space="\t", level=0)	1✔
75	to_write.write(opath, encoding='utf-8')	1✔
76
77	def _add_grafts(self,	1✔
78	config: dict,
79	tree: ET.Element) -> None:
80
81	graft_parent = tree.find(config['child_grafts_parent'])	1✔
82	assert graft_parent is not None, \	1✔
83	f"Bad parent '{graft_parent}' for grafting"
84
85	for g in config['child_grafts']:	1✔
86	self.logger.trace("Graft tree@'%s' as child under '%s'", # type: ignore	1✔
87	g,
88	graft_parent)
89	elt = self.root.find(g)	1✔
90	graft_parent.append(elt)	1✔
91
92	def _add_new_children(self,	1✔
93	config: dict,
94	tree: ET.ElementTree) -> None:
95	"""Given the experiment definition, add new children as configured.
96
97	We operate on the whole definition in-situ, rather than creating a new
98	subtree with all the children because that is less error prone in terms
99	of grafting the new subtree back into the experiment definition.
100	"""
101
102	parent = tree.find(config['new_children_parent'])	1✔
103
104	for spec in config['new_children']:	1✔
105	if spec.as_root_elt:	1✔
106	# Special case: Adding children to an empty tree
107	tree = ET.Element(spec.path, spec.attr)	1✔
108	continue	1✔
109
110	elt = parent.find(spec.path)	×
111
112	assert elt is not None, \	×
113	(f"Could not find parent '{spec.path}' of new child element '{spec.tag}' "
114	"to add")
115
116	ET.SubElement(elt, spec.tag, spec.attr)	×
117
118	self.logger.trace("Create child element '%s' under '%s'", # type: ignore	×
119	spec.tag,
120	spec.path)
121
122	def _prepare_tree(self,	1✔
123	base_opath: pathlib.Path,
124	config: dict) -> tp.Tuple[tp.Optional[ET.Element],
125	str,
126	pathlib.Path]:
127	assert 'src_parent' in config, "'src_parent' key is required"	1✔
128	assert 'src_tag' in config and config['src_tag'] is not None, \	1✔
129	"'src_tag' key is required"
130
131	if config['src_parent'] is None:	1✔
132	src_root = config['src_tag']	1✔
133	else:
134	src_root = "{0}/{1}".format(config['src_parent'],	1✔
135	config['src_tag'])
136
137	tree_out = self.tree.getroot().find(src_root)	1✔
138
139	# Customizing the output write path is not required
140	opath = base_opath	1✔
141	if 'opath_leaf' in config and config['opath_leaf'] is not None:	1✔
142	opath = base_opath.with_name(base_opath.name + str(config['opath_leaf']))	1✔
143
144	self.logger.trace("Preparing subtree write of '%s' to '%s', root='%s'",	1✔
145	tree_out,
146	opath,
147	tree_out)
148
149	return (tree_out, src_root, opath)	1✔
150
151
152	def root_querypath() -> str:	1✔
153	return "."	1✔
154
155
156	@implements.implements(definition.BaseExpDef)	1✔
157	class ExpDef:	1✔
158	"""Read, write, and modify parsed XML files into experiment definitions.
159	"""
160
161	def __init__(self,	1✔
162	input_fpath: pathlib.Path,
163	write_config: tp.Optional[definition.WriterConfig] = None) -> None:
164
165	self.write_config = write_config	1✔
166	self.input_fpath = input_fpath	1✔
167	self.tree = ET.parse(self.input_fpath)	1✔
168	self.root = self.tree.getroot()	1✔
169	self.element_adds = definition.ElementAddList()	1✔
170	self.attr_chgs = definition.AttrChangeSet()	1✔
171
172	self.logger = logging.getLogger(__name__)	1✔
173
174	def write_config_set(self, config: definition.WriterConfig) -> None:	1✔
175	"""Set the write config for the object.
176
177	Provided for cases in which the configuration is dependent on whether or
178	not certain tags/element are present in the input file.
179
180	"""
181	self.write_config = config	1✔
182
183	def write(self, base_opath: pathlib.Path) -> None:	1✔
184	assert self.write_config is not None, \	1✔
185	"Can't write without write config"
186
187	writer = Writer(self.tree)	1✔
188	writer(self.write_config, base_opath)	1✔
189
190	def flatten(self, keys: tp.List[str]) -> None:	1✔
191	raise NotImplementedError(
192	"The XML expdef plugin does not support flattening")
193
194	def attr_get(self, path: str, attr: str) -> tp.Optional[tp.Union[str, int, float]]:	1✔
195	el = self.root.find(path)	1✔
196	if el is not None and attr in el.attrib:	1✔
197	return el.attrib[attr]	1✔
198	return None	×
199
200	def attr_change(self,	1✔
201	path: str,
202	attr: str,
203	value: tp.Union[str, int, float],
204	noprint: bool = False) -> bool:
205	el = self.root.find(path)	1✔
206	if el is None:	1✔
207	if not noprint:	1✔
208	self.logger.warning("Parent element '%s' not found", path)	1✔
209	return False	1✔
210
211	if attr not in el.attrib:	1✔
212	if not noprint:	×
213	self.logger.warning("Attribute '%s' not found in path '%s'",	×
214	attr,
215	path)
216	return False	×
217
218	el.attrib[attr] = value	1✔
219	self.logger.trace("Modify attr: '%s/%s' = '%s'", # type: ignore	1✔
220	path,
221	attr,
222	value)
223
224	self.attr_chgs.add(definition.AttrChange(path, attr, str(value)))	1✔
225	return True	1✔
226
227	def attr_add(self,	1✔
228	path: str,
229	attr: str,
230	value: tp.Union[str, int, float],
231	noprint: bool = False) -> bool:
232	el = self.root.find(path)	×
233	if el is None:	×
234	if not noprint:	×
235	self.logger.warning("Parent element '%s' not found", path)	×
236	return False	×
237
238	if attr in el.attrib:	×
239	if not noprint:	×
240	self.logger.warning("Attribute '%s' already in path '%s'",	×
241	attr,
242	path)
243	return False	×
244
245	el.set(attr, value)	×
246	self.logger.trace("Add new attribute: '%s/%s' = '%s'", # type: ignore	×
247	path,
248	attr,
249	value)
NEW 250	self.attr_chgs.add(definition.AttrChange(path, attr, str(value)))	×
251	return True	×
252
253	def has_element(self, path: str) -> bool:	1✔
254	return self.root.find(path) is not None	1✔
255
256	def has_attr(self, path: str, attr: str) -> bool:	1✔
257	el = self.root.find(path)	×
258	if el is None:	×
259	return False	×
260	return attr in el.attrib	×
261
262	def element_change(self, path: str, tag: str, value: str) -> bool:	1✔
263	el = self.root.find(path)	1✔
264	if el is None:	1✔
265	self.logger.warning("Parent element '%s' not found", path)	×
266	return False	×
267
268	for child in el:	1✔
269	if child.tag == tag:	1✔
270	child.tag = value	1✔
271	self.logger.trace("Modify element: '%s/%s' = '%s'", # type: ignore	1✔
272	path,
273	tag,
274	value)
275	return True	1✔
276
277	self.logger.warning("No such element '%s' found in '%s'", tag, path)	1✔
278	return False	1✔
279
280	def element_remove(self, path: str, tag: str, noprint: bool = False) -> bool:	1✔
281	parent = self.root.find(path)	1✔
282
283	if parent is None:	1✔
284	if not noprint:	×
285	self.logger.warning("Parent node '%s' not found", path)	×
286	return False	×
287
288	victim = parent.find(tag)	1✔
289	if victim is None:	1✔
290	if not noprint:	1✔
291	self.logger.warning("No victim '%s' found in parent '%s'",	1✔
292	tag,
293	path)
294	return False	1✔
295
296	parent.remove(victim)	1✔
297	return True	1✔
298
299	def element_remove_all(self,	1✔
300	path: str,
301	tag: str,
302	noprint: bool = False) -> bool:
303
304	parent = self.root.find(path)	1✔
305
306	if parent is None:	1✔
307	if not noprint:	×
308	self.logger.warning("Parent element '%s' not found", path)	×
309	return False	×
310
311	victims = parent.findall(tag)	1✔
312	if not victims:	1✔
313	if not noprint:	×
314	self.logger.warning("No victims matching '%s' found in parent '%s'",	×
315	tag,
316	path)
317	return False	×
318
319	for victim in victims:	1✔
320	parent.remove(victim)	1✔
321	self.logger.trace("Remove matching element: '%s/%s'", # type: ignore	1✔
322	path,
323	tag)
324
325	return True	1✔
326
327	def element_add(self,	1✔
328	path: str,
329	tag: str,
330	attr: tp.Optional[types.StrDict] = None,
331	allow_dup: bool = True,
332	noprint: bool = False) -> bool:
333	"""
334	Add tag name as a child element of enclosing parent.
335	"""
336	parent = self.root.find(path)	1✔
337
338	if parent is None:	1✔
339	if not noprint:	×
340	self.logger.warning("Parent element '%s' not found", path)	×
341	return False	×
342
343	if not allow_dup:	1✔
344	if parent.find(tag) is not None:	1✔
345	if not noprint:	1✔
346	self.logger.warning("Child element '%s' already in parent '%s'",	1✔
347	tag,
348	path)
349	return False	1✔
350
351	ET.SubElement(parent, tag, attrib=attr if attr else {})	1✔
352	self.logger.trace("Add new unique element: '%s/%s' = '%s'", # type: ignore	1✔
353	path,
354	tag,
355	str(attr))
356	else:
357	# Use ET.Element instead of ET.SubElement so that child nodes with
358	# the same 'tag' don't overwrite each other.
359	child = ET.Element(tag, attrib=attr if attr else {})	1✔
360	parent.append(child)	1✔
361	self.logger.trace("Add new element: '%s/%s' = '%s'", # type: ignore	1✔
362	path,
363	tag,
364	str(attr))
365
366	self.element_adds.append(definition.ElementAdd(path, tag, attr, allow_dup))	1✔
367	return True	1✔
368
369
370	def unpickle(fpath: pathlib.Path) -> tp.Optional[tp.Union[definition.AttrChangeSet,	1✔
371	definition.ElementAddList]]:
372	"""Unickle all XML modifications from the pickle file at the path.
373
374	You don't know how many there are, so go until you get an exception.
375
376	"""
377	try:	1✔
378	return definition.AttrChangeSet.unpickle(fpath)	1✔
379	except EOFError:	×
380	pass	×
381
382	try:	×
383	return definition.ElementAddList.unpickle(fpath)	×
384	except EOFError:	×
385	pass	×
386
387	raise NotImplementedError
388
389
390	__all__ = [	1✔
391	'ExpDef',
392	'unpickle'
393	]

jharwell / sierra / 14916511482

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous