Skip to content

muxtools.utils.binaries.runner

__all__ module-attribute 

__all__ = ['get_managed_executable']

TIER_FEATURES module-attribute 

TIER_FEATURES = {'baseline': {'cx16', 'lahf_lm', 'popcnt', 'sse3', 'ssse3', 'sse4_1', 'sse4_2'}, 'avx2': {'avx', 'avx2', 'bmi1', 'bmi2', 'f16c', 'fma', 'lzcnt', 'movbe', 'xsave'}, 'avx512': {'avx512f', 'avx512bw', 'avx512cd', 'avx512dq', 'avx512vl'}, 'zn4': {'aes', 'pclmul', 'rdrand', 'rdseed', 'adx', 'sha', 'clflushopt', 'clwb', 'avx512ifma', 'avx512vbmi', 'avx512vbmi2', 'avx512vnni', 'avx512bitalg', 'avx512vpopcntdq', 'gfni', 'vaes', 'vpclmulqdq', 'rdpid', 'prfchw'}}

supports 

supports(tier: str, features: set[str]) -> bool
Source code in muxtools/utils/binaries/runner.py 
48
49
50
51
52
53
54
def supports(tier: str, features: set[str]) -> bool:
    required: set[str] = set()
    for level, additions in TIER_FEATURES.items():
        required.update(additions)
        if level == tier:
            return required <= features
    raise KeyError(tier)

cpu_state 

cpu_state() -> set[str]
Source code in muxtools/utils/binaries/runner.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def cpu_state() -> set[str]:
    aliases = {
        "pni": "sse3",
        "abm": "lzcnt",
        "pclmulqdq": "pclmul",
        "rdrnd": "rdrand",
        "sha_ni": "sha",
        "3dnowprefetch": "prfchw",
    }
    try:
        cpuinfo = get_cpu_info()
        features = {aliases.get(flag, flag).replace("avx512_", "avx512") for flag in cpuinfo.get("flags", [])}  # type: ignore
        # cpuinfo merges CPUID and OS flags, so remove vector features the OS cannot use.
        if sys.platform == "win32":
            import ctypes

            enabled = ctypes.WinDLL("kernel32").GetEnabledXStateFeatures
            enabled.restype = ctypes.c_uint64
            enabled.argtypes = []
            xstate = enabled()
            avx_enabled = xstate & 6 == 6
            avx512_enabled = xstate & 0xE6 == 0xE6
        elif platform.system() == "Linux":
            flags = re.findall(r"^flags\s*:\s*(.*)$", Path("/proc/cpuinfo").read_text(), re.MULTILINE)
            enabled = set.intersection(*(set(line.split()) for line in flags)) if flags else set()
            avx_enabled = "avx" in enabled
            avx512_enabled = "avx512f" in enabled
        else:
            return set()
        # Each higher tier includes these required features, so clearing them gates that tier and above.
        if not avx_enabled:
            features.discard("avx")
        if not avx512_enabled:
            features.discard("avx512f")
        return features
    except Exception as error:
        warn(f"CPU detection unavailable; skipping optimized variants: {error}")
    return set()

get_managed_executable 

get_managed_executable(name: str, config: ProjectConfig | None = None) -> str | None
Source code in muxtools/utils/binaries/runner.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def get_managed_executable(name: str, config: ProjectConfig | None = None) -> str | None:
    effective_config = resolve_binary_config(config if config is not None else discover_config())
    if effective_config is None or effective_config.mode == "system":
        return shutil.which(name)
    root = scope_path(effective_config)
    catalog = None
    try:
        catalog = load_catalog(offline=True)
    except ValueError:
        pass
    installed_items = _installed_metadata(root)
    for raw in effective_config.packages:
        spec = parse_spec(raw)
        package = resolve_name(spec.name, catalog) if catalog else resolve_installed_name(spec.name, installed_items)
        items = [item for item in installed_items if item["name"] == package]
        if catalog:
            provided = catalog["packages"][package].get("provides", [])
        else:
            provided = [key for item in items for key in item.get("binaries", {})]
        if name not in provided:
            continue
        required = version_code(Spec(package, spec.operator, spec.version), catalog["packages"].get(package, {}) if catalog else {}, root)
        valid = [item for item in items if satisfies(item["version_code"], spec.operator, required)]
        for item in sorted(valid, key=lambda value: value["version_code"], reverse=True):
            try:
                return str(_get_fitting_variant(item["binaries"][name], Path(item["_path"])))
            except (KeyError, ValueError):
                continue
        raise FileNotFoundError(f"Configured binary {name!r} is missing; run 'muxtools binaries sync'")
    return shutil.which(name)