#!/usr/bin/env python3 """Generate callable, failure-only Rust shims from the authoritative catalog.""" from __future__ import annotations import argparse import csv import json import re import subprocess from collections import Counter from pathlib import Path import generate_rust_mapping as mapping ROOT = Path(__file__).resolve().parents[1] CATALOG = ROOT / "api/public-api.json" LEDGER = ROOT / "api/RUST-MAPPING.tsv" COVERAGE = ROOT / "api/SHIM-COVERAGE.md" TARGETS = { "LibreMetaverse.Types": ROOT / "crates/libremetaverse-types/src/generated.rs", "LibreMetaverse.StructuredData": ROOT / "crates/libremetaverse-structured-data/src/generated.rs", } UNDERLYING = { "System.SByte": "i8", "System.Byte": "u8", "System.Int16": "i16", "System.UInt16": "u16", "System.Int32": "i32", "System.UInt32": "u32", "System.Int64": "i64", "System.UInt64": "u64", } PRIVATE_LAYOUTS = { "T:LibreMetaverse.CacheDictionary`2": [("entries", "std::collections::HashMap")], "T:LibreMetaverse.DoubleDictionary`3": [ ("by_first", "std::collections::HashMap"), ("by_second", "std::collections::HashMap"), ], "T:LibreMetaverse.EmptyRemovalStrategy`1": [("keys", "Vec")], "T:LibreMetaverse.ExpiringCache`2": [("entries", "std::collections::HashMap")], "T:LibreMetaverse.LruRemovalStrategy`1": [("keys", "Vec")], "T:LibreMetaverse.MruRemovalStrategy`1": [("keys", "Vec")], "T:LibreMetaverse.MultiValueDictionary`2": [("entries", "std::collections::HashMap>")], "T:LibreMetaverse.TokenBucket": [("content", "i32"), ("max_burst", "i32"), ("drip_rate", "i32")], "T:LibreMetaverse.UUID": [("bytes", "[u8; 16]")], "T:LibreMetaverse.StructuredData.OSDArray": [("values", "Vec")], "T:LibreMetaverse.StructuredData.OSDBinary": [("value", "Vec")], "T:LibreMetaverse.StructuredData.OSDBoolean": [("value", "bool")], "T:LibreMetaverse.StructuredData.OSDDate": [("value", "std::time::SystemTime")], "T:LibreMetaverse.StructuredData.OSDException": [("message", "String")], "T:LibreMetaverse.StructuredData.OSDInteger": [("value", "i32")], "T:LibreMetaverse.StructuredData.OSDMap": [("values", "std::collections::HashMap")], "T:LibreMetaverse.StructuredData.OSDReal": [("value", "f64")], "T:LibreMetaverse.StructuredData.OSDString": [("value", "String")], "T:LibreMetaverse.StructuredData.OSDUUID": [("value", "libremetaverse_types::UUID")], "T:LibreMetaverse.StructuredData.OSDUri": [("value", "libremetaverse_types::compat::Uri")], } def generic_names(item: dict) -> list[str]: return [parameter["name"] for parameter in item.get("generic_parameters", [])] def format_rust(source: str) -> str: result = subprocess.run( ["rustfmt", "--edition", "2024", "--emit", "stdout"], input=source, text=True, capture_output=True, check=True, ) return result.stdout def type_name(type_row: dict) -> str: return type_row["rust_path"].rsplit("::", 1)[-1] def generic_suffix(names: list[str]) -> str: return "<" + ", ".join(names) + ">" if names else "" def flags_enum(item: dict) -> bool: return any(attribute["type"] == "System.FlagsAttribute" for attribute in item.get("attributes", [])) def duplicate_enum_values(item: dict) -> bool: values = [member["value"]["value"] for member in item["members"] if member["kind"] == "enum_value"] return len(values) != len(set(values)) def render_enum(item: dict, rust_name: str) -> list[str]: underlying = UNDERLYING[item["enum_underlying_type"]] values = [member for member in item["members"] if member["kind"] == "enum_value"] lines = [f"/// C# type: `{item['doc_id']}`."] if flags_enum(item) or duplicate_enum_values(item): lines += [ "#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]", "#[repr(transparent)]", f"pub struct {rust_name}(pub {underlying});", f"impl {rust_name} {{", ] for member in values: name = mapping.snake(member["name"]).upper() value = member["value"]["value"] lines += [f" /// C# member: `{member['doc_id']}`.", f" pub const {name}: Self = Self({value});"] lines.append("}") return lines lines += [ "#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]", f"#[repr({underlying})]", f"pub enum {rust_name} {{", ] for member in values: lines += [f" /// C# member: `{member['doc_id']}`.", f" {mapping.pascal(member['name'])} = {member['value']['value']},"] lines.append("}") return lines def render_body(signature: str, member_id: str, error_model: str, trait: bool) -> str: if trait: signature = signature.removeprefix("pub ") failure = ( f"libremetaverse_types::not_implemented({json.dumps(member_id)})" if error_model.startswith("Result") else f"libremetaverse_types::unimplemented_api!({json.dumps(member_id)})" ) return f" {signature} {{ {failure} }}" def render_type(item: dict, type_row: dict, member_rows: dict[str, dict[str, str]], mapper: mapping.Mapper) -> str: rust_name = type_name(type_row) if item["kind"] == "enum": return "\n".join(render_enum(item, rust_name)) names = generic_names(item) suffix = generic_suffix(names) trait = item["kind"] == "interface" lines = [f"/// C# type: `{item['doc_id']}`."] if trait: lines.append(f"pub trait {rust_name}{suffix} {{") else: fields = [member for member in item["members"] if member["kind"] == "field" and not member.get("static")] private_fields = PRIVATE_LAYOUTS.get(item["doc_id"], []) if item["doc_id"] == "T:LibreMetaverse.StructuredData.OSD": lines += [ "#[non_exhaustive]", "pub enum OSD {", " Undefined, Boolean(bool), Integer(i32), Real(f64), String(String),", " UUID(libremetaverse_types::UUID), Date(std::time::SystemTime),", " Uri(libremetaverse_types::compat::Uri), Binary(Vec),", " Array(Vec), Map(std::collections::HashMap), LlsdXml(String),", "}", ] elif not fields and not names and not private_fields: lines.append(f"pub struct {rust_name};") else: lines.append(f"pub struct {rust_name}{suffix} {{") generics = set(names) for field in fields: field_type = mapper.type(field["type"], field.get("nullability"), generics) lines += [f" /// C# member: `{field['doc_id']}`.", f" pub {mapping.snake(field['name'])}: {field_type},"] for field_name, field_type in private_fields: lines.append(f" {field_name}: {field_type},") if names: tuple_type = "(" + ", ".join(names) + ("," if len(names) == 1 else "") + ")" lines.append(f" _marker: std::marker::PhantomData<{tuple_type}>,") lines.append("}") lines += [f"impl{suffix} {rust_name}{suffix} {{"] for member in item["members"]: if member["kind"] in {"enum_value"} or (member["kind"] == "field" and not member.get("static")): continue row = member_rows[member["doc_id"]] lines.append(f" /// C# member: `{member['doc_id']}`.") if member["kind"] == "constant": value = member["value"]["value"] lines.append(f" {row['rust_signature']} = {value};") continue signatures = row["rust_signature"].split(" ; ") for index, signature in enumerate(signatures): if index: lines.append(f" /// Setter for C# member: `{member['doc_id']}`.") lines.append(render_body(signature, member["doc_id"], row["error_model"], trait)) lines.append("}") if not trait: for interface in item.get("interfaces", []): base, arguments, _ = mapping.type_parts(interface) if base not in mapper.interfaces: continue mapped_arguments = [mapper.type(argument, None, set(names)) for argument in arguments] trait_path = mapper.resolved[base] + generic_suffix(mapped_arguments) lines.append(f"impl{suffix} {trait_path} for {rust_name}{suffix} {{}}") return "\n".join(lines) def generate_sources(catalog: dict) -> tuple[dict[Path, str], dict[str, tuple[int, int]]]: type_rows, resolved = mapping.build_type_rows(catalog) type_by_id = {row["csharp_type_id"]: row for row in type_rows} interfaces = {mapping.type_parts(item["signature"])[0] for assembly in catalog["assemblies"] for item in assembly["types"] if item["kind"] == "interface"} mapper = mapping.Mapper(resolved, interfaces) member_rows = {row["csharp_id"]: row for row in csv.DictReader(LEDGER.open(), delimiter="\t")} outputs = {} coverage = {} for assembly in catalog["assemblies"]: name = assembly["identity"]["name"] if name not in TARGETS: continue chunks = [ "// @generated by tools/generate_api_shims.py; do not edit by hand.", "#![allow(clippy::missing_errors_doc)]", "#![allow(clippy::must_use_candidate)]", "#![allow(clippy::all, clippy::pedantic, dead_code, unused_variables)]", "#![allow(non_camel_case_types)]", "#![allow(non_snake_case)]", "", ] for item in assembly["types"]: chunks += [render_type(item, type_by_id[item["doc_id"]], member_rows, mapper), ""] source = format_rust("\n".join(chunks).rstrip() + "\n") expected_types = {item["doc_id"] for item in assembly["types"]} expected_members = {member["doc_id"] for item in assembly["types"] for member in item["members"]} emitted_types = set(re.findall(r"C# type: `(.*)`\.", source)) emitted_members = set(re.findall(r"C# member: `(.*)`\.", source)) if emitted_types != expected_types or emitted_members != expected_members: raise ValueError( f"generated catalog ID mismatch for {name}: " f"types missing/stale={len(expected_types - emitted_types)}/{len(emitted_types - expected_types)}, " f"members missing/stale={len(expected_members - emitted_members)}/{len(emitted_members - expected_members)}" ) outputs[TARGETS[name]] = source coverage[name] = (len(assembly["types"]), sum(len(item["members"]) for item in assembly["types"])) return outputs, coverage def coverage_report(catalog: dict, coverage: dict[str, tuple[int, int]]) -> str: lines = [ "# Callable shim coverage", "", "Generated by `python3 tools/generate_api_shims.py`; do not edit by hand.", "", "| Assembly | Types | Members | Status |", "|---|---:|---:|---|", ] total_types = total_members = 0 for assembly in catalog["assemblies"]: name = assembly["identity"]["name"] types = len(assembly["types"]) members = sum(len(item["members"]) for item in assembly["types"]) if name in coverage: got_types, got_members = coverage[name] if (types, members) != (got_types, got_members): raise ValueError(f"incomplete generated slice for {name}") status = "callable failure-only shim" total_types += types total_members += members else: status = "pending milestone issue" lines.append(f"| `{name}` | {types:,} | {members:,} | {status} |") lines += ["", f"Current callable coverage: **{total_types:,} types / {total_members:,} members**.", ""] return "\n".join(lines) def generate() -> dict[Path, str]: catalog = json.loads(CATALOG.read_text()) sources, coverage = generate_sources(catalog) sources[COVERAGE] = coverage_report(catalog, coverage) return sources def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--check", action="store_true") args = parser.parse_args() outputs = generate() if args.check: stale = [str(path.relative_to(ROOT)) for path, text in outputs.items() if not path.exists() or path.read_text() != text] if stale: raise SystemExit("stale generated API shims: " + ", ".join(stale)) else: for path, text in outputs.items(): path.write_text(text) print("generated " + ", ".join(str(path.relative_to(ROOT)) for path in outputs)) if __name__ == "__main__": main()