13317087927

Committed 13 Feb 2025 09:07PM UTC coverage: 85.878% (-0.05%) from 85.929%

Build # 13317087927

Build Type

Pull #882

github

Committed by

web-flow

Commit Message

Merge 79a5d704c into 2262c526d

Pull Request Pull Request #882: fix `zntrack list` for nested groups

Run Details

226 of 299 branches covered (75.59%)

Branch coverage included in aggregate %.

10 of 12 new or added lines in 1 file covered. (83.33%)

1495 of 1705 relevant lines covered (87.68%)

3.5 hits per line

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

95.45

/zntrack/utils/cli.py

import json
import pathlib
from typing import Tuple

from dvc.api import DVCFileSystem

from zntrack.config import ZNTRACK_FILE_PATH


def get_groups(remote, rev) -> Tuple[dict, list]:
    """Get the group names and the nodes in each group from the remote.

    Arguments:
    ---------
    remote : str
        The remote to get the group names from.
    rev : str
        The revision to use.

    Returns:
    -------
    groups : dict
        A nested dictionary with the group names as keys and the nodes in each group as
        values. Contains "short-name -> long-name" if inside a group.
    node_names: list
        A list of all node names in the project.

    """
    fs = DVCFileSystem(url=remote, rev=rev)
    with fs.open(ZNTRACK_FILE_PATH) as f:
        config = json.load(f)

    true_groups = {}
    node_names = []

    def add_to_group(groups, grp_names, node_name):
        """Recursively add node_name into the correct nested group structure."""
        if not grp_names:
            return
        
        current_group = grp_names[0]

        # If this is the last level, add the node directly
        if len(grp_names) == 1:
            if current_group not in groups:
                groups[current_group] = []
            groups[current_group].append(node_name)
        else:
            # Ensure the current group contains a dictionary inside a list
            if current_group not in groups:
                groups[current_group] = [{}]
            elif not isinstance(groups[current_group][0], dict):
                groups[current_group].insert(0, {})

            add_to_group(groups[current_group][0], grp_names[1:], node_name)

    for node_name, node_config in config.items():
        nwd = pathlib.Path(node_config["nwd"]["value"])
        grp_names = nwd.parent.as_posix().split("/")[1:]

        if not grp_names:
            node_names.append(node_name)
            grp_names = ["nodes"]
        else:
            for grp_name in grp_names:
                node_name = node_name.replace(f"{grp_name}_", "")

            node_names.append(f"{'_'.join(grp_names)}_{node_name}")
            node_name = f"{node_name} -> {node_names[-1]}"

        add_to_group(true_groups, grp_names, node_name)

    return true_groups, node_names

1	import json	4✔
2	import pathlib	4✔
3	from typing import Tuple	4✔
4
5	from dvc.api import DVCFileSystem	4✔
6
7	from zntrack.config import ZNTRACK_FILE_PATH	4✔
8
9
10	def get_groups(remote, rev) -> Tuple[dict, list]:	4✔
11	"""Get the group names and the nodes in each group from the remote.
12
13	Arguments:
14	---------
15	remote : str
16	The remote to get the group names from.
17	rev : str
18	The revision to use.
19
20	Returns:
21	-------
22	groups : dict
23	A nested dictionary with the group names as keys and the nodes in each group as
24	values. Contains "short-name -> long-name" if inside a group.
25	node_names: list
26	A list of all node names in the project.
27
28	"""
29	fs = DVCFileSystem(url=remote, rev=rev)	4✔
30	with fs.open(ZNTRACK_FILE_PATH) as f:	4✔
31	config = json.load(f)	4✔
32
33	true_groups = {}	4✔
34	node_names = []	4✔
35
36	def add_to_group(groups, grp_names, node_name):	4✔
37	"""Recursively add node_name into the correct nested group structure."""
38	if not grp_names:	4✔
NEW 39	return	×
40
41	current_group = grp_names[0]	4✔
42
43	# If this is the last level, add the node directly
44	if len(grp_names) == 1:	4✔
45	if current_group not in groups:	4✔
46	groups[current_group] = []	4✔
47	groups[current_group].append(node_name)	4✔
48	else:
49	# Ensure the current group contains a dictionary inside a list
50	if current_group not in groups:	4✔
NEW 51	groups[current_group] = [{}]	×
52	elif not isinstance(groups[current_group][0], dict):	4✔
53	groups[current_group].insert(0, {})	4✔
54
55	add_to_group(groups[current_group][0], grp_names[1:], node_name)	4✔
56
57	for node_name, node_config in config.items():	4✔
58	nwd = pathlib.Path(node_config["nwd"]["value"])	4✔
59	grp_names = nwd.parent.as_posix().split("/")[1:]	4✔
60
61	if not grp_names:	4✔
62	node_names.append(node_name)	4✔
63	grp_names = ["nodes"]	4✔
64	else:
65	for grp_name in grp_names:	4✔
66	node_name = node_name.replace(f"{grp_name}_", "")	4✔
67
68	node_names.append(f"{'_'.join(grp_names)}_{node_name}")	4✔
69	node_name = f"{node_name} -> {node_names[-1]}"	4✔
70
71	add_to_group(true_groups, grp_names, node_name)	4✔
72
73	return true_groups, node_names	4✔

zincware / ZnTrack / 13317087927

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