generated from coulomb/repo-seed
tegwick
270065ff58
Some checks failed
ci / validate-registry (push) Has been cancelled
Add hub sync and report cohorts CLI commands with pytest coverage, document sibling index publish contract and hub hardening path, align INTENT layout, raise external evidence on three registry entries, and close gap priorities 19-23 (priority 18 deferred on sibling index blocks).
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from reuse_surface.registry import level_at_least, load_index, parse_vector
|
|
|
|
|
|
def _availability_at_most(current: str, maximum: str) -> bool:
|
|
from reuse_surface.registry import LEVEL_ORDERS
|
|
|
|
order = LEVEL_ORDERS["availability"]
|
|
return order.index(current) <= order.index(maximum)
|
|
|
|
|
|
def cohort_filters_from_args(args: Any) -> dict[str, str | None]:
|
|
filters: dict[str, str | None] = {
|
|
"discovery_min": getattr(args, "discovery_min", None),
|
|
"availability_min": getattr(args, "availability_min", None),
|
|
"availability_max": getattr(args, "availability_max", None),
|
|
"domain": getattr(args, "domain", None),
|
|
}
|
|
if getattr(args, "planning_min", None):
|
|
filters["discovery_min"] = args.planning_min
|
|
filters["availability_max"] = filters["availability_max"] or "A1"
|
|
if getattr(args, "implementation_min", None):
|
|
filters["availability_min"] = args.implementation_min
|
|
return filters
|
|
|
|
|
|
def select_cohort(
|
|
filters: dict[str, str | None],
|
|
index: dict[str, Any] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
data = index or load_index()
|
|
matches: list[dict[str, Any]] = []
|
|
for item in data.get("capabilities", []):
|
|
vector = parse_vector(item["vector"])
|
|
if filters.get("discovery_min") and not level_at_least(
|
|
"discovery", vector["discovery"], filters["discovery_min"]
|
|
):
|
|
continue
|
|
if filters.get("availability_min") and not level_at_least(
|
|
"availability", vector["availability"], filters["availability_min"]
|
|
):
|
|
continue
|
|
if filters.get("availability_max") and not _availability_at_most(
|
|
vector["availability"], filters["availability_max"]
|
|
):
|
|
continue
|
|
if filters.get("domain") and item.get("domain") != filters["domain"]:
|
|
continue
|
|
matches.append(item)
|
|
return matches
|
|
|
|
|
|
def format_cohort_markdown(
|
|
matches: list[dict[str, Any]],
|
|
filters: dict[str, str | None],
|
|
) -> str:
|
|
lines = ["# Capability cohort report", ""]
|
|
active = {key: value for key, value in filters.items() if value}
|
|
if active:
|
|
lines.append("Filters:")
|
|
for key, value in sorted(active.items()):
|
|
lines.append(f"- `{key}`: `{value}`")
|
|
lines.append("")
|
|
if not matches:
|
|
lines.append("_No capabilities matched._")
|
|
return "\n".join(lines) + "\n"
|
|
lines.append("| ID | Vector | Consumption modes |")
|
|
lines.append("|---|---|---|")
|
|
for item in matches:
|
|
modes = ", ".join(item.get("consumption_modes", []))
|
|
lines.append(f"| `{item['id']}` | {item['vector']} | {modes} |")
|
|
lines.append("")
|
|
lines.append(f"**{len(matches)}** capabilit{'y' if len(matches) == 1 else 'ies'}.")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def format_cohort_json(matches: list[dict[str, Any]], filters: dict[str, str | None]) -> str:
|
|
payload = {
|
|
"count": len(matches),
|
|
"filters": {key: value for key, value in filters.items() if value},
|
|
"capabilities": matches,
|
|
}
|
|
return json.dumps(payload, indent=2, sort_keys=True) |