Adds dependency resolver for recursive project handling

Introduces a dependency resolver to clone and manage project dependencies recursively.
Updates CLI to include a new "dep" command for dependency resolution.
Enhances source resolver to handle dependencies and expand all sources.
Improves console utilities with live mode for real-time feedback.
Marks dependency names in project configuration as optional.

Enables streamlined multi-repository workflows and dependency management.
This commit is contained in:
2025-04-26 18:11:31 +00:00
parent 3cf3fc1437
commit 4f1f2e7d51
8 changed files with 239 additions and 30 deletions

View File

@@ -12,6 +12,7 @@ pyyaml = "^6.0.2"
doit = "^0.36.0" doit = "^0.36.0"
pydantic = "^2.11.3" pydantic = "^2.11.3"
rich = "^14.0.0" rich = "^14.0.0"
gitpython = "^3.1.44"
[build-system] [build-system]

View File

@@ -0,0 +1,99 @@
# src/hdlbuild/dependency/resolver.py
from typing import List, Set
from git import Repo
from models.config import DIRECTORIES, GIT
from models.project import ProjectConfig
from models.dependency import ResolvedDependency
import os
from utils.console_utils import ConsoleUtils
from utils.project_loader import load_project_config
class DependencyResolver:
def __init__(self, root_project: ProjectConfig, offline_mode: bool = False):
self.root_project = root_project
self.offline_mode = offline_mode
self.resolved: List[ResolvedDependency] = []
self.visited_urls: Set[str] = set()
self.console = ConsoleUtils(live=True)
self.console.start_live()
def resolve_all(self):
"""Startet das Auflösen aller Abhängigkeiten (rekursiv)."""
self._resolve_project(self.root_project)
self.console.stop_live("[bold green]Alle Abhängigkeiten aufgelöst.[/bold green]")
def _resolve_project(self, project: ProjectConfig):
"""Löst die Abhängigkeiten eines einzelnen Projekts auf."""
for dep in project.dependencies or []:
if dep.git in self.visited_urls:
continue
self.visited_urls.add(dep.git)
local_path = self._clone_or_use_existing(dep.git, dep.rev)
dep_project = self._load_project_config(os.path.join(local_path, "project.yml"))
# Speichern als ResolvedDependency
self.resolved.append(ResolvedDependency(project=dep_project, local_path=local_path))
self._resolve_project(dep_project)
def _clone_or_use_existing(self, git_url: str, rev: str) -> str:
folder_name = os.path.basename(git_url.rstrip("/")).replace(".git", "")
local_path = os.path.join(DIRECTORIES.dependency, folder_name)
if os.path.exists(local_path):
# Lokales Repo vorhanden
self.console.print(f"[bold green]Benutze vorhandenes Repository: {folder_name}[/bold green]")
repo = Repo(local_path)
if not self.offline_mode:
try:
self.console.print(f"[bold green]Aktualisiere {folder_name}...[/bold green]")
# Fetch Remote Updates
repo.remotes.origin.fetch()
# Prüfen, ob HEAD und origin/branch unterschiedlich sind
local_commit = repo.head.commit
remote_ref = repo.remotes.origin.refs[repo.active_branch.name]
remote_commit = remote_ref.commit
if local_commit.hexsha != remote_commit.hexsha:
self.console.print(f"[bold yellow]Änderungen erkannt! Force-Pull wird durchgeführt...[/bold yellow]")
repo.git.reset('--hard', remote_commit.hexsha)
else:
self.console.print(f"[bold green]Repository {folder_name} ist aktuell.[/bold green]")
except Exception as e:
self.console.print(f"[bold red]Warnung beim Aktualisieren: {e}[/bold red]")
else:
# Lokales Repo fehlt → nur dann klonen
if self.offline_mode:
raise FileNotFoundError(f"Repository {folder_name} existiert lokal nicht und offline_mode ist aktiv.")
else:
self.console.print(f"[bold green]Klone {git_url}...[/bold green]")
repo = Repo.clone_from(git_url, local_path)
# Immer: Auf den richtigen Commit/Branch wechseln
self.console.print(f"[bold green]Checkout auf[/bold green] [yellow]{rev}[/yellow] in {folder_name}")
repo.git.checkout(rev)
return local_path
def _load_project_config(self, path: str) -> ProjectConfig:
"""
Lädt eine project.yml aus einem lokalen Ordner.
Args:
path (str): Basisverzeichnis des geklonten Projekts.
Returns:
ProjectConfig: Das geladene Projekt.
"""
self.console.print(f"Lade project.yml aus {path}...")
return load_project_config(path)

View File

@@ -1,6 +1,7 @@
import argparse import argparse
import sys import sys
from dependencies.resolver import DependencyResolver
from tools.xilinx_ise.main import xilinx_ise_all, xilinx_ise_synth from tools.xilinx_ise.main import xilinx_ise_all, xilinx_ise_synth
from utils.console_utils import ConsoleUtils from utils.console_utils import ConsoleUtils
from utils.directory_manager import clear_build_directories, clear_directories, ensure_directories_exist from utils.directory_manager import clear_build_directories, clear_directories, ensure_directories_exist
@@ -32,6 +33,12 @@ def synth(args):
ensure_directories_exist() ensure_directories_exist()
xilinx_ise_synth(project) xilinx_ise_synth(project)
def dep(args):
"""Starts the dependencies process."""
console_utils.print("Starting dependencies process...")
DependencyResolver(project).resolve_all()
def main(): def main():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="hdlbuild - Build management tool for FPGA projects", description="hdlbuild - Build management tool for FPGA projects",
@@ -63,6 +70,10 @@ def main():
parser_build = subparsers.add_parser("synth", help="Start the synth process") parser_build = subparsers.add_parser("synth", help="Start the synth process")
parser_build.set_defaults(func=synth) parser_build.set_defaults(func=synth)
# Dependencies command
parser_build = subparsers.add_parser("dep", help="Start the dependencies process")
parser_build.set_defaults(func=dep)
args = parser.parse_args() args = parser.parse_args()
args.func(args) args.func(args)

8
src/models/dependency.py Normal file
View File

@@ -0,0 +1,8 @@
# models/dependency.py
from pydantic import BaseModel
from models.project import ProjectConfig
class ResolvedDependency(BaseModel):
project: ProjectConfig
local_path: str

View File

@@ -16,7 +16,7 @@ class ToolOptions(BaseModel):
fuse: List[str] = Field(default_factory=list) fuse: List[str] = Field(default_factory=list)
class Dependency(BaseModel): class Dependency(BaseModel):
name: str name: Optional[str] = None # Name ist jetzt optional
git: str git: str
rev: str rev: str
library: str = "work" # Default auf 'work' library: str = "work" # Default auf 'work'

View File

@@ -1,7 +1,8 @@
from typing import Optional from typing import Optional
from dependencies.resolver import DependencyResolver
from models.config import DIRECTORIES from models.config import DIRECTORIES
from tools.xilinx_ise.common import copy_file, run_tool from tools.xilinx_ise.common import copy_file, run_tool
from utils.source_resolver import expand_sources from utils.source_resolver import expand_all_sources
from models.project import ProjectConfig from models.project import ProjectConfig
import subprocess import subprocess
import os import os
@@ -12,12 +13,15 @@ def generate_xst_project_file(project: ProjectConfig, output_path: str):
Generiert die XST .prj-Datei mit allen Quellcodes. Generiert die XST .prj-Datei mit allen Quellcodes.
""" """
with open(output_path, "w") as f: with open(output_path, "w") as f:
# VHDL-Sources resolver = DependencyResolver(project, offline_mode=True)
for lib, src in expand_sources(project.sources.vhdl): resolver.resolve_all()
f.write(f"vhdl {lib} \"{DIRECTORIES.get_relative_prefix()}/{src}\"\n") vhdl_sources, verilog_sources = expand_all_sources(project, resolver.resolved)
# Verilog-Sources
for lib, src in expand_sources(project.sources.verilog): for lib, file in vhdl_sources:
f.write(f"verilog {lib} \"{DIRECTORIES.get_relative_prefix()}/{src}\"\n") f.write(f"vhdl {lib} \"{DIRECTORIES.get_relative_prefix()}{file}\"\n")
for lib, file in verilog_sources:
f.write(f"verilog {lib} \"{DIRECTORIES.get_relative_prefix()}{file}\"\n")
# Optionale Dependencies # Optionale Dependencies
if project.dependencies: if project.dependencies:

View File

@@ -1,3 +1,4 @@
import sys
import threading import threading
import time import time
import subprocess import subprocess
@@ -6,6 +7,7 @@ from typing import List, Optional
from rich.console import Console from rich.console import Console
from rich.live import Live from rich.live import Live
from rich.text import Text from rich.text import Text
from rich.markup import render
class ConsoleTask: class ConsoleTask:
def __init__(self, prefix:str, title: str, step_number: Optional[int] = None, total_steps: Optional[int] = None, max_log_lines: int = 10): def __init__(self, prefix:str, title: str, step_number: Optional[int] = None, total_steps: Optional[int] = None, max_log_lines: int = 10):
@@ -18,6 +20,7 @@ class ConsoleTask:
self.stop_event = threading.Event() self.stop_event = threading.Event()
self.spinner_thread: Optional[threading.Thread] = None self.spinner_thread: Optional[threading.Thread] = None
self.output_lines: List[str] = [] self.output_lines: List[str] = []
self.all_lines: List[str] = []
self._stdout_lock = threading.Lock() self._stdout_lock = threading.Lock()
self.console = Console() self.console = Console()
self.live: Optional[Live] = None self.live: Optional[Live] = None
@@ -58,6 +61,7 @@ class ConsoleTask:
def log(self, message: str): def log(self, message: str):
with self._stdout_lock: with self._stdout_lock:
self.all_lines.append(message)
self.output_lines.append(message) self.output_lines.append(message)
if len(self.output_lines) > self.max_log_lines: if len(self.output_lines) > self.max_log_lines:
self.output_lines = self.output_lines[-self.max_log_lines:] self.output_lines = self.output_lines[-self.max_log_lines:]
@@ -89,12 +93,7 @@ class ConsoleTask:
self.log(line.rstrip()) self.log(line.rstrip())
success = (process.returncode == 0) success = (process.returncode == 0)
if not success:
raise subprocess.CalledProcessError(process.returncode, cmd)
except subprocess.CalledProcessError:
success = False
raise
finally: finally:
self.stop_event.set() self.stop_event.set()
if self.spinner_thread: if self.spinner_thread:
@@ -102,11 +101,18 @@ class ConsoleTask:
duration = time.time() - start_time duration = time.time() - start_time
# Finalize output
with self._stdout_lock: with self._stdout_lock:
self._finalize_output(success, duration) self._finalize_output(success, duration)
return 0 if success else 1 if not success:
# Schöne Fehlerausgabe und kontrolliertes Beenden
self.console.print("\n[bold red]❌ Fehler beim Ausführen des Kommandos:[/bold red]")
for line in self.all_lines:
self.console.print(f"[red]{line}[/red]")
sys.exit(1) # ❗ Hier: hartes, aber sauberes Beenden des Programms
return 0
def _finalize_output(self, success: bool, duration: float): def _finalize_output(self, success: bool, duration: float):
if self.live: if self.live:
@@ -122,18 +128,49 @@ class ConsoleTask:
self.console.print(final_line) self.console.print(final_line)
class ConsoleUtils: class ConsoleUtils:
def __init__(self, def __init__(
prefix: str = "", self,
prefix: str = "hdlbuild",
step_number: Optional[int] = None, step_number: Optional[int] = None,
total_steps: Optional[int] = None total_steps: Optional[int] = None,
live: bool = False
): ):
self.prefix = prefix self.prefix = prefix
self.step_number = step_number self.step_number = step_number
self.total_steps = total_steps self.total_steps = total_steps
self.console = Console() self.console = Console()
self.live_mode = live
self.live: Optional[Live] = None
self.messages: List[str] = []
def start_live(self):
"""Startet den Live-Modus."""
if self.live_mode and self.live is None:
self.live = Live(console=self.console, refresh_per_second=10, transient=True)
self.live.start()
def print(self, message: str): def print(self, message: str):
prefix = f"[grey50]\[{self.prefix}][/grey50]" if self.prefix else "" prefix = f"[grey50]\[{self.prefix}][/grey50]" if self.prefix else ""
step_info = f"[bold blue]Step {self.step_number}/{self.total_steps}[/bold blue]" if self.step_number and self.total_steps else "" step_info = f"[bold blue]Step {self.step_number}/{self.total_steps}[/bold blue]" if self.step_number and self.total_steps else ""
message_text = f"{prefix} {step_info} {message}" full_message = f"{prefix} {step_info} {message}"
self.console.print(message_text)
if self.live_mode and self.live:
self.messages.append(full_message)
rendered_lines = [Text.from_markup(line) for line in self.messages]
combined = Text()
for line in rendered_lines:
combined.append(line)
combined.append("\n")
self.live.update(combined)
else:
self.console.print(full_message)
def stop_live(self, final_message: Optional[str] = None):
"""Beendet den Live-Modus, löscht alte Ausgaben und zeigt eine Abschlussnachricht."""
if self.live_mode and self.live:
self.live.stop()
self.live = None
self.messages.clear() # Alte Messages verwerfen
if final_message:
self.console.print(final_message)

View File

@@ -1,19 +1,68 @@
# src/hdlbuild/utils/source_resolver.py
import glob import glob
import os import os
from typing import List, Tuple from typing import List, Tuple
from models.project import SourceFile from models.project import SourceFile, ProjectConfig
from models.dependency import ResolvedDependency
def expand_sources(sources: List[SourceFile]) -> List[Tuple[str, str]]: def _expand_project_sources(project: ProjectConfig, project_root: str) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
""" """
Expandiert eine Liste von SourceFile-Objekten mit Wildcards in echte Pfade. Expandiert die Quellen eines einzelnen Projekts, getrennt nach VHDL und Verilog.
Args:
project (ProjectConfig): Das Projekt, dessen Quellen expandiert werden sollen.
project_root (str): Basisverzeichnis, von dem aus die Pfade aufgelöst werden.
Returns: Returns:
List of (library, filepath) Tuple: (List of (library, filepath) für VHDL, List of (library, filepath) für Verilog)
""" """
expanded = [] vhdl_expanded = []
for source in sources: verilog_expanded = []
matched_files = glob.glob(source.path, recursive=True)
# VHDL-Sources
for source in project.sources.vhdl:
full_pattern = os.path.join(project_root, source.path)
matched_files = glob.glob(full_pattern, recursive=True)
for file in matched_files: for file in matched_files:
normalized_path = os.path.normpath(file) normalized_path = os.path.normpath(file)
expanded.append((source.library, normalized_path)) vhdl_expanded.append((source.library, normalized_path))
return expanded
# Verilog-Sources
for source in project.sources.verilog:
full_pattern = os.path.join(project_root, source.path)
matched_files = glob.glob(full_pattern, recursive=True)
for file in matched_files:
normalized_path = os.path.normpath(file)
verilog_expanded.append((source.library, normalized_path))
return vhdl_expanded, verilog_expanded
def expand_all_sources(root_project: ProjectConfig, resolved_dependencies: List[ResolvedDependency]) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
"""
Expandiert alle Quellen aus dem Root-Projekt und allen Dependencies, getrennt nach VHDL und Verilog.
Args:
root_project (ProjectConfig): Das Hauptprojekt
resolved_dependencies (List[ResolvedDependency]): Alle rekursiv aufgelösten Dependencies
Returns:
Tuple:
- List of (library, filepath) für VHDL
- List of (library, filepath) für Verilog
"""
all_vhdl_sources = []
all_verilog_sources = []
# Root-Projekt expandieren
vhdl_sources, verilog_sources = _expand_project_sources(root_project, ".")
all_vhdl_sources.extend(vhdl_sources)
all_verilog_sources.extend(verilog_sources)
# Dependencies expandieren
for dep in resolved_dependencies:
vhdl_dep, verilog_dep = _expand_project_sources(dep.project, dep.local_path)
all_vhdl_sources.extend(vhdl_dep)
all_verilog_sources.extend(verilog_dep)
return all_vhdl_sources, all_verilog_sources