Example scripts¶

"""Getting Started path: upload cases → inspect → recipe → pandas DataFrame.

Requires ``miura-nexus[pandas,sigma]`` and a live Nexus deployment.
"""

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
import sys
from tempfile import TemporaryDirectory

from miura import Nexus
from miura.adapters.dataframe import to_dataframe
from miura.display import print_upload_result
from miura.sdk import LocalDataSource, RecipeSpec, Schema
from miura.sdk.sigma import sigma

_examples = Path(__file__).resolve().parent
if str(_examples) not in sys.path:
    sys.path.insert(0, str(_examples))

from schemas import xdmf_case_schema, xdmf_case_structure


def _xdmf_stub(*, param_value: float, param_name: str = "param_1") -> str:
    # GlobalExtractionSchema / ExtractGlobal reads XDMF Information keys.
    return f"""<?xml version="1.0" ?>
<Xdmf Version="3.0">
  <Domain>
    <Grid Name="grid" GridType="Uniform">
      <Topology TopologyType="Triangle" NumberOfElements="1">
        <DataItem Dimensions="1 3" NumberType="Int" Format="XML">0 1 2</DataItem>
      </Topology>
      <Geometry GeometryType="XYZ">
        <DataItem Dimensions="3 3" NumberType="Float" Precision="8" Format="XML">
          0 0 0
          1 0 0
          0 1 0
        </DataItem>
      </Geometry>
    </Grid>
  </Domain>
  <Information Name="{param_name}" Value="{param_value}"/>
</Xdmf>
"""


def write_xdmf_cases(root: Path) -> None:
    for index, value in enumerate((1.02, 0.97), start=1):
        case_dir = root / f"case_{index:06d}"
        case_dir.mkdir(parents=True, exist_ok=True)
        (case_dir / "field.xdmf").write_text(
            _xdmf_stub(param_value=value),
            encoding="utf-8",
        )


def main() -> None:
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")

    with TemporaryDirectory(prefix="epico-quickstart-") as tmp:
        data_path = Path(tmp)
        write_xdmf_cases(data_path)

        with Nexus() as nexus:
            project = nexus.create_project(f"epico-quickstart-{run_id}")
            source = project.create_collection("cases", schema=xdmf_case_schema())
            target = project.create_collection("tables", schema=Schema.permissive())

            # upload() plans then applies; use source.plan(...) to inspect gates first
            structure = xdmf_case_structure()
            upload = source.upload(
                LocalDataSource(data_path),
                case_grouping=structure.case_grouping(),
            )
            print_upload_result(upload)

            # Browse the admitted cases and the logical assets they published
            case_set = source.case_set(upload.cases)
            try:
                case_set.get_projection()
            except Exception:
                case_set.materialize_projection()
            cases = case_set.list_cases()
            assets = case_set.list_logical_assets()
            print(f"cases: {len(cases.items)}")
            for case in cases.items:
                print(f"  {case.case_group_id}: {dict(case.coordinates)}")
            print(f"logical_assets: {len(assets.items)}")
            for asset in assets.items:
                print(f"  {asset.role} → {asset.modality_id}")

            # Extract param_1 from each case's XDMF Information entry
            spec = RecipeSpec.from_sigma_dataset_generator(
                source=source.name,
                target=target.name,
                dataset_generator=sigma.DataSetGenerator(
                    sigma.GlobalExtractionSchema(
                        sealed_algorithm={"global_param_name": "param_1"},
                    )
                ),
                writer="parquet_tabular",
            )
            recipe = project.deploy_recipe(
                "cases-to-tables",
                spec,
                description="Quick start cases to tables",
                cases=upload.cases,  # pin to the admitted cases from the upload
            )
            recipe.run().wait()

            df = to_dataframe(target, limit_rows=10)
            print(df.head())


if __name__ == "__main__":
    main()

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
import sys
from tempfile import TemporaryDirectory

from miura import Nexus
from miura.display import print_ingestion_plan, print_items, print_upload_result
from miura.nexus.case_grouping import CaseGrouping
from miura.sdk import LocalDataSource, UploadMode

_examples = Path(__file__).resolve().parent
if str(_examples) not in sys.path:
    sys.path.insert(0, str(_examples))

from openfoam_data import transport_properties, write_case
from schemas import openfoam_case_schema


def write_demo_tree(root: Path, *, version: int) -> None:
    write_case(root / "case_000001", case_index=1)
    if version >= 2:
        write_case(root / "case_000002", case_index=2)


def main():
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")

    with TemporaryDirectory(prefix="epico-upload-") as tmp:
        data_root = Path(tmp)
        write_demo_tree(data_root, version=1)

        with Nexus() as nexus:
            project = nexus.create_project(f"epico-upload-{run_id}")

            collection = project.create_collection(
                collection_name="upload-demo",
                schema=openfoam_case_schema(),
            )
            datasource = LocalDataSource(data_root)

            print("Step 1: Plan a replace upload")
            replace_plan = collection.plan(datasource, mode=UploadMode.REPLACE)
            print_ingestion_plan(replace_plan, data_root=datasource.path)

            print("\nStep 1b: Optional case grouping on the same plan")
            # Folder-per-case depth under the collection root; omit when you only
            # need files in the collection without admitting cases.
            grouping = CaseGrouping.root_path("/", 1)
            grouped_plan = collection.plan(
                datasource,
                mode=UploadMode.REPLACE,
                case_grouping=grouping,
            )
            print(
                f"  gate={grouped_plan.gate_status} "
                f"admissible={grouped_plan.is_admissible} "
                f"cases={grouped_plan.cases is not None}"
            )

            print("\nStep 2: Apply the plan")
            replace_result = replace_plan.apply()
            print_upload_result(replace_result)

            print("\nStep 3: Add another case with UPSERT")
            write_demo_tree(data_root, version=2)
            upsert_result = collection.upload(datasource, mode=UploadMode.UPSERT)
            print_upload_result(upsert_result)

            print("\nStep 4: Inspect uploaded files")
            items = collection.list_items(
                path=None,
                include_dirs=False,
                sort_by="item_uri",
            )
            print_items(items, max_items=20)

            print("\nStep 5: Append a new case folder with INSERT")
            write_case(data_root / "case_000003", case_index=3)
            insert_plan = collection.plan(datasource, mode=UploadMode.INSERT)
            print_ingestion_plan(insert_plan, data_root=datasource.path)
            if insert_plan.is_admissible and insert_plan.requires_execution:
                insert_result = insert_plan.apply()
                print_upload_result(insert_result)

            print("\nStep 6: Refresh an existing file with UPDATE")
            transport_path = data_root / "case_000001" / "constant" / "transportProperties"
            transport_path.write_text(
                transport_properties(nu="2.5e-05"),
                encoding="utf-8",
            )
            update_plan = collection.plan(datasource, mode=UploadMode.UPDATE)
            print_ingestion_plan(update_plan, data_root=datasource.path)
            if update_plan.is_admissible and update_plan.requires_execution:
                update_result = update_plan.apply()
                print_upload_result(update_result)

            print("\nAvailable ingestion modes:")
            for mode in (
                UploadMode.INSERT,
                UploadMode.UPDATE,
                UploadMode.UPSERT,
                UploadMode.REPLACE,
            ):
                print(f"  {mode.value}")


if __name__ == "__main__":
    main()

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
import sys

from miura import Nexus
from miura.display import print_items, print_upload_result
from miura.sdk import LocalDataSource, UploadMode

_examples = Path(__file__).resolve().parent
if str(_examples) not in sys.path:
    sys.path.insert(0, str(_examples))

from schemas import openfoam_case_schema
from openfoam_data import openfoam_data_root


def take(source, limit):
    rows = []
    for row in source:
        rows.append(row)
        if len(rows) >= limit:
            break
    return rows


def print_uris(label, items):
    print(label)
    for item in items:
        suffix = "/" if item.is_folder and not item.item_uri.endswith("/") else ""
        print(f"  {item.item_uri}{suffix}")


def main():
    examples_root = Path(__file__).resolve().parent
    data_path = openfoam_data_root(examples_root)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")

    with Nexus() as nexus:
        project = nexus.create_project(f"epico-items-{run_id}")

        print("Step 1: Create a collection and upload OpenFOAM cases")
        collection = project.create_collection(
            collection_name="openfoam-items",
            schema=openfoam_case_schema(),
        )
        upload_result = collection.upload(
            LocalDataSource(data_path),
            mode=UploadMode.REPLACE,
        )
        print_upload_result(upload_result)

        print("\nStep 2: List case folders at the root (one level)")
        cases = collection.list_items(path="/", include_dirs=True, sort_by="item_uri")
        print_uris("Top-level items:", cases[:5])

        first_folder = next((item for item in cases if item.is_folder), None)
        if first_folder is None:
            print("No folder found under root.")
            return

        print("\nStep 3: Navigate down with .children")
        folder_children = take(first_folder.children, limit=5)
        print_uris(f"Children of {first_folder.item_uri}:", folder_children)

        first_file = next(
            (item for item in folder_children if not item.is_folder), None
        )
        if first_file is None:
            for child in folder_children:
                if not child.is_folder:
                    continue
                nested = take(child.children, limit=5)
                first_file = next((item for item in nested if not item.is_folder), None)
                if first_file is not None:
                    break
        if first_file is None:
            first_file = next(
                collection.walk_items(path=first_folder.item_uri, include_dirs=False),
                (None, None),
            )[0]
        if first_file is None:
            print(f"No file found under {first_folder.item_uri}.")
            return

        print("\nStep 4: Navigate up with .parent")
        parent = first_file.parent
        print(f"Item: {first_file.item_uri}")
        print(f"Parent: {parent.item_uri if parent else None}")

        print("\nStep 5: Navigate sideways with .siblings")
        siblings = take(first_file.siblings, limit=5)
        print_uris(f"Siblings of {first_file.item_uri}:", siblings)

        print("\nStep 6: Fetch the same item directly by path")
        same_item = collection.get_item(first_file.item_uri)
        print(f"Direct lookup: {same_item.item_uri if same_item else None}")

        print("\nStep 7: Walk all files under the collection")
        for item, depth in take(collection.walk_items(path="/", include_dirs=False), limit=5):
            indent = "  " * depth
            print(f"{indent}{item.item_uri} ({item.file_size} B)")

        print("\nStep 8: Climb toward the root from one file")
        current = first_file
        while current is not None:
            print(f"  {current.item_uri}")
            current = current.parent


if __name__ == "__main__":
    main()

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
import sys

from miura import Nexus
from miura.api.annotations import ItemAttributes, ItemSelect
from miura.display import print_items, print_upload_result
from miura.sdk import LocalDataSource, UploadMode

_examples = Path(__file__).resolve().parent
if str(_examples) not in sys.path:
    sys.path.insert(0, str(_examples))

from schemas import openfoam_case_schema
from openfoam_data import openfoam_data_root


def first_files(collection, limit):
    files = []
    for item in collection.iter_items(
        path=None,
        include_dirs=False,
        sort_by="item_uri",
        page_size=20,
    ):
        files.append(item)
        if len(files) >= limit:
            break
    return files


def main():
    examples_root = Path(__file__).resolve().parent
    data_path = openfoam_data_root(examples_root)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")

    with Nexus() as nexus:
        project = nexus.create_project(f"epico-annotations-{run_id}")

        print("Step 1: Create a collection and upload OpenFOAM cases")
        collection = project.create_collection(
            collection_name="openfoam-annotations",
            schema=openfoam_case_schema(),
        )
        upload_result = collection.upload(
            LocalDataSource(data_path),
            mode=UploadMode.REPLACE,
        )
        print_upload_result(upload_result)

        files = first_files(collection, limit=6)
        if not files:
            print("No files found to annotate")
            return

        print("\nStep 2: Add attributes to one item")
        baseline = files[0]
        baseline.set_attributes(
            split="train",
            labels={"solver": "simpleFoam", "stage": "baseline"},
            annotations={"note": "single item update"},
        )
        print(f"Tagged one item: {baseline.item_uri}")

        print("\nStep 3: Add attributes to several items at once")
        batch_rows = []
        for index, item in enumerate(files[1:]):
            split = "train" if index < 3 else "val"
            stage = "baseline" if index % 2 == 0 else "candidate"
            batch_rows.append(
                ItemAttributes(
                    item_id=item.uuid,
                    split=split,
                    labels={"solver": "simpleFoam", "stage": stage},
                    annotations={"note": f"batch update {index + 1}"},
                )
            )

        result = collection.set_item_attributes(batch_rows)
        print(
            "Updated "
            f"{result.get('success_count', 0)} of {result.get('total_count', len(batch_rows))} items"
        )

        print("\nStep 4: Read attributes back from one item")
        attributes = collection.get_item_attributes(baseline.uuid)
        print(attributes)

        print("\nStep 5: Filter items with ItemSelect")
        train_baseline = collection.list_items(
            include_dirs=False,
            sort_by="item_uri",
            select=ItemSelect(
                split="train",
                labels={"solver": "simpleFoam", "stage": "baseline"},
            ),
        )
        print_items(train_baseline, style="table", show_attributes=True)

        print("\nStep 6: Count selected items")
        train_count = collection.count_items(select=ItemSelect(split="train"))
        baseline_count = collection.count_items(
            select=ItemSelect(labels={"stage": "baseline"})
        )
        print(f"Train files: {train_count}")
        print(f"Baseline files: {baseline_count}")

        print("\nStep 7: Clear one field")
        collection.clear_item_attributes(baseline.uuid, keys=["annotations"])
        refreshed = collection.get_item_attributes(baseline.uuid)
        print(refreshed)


if __name__ == "__main__":
    main()

"""Build and validate transformation specs (converter and dataset generator).

Section 3 execution-layer demo: RecipeSpec authoring + ``validate_recipe``.
Pin admitted cases at ``deploy_recipe(..., cases=upload.cases)`` — not via a raw
``CaseGrouping`` on the spec. For deploy / conflict / update / run see
``recipe_lifecycle.py`` and ``3_describe_transformation.md``.
"""

from __future__ import annotations

from datetime import datetime, timezone

from miura import Nexus
from miura.display import print_capabilities, print_json, print_validation_result
from miura.nexus.recipe.sigma_converter import (
    RemoteSigmaConverter,
    SigmaClassRef,
    SigmaSetMethod,
)
from miura.nexus.recipe.sigma_handoff import TransformEntry, default_multipart_capability
from miura.sdk import Schema
from miura.sdk.exceptions import MissingOptionalDependencyError, ValidationError
from miura.sdk.recipe import (
    CaseAdmissionPolicy,
    CaseSelection,
    RecipeSpec,
)


def simple_vtk_converter() -> RemoteSigmaConverter:
    return RemoteSigmaConverter(
        required_extensions=[".vtk"],
        reader=SigmaClassRef(class_name="vtkDataSetReader"),
        writer=SigmaClassRef(class_name="vtkXdmfWriter"),
    )


def mapdl_thermal_converter() -> RemoteSigmaConverter:
    return RemoteSigmaConverter(
        required_extensions=[".rth"],
        reader=SigmaClassRef(
            class_name="vtkPymapdlReader",
            set_methods={
                "SetReaderProfile": SigmaSetMethod(args=("semantic_temperature",)),
            },
        ),
        writer=SigmaClassRef(
            class_name="vtkXdmfWriter",
            set_methods={
                "SetMeshStaticOverTime": SigmaSetMethod(args=(True,)),
            },
        ),
        custom_algorithm=SigmaClassRef(
            class_name="SplitHomogeneousMeshBlocks",
            options={"cell_types": [12]},
        ),
    )


def create_dataset_generator():
    from miura.sdk.sigma import sigma

    return sigma.DataSetGenerator(
        sigma.GlobalExtractionSchema(
            sealed_algorithm={"global_param_name": "param_1"},
        )
    )


def validate(project, label: str, spec: RecipeSpec) -> None:
    print(f"\nValidating {label}")
    print_validation_result(project.validate_recipe(spec))


def main():
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")

    with Nexus() as nexus:
        project = nexus.create_project(f"epico-transform-{run_id}")
        schema = Schema.permissive()
        source_vtk = project.create_collection("source-vtk", schema=schema)
        target_xdmf = project.create_collection("target-xdmf", schema=schema)
        target_parquet = project.create_collection("target-parquet", schema=schema)

        print("Step 1: Transform files into other files (converter)")
        converter_spec = RecipeSpec.from_sigma_converter(
            source=source_vtk.name,
            target=target_xdmf.name,
            converter=simple_vtk_converter(),
        )
        print_json(
            converter_spec.to_payload()["execution_payload"]["data"],
            title="Converter payload:",
        )
        validate(project, "VTK converter", converter_spec)

        print("\nStep 2: Remote reader + custom algorithm")
        mapdl_spec = RecipeSpec.from_sigma_converter(
            source=source_vtk.name,
            target=target_xdmf.name,
            converter=mapdl_thermal_converter(),
        )
        print_json(
            mapdl_spec.to_payload()["execution_payload"]["data"]["custom_algorithm"],
            title="Custom algorithm:",
        )
        validate(project, "remote converter", mapdl_spec)

        print("\nStep 3: CaseSelection on the spec (pin admitted cases at deploy)")
        # Labels align with annotations.py (solver=simpleFoam, split=train).
        # Admit grouping on upload/plan; pin with deploy_recipe(..., cases=upload.cases).
        multi_case_spec = RecipeSpec.from_sigma_converter(
            source=source_vtk.name,
            target=target_xdmf.name,
            converter=simple_vtk_converter(),
            case_selection=CaseSelection.where(
                split="train",
                labels={"solver": "simpleFoam"},
            ),
            case_admission_policy=CaseAdmissionPolicy(
                skip_unchanged_before_transform=False,
                zero_admitted_is_success_noop=True,
            ),
        )
        orchestration = multi_case_spec.to_payload()["execution_payload"]["data_ref"][
            "orchestration"
        ]
        print_json(orchestration, title="Orchestration:")
        validate(project, "multi-case converter", multi_case_spec)

        print("\nStep 3b: Optional TransformEntry (modality handoff on the spec)")
        # Default converter path omits this — Sigma picks staged files by extension.
        # Use TransformEntry when the run must bind by published logical assets.
        handoff_spec = RecipeSpec.from_sigma_converter(
            source=source_vtk.name,
            target=target_xdmf.name,
            converter=simple_vtk_converter(),
            transform_entry=TransformEntry(
                primary_modality_id="field",
                required_companion_modalities=["mesh"],
            ),
            sigma_handoff_capability=default_multipart_capability(),
        )
        handoff_orch = handoff_spec.to_payload()["execution_payload"]["data_ref"][
            "orchestration"
        ]
        print_json(
            {
                "transform_entry": handoff_orch.get("transform_entry"),
                "sigma_handoff": handoff_orch.get("sigma_handoff"),
            },
            title="Modality handoff:",
        )
        validate(project, "transform-entry converter", handoff_spec)

        print("\nStep 4: Extract structured tables (dataset generator + writer)")
        capabilities = project.get_recipe_capabilities()
        print_capabilities(capabilities, reader_limit=5)
        print(f"available_writers: {list(project.available_writers)}")
        try:
            dataset_generator = create_dataset_generator()
        except MissingOptionalDependencyError as exc:
            print(f"Skipping dataset-generator spec (install miura-nexus[sigma]): {exc}")
            return

        try:
            project.validate_recipe(
                RecipeSpec.from_sigma_dataset_generator(
                    source=target_xdmf.name,
                    target=target_parquet.name,
                    dataset_generator=dataset_generator,
                    writer="parquet_tabluar",  # intentional typo
                )
            )
        except ValidationError as exc:
            print(f"bad writer rejected: {exc}")

        dataset_spec = RecipeSpec.from_sigma_dataset_generator(
            source=target_xdmf.name,
            target=target_parquet.name,
            dataset_generator=dataset_generator,
            writer="parquet_tabular",
            writer_options={
                "flush": {
                    "mode": "rows",
                    "rows_per_file": 1_000_000,
                    "target_row_group_rows": 1_000_000,
                }
            },
        )
        print_json(dataset_spec.to_payload()["writer"], title="DataSetGenerator writer:")
        validate(project, "dataset generator", dataset_spec)


if __name__ == "__main__":
    main()

"""Recipe lifecycle: deploy / conflict / update (ADR-0019 V2).

Preferred authoring path for versioned recipes:

1. ``project.deploy_recipe(name, spec, cases=…)`` — create, no-op, or revise
2. ``get_or_create_recipe`` raises ``RecipeSpecConflictError`` on mismatch
3. Typed rebind with ``recipe.update(cases=…)`` when you hold a handle
4. Stale handle after concurrent revise → ``RecipeConflictError``

Default ``recipe.run()`` / ``run_recipe(recipe)`` uses ``inputs="pinned"``. Automatic
``inputs="latest"`` (re-admit against collection head) is not wired yet —
after new membership, re-admit via ``plan(..., case_grouping=…)``, then
``recipe.update(cases=upload.cases)`` or ``deploy_recipe(..., cases=…)``, then run.
"""

from __future__ import annotations

from datetime import datetime, timezone

from miura import Nexus
from miura.display import print_json
from miura.nexus.case_grouping.admission import CaseGroupingResult, MembershipCommitment
from miura.nexus.recipe.sigma_converter import RemoteSigmaConverter, SigmaClassRef
from miura.sdk import Schema
from miura.sdk.exceptions import RecipeConflictError, RecipeSpecConflictError
from miura.sdk.recipe import RecipeSpec


def _vtk_converter() -> RemoteSigmaConverter:
    return RemoteSigmaConverter(
        required_extensions=[".vtk"],
        reader=SigmaClassRef(class_name="vtkDataSetReader"),
        writer=SigmaClassRef(class_name="vtkXdmfWriter"),
    )


def _demo_cgr(result_id: str) -> CaseGroupingResult:
    """Construct an admitted-shaped pin for lifecycle demos (not a live CS admit)."""
    digest = "sha256:" + ("a" * 64)
    return CaseGroupingResult(
        result_id=result_id,
        membership_commitment=MembershipCommitment(
            canonical_membership_schema_version="1",
            case_group_stream_sha256=digest,
            member_edge_stream_sha256=digest,
            case_group_id_set_sha256=digest,
            membership_merkle_root=digest,
        ),
    )


def main() -> None:
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")
    name = f"lifecycle-demo-{run_id}"

    with Nexus() as nexus:
        project = nexus.create_project(f"epico-recipe-lifecycle-{run_id}")
        schema = Schema.permissive()
        source = project.create_collection("source-vtk", schema=schema)
        target = project.create_collection("target-xdmf", schema=schema)

        spec_a = RecipeSpec.from_sigma_converter(
            source=source.name,
            target=target.name,
            converter=_vtk_converter(),
        )
        # Production code gets this handle from upload.cases, plan.cases, or
        # collection.case_grouping.admit(...).
        initial_cases = source.admitted_cases(_demo_cgr(f"cgr_initial_{run_id}"))
        print("Step 1: deploy_recipe creates the first revision")
        recipe = project.deploy_recipe(name, spec_a, cases=initial_cases)
        print(f"  recipe={recipe.name} revision={recipe.current_revision_id}")

        print("\nStep 2: deploy_recipe is a no-op when the effective spec matches")
        again = project.deploy_recipe(name, spec_a, cases=initial_cases)
        print(f"  same revision={again.current_revision_id}")

        print("\nStep 3: get_or_create_recipe conflicts on a different executable spec")
        # Source/target names are not part of the Converter execution payload —
        # change the reader so the effective executable spec actually differs.
        spec_b = RecipeSpec.from_sigma_converter(
            source=source.name,
            target=target.name,
            converter=RemoteSigmaConverter(
                required_extensions=[".vtk"],
                reader=SigmaClassRef(class_name="vtkXMLUnstructuredGridReader"),
                writer=SigmaClassRef(class_name="vtkXdmfWriter"),
            ),
        )
        try:
            project.get_or_create_recipe(name, spec_b)
            raise SystemExit("FAIL: expected RecipeSpecConflictError")
        except RecipeSpecConflictError as exc:
            print(f"  caught {type(exc).__name__}: {exc}")

        print("\nStep 4: deploy_recipe revises intentionally")
        revised = project.deploy_recipe(name, spec_b, cases=initial_cases)
        print(f"  new revision={revised.current_revision_id}")

        print("\nStep 5: recipe.update(cases=…) typed rebind")
        # In production this is upload.cases / plan.cases straight from admission.
        cgr = _demo_cgr(f"cgr_demo_{run_id}")
        cases = source.admitted_cases(cgr)
        rebound = revised.update(cases=cases)
        print(f"  rebound revision={rebound.current_revision_id} cases={cases.result_id}")

        print("\nStep 6: stale handle after revise raises RecipeConflictError")
        try:
            revised.update(cases=source.admitted_cases(_demo_cgr(f"cgr_stale_{run_id}")))
            raise SystemExit("FAIL: expected RecipeConflictError")
        except RecipeConflictError as exc:
            print(f"  caught {type(exc).__name__}: {exc}")

        print_json(
            {
                "name": rebound.name,
                "current_revision_id": rebound.current_revision_id,
                "effective_spec_commitment": rebound.effective_spec_commitment,
                "case_grouping_result_id": cgr.result_id,
            },
            title="Current recipe:",
        )

        print("\nLifecycle OK (deploy → conflict → deploy revise → update → stale CAS).")


if __name__ == "__main__":
    main()

"""Browse an admitted CaseSet (cases, roles, logical assets).

Creates a small companion layout (field + mesh in each case folder), uploads
with ``case_grouping=`` and ``require=[...]``, then browses the admitted
universe — no transform entry on a recipe.

Requires a live Nexus deployment (``MIURA_NEXUS_API_KEY`` / ``miura-auth``).
"""

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
from tempfile import TemporaryDirectory

from miura import Nexus
from miura.display import print_upload_result
from miura.nexus.case_grouping import CaseGrouping, ValidationProfile, require_role
from miura.sdk import LocalDataSource, Schema


def _vtk_stub(*, title: str) -> str:
    return (
        "# vtk DataFile Version 3.0\n"
        f"{title}\n"
        "ASCII\n"
        "DATASET POLYDATA\n"
        "POINTS 3 float\n"
        "0 0 0\n"
        "1 0 0\n"
        "0 1 0\n"
        "POLYGONS 1 4\n"
        "3 0 1 2\n"
    )


def write_companion_cases(root: Path, *, case_count: int = 2) -> None:
    """Guide layout: ``case_000/case_000.vtk`` + ``case_000/case_000_mesh.vtk``."""
    for index in range(case_count):
        case_id = f"case_{index:03d}"
        case_dir = root / case_id
        case_dir.mkdir(parents=True, exist_ok=True)
        (case_dir / f"{case_id}.vtk").write_text(
            _vtk_stub(title=f"{case_id} field"),
            encoding="utf-8",
        )
        (case_dir / f"{case_id}_mesh.vtk").write_text(
            _vtk_stub(title=f"{case_id} mesh"),
            encoding="utf-8",
        )


def companion_case_grouping() -> CaseGrouping:
    # Processing validation rides on the projected grouping (not CaseStructure).
    profile = ValidationProfile.of(
        "companion_roles_present",
        require_role("boundary_solution"),
        require_role("mesh"),
    )
    return CaseGrouping.join_by_path_id(
        "case_id",
        {
            "boundary_solution": "{case_id}/{case_id}.vtk",
            "mesh": "{case_id}/{case_id}_mesh.vtk",
        },
        require=["boundary_solution", "mesh"],
    ).with_validation_profile(profile)


def main() -> None:
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")

    with TemporaryDirectory(prefix="epico-caseset-") as tmp:
        data_path = Path(tmp)
        write_companion_cases(data_path)

        with Nexus() as nexus:
            project = nexus.create_project(f"epico-caseset-{run_id}")
            collection = project.create_collection(
                "companion-cases",
                schema=Schema.permissive(),
            )

            print("Step 1: Upload with companions required in each case")
            grouping = companion_case_grouping()
            profile = getattr(grouping.strategy, "validation_profile", None)
            print(
                f"  validation_profile={getattr(profile, 'profile_id', None)} "
                f"rules={len(getattr(profile, 'rules', ()) or ())}"
            )
            upload = collection.upload(
                LocalDataSource(data_path),
                case_grouping=grouping,
            )
            print_upload_result(upload)

            print("\nStep 2: Browse admitted cases and logical assets")
            case_set = collection.case_set(upload.cases)
            try:
                projection = case_set.get_projection()
            except Exception:
                projection = case_set.materialize_projection()
            print(
                f"projection: {projection.projection_id} "
                f"state={projection.state} "
                f"cases={projection.case_count} assets={projection.logical_asset_count}"
            )
            cases = case_set.list_cases()
            assets = case_set.list_logical_assets()
            print(f"cases: {len(cases.items)}")
            for case in cases.items:
                coords = ", ".join(f"{k}={v}" for k, v in sorted(case.coordinates.items()))
                print(f"  {case.case_group_id} ({coords})")
            print(f"logical_assets: {len(assets.items)}")
            for asset in assets.items:
                print(f"  {asset.role} → {asset.modality_id}")


if __name__ == "__main__":
    main()

"""Admit case grouping after files already exist in a collection.

This is the NEXUS-392 path: upload without ``case_grouping=``, then snapshot
the existing collection membership with ``collection.case_grouping.admit``.
Admission moves no bytes and creates no ingestion plan. The returned
``AdmittedCases`` binds directly to ``deploy_recipe(..., cases=...)``.

Requires a live Nexus deployment (``MIURA_NEXUS_API_KEY`` / ``miura-auth``).
"""

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
from tempfile import TemporaryDirectory

from miura import Nexus
from miura.nexus.case_grouping import CaseGrouping
from miura.nexus.recipe.sigma_converter import RemoteSigmaConverter, SigmaClassRef
from miura.sdk import LocalDataSource, Schema, UploadMode
from miura.sdk.recipe import RecipeSpec


def _vtk_stub(*, title: str) -> str:
    return (
        "# vtk DataFile Version 3.0\n"
        f"{title}\n"
        "ASCII\n"
        "DATASET POLYDATA\n"
        "POINTS 3 float\n"
        "0 0 0\n"
        "1 0 0\n"
        "0 1 0\n"
        "POLYGONS 1 4\n"
        "3 0 1 2\n"
    )


def write_existing_cases(root: Path) -> None:
    for case_id in ("case_001", "case_002"):
        case_dir = root / case_id
        case_dir.mkdir(parents=True, exist_ok=True)
        (case_dir / "input.vtk").write_text(
            _vtk_stub(title=case_id),
            encoding="utf-8",
        )


def _converter_spec(*, source: str, target: str) -> RecipeSpec:
    return RecipeSpec.from_sigma_converter(
        source=source,
        target=target,
        converter=RemoteSigmaConverter(
            required_extensions=[".vtk"],
            reader=SigmaClassRef(class_name="vtkDataSetReader"),
            writer=SigmaClassRef(class_name="vtkXdmfWriter"),
        ),
    )


def main() -> None:
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")

    with TemporaryDirectory(prefix="epico-existing-cases-") as tmp:
        data_path = Path(tmp)
        write_existing_cases(data_path)

        with Nexus() as nexus:
            project = nexus.create_project(f"epico-existing-cases-{run_id}")
            source = project.create_collection("existing-cases", schema=Schema.permissive())
            target = project.create_collection("converted-cases", schema=Schema.permissive())

            print("Step 1: Upload files without case grouping")
            upload = source.upload(
                LocalDataSource(data_path),
                mode=UploadMode.INSERT,
            )
            if upload.status != "success":
                raise SystemExit(f"FAIL: upload status={upload.status}")
            before = source.count_items()
            print(f"  existing items={before}")

            print("\nStep 2: Admit grouping against frozen existing membership")
            cases = source.case_grouping.admit(
                CaseGrouping.root_path("/", depth=1),
                group_name="data",
                schema_extras="include",
            )
            result = cases.case_grouping_result
            after = source.count_items()
            if after != before:
                raise SystemExit(f"FAIL: admission changed membership ({before} → {after})")
            if not result.has_inventory_pins:
                raise SystemExit("FAIL: admission did not return complete inventory pins")
            print(f"  result_id={result.result_id}")
            print(f"  membership unchanged={before == after}")
            print(f"  snapshot={result.collection_snapshot_ref}")

            print("\nStep 3: Bind admitted cases to a recipe")
            recipe = project.deploy_recipe(
                "process-existing-cases",
                _converter_spec(source=source.name, target=target.name),
                cases=cases,
            )
            binding = recipe.recipe["execution_payload"]["data_ref"]["orchestration"]["case_grouping_binding"]
            if binding.get("version") != 2:
                raise SystemExit(f"FAIL: expected binding version 2, got {binding.get('version')!r}")
            print(f"  recipe={recipe.name}")
            print(f"  binding version={binding['version']}")
            print(f"  bound result={binding['result_id']}")


if __name__ == "__main__":
    main()

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
from tempfile import TemporaryDirectory

from miura import Nexus
from miura.adapters import DatasetSource
from miura.adapters.dataframe import to_dataframe
from miura.display import print_items, print_upload_result
from miura.sdk import LocalDataSource, Schema, UploadMode


def write_parquet_tree(root: Path) -> None:
    import pyarrow as pa
    import pyarrow.parquet as pq

    root.mkdir(parents=True, exist_ok=True)
    for shard in range(2):
        table = pa.table(
            {
                "case_id": [shard] * 5,
                "temperature": [20.0 + shard + i for i in range(5)],
                "pressure": [1010.0 + 2.0 * i for i in range(5)],
                "label": [int(i % 2 == 0) for i in range(5)],
            }
        )
        pq.write_table(table, root / f"part-{shard:04d}.parquet")


def main():
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")

    with TemporaryDirectory(prefix="epico-parquet-") as tmp:
        data_root = Path(tmp)
        write_parquet_tree(data_root)

        with Nexus() as nexus:
            project = nexus.create_project(f"epico-parquet-{run_id}")

            print("Step 1: Upload generated Parquet files")
            collection = project.create_collection(
                collection_name="direct-parquet",
                schema=Schema.permissive(),
            )
            result = collection.upload(
                LocalDataSource(data_root),
                mode=UploadMode.REPLACE,
            )
            print_upload_result(result)

            print("\nStep 2: Inspect files")
            items = collection.list_items(
                path=None,
                include_dirs=False,
                sort_by="item_uri",
            )
            print_items(items)

            print("\nStep 3: Read through the DataFrame adapter")
            df = to_dataframe(
                collection,
                columns=["temperature", "pressure", "label"],
                limit_rows=6,
            )
            print(df.to_string(index=False))

            print("\nStep 4: Build a dataset wrapper for downstream adapters")
            source = DatasetSource.from_collection(collection, path="/")
            print(source)

            print("\nStep 5: Keep catalog reads for published recipe outputs")
            print(
                "Direct uploads are readable as files. Published recipe outputs can "
                "also expose catalog assets for batch-oriented reads."
            )

            # print(f"\nCleaning up project: {project.name}")
            # project.delete_project()


if __name__ == "__main__":
    main()

"""
Catalog read wiring without a live catalog asset.

Builds a CatalogAssetRef from a fresh project/collection, previews batch shape
locally, and wires DatasetSource + shard policy. For live Flight reads after a
published table exists, use catalog_stream.py.

Run:
  uv run python docs/examples/catalog.py
"""

from __future__ import annotations

from datetime import datetime, timezone

import pyarrow as pa

from miura import Nexus
from miura.adapters import DatasetSource
from miura.display import (
    print_catalog_read_batches,
    print_catalog_read_complete,
    print_catalog_read_scope,
    print_json,
)
from miura.nexus.catalog import CatalogAssetRef, CatalogReadShard
from miura.sdk import Schema

DEMO_COLUMNS = ("temperature", "pressure", "smoke_scalar")


def local_batches(columns: list[str]) -> list[pa.RecordBatch]:
    values = {
        "smoke_scalar": pa.array([0.1, 0.2, 0.3, 0.4]),
        "pressure": pa.array([1010.0, 1012.5, 1015.0, 1018.5]),
        "temperature": pa.array([293.0, 294.5, 296.0, 297.5]),
    }
    arrays = [values.get(name, pa.array([None, None, None, None])) for name in columns]
    return [pa.record_batch(arrays, names=columns)]


def main() -> None:
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")
    columns = list(DEMO_COLUMNS)
    batch_size = 4096
    max_batches = 2

    with Nexus() as nexus:
        project = nexus.create_project(f"epico-catalog-{run_id}")
        collection = project.create_collection(
            collection_name="catalog-wiring-demo",
            schema=Schema.permissive(),
        )
        asset_ref = CatalogAssetRef(
            project_id=project.id,
            collection_id=collection.id,
            asset_id="demo-catalog-asset",
        )

        print("Step 1: Build a catalog asset ref from the current project")
        print_catalog_read_scope(
            project_id=project.id,
            collection_id=collection.id,
            asset_id=asset_ref.asset_id,
            tenant_id="demo",
            columns=columns,
        )

        print("Step 2: Preview the projected batch shape (local sample rows)")
        batches = local_batches(columns)
        batch_count = min(len(batches), max_batches)
        row_count = sum(batch.num_rows for batch in batches[:batch_count])
        first_batch = batches[0] if batches else None

        print_catalog_read_batches(
            batch_count=batch_count,
            row_count=row_count,
            column_names=first_batch.column_names if first_batch is not None else None,
            sample_batch=first_batch,
        )

        print("Step 3: Reuse the asset ref as an adapter source")
        source = DatasetSource.from_catalog_asset(
            asset_ref,
            columns=columns,
            catalog_batch_size=batch_size,
            shard_world_size=2,
        )
        print_json(
            {
                "readiness": source.readiness,
                "columns": source.columns,
                "batch_size": source.catalog_batch_size,
                "shard_world_size": source.shard_world_size,
                "asset_ref": source.asset_ref,
            },
            title="DatasetSource:",
        )

        print("\nStep 4: Plan one deterministic shard")
        shard = CatalogReadShard(strategy="rank_partition", rank=0, world_size=2)
        print_json(shard.as_policy(), title="Shard policy:")

        print_catalog_read_complete(
            label="catalog.py",
            batch_count=batch_count,
            row_count=row_count,
        )


if __name__ == "__main__":
    main()

"""Live catalog asset read (Flight-backed batches).

Uploads Parquet so the platform publishes a catalog asset, waits until the
collection binding is queryable, then streams rows through catalog read
(``to_dataframe_async`` / ``nexus.catalog.iter_batches``).

After a dataset-generator recipe, resolve the same way with
``target.catalog_asset()`` on the recipe target collection — see ``4_consume.md``.

Requires ``miura-nexus[pandas]``, ``pyarrow``, and a live Nexus deployment.
"""

from __future__ import annotations

import asyncio
from datetime import datetime, timezone
from pathlib import Path
from tempfile import TemporaryDirectory

import pyarrow as pa
import pyarrow.parquet as pq

from miura import Nexus
from miura.adapters import DatasetSource
from miura.adapters.dataframe import to_dataframe_async
from miura.api import AsyncNexus
from miura.display import (
    print_catalog_read_batches,
    print_catalog_read_complete,
    print_catalog_read_scope,
    print_json,
    print_upload_result,
)
from miura.nexus.catalog import CatalogReadShard
from miura.sdk import LocalDataSource, Schema, UploadMode

COLUMNS = ("temperature", "pressure", "smoke_scalar")


def write_parquet_tree(root: Path) -> None:
    root.mkdir(parents=True, exist_ok=True)
    table = pa.table(
        {
            "temperature": [293.0, 294.5, 296.0, 297.5, 299.0, 300.5],
            "pressure": [1010.0, 1012.5, 1015.0, 1018.5, 1020.0, 1022.5],
            "smoke_scalar": [0.10, 0.20, 0.30, 0.40, 0.50, 0.60],
        }
    )
    pq.write_table(table, root / "part-0000.parquet")


async def stream_batches(asset, *, columns: list[str], batch_size: int) -> tuple[int, int]:
    """Return (batch_count, row_count) from a live Flight-backed stream."""
    ref = asset.logical_ref
    batch_count = 0
    row_count = 0
    first_batch = None
    async with AsyncNexus() as nexus:
        async for batch in nexus.catalog.iter_batches(
            project_id=ref.project_id,
            collection_id=ref.collection_id,
            asset_id=ref.asset_id,
            columns=columns,
            batch_size=batch_size,
        ):
            if first_batch is None:
                first_batch = batch
            batch_count += 1
            row_count += batch.num_rows
    print_catalog_read_batches(
        batch_count=batch_count,
        row_count=row_count,
        column_names=list(first_batch.column_names) if first_batch is not None else None,
        sample_batch=first_batch,
    )
    return batch_count, row_count


async def stream_shard(asset, *, columns: list[str], batch_size: int) -> int:
    """One rank of a rank_partition world — returns row count for this shard."""
    ref = asset.logical_ref
    shard = CatalogReadShard(strategy="rank_partition", rank=0, world_size=2)
    rows = 0
    async with AsyncNexus() as nexus:
        async for batch in nexus.catalog.iter_batches(
            project_id=ref.project_id,
            collection_id=ref.collection_id,
            asset_id=ref.asset_id,
            columns=columns,
            batch_size=batch_size,
            shard=shard,
        ):
            rows += batch.num_rows
    return rows


def main() -> None:
    run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f")
    columns = list(COLUMNS)
    batch_size = 4096

    with TemporaryDirectory(prefix="epico-catalog-stream-") as tmp:
        data_root = Path(tmp)
        write_parquet_tree(data_root)

        with Nexus() as nexus:
            project = nexus.create_project(f"epico-catalog-stream-{run_id}")
            collection = project.create_collection(
                collection_name="published-tables",
                schema=Schema.permissive(),
            )

            print("Step 1: Upload Parquet (platform publishes the catalog asset)")
            upload = collection.upload(
                LocalDataSource(data_root),
                mode=UploadMode.REPLACE,
            )
            print_upload_result(upload)

            print("\nStep 2: Wait until the published table is queryable")
            # After a dataset-generator recipe: asset = target.catalog_asset(...)
            asset = collection.catalog_asset(timeout=180.0, poll_interval=2.0)
            print_catalog_read_scope(
                project_id=asset.logical_ref.project_id,
                collection_id=asset.logical_ref.collection_id,
                asset_id=asset.logical_ref.asset_id,
                columns=columns,
            )
            print_json(
                {
                    "asset_id": asset.logical_ref.asset_id,
                    "fileset_snapshot_id": asset.fileset_snapshot_id,
                    "catalog_registration_id": asset.catalog_registration_id,
                },
                title="Queryable asset:",
            )

            print("\nStep 3: Read as a pandas DataFrame (Flight-backed)")
            source = DatasetSource.from_catalog_asset(
                asset,
                columns=columns,
                catalog_batch_size=batch_size,
            )
            df = asyncio.run(to_dataframe_async(source, limit_rows=10))
            print(df.to_string(index=False))

            print("\nStep 4: Stream RecordBatches via nexus.catalog.iter_batches")
            batch_count, row_count = asyncio.run(stream_batches(asset, columns=columns, batch_size=batch_size))

            print("\nStep 5: Optional shard (rank 0 of world_size=2)")
            shard_rows = asyncio.run(stream_shard(asset, columns=columns, batch_size=batch_size))
            print(f"  shard rows (rank 0 / 2): {shard_rows}")

            print_catalog_read_complete(
                label="catalog_stream.py",
                batch_count=batch_count,
                row_count=row_count,
            )


if __name__ == "__main__":
    main()