File: _node_info.py

package info (click to toggle)
python-asdf 2.14.3-1%2Bdeb12u1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm
  • size: 2,280 kB
  • sloc: python: 16,612; makefile: 124
file content (309 lines) | stat: -rw-r--r-- 10,289 bytes parent folder | download
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
import re
from collections import namedtuple

from .schema import load_schema
from .treeutil import get_children


def _filter_tree(info, filters):
    """
    Keep nodes in tree which pass the all of the filters.

    Mutates the tree.
    """
    filtered_children = []
    for child in info.children:
        if _filter_tree(child, filters):
            filtered_children.append(child)
    info.children = filtered_children

    return len(info.children) > 0 or all(f(info.node, info.identifier) for f in filters)


def create_tree(key, node, identifier="root", filters=[], refresh_extension_manager=False):
    """
    Create a `NodeSchemaInfo` tree which can be filtered from a base node.

    Parameters
    ----------
    key : str
        The key to look up.
    node : dict
        The asdf tree to search.
    filters : list of functions
        A list of functions that take a node and identifier and return True if the node should be included in the tree.
    preserve_list : bool
        If True, then lists are preserved. Otherwise, they are turned into dicts.
    refresh_extension_manager : bool
        If `True`, refresh the extension manager before looking up the
        key.  This is useful if you want to make sure that the schema
        data for a given key is up to date.
    """

    schema_info = NodeSchemaInfo.from_root_node(
        key, identifier, node, refresh_extension_manager=refresh_extension_manager
    )

    if len(filters) > 0:
        if not _filter_tree(schema_info, filters):
            return

    return schema_info


def collect_schema_info(
    key, path, node, identifier="root", filters=[], preserve_list=True, refresh_extension_manager=False
):
    """
    Collect from the underlying schemas any of the info stored under key, relative to the path

    Parameters
    ----------
    key : str
        The key to look up.
    path : str or None
        A dot-separated path to the parameter to find the key information on.
        If None return full dictionary.
    node : dict
        The asdf tree to search.
    filters : list of functions
        A list of functions that take a node and identifier and return True if the node should be included in the tree.
    preserve_list : bool
        If True, then lists are preserved. Otherwise, they are turned into dicts.
    refresh_extension_manager : bool
        If `True`, refresh the extension manager before looking up the
        key.  This is useful if you want to make sure that the schema
        data for a given key is up to date.
    """

    schema_info = create_tree(
        key, node, identifier=identifier, filters=filters, refresh_extension_manager=refresh_extension_manager
    )

    info = schema_info.collect_info(preserve_list=preserve_list)

    if path is not None:
        for path_key in path.split("."):
            info = info.get(path_key, None)

            if info is None:
                return

    return info


def _get_extension_manager(refresh_extension_manager):
    from .asdf import AsdfFile, get_config
    from .extension import ExtensionManager

    af = AsdfFile()
    if refresh_extension_manager:
        config = get_config()
        af._extension_manager = ExtensionManager(config.extensions)

    return af.extension_manager


_SchemaInfo = namedtuple("SchemaInfo", ["info", "value"])


class SchemaInfo(_SchemaInfo):
    """
    A class to hold the schema info and the value of the node.

    Parameters
    ----------
    info : dict
        The schema info.
    value : object
        The value of the node.
    """

    def __repr__(self):
        return f"{self.info}"


class NodeSchemaInfo:
    """
    Container for keyed information collected from a schema about a node of an ASDF file tree.

        This contains node alongside the parent and child nodes of that node in the ASDF file tree.
        Effectively this means that each of these "node" objects represents the a subtree of the file tree
        rooted at the node in question alongside methods to access the underlying schemas for the portions
        of the ASDF file in question.

    This is used for a variety of general purposes, including:
    - Providing the long descriptions for nodes as described in the schema.
    - Assisting in traversing an ASDF file like trees, to search nodes.
    - Providing a way to pull static information about an ASDF file which has
      been stored within the schemas for that file.

    Parameters
    ----------
    key : str
        The key for the information to be collected from the underlying schema(s).

    parent : NodeSchemaInfo
        The parent node of this node. None if this is the root node.

    identifier : str
        The identifier for this node in the ASDF file tree.

    node : any
        The value of the node in the ASDF file tree.

    depth : int
        The depth of this node in the ASDF file tree.

    recursive : bool
        If this node has already been visited, then this is set to True. Default is False.

    visible : bool
        If this node will be made visible in the output. Default is True.

    children : list
        List of the NodeSchemaInfo objects for the children of this node. This is a leaf node if this is empty.

    schema : dict
        The portion of the underlying schema corresponding to the node.
    """

    def __init__(self, key, parent, identifier, node, depth, recursive=False, visible=True):
        self.key = key
        self.parent = parent
        self.identifier = identifier
        self.node = node
        self.depth = depth
        self.recursive = recursive
        self.visible = visible
        self.children = []
        self.schema = None

    @classmethod
    def traversable(cls, node):
        """
        This method determines if the node is an instance of a class that
        supports introspection by the info machinery. This determined by
        the presence of a __asdf_traverse__ method.
        """
        return hasattr(node, "__asdf_traverse__")

    @property
    def visible_children(self):
        return [c for c in self.children if c.visible]

    @property
    def parent_node(self):
        if self.parent is not None:
            return self.parent.node

    @property
    def info(self):
        if self.schema is not None:
            return self.schema.get(self.key, None)

    def get_schema_for_property(self, identifier):
        subschema = self.schema.get("properties", {}).get(identifier, None)
        if subschema is not None:
            return subschema

        subschema = self.schema.get("properties", {}).get("patternProperties", None)
        if subschema:
            for key in subschema:
                if re.search(key, identifier):
                    return subschema[key]
        return {}

    def set_schema_for_property(self, parent, identifier):
        """Extract a subschema from the parent for the identified property"""

        self.schema = parent.get_schema_for_property(identifier)

    def set_schema_from_node(self, node, extension_manager):
        """Pull a tagged schema for the node"""

        tag_def = extension_manager.get_tag_definition(node._tag)
        schema_uri = tag_def.schema_uris[0]
        schema = load_schema(schema_uri)

        self.schema = schema

    @classmethod
    def from_root_node(cls, key, root_identifier, root_node, schema=None, refresh_extension_manager=False):
        """
        Build a NodeSchemaInfo tree from the given ASDF root node.
        Intentionally processes the tree in breadth-first order so that recursively
        referenced nodes are displayed at their shallowest reference point.
        """
        extension_manager = _get_extension_manager(refresh_extension_manager)

        current_nodes = [(None, root_identifier, root_node)]
        seen = set()
        root_info = None
        current_depth = 0
        while True:
            next_nodes = []

            for parent, identifier, node in current_nodes:
                if (isinstance(node, dict) or isinstance(node, tuple) or cls.traversable(node)) and id(node) in seen:
                    info = NodeSchemaInfo(key, parent, identifier, node, current_depth, recursive=True)
                    parent.children.append(info)

                else:
                    info = NodeSchemaInfo(key, parent, identifier, node, current_depth)

                    if root_info is None:
                        root_info = info

                    if parent is not None:
                        if parent.schema is not None and not cls.traversable(node):
                            info.set_schema_for_property(parent, identifier)

                        parent.children.append(info)

                    seen.add(id(node))

                    if cls.traversable(node):
                        t_node = node.__asdf_traverse__()
                        info.set_schema_from_node(node, extension_manager)

                    else:
                        t_node = node

                    if parent is None:
                        info.schema = schema

                    for child_identifier, child_node in get_children(t_node):
                        next_nodes.append((info, child_identifier, child_node))

            if len(next_nodes) == 0:
                break

            current_nodes = next_nodes
            current_depth += 1

        return root_info

    def collect_info(self, preserve_list=True):
        """
        Collect the information from the NodeSchemaInfo tree, and return it as nested dict.

        Parameters
        ----------

        preserve_list : bool
            If True, then lists are preserved. Otherwise, they are turned into dicts.
        """
        if preserve_list and (isinstance(self.node, list) or isinstance(self.node, tuple)) and self.info is None:
            info = [c_info for child in self.visible_children if len(c_info := child.collect_info(preserve_list)) > 0]
        else:
            info = {
                child.identifier: c_info
                for child in self.visible_children
                if len(c_info := child.collect_info(preserve_list)) > 0
            }

            if self.info is not None:
                info[self.key] = SchemaInfo(self.info, self.node)

        return info