Read an LCG Key4hep nightly's identity and git provenance off CVMFS.
LCG rotates weekday-named views, so their generated timestamp is the release
identity. Package revisions come from the manifest's install paths and the
.buildinfo_* file installed with each HEAD build. Metadata failures return
an empty result rather than failing the benchmark they describe.
RepoRef
dataclass
RepoRef(forge: str, host: str, slug: str)
A package's upstream repository, parsed far enough to build links.
stack_identity
stack_identity(stack_setup: str | Path) -> tuple[str, str]
Return (publication_date, platform) from an LCG view setup.
Source code in k4bench/provenance/stack.py
| def stack_identity(stack_setup: str | Path) -> tuple[str, str]:
"""Return ``(publication_date, platform)`` from an LCG view setup."""
path = Path(stack_setup).resolve()
try:
match = _GENERATED_RE.search(path.read_text())
if match:
return parsedate_to_datetime(match.group(1)).date().isoformat(), path.parent.name
except (OSError, TypeError, ValueError):
pass
return "", "unknown"
|
parse_repo
parse_repo(url: str | None) -> RepoRef | None
Parse a GitHub/GitLab repository URL used by attribution links.
Source code in k4bench/provenance/stack.py
| def parse_repo(url: str | None) -> RepoRef | None:
"""Parse a GitHub/GitLab repository URL used by attribution links."""
match = _URL_RE.match(url.strip()) if url else None
if not match:
return None
host, slug = match.group("host"), match.group("slug")
if host == "github.com":
if slug.count("/") != 1:
return None
forge = "github"
elif "gitlab" in host:
forge = "gitlab"
else:
return None
return RepoRef(forge, host, slug)
|
read_stack
read_stack(stack_setup: str | Path) -> tuple[Path | None, dict[str, dict]]
Return the LCG manifest and git-built HEAD packages for a view setup.
Source code in k4bench/provenance/stack.py
| def read_stack(stack_setup: str | Path) -> tuple[Path | None, dict[str, dict]]:
"""Return the LCG manifest and git-built HEAD packages for a view setup."""
setup = Path(stack_setup).resolve()
manifest = _manifest(setup)
if manifest is None:
return None, {}
try:
rows = manifest.read_text().splitlines()
except OSError as exc:
_log.warning("cannot read LCG manifest '%s' (%s)", manifest, exc)
return None, {}
packages: dict[str, dict] = {}
build_revisions: dict[str, str] = {}
for row in rows:
fields = [field.strip() for field in row.split(";")]
if len(fields) < 4 or fields[2] != "HEAD":
continue
name, install = fields[0], Path(fields[3])
try:
buildinfo = next(install.glob(".buildinfo_*.txt")).read_text()
except (OSError, StopIteration):
continue
config = _GITHASH_RE.search(buildinfo)
revision = _REVISION_RE.search(buildinfo)
if revision:
build_revisions[name] = config.group(1) if config else ""
packages[name] = {
"commit": revision.group(1),
"version": "HEAD",
"repo_url": None,
}
# Incremental builds can use different toolchains within one view. Resolve
# each package against its own build, fetching each distinct revision once.
repos_by_revision = {
revision: _repository_urls(setup.parents[2].name, revision)
for revision in dict.fromkeys(build_revisions.values()) if revision
}
for name, package in packages.items():
package["repo_url"] = repos_by_revision.get(build_revisions[name], {}).get(name.lower())
return manifest, packages
|