Materialize the KirkpatrickBaez Assembly at CDI + IOS (first two fixtures)

apps/api/tests/integration/scenarios/test_cdi_kb_mirror_setup.py

@@ -0,0 +1,182 @@
+"""KB focusing-pair deployment at NSLS-II CDI/CHX (Assembly + Fixture).
+
+cluster: Commissioning
+archetype: setup
+bc_primary: Equipment
+bc_touches: Equipment
+
+Materializes CDI's Kirkpatrick-Baez nanofocusing mirror pair as the FIRST binding
+of the catalog KirkpatrickBaez Assembly (which shipped with zero Fixtures),
+end-to-end against Postgres, from the assets CDI's deployment descriptor carries
+(`deployments/cdi/beamline.yaml`, KB-1).
+
+CDI's descriptor collapses the pair into ONE `KBMirror` Mirror Asset, but its note
+names both physical mirrors with their real PVs: a vertical VKB at `Mir:KBv` and a
+horizontal HKB at `Mir:KBh`. The KirkpatrickBaez Assembly declares two Exactly1
+Mirror slots (vertical_mirror + horizontal_mirror), and the catalog note sanctions
+exactly this move: "deployments that today carry the pair as one collapsed Mirror
+Asset split it into the two mirrors that physically exist when they materialize the
+Fixture." So this scenario binds:
+
+  - VKB (Mir:KBv, Family Mirror) -> Exactly1 vertical_mirror slot
+  - HKB (Mir:KBh, Family Mirror) -> Exactly1 horizontal_mirror slot
+
+The Assembly presents no Role (presents_as: []): a KB pair focuses the beam and
+none of the catalog Roles names that function, so no role_lookup registration is
+needed (unlike the Positioner-presenting Diffractometer fixtures). The exit window
+(Wnd:Exit), defining slits, and fluorescent screens on the KB module are orthogonal
+to the focusing composition and are not bound; focal size and coating are pending
+(KB-1).
+"""
+
+# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
+
+from datetime import UTC, datetime
+from uuid import UUID, uuid4
+
+import asyncpg
+import pytest
+
+from cora.equipment.aggregates.assembly import (
+    SlotCardinality,
+    SlotName,
+    TemplateSlot,
+)
+from cora.equipment.aggregates.family import FamilyName, family_stream_id
+from cora.equipment.aggregates.fixture import SlotAssetBinding
+from cora.equipment.features.attach_asset_to_fixture import AttachAssetToFixture
+from cora.equipment.features.attach_asset_to_fixture import bind as bind_attach_asset_to_fixture
+from cora.equipment.features.define_assembly import DefineAssembly
+from cora.equipment.features.define_assembly import bind as bind_define_assembly
+from cora.equipment.features.register_fixture import RegisterFixture
+from cora.equipment.features.register_fixture import bind as bind_register_fixture
+from tests.integration._equipment_helpers import install_existing_asset_into_fresh_mount
+from tests.integration._helpers import build_postgres_deps, make_pg_profile_store
+from tests.integration.scenarios._facility_fixture import (
+    DeviceSpec,
+    facility_id_prefix,
+    install_aps_unit,
+    operator_for,
+)
+
+_NOW = datetime(2026, 5, 28, 12, 0, 0, tzinfo=UTC)
+_PRINCIPAL_ID = operator_for(__file__)
+_CORRELATION_ID = UUID("01900000-0000-7000-8000-000000cd00bb")
+
+# Facility hierarchy (scenario tag cd = CDi).
+_CDI_UNIT_ID = UUID("01900000-0000-7000-8000-000000cd0a01")
+
+# Family id (deterministic uuid5 from the name).
+_CAP_MIRROR_ID = family_stream_id(FamilyName("Mirror"))
+
+# The two KB mirror Assets (scenario-supplied ids), split from the collapsed
+# `KBMirror` device in deployments/cdi/beamline.yaml per its note (Mir:KBv +
+# Mir:KBh) and the catalog KirkpatrickBaez split-on-materialize sanction.
+_ASSET_VKB_ID = UUID("01900000-0000-7000-8000-000000cd0a11")
+_ASSET_HKB_ID = UUID("01900000-0000-7000-8000-000000cd0a21")
+
+_DEVICES = (
+    DeviceSpec("VKB", _ASSET_VKB_ID, "Mirror", _CAP_MIRROR_ID),
+    DeviceSpec("HKB", _ASSET_HKB_ID, "Mirror", _CAP_MIRROR_ID),
+)
+
+# Slot bindings (slot_name -> asset_id): the two Exactly1 mirror slots.
+_BINDINGS: tuple[tuple[str, UUID], ...] = (
+    ("vertical_mirror", _ASSET_VKB_ID),
+    ("horizontal_mirror", _ASSET_HKB_ID),
+)
+
+
+def _id_queue() -> list[UUID]:
+    return [
+        *facility_id_prefix(unit_id=_CDI_UNIT_ID, devices=_DEVICES),
+        *[uuid4() for _ in range(200)],
+    ]
+
+
+@pytest.mark.integration
+async def test_kb_pair_materializes_vertical_and_horizontal_mirrors(db_pool: asyncpg.Pool) -> None:
+    """Compose CDI's KB nanofocusing pair as the catalog KirkpatrickBaez Assembly
+    materialized by one Fixture binding a vertical (VKB) and a horizontal (HKB) Mirror
+    Asset to the two Exactly1 mirror slots. Assert the Assembly stream (no Role, flat),
+    the Fixture stream with its two-slot binding, and the per-Asset fixture
+    back-references."""
+    deps = build_postgres_deps(db_pool, now=_NOW, ids=_id_queue())
+
+    # ----- Facility install (CDI Unit + the two KB mirror Assets) -----
+    await install_aps_unit(
+        deps,
+        profile_store=make_pg_profile_store(db_pool),
+        correlation_id=_CORRELATION_ID,
+        unit_id=_CDI_UNIT_ID,
+        devices=_DEVICES,
+        unit_name="CDI",
+    )
+
+    # ----- KirkpatrickBaez Assembly (flat, presents no Role) -----
+    def _slot(name: str) -> TemplateSlot:
+        return TemplateSlot(
+            slot_name=SlotName(name),
+            required_family_ids=frozenset({_CAP_MIRROR_ID}),
+            cardinality=SlotCardinality.EXACTLY_1,
+        )
+
+    assembly_id = await bind_define_assembly(deps)(
+        DefineAssembly(
+            name="KirkpatrickBaez",
+            presents_as=frozenset(),
+            required_slots=frozenset({_slot("vertical_mirror"), _slot("horizontal_mirror")}),
+        ),
+        principal_id=_PRINCIPAL_ID,
+        correlation_id=_CORRELATION_ID,
+    )
+
+    # ----- Install each bound Asset in a lightweight Mount (register_fixture
+    #       install precondition); the helper runs on its own id pool. -----
+    for i, (slot_name, asset_id) in enumerate(_BINDINGS):
+        await install_existing_asset_into_fresh_mount(
+            db_pool, now=_NOW, asset_id=asset_id, slot_code=f"kb_{slot_name}_{i}"
+        )
+
+    # ----- Register the Fixture (binds the two mirrors across the two slots) -----
+    fixture_id = await bind_register_fixture(deps)(
+        RegisterFixture(
+            assembly_id=assembly_id,
+            slot_asset_bindings=frozenset(
+                SlotAssetBinding(slot_name=slot_name, asset_id=asset_id)
+                for slot_name, asset_id in _BINDINGS
+            ),
+        ),
+        principal_id=_PRINCIPAL_ID,
+        correlation_id=_CORRELATION_ID,
+    )
+    for _, asset_id in _BINDINGS:
+        await bind_attach_asset_to_fixture(deps)(
+            AttachAssetToFixture(asset_id=asset_id, fixture_id=fixture_id),
+            principal_id=_PRINCIPAL_ID,
+            correlation_id=_CORRELATION_ID,
+        )
+
+    # ===== Assertions =====
+
+    # Assembly stream: AssemblyDefined, two leaf slots, presents no Role, flat.
+    assembly_events, _ = await deps.event_store.load("Assembly", assembly_id)
+    assert [e.event_type for e in assembly_events] == ["AssemblyDefined"]
+    payload = assembly_events[0].payload
+    assert payload["presents_as"] == []
+    assert len(payload["required_slots"]) == 2
+    assert payload["required_sub_assemblies"] == []
+
+    # Fixture stream: FixtureRegistered binding the two mirrors to the two slots.
+    fixture_events, _ = await deps.event_store.load("Fixture", fixture_id)
+    assert [e.event_type for e in fixture_events] == ["FixtureRegistered"]
+    bindings = fixture_events[0].payload["slot_asset_bindings"]
+    assert len(bindings) == 2
+    assert {b["slot_name"] for b in bindings} == {"vertical_mirror", "horizontal_mirror"}
+    assert fixture_events[0].payload["assembly_id"] == str(assembly_id)
+
+    # Each bound Asset carries the fixture back-reference.
+    for slot_name, asset_id in _BINDINGS:
+        events, _ = await deps.event_store.load("Asset", asset_id)
+        types = [e.event_type for e in events]
+        assert "AssetAttachedToFixture" in types, f"{slot_name}: expected fixture attach"

apps/api/tests/integration/scenarios/test_ios_kb_mirror_setup.py

@@ -0,0 +1,164 @@
+"""KB focusing-pair deployment at NSLS-II IOS (Assembly + Fixture).
+
+cluster: Commissioning
+archetype: setup
+bc_primary: Equipment
+bc_touches: Equipment
+
+Materializes IOS's Kirkpatrick-Baez focusing mirror pair as a binding of the
+catalog KirkpatrickBaez Assembly, end-to-end against Postgres, from the assets
+IOS's deployment descriptor carries (`deployments/ios/beamline.yaml`, OPT-1).
+
+Unlike CDI (which collapses its KB pair into one Asset that the Fixture splits),
+IOS's descriptor already models the two mirrors as distinct Mirror Assets, so the
+Fixture binds them directly:
+
+  - KBMirror_Vertical   (Family Mirror) -> Exactly1 vertical_mirror slot
+  - KBMirror_Horizontal (Family Mirror) -> Exactly1 horizontal_mirror slot
+
+The Assembly presents no Role (presents_as: []): a KB pair focuses the beam and
+none of the catalog Roles names that function, so no role_lookup registration is
+needed. Focal spot and working distance are pending (OPT-1).
+"""
+
+# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
+
+from datetime import UTC, datetime
+from uuid import UUID, uuid4
+
+import asyncpg
+import pytest
+
+from cora.equipment.aggregates.assembly import (
+    SlotCardinality,
+    SlotName,
+    TemplateSlot,
+)
+from cora.equipment.aggregates.family import FamilyName, family_stream_id
+from cora.equipment.aggregates.fixture import SlotAssetBinding
+from cora.equipment.features.attach_asset_to_fixture import AttachAssetToFixture
+from cora.equipment.features.attach_asset_to_fixture import bind as bind_attach_asset_to_fixture
+from cora.equipment.features.define_assembly import DefineAssembly
+from cora.equipment.features.define_assembly import bind as bind_define_assembly
+from cora.equipment.features.register_fixture import RegisterFixture
+from cora.equipment.features.register_fixture import bind as bind_register_fixture
+from tests.integration._equipment_helpers import install_existing_asset_into_fresh_mount
+from tests.integration._helpers import build_postgres_deps, make_pg_profile_store
+from tests.integration.scenarios._facility_fixture import (
+    DeviceSpec,
+    facility_id_prefix,
+    install_aps_unit,
+    operator_for,
+)
+
+_NOW = datetime(2026, 5, 28, 12, 0, 0, tzinfo=UTC)
+_PRINCIPAL_ID = operator_for(__file__)
+_CORRELATION_ID = UUID("01900000-0000-7000-8000-000000e500bb")
+
+# Facility hierarchy (scenario tag e5 = ios).
+_IOS_UNIT_ID = UUID("01900000-0000-7000-8000-000000e50a01")
+
+# Family id (deterministic uuid5 from the name).
+_CAP_MIRROR_ID = family_stream_id(FamilyName("Mirror"))
+
+# The two KB mirror Assets (scenario-supplied ids), mirroring the distinct named
+# devices in deployments/ios/beamline.yaml.
+_ASSET_VKB_ID = UUID("01900000-0000-7000-8000-000000e50a11")
+_ASSET_HKB_ID = UUID("01900000-0000-7000-8000-000000e50a21")
+
+_DEVICES = (
+    DeviceSpec("KBMirror_Vertical", _ASSET_VKB_ID, "Mirror", _CAP_MIRROR_ID),
+    DeviceSpec("KBMirror_Horizontal", _ASSET_HKB_ID, "Mirror", _CAP_MIRROR_ID),
+)
+
+_BINDINGS: tuple[tuple[str, UUID], ...] = (
+    ("vertical_mirror", _ASSET_VKB_ID),
+    ("horizontal_mirror", _ASSET_HKB_ID),
+)
+
+
+def _id_queue() -> list[UUID]:
+    return [
+        *facility_id_prefix(unit_id=_IOS_UNIT_ID, devices=_DEVICES),
+        *[uuid4() for _ in range(200)],
+    ]
+
+
+@pytest.mark.integration
+async def test_kb_pair_materializes_vertical_and_horizontal_mirrors(db_pool: asyncpg.Pool) -> None:
+    """Compose IOS's KB focusing pair as the catalog KirkpatrickBaez Assembly
+    materialized by one Fixture binding the descriptor's two distinct Mirror Assets
+    (vertical + horizontal) to the two Exactly1 mirror slots. Assert the Assembly
+    stream (no Role, flat), the Fixture stream with its two-slot binding, and the
+    per-Asset fixture back-references."""
+    deps = build_postgres_deps(db_pool, now=_NOW, ids=_id_queue())
+
+    await install_aps_unit(
+        deps,
+        profile_store=make_pg_profile_store(db_pool),
+        correlation_id=_CORRELATION_ID,
+        unit_id=_IOS_UNIT_ID,
+        devices=_DEVICES,
+        unit_name="IOS",
+    )
+
+    def _slot(name: str) -> TemplateSlot:
+        return TemplateSlot(
+            slot_name=SlotName(name),
+            required_family_ids=frozenset({_CAP_MIRROR_ID}),
+            cardinality=SlotCardinality.EXACTLY_1,
+        )
+
+    assembly_id = await bind_define_assembly(deps)(
+        DefineAssembly(
+            name="KirkpatrickBaez",
+            presents_as=frozenset(),
+            required_slots=frozenset({_slot("vertical_mirror"), _slot("horizontal_mirror")}),
+        ),
+        principal_id=_PRINCIPAL_ID,
+        correlation_id=_CORRELATION_ID,
+    )
+
+    for i, (slot_name, asset_id) in enumerate(_BINDINGS):
+        await install_existing_asset_into_fresh_mount(
+            db_pool, now=_NOW, asset_id=asset_id, slot_code=f"kb_{slot_name}_{i}"
+        )
+
+    fixture_id = await bind_register_fixture(deps)(
+        RegisterFixture(
+            assembly_id=assembly_id,
+            slot_asset_bindings=frozenset(
+                SlotAssetBinding(slot_name=slot_name, asset_id=asset_id)
+                for slot_name, asset_id in _BINDINGS
+            ),
+        ),
+        principal_id=_PRINCIPAL_ID,
+        correlation_id=_CORRELATION_ID,
+    )
+    for _, asset_id in _BINDINGS:
+        await bind_attach_asset_to_fixture(deps)(
+            AttachAssetToFixture(asset_id=asset_id, fixture_id=fixture_id),
+            principal_id=_PRINCIPAL_ID,
+            correlation_id=_CORRELATION_ID,
+        )
+
+    # ===== Assertions =====
+
+    assembly_events, _ = await deps.event_store.load("Assembly", assembly_id)
+    assert [e.event_type for e in assembly_events] == ["AssemblyDefined"]
+    payload = assembly_events[0].payload
+    assert payload["presents_as"] == []
+    assert len(payload["required_slots"]) == 2
+    assert payload["required_sub_assemblies"] == []
+
+    fixture_events, _ = await deps.event_store.load("Fixture", fixture_id)
+    assert [e.event_type for e in fixture_events] == ["FixtureRegistered"]
+    bindings = fixture_events[0].payload["slot_asset_bindings"]
+    assert len(bindings) == 2
+    assert {b["slot_name"] for b in bindings} == {"vertical_mirror", "horizontal_mirror"}
+    assert fixture_events[0].payload["assembly_id"] == str(assembly_id)
+
+    for slot_name, asset_id in _BINDINGS:
+        events, _ = await deps.event_store.load("Asset", asset_id)
+        types = [e.event_type for e in events]
+        assert "AssetAttachedToFixture" in types, f"{slot_name}: expected fixture attach"