Skip to content

QA Results Viewer esgf_qa.qaviewer

QCViewer

Bases: App

Source code in esgf_qa/qaviewer.py 
 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
class QCViewer(App):
    BINDINGS = [
        ("q", "quit", "Quit"),
        ("/", "focus_search", "Search"),
        ("n", "next_match", "Next Match"),
        ("p", "prev_match", "Prev Match"),
    ]

    def __init__(self, data):
        super().__init__()
        self.data = data
        self.matches = []
        self.match_index = -1
        self.qc_tree = None  # will be created in compose()
        self.last_highlighted_node = None

    def compose(self) -> ComposeResult:
        yield Header()
        self.qc_tree = Tree("QC Results", id="qc-tree")
        yield self.qc_tree
        yield Input(placeholder="Search...", id="search")
        yield Static("", id="status")
        yield QCFooter()

    def on_tree_node_selected(self, event: Tree.NodeSelected):
        # Right-click simulation: expand all children
        # Unfortunately, Tree.NodeSelected does not carry button info
        # We rely on a right-click flag set in on_mouse_down
        if getattr(self, "_right_click_pending", False):
            self.toggle_expand_node(event.node)
            self._right_click_pending = False

    def on_mouse_up(self, event: MouseUp):
        # Set a flag if right-click
        if event.button == 3:  # right click
            self._right_click_pending = True
            # node = self.qc_tree.cursor_node
            # if node:
            #    self.toggle_expand_node(node)

    def toggle_expand_node(self, node):
        """Right-click: expand if collapsed, collapse if expanded, recursively."""
        if hasattr(node, "_expanded_state"):
            # toggle previous state
            expanding = not node._expanded_state
        else:
            # first time, check actual state
            expanding = not node.is_expanded

        if not expanding:
            # Collapse node and all children
            self._collapse_tree(node)
        else:
            # Expand node and all children, potentially only up to certain level
            depth = self.get_node_depth(node)
            if depth <= 1:
                self._expand_tree_up_to_depth(node, current_lvl=depth, target_lvl=2)
            else:
                self._expand_tree(node)

    def get_node_depth(self, node):
        """Return depth of node (root=0)."""
        depth = 0
        parent = node.parent
        while parent:
            depth += 1
            parent = parent.parent
        return depth

    def _expand_tree_up_to_depth(self, node, current_lvl, target_lvl):
        """Recursively expand node up to certain depth level."""
        node.expand()
        node._expanded_state = True
        if current_lvl >= target_lvl:
            return
        for child in node.children:
            self._expand_tree_up_to_depth(child, current_lvl + 1, target_lvl)

    def _expand_tree(self, node):
        """Recursively expand node and all children."""
        node.expand()
        node._expanded_state = True
        for child in node.children:
            self._expand_tree(child)

    def _collapse_tree(self, node):
        """Recursively collapses node and all children."""
        for child in node.children:
            self._collapse_tree(child)
        node.collapse()
        node._expanded_state = False

    def on_mount(self):
        self.populate_tree(self.qc_tree.root, self.data)
        self.qc_tree.root.expand()

    def populate_tree(self, node, data):
        if isinstance(data, dict):
            for k, v in data.items():
                child = node.add(k, expand=False)
                self.populate_tree(child, v)
        elif isinstance(data, list):
            for i, v in enumerate(data):
                child = node.add(f"[{i}]", expand=False)
                self.populate_tree(child, v)
        else:
            node.add(repr(data))

    def action_focus_search(self):
        self.query_one("#search").focus()

    def on_input_submitted(self, event: Input.Submitted):
        """Called when the user submits a search in the input."""
        query = event.value.strip()
        self.matches = []
        self.match_index = -1

        if query:
            # Collect all matching nodes
            for node in iter_nodes(self.qc_tree.root):
                if query.lower() in str(node.label).lower():
                    self.matches.append(node)

        if self.matches:
            # Start at first match
            self.match_index = 0
            # Jump to it (expand path, collapse old if needed)
            self.jump_to_match()
            # Return focus to tree so n/p work
            self.set_focus(self.qc_tree)
        else:
            self.query_one("#status", Static).update(f"No matches for '{query}'")

    def focus_match(self):
        node = self.matches[self.match_index]
        node.expand_all()
        self.qc_tree.select_node(node)
        self.qc_tree.scroll_to_node(node)
        self.query_one("#status", Static).update(
            f"Match {self.match_index+1}/{len(self.matches)}: {node.label}"
        )

    def action_next_match(self):
        if self.matches:
            self.match_index = (self.match_index + 1) % len(self.matches)
            self.jump_to_match()

    def action_prev_match(self):
        if self.matches:
            self.match_index = (self.match_index - 1) % len(self.matches)
            self.jump_to_match()

    def jump_to_match(self) -> None:
        """Jump to the current match, expanding its parents and collapsing previous."""
        if not self.matches or self.match_index < 0:
            return

        # Collapse the previously focused node if any
        if hasattr(self, "current_match_node") and self.current_match_node is not None:
            try:
                self.current_match_node.collapse()
            except Exception:
                pass

        # Get the current match node
        node = self.matches[self.match_index]
        self.current_match_node = node

        # Expand all parents so the node is visible
        parent = node.parent
        while parent:
            parent.expand()
            parent = parent.parent

        # Expand this node itself too
        node.expand()

        # Scroll to and select it
        self.qc_tree.select_node(node)
        self.qc_tree.scroll_to_node(node)
        self.qc_tree.select_node(node)

        # Status line
        self.query_one("#status", Static).update(
            f"Match {self.match_index+1}/{len(self.matches)}: {node.label}"
        )

get_node_depth(node)

Return depth of node (root=0).

Source code in esgf_qa/qaviewer.py 
127
128
129
130
131
132
133
134
def get_node_depth(self, node):
    """Return depth of node (root=0)."""
    depth = 0
    parent = node.parent
    while parent:
        depth += 1
        parent = parent.parent
    return depth

jump_to_match()

Jump to the current match, expanding its parents and collapsing previous.

Source code in esgf_qa/qaviewer.py 
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
def jump_to_match(self) -> None:
    """Jump to the current match, expanding its parents and collapsing previous."""
    if not self.matches or self.match_index < 0:
        return

    # Collapse the previously focused node if any
    if hasattr(self, "current_match_node") and self.current_match_node is not None:
        try:
            self.current_match_node.collapse()
        except Exception:
            pass

    # Get the current match node
    node = self.matches[self.match_index]
    self.current_match_node = node

    # Expand all parents so the node is visible
    parent = node.parent
    while parent:
        parent.expand()
        parent = parent.parent

    # Expand this node itself too
    node.expand()

    # Scroll to and select it
    self.qc_tree.select_node(node)
    self.qc_tree.scroll_to_node(node)
    self.qc_tree.select_node(node)

    # Status line
    self.query_one("#status", Static).update(
        f"Match {self.match_index+1}/{len(self.matches)}: {node.label}"
    )

on_input_submitted(event)

Called when the user submits a search in the input.

Source code in esgf_qa/qaviewer.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def on_input_submitted(self, event: Input.Submitted):
    """Called when the user submits a search in the input."""
    query = event.value.strip()
    self.matches = []
    self.match_index = -1

    if query:
        # Collect all matching nodes
        for node in iter_nodes(self.qc_tree.root):
            if query.lower() in str(node.label).lower():
                self.matches.append(node)

    if self.matches:
        # Start at first match
        self.match_index = 0
        # Jump to it (expand path, collapse old if needed)
        self.jump_to_match()
        # Return focus to tree so n/p work
        self.set_focus(self.qc_tree)
    else:
        self.query_one("#status", Static).update(f"No matches for '{query}'")

toggle_expand_node(node)

Right-click: expand if collapsed, collapse if expanded, recursively.

Source code in esgf_qa/qaviewer.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def toggle_expand_node(self, node):
    """Right-click: expand if collapsed, collapse if expanded, recursively."""
    if hasattr(node, "_expanded_state"):
        # toggle previous state
        expanding = not node._expanded_state
    else:
        # first time, check actual state
        expanding = not node.is_expanded

    if not expanding:
        # Collapse node and all children
        self._collapse_tree(node)
    else:
        # Expand node and all children, potentially only up to certain level
        depth = self.get_node_depth(node)
        if depth <= 1:
            self._expand_tree_up_to_depth(node, current_lvl=depth, target_lvl=2)
        else:
            self._expand_tree(node)

iter_nodes(node)

Recursively yield all nodes starting from this one.

Source code in esgf_qa/qaviewer.py 
52
53
54
55
56
def iter_nodes(node):
    """Recursively yield all nodes starting from this one."""
    yield node
    for child in node.children:
        yield from iter_nodes(child)

transform_keys(data)

Apply similar renaming logic than in display_qc_results.html

Source code in esgf_qa/qaviewer.py 
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
def transform_keys(data):
    """Apply similar renaming logic than in display_qc_results.html"""
    weight_map = {3: "Required", 2: "Recommended", 1: "Suggested"}
    result = {}
    info_map = {
        "id": "Dataset-ID",
        "date": "Date",
        "parent_dir": "Root Directory",
        "files": "# Files",
        "datasets": "# Datasets",
        "cc_version": "Compliance Checker Version",
        "checkers": "Applied Checkers",
        "inter_ds_con_checks_ref": "Reference Datasets (inter-dataset consistency checks)",
    }

    if "info" in data and data["info"]:
        result["Info"] = {}
        for key, val in info_map.items():
            result["Info"][val] = data["info"].get(key, "UNSPECIFIED")

    if "error" in data and data["error"]:
        result["Runtime Errors"] = data["error"]

    if "fail" in data and data["fail"]:
        fail_section = {}
        for w, name in weight_map.items():
            fail_section[name] = data["fail"].get(str(w), data["fail"].get(w, {}))
        result["Failed Checks"] = fail_section

    if "pass" in data and data["pass"]:
        pass_section = {}
        for w, name in weight_map.items():
            pass_section[name] = data["pass"].get(str(w), data["pass"].get(w, {}))
        result["Passed Checks"] = pass_section

    return result