Processing

def dump(node: Any, names: dict[int, str] | None = None) -> str:
    """Write `node` in the rule files' style.

    Args:
        node: Plain data: dicts, lists, strings, numbers and `Comment`s.
        names: Anchor names by `id()` of the object to anchor. An object that appears more
            than once is anchored where it is first written and aliased everywhere else;
            one not named here gets PyYAML's `id001`.

    Returns:
        The YAML text.
    """
    anchors = names or {}

    class Dumper(yaml.SafeDumper):
        def __init__(self, *args: Any, **kwargs: Any) -> None:
            super().__init__(*args, **kwargs)
            self.anchor_names: dict[Any, str] = {}

        def increase_indent(self, flow: bool = False, indentless: bool = False) -> None:
            # a list sits two spaces under its key, not level with it
            return super().increase_indent(flow, False)

        def represent_data(self, data: Any) -> Any:
            node = super().represent_data(data)
            if id(data) in anchors:
                self.anchor_names[node] = anchors[id(data)]
            return node

        def generate_anchor(self, node: Any) -> Any:
            return self.anchor_names.get(node) or super().generate_anchor(node)

        def choose_scalar_style(self) -> Any:
            style = super().choose_scalar_style()
            return '"' if style == "'" else style

        def write_folded(self, text: str) -> None:
            width, self.best_width = self.best_width, COMMENT_WIDTH
            try:
                super().write_folded(text)
            finally:
                self.best_width = width

    # a note that fits on its key's line stays there; anything longer is folded under it
    Dumper.add_representer(
        Comment,
        lambda d, v: d.represent_scalar(
            "tag:yaml.org,2002:str",
            str(v),
            style=">" if "\n" in v or len(v) > COMMENT_WIDTH - 30 else None,
        ),
    )
    return yaml.dump(
        node, Dumper=Dumper, sort_keys=False, allow_unicode=True, width=4096
    )

def fold_comments(node: Any) -> None:
    """Mark every `comment` value below `node` as a `Comment`, in place.

    In place, so an object shared by several places stays one object and keeps its anchor.
    """
    seen: set[int] = set()
    stack = [node]
    while stack:
        current = stack.pop()
        if id(current) in seen:
            continue
        seen.add(id(current))
        if isinstance(current, dict):
            if isinstance(current.get("comment"), str):
                current["comment"] = Comment(current["comment"])
            stack.extend(current.values())
        elif isinstance(current, list):
            stack.extend(current)

def load_yaml(text: str) -> Any:
    """Read a rule file's YAML.

    Args:
        text: The file's contents.

    Raises:
        RuleFileError: A key is written twice, is a merge key or reads as a boolean.
        yaml.YAMLError: `text` is not YAML.

    Returns:
        The document, with every alias resolved to the object its anchor names.
    """
    # B506 matches the name `Loader=` is given, not what it derives from: UniqueKeyLoader
    # is a SafeLoader, and yaml.safe_load is exactly this call with the plain one
    return yaml.load(text, Loader=UniqueKeyLoader)  # nosec B506

def normalize_document[D](
    text: str,
    *,
    validate: Callable[[Any], D],
    write: Callable[[Any, D], str],
    same: Callable[[D, D], bool] = operator.eq,
) -> str:
    """Return `text` in canonical form, having checked it still says the same thing.

    Args:
        text: The rule file's contents.
        validate: Builds the document's model from the loaded YAML.
        write: Writes the file below its header, given the loaded YAML and the model.
        same: Whether two models say the same thing. Equality unless the model holds
            something that normalizing is allowed to change.

    Raises:
        NormalizeError: The file is not valid, holds a YAML comment below its header, or
            would no longer say the same thing once written.

    Returns:
        The canonical text, the header comment block included.
    """
    lines = text.splitlines()
    header = _header(lines)
    _refuse_yaml_comments(lines, len(header))
    raw = _load(text)
    document = _validate(validate, raw)
    head = "\n".join(line for line in header if line.strip())
    result = (head + "\n\n" if head else "") + write(raw, document)
    if not same(_validate(validate, _load(result)), document):
        raise NormalizeError(
            "normalizing would change what the rules say; nothing was written"
        )
    return result

def run_normalizer(
    argv: Sequence[str] | None,
    *,
    normalize: Callable[[str], str],
    default: Path,
    description: str,
) -> int:
    """Normalize the given files in place, or report with `--check`.

    Args:
        argv: The command line, without the program name; None reads `sys.argv`.
        normalize: Turns a file's text into its canonical form.
        default: The file to normalize when none is given.
        description: The first line of the script's help.

    Returns:
        0 when nothing changed, 1 when a file was (or with `--check`, would be) rewritten,
        2 on an error.
    """
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument("--check", action="store_true", help="report, do not write")
    parser.add_argument("files", nargs="*", type=Path, default=[default])
    args = parser.parse_args(argv)
    status = 0
    for path in args.files:
        text = path.read_text("utf-8")
        try:
            result = normalize(text)
        except NormalizeError as e:
            print(f"{path}: {e}", file=sys.stderr)
            return 2
        if result == text:
            continue
        status = 1
        if args.check:
            sys.stdout.writelines(
                difflib.unified_diff(
                    text.splitlines(keepends=True),
                    result.splitlines(keepends=True),
                    str(path),
                    str(path),
                )
            )
        else:
            path.write_text(result, "utf-8")
            print(f"normalized {path}")
    return status

def migrate(
    collection: FeatureCollection,
    to_version: str,
    *,
    report: MigrationReport | None = None,
) -> MigrationResult:
    """Migrate a collection to `to_version`, however many hops that takes.

    One report covers the whole run: every issue records the hop that raised it, so a
    4.1 -> 6.1 migration still says which step dropped an attribute.

    Args:
        collection: The collection to migrate.
        to_version: The wanted XPlanung version.
        report: An existing report to collect into, if the caller is already holding one.

    Raises:
        ValueError: `to_version` is not reachable from the collection's version.
        MigrationError: A hop failed; the report says which features and why.
        UnsupportedFeatureError: The collection holds a feature with no mapping at all.

    Returns:
        The migrated collection and the report.
    """
    from xmas_core.processing.transform.engine import run_hop
    from xmas_core.processing.transform.rules import load

    report = report if report is not None else MigrationReport()
    version = collection.appschema.version
    for step in migration_path(version, to_version):
        logger.info(f"Migrating from version {version} to {step}")
        collection = run_hop(collection, load(version, step), report)
        version = step
    return MigrationResult(collection=collection, report=report)

def migration_path(from_version: str, to_version: str) -> list[str]:
    """The versions to migrate through to get from `from_version` to `to_version`.

    Args:
        from_version: The collection's current version.
        to_version: The wanted version.

    Raises:
        ValueError: `to_version` is not reachable from `from_version`.

    Returns:
        The intermediate targets, in order, excluding `from_version`. Empty if the
        collection is already at `to_version`.
    """
    path: list[str] = []
    current = from_version
    while current != to_version:
        if not (nxt := MIGRATIONS.get(current)):
            raise ValueError(
                f"Migration from version {from_version} to {to_version} not yet implemented"
            )
        path.append(nxt)
        current = nxt
    return path

def apply_ops(ops: list[Op], obj: dict[str, Any], ctx: RuleContext) -> None:
    """Run `ops` against `obj` in order, stopping if one removes the feature."""
    for op in ops:
        op.apply(obj, ctx)
        if not obj:
            return

def run_collection_ops(
    ops: list[CollectionOp], staged: dict[UUID, Staged]
) -> dict[UUID, Staged]:
    """Run every operation in order over the staged features, which are mutated in place.

    Nothing is validated here, in either slot. A `pre` operation writes into source-shaped
    data the rules are about to rewrite - one of them moves a Rasterdarstellung onto the
    plan's `texte`, where it becomes a TextAbschnitt in the same hop, which the source schema
    does not declare - and a `post` operation writes into target-shaped data the target
    models are about to be built from. Either way the check that matters is the one at the
    end of the hop, against the version the document is becoming.
    """
    for op in ops:
        op.run(staged)
    return staged

def apply_ops(ops: list[Op], obj: dict[str, Any], ctx: RuleContext) -> None:
    """Run `ops` against `obj` in order, stopping if one removes the feature."""
    for op in ops:
        op.apply(obj, ctx)
        if not obj:
            return

def guards_last(data: dict[str, Any]) -> dict[str, Any]:
    """`data` with `when` and `then` moved to the end, in that order."""
    return {k: v for k, v in data.items() if k not in ("when", "then")} | {
        k: data[k] for k in ("when", "then") if k in data
    }

def one_or_many_schema(schema: dict[str, Any]) -> None:
    """Publish a `when` list as one clause or a list of them, as `_one_or_many` reads it."""
    items = schema.pop("items")
    del schema["type"]
    schema["anyOf"] = [items, {"type": "array", "items": items}]

@functools.cache
def value_types(names: tuple[str, ...]) -> tuple[type, ...]:
    """The value classes a `geom:` predicate was resolved to, looked up once.

    Imported here rather than at module scope so the vocabulary stays independent of the
    model layer: every other part of it works on the dict alone.
    """
    from xmas_core.model.appschema import definitions

    return tuple(getattr(definitions, name) for name in names)

def run_collection_ops(
    ops: list[CollectionOp], staged: dict[UUID, Staged]
) -> dict[UUID, Staged]:
    """Run every operation in order over the staged features, which are mutated in place.

    Nothing is validated here, in either slot. A `pre` operation writes into source-shaped
    data the rules are about to rewrite - one of them moves a Rasterdarstellung onto the
    plan's `texte`, where it becomes a TextAbschnitt in the same hop, which the source schema
    does not declare - and a `post` operation writes into target-shaped data the target
    models are about to be built from. Either way the check that matters is the one at the
    end of the hop, against the version the document is becoming.
    """
    for op in ops:
        op.run(staged)
    return staged