core ¶

def build_dependency_graph(
    source_path: Path,
    verbose: bool = False,
    log_level: str = "WARNING",
    log_file: Path | None = None,
    rich_handler: bool = False,
) -> nx.DiGraph:
    """
    Build a recursive dependency graph from YAML files.
    """
    # # Only set up logging if not already configured
    # if not any(
    #     handler.levelno <= getattr(logger, log_level.upper())
    #     for handler in logger._core.handlers.values()
    # ):
    #  Initialize logging
    # setup_factory_logger(
    #     verbose=verbose,
    #     level=log_level,
    #     file=log_file,
    #     rich_handler=rich_handler,
    # )

    base_dir = Path(source_path.parts[0])
    logger.debug(f"Base directory : {base_dir=}")
    # base_dir = Path(source_path.parts[0]) if source_path.is_dir() else source_path.parents[1]

    graph = nx.DiGraph()
    visited = {}  # Maps require path -> model name
    queue = deque()

    if source_path.is_file() and source_path.suffix == '.yaml':
        queue.append((source_path.name, source_path))
        logger.debug(f"Data appended to queue which now is {queue=}")

    elif source_path.is_dir():
        yaml_files = source_path.rglob("*.yaml")
        for yaml_file in track(yaml_files, description="Queueing data models for processing\n" ):
            queue.append((yaml_file.name, yaml_file))

    while queue:
        # logger.debug(f"[underline]The queue is now[/underline]\n\n   {queue=}\n")
        req_name, yaml_path = queue.popleft()
        # logger.debug(f"[dim]After poping\n\n  {queue=}[/dim]\n")
        logger.debug(f"In the queue : {req_name=} {yaml_path=}")
        if req_name in visited:
            logger.debug(f"[red dim]{req_name=} already processed ![/red dim]")
            continue  # Already processed

        process_model(
            graph=graph,
            base_dir=base_dir,
            require_path=req_name,
            yaml_path=yaml_path,
            queue=queue,
            visited=visited,
        )

    # logger.debug("Return dependency graph G\n\n{graph}", G=graph)
    # logger.debug(f"Return {graph.nodes()=}")
    return graph

def process_model(
    graph: nx.DiGraph,
    base_dir: Path,
    require_path: str,
    yaml_path: Path,
    queue: deque,
    visited: dict,
):
    """
    Process a YAML file and its dependencies
    """
    logger.debug(f"Input graph\n   {graph.nodes=}\n   {graph.edges=}\n")
    # logger.debug(
    #     f"Processing\n\n   {graph.nodes()=}\n\n   YAML file {yaml_path=}",
    # )
    logger.debug(
        "Processing YAML file {yaml_path}",
        yaml_path=yaml_path,
        alt=f"Processing YAML file {yaml_path=}"
    )
    with open(yaml_path, 'r', encoding='utf-8') as f:
        data = yaml.safe_load(f)
        logger.debug(f"Loaded data\n   {data=}")

    model_name = data.get('name')  # required
    visited[require_path] = model_name

    # Merge parent attributes if inheriting
    if 'require' in data:
        for parent_require in track(data['require'], description="Resolving requirements"):
            logger.debug(f"Processing require directive {base_dir=} / {require_path=} = {parent_require=}")
            parent_path = resolve_require_path(base_dir=base_dir, require_path=parent_require)
            logger.debug(f"Path to parent node {parent_path=}")
            if parent_path.exists():
                logger.debug(f"Loading {parent_path=}")
                with open(parent_path, 'r') as pf:
                    parent_data = yaml.safe_load(pf)
                    # Recursive attribute merging
                logger.debug(f"Merging\n\n   {parent_data}\n\nand\n\n   {data=}\n")
                data = deep_merge(
                    base=parent_data,
                    override=data,
                )
                logger.debug(f"Merged\n\n   {data=}\n")

    # Get meaninfgul attributes
    model_symbol = data.get('symbol', '')
    model_label = data.get('label', '')
    model_label += f" {model_symbol}"
    model_description = data.get('description')  # required
    model_attributes = data.get('sections', '')
    model_color = data.get('color', 'white')

    # Add node with merged attributes
    logger.debug(
            f"Adding {model_name=} to graph"
            )

    # ---------------------------------------------------------------------
    # graph.add_node(
    #     node_for_adding=model_name,
    #     **{key: value for key, value in data.items() if key != "require"},
    #     _source_path=str(yaml_path),
    # )
    # ---------------------------------------------------------------------
    graph.add_node(
        node_for_adding=model_name,
        label=model_label,
        description=model_description,
        symbol=model_symbol,
        attributes=model_attributes,
        color=model_color,
        border_color='lightgray',
        border_size=1,
        hover=model_description,
        click=yaml.dump(data=model_attributes, allow_unicode=True),#, encoding='utf-8'),
    )

    logger.debug(
        f"   [bold dim]Updated graph nodes\n      {graph.nodes[model_name]=}\n"
    )

    # Process dependencies
    requires = data.get('require', [])
    logger.debug(f"[bold blue]Require directives to process[/bold blue]\n\n   {requires=}\n")

    for parent_node in requires:

        logger.debug(f"[bold blue]Processing parent node[/bold blue] {parent_node=}")
        parent_node_path = resolve_require_path(base_dir, parent_node)
        parent_node_label = parent_node_path.parts[-2]

        # logger.debug(f"{parent_node_path.exists()=}")
        if not parent_node_path.exists():
            logger.debug(f"Continue ?")
            continue

        if parent_node not in visited:
            logger.debug(f"[blue dim]Appending to processing queue[/blue dim] {parent_node=}")
            queue.append((parent_node, parent_node_path))
            with open(parent_node_path, 'r', encoding='utf-8') as parent_node_file:
                parent_node = yaml.safe_load(parent_node_file)
            parent_node_name = parent_node.get('name', '<Unnamed node>')

        else:
            logger.debug(f"[blue]Already visited[/blue] {parent_node=}, see\n\n   {visited=}")
            parent_node_name = visited[parent_node]


        # Add edge even if parent not processed yet
        graph.add_edge(
            u_of_edge=model_name,
            v_of_edge=parent_node_name or parent_node,
            label=parent_node_label,
            color='lightgray',
        )
        # logger.debug(f"Updated edges\n      {graph.edges=}\n")
        # logger.debug(f"Nodes after new edge addition\n      {graph.nodes=}\n")

    logger.debug(f"Updated graph\n   {graph.nodes=}\n   {graph.edges=}\n")

def resolve_require_path(base_dir: Path, require_path: str) -> Path:
    """
    Resolve require path (e.g., 'sun/position') to a YAML file (e.g., 'sun/position.yaml').
    """
    path = (base_dir / require_path).with_suffix('.yaml')
    # logger.debug(f"[blue]Resolved path to parent node[/blue] {path=}")
    return path

def visualise_circular_tree(
    graph: nx.DiGraph,
    output_file: str = "data_model_dependency_graph",
    node_size: int = 20,
) -> None:
    """ """
    pos = nx.nx_agraph.graphviz_layout(
        G=graph,
        prog="twopi",
        args="",
    )

    # plt.figure(figsize=(11.69, 11.69))
    plt.figure(figsize=(10, 10))
    nx.draw(
        G=graph,
        pos=pos,
        node_size=node_size,
        alpha=0.5,
        node_color="blue",
        with_labels=False,
    )

    leaf_nodes = [n for n in graph.nodes if graph.out_degree(n) == 0]

    # leaf_nodes = set()
    # parent_nodes = set()
    # parent_paths = {}

    # for u, v, data in graph.edges(data=True):
    #     leaf_nodes.add(u)
    #     parent_nodes.add(v)
    #     parent_paths[v] = data.get('label', 'unknown')

    logger.info(
        "Leaf nodes\n\n{leaf_nodes}",
        # "Leaf nodes\n\n{leaf_nodes}\n\nParent nodes\n\n{parent_nodes}",
        leaf_nodes=leaf_nodes,
        # parent_nodes=parent_nodes,
    )

    # Create labels only for leaf nodes
    # labels = {n: str(n) for n in leaf_nodes}
    labels = {
        # n: n + f"\n{data.get('symbol', '')}"
        node: f"{data.get('symbol', 'Symbol')}"
        for node, data in graph.nodes(data=True)
        if node in leaf_nodes
    }

    # Draw labels for leaf nodes
    nx.draw_networkx_labels(
        G=graph,
        pos=pos,
        labels=labels,
        font_size=7,
        font_color="red",
        # font_weight="bold",
        horizontalalignment='center'
    )

    # Titles
    plt.suptitle(
        t="Data Model Dependency Graph ⎄",
        fontsize=12,
        fontweight='bold',
        color='#2F3131',  #darkgray',
    )

    # plt.title(
    #     label="Lighter Colors = Path Hierarchy | Salmon = Leaf Node",
    #     fontsize=9,
    #     color='#2F4F4F',  #'darkgray',
    #     # pad=10
    # )

    # plt.axis("off")
    plt.axis("equal")
    # plt.tight_layout(rect=[0, 0, 1, 0.98])

    # Save
    output_file += ".png"
    plt.savefig(output_file, dpi=300, bbox_inches='tight')
    plt.close()
    logger.info(f"Graph saved to {output_file}")

def generate_color_from_path(path: str, max_levels: int = 10):
    """
    Generate a visually distinct color for a given path.
    - Root component determines the base hue.
    - Each subpath level modifies saturation and lightness.
    """
    components = path.split("/")
    root = components[0]

    # Generate a consistent base hue from root
    hash_obj = hashlib.md5(root.encode())
    hue = int(hash_obj.hexdigest(), 16) / (16**32)  # Normalize to [0,1]

    # Lighter base lightness and subtle contrast with depth
    base_lightness = 0.85
    lightness = base_lightness - 0.02 * min(len(components), max_levels)

    # Slightly increase saturation with depth for contrast
    saturation = 0.4 + 0.03 * min(len(components), max_levels)

    # Convert to RGB
    r, g, b = colorsys.hls_to_rgb(hue, lightness, saturation)
    return (r, g, b)

def generate_circular_tree(
    source_path: Path,
    # yaml_file: Path,
    node_size: int = 20,
) -> None:
    """
    """
    # Build graph
    graph = build_dependency_graph(
        source_path=source_path,
    )

    # # Topological sort
    # import networkx as nx
    # def topological_sort(G: nx.DiGraph) -> list:
    #     try:
    #         return list(nx.topological_sort(G))
    #     except nx.NetworkXUnfeasible:
    #         raise ValueError("Graph contains a cycle")

    # order = topological_sort(graph)
    # logger.debug("Topological Order:", order)

    # Visualize
    visualise_circular_tree(
        graph=graph,
        node_size=node_size,
    )

def generate_graph(
    source_path: Path,
    # yaml_file: Path,
    node_size: int = 2400,
    parent_node_size: int = 1200,
) -> None:
    """
    """
    # Build graph
    graph = build_dependency_graph(
        source_path=source_path,
    )

    # # Topological sort
    # import networkx as nx
    # def topological_sort(G: nx.DiGraph) -> list:
    #     try:
    #         return list(nx.topological_sort(G))
    #     except nx.NetworkXUnfeasible:
    #         raise ValueError("Graph contains a cycle")

    # order = topological_sort(graph)
    # logger.debug("Topological Order:", order)

    # Visualize
    visualise_graph(
        graph=graph,
        node_size=node_size,
        parent_node_size=parent_node_size,
    )