Skip to content

k4bench.provenance.sources

k4bench.provenance.sources

Where a run's components are defined: the stack package whose install holds each component's library, and that package's repository at the revision the stack built.

Links point at the commit the package's build recorded. A HEAD package's repository is the one the stack's metadata names. A package LCG builds from a release tarball has none recorded, so it is linked only where :data:RELEASE_REPOS names its repository, and otherwise named with its version alone.

PackageSource dataclass 

PackageSource(package: str, ref: str | None, repo: RepoRef | None, version: str | None = None)

One stack package: its name, the commit built and its repository.

search_url 

search_url(cls: str) -> str | None

Where class cls (namespaced or templated alike) is defined.

Source code in k4bench/provenance/sources.py 
51
52
53
54
55
def search_url(self, cls: str) -> str | None:
    """Where class *cls* (namespaced or templated alike) is defined."""
    if not self.repo or not self.ref:
        return None
    return self.repo.code_search_url(class_symbol(cls), self.ref)

class_symbol 

class_symbol(cls: str) -> str

The unqualified name of class cls, its template arguments dropped.

Source code in k4bench/provenance/sources.py 
58
59
60
def class_symbol(cls: str) -> str:
    """The unqualified name of class *cls*, its template arguments dropped."""
    return re.split(r"<", cls, maxsplit=1)[0].rsplit("::", 1)[-1].strip()

load_run_packages 

load_run_packages(run_dir: str | Path) -> dict[str, dict]

The packages run_dir's stack holds, with install paths where its manifests were saved; empty for a run without run_info.json.

Source code in k4bench/provenance/sources.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def load_run_packages(run_dir: str | Path) -> dict[str, dict]:
    """The packages *run_dir*'s stack holds, with install paths where its
    manifests were saved; empty for a run without ``run_info.json``."""
    run_dir = Path(run_dir)
    try:
        info = json.loads((run_dir / "run_info.json").read_text())
    except (OSError, ValueError):
        return {}
    manifests = []
    for name in STACK_MANIFESTS:
        try:
            manifests.append(gzip.decompress((run_dir / name).read_bytes()).decode())
        except (OSError, EOFError, UnicodeDecodeError, zlib.error):
            continue
    return stack_snapshot(info, {}, manifests).packages

package_of 

package_of(library: str, packages: dict[str, dict]) -> PackageSource | None

The package whose install holds library, by its longest install path.

The auditor records library resolved through symlinks, so it is matched against the install path the manifest names and against the directory that path resolved to when the run was captured. Paths are compared as recorded: LCG's rotating view names may point elsewhere now.

Source code in k4bench/provenance/sources.py 
 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
def package_of(library: str, packages: dict[str, dict]) -> PackageSource | None:
    """The package whose install holds *library*, by its longest install path.

    The auditor records *library* resolved through symlinks, so it is matched
    against the install path the manifest names and against the directory that
    path resolved to when the run was captured. Paths are compared as
    recorded: LCG's rotating view names may point elsewhere now.
    """
    if not library:
        return None
    match = max(
        (
            (len(prefix), name)
            for name, package in packages.items()
            for key in ("install_path", "install_realpath")
            if (prefix := package.get(key, "").rstrip("/") + "/") != "/"
            and library.startswith(prefix)
        ),
        default=None,
    )
    if match is None:
        return None
    name = match[1]
    package = packages[name]
    if package.get("version") == "HEAD":
        return PackageSource(name, package.get("commit"), parse_repo(package.get("repo_url")))
    return PackageSource(
        name, package.get("commit"), parse_repo(RELEASE_REPOS.get(name)), package.get("version"),
    )