Source code for fdtdx.fdtd.symmetry

"""Automatic mirror-symmetry support for FDTD simulations.

When :attr:`fdtdx.SimulationConfig.symmetry` has a nonzero entry, :func:`fdtdx.place_objects`
reduces the domain to the symmetric half/quarter/octant before running the FDTD. This module
provides the two halves of that machinery:

* **Reduction** (used internally by ``place_objects``): clip every resolved grid slice onto the
  kept upper half along each symmetric axis, drop objects that fall entirely in the discarded
  half, and insert a PEC wall on each electric symmetry plane.
* **Unfolding** (public helpers): reconstruct full-domain field and detector arrays from the
  reduced simulation, applying the correct per-component mirror parity.

Symmetry encoding (per axis, order ``(x, y, z)``): ``0`` = none, ``-1`` = PEC (electric wall),
``+1`` = PMC (magnetic wall). The symmetry plane is the geometric center of the axis; the upper
half is kept so the plane lands at the reduced domain's min edge.

**Where the two wall types sit.** An electric plane sits *on* the reduced domain's min edge: the
tangential ``E`` components are sampled there and the odd symmetry makes them vanish, which is a
PEC face. A magnetic plane sits *half a cell below* it: sources and materials are rasterized per
cell, so the discrete problem is mirror symmetric about the tangential-``H`` node one cell out,
where tangential ``H`` vanishes — already supplied by the zero halo of the field padding, so no wall
object is created for it (see :func:`make_symmetry_walls`). The same distinction sets the mirror
index map used when unfolding (:func:`~fdtdx.core.physics.symmetry.mirror_pairs_on_plane`): across
an electric plane, components sampled on it pair as ``m ± j``; across a magnetic plane every
component mirrors one-to-one.

Mode sources and mode-overlap detectors clipped by a symmetry plane do not use the mode solver's own
symmetric solve; they mirror the reduced cross-section back to the full one, solve there and
restrict (see :func:`~fdtdx.core.physics.modes.compute_mode_symmetry_reduced`).

The parity table itself (:func:`field_component_parity`) lives in
:mod:`fdtdx.core.physics.symmetry` so that source and detector classes can use it without an
import cycle; it is re-exported here, where it is documented.
"""

from typing import Literal

import jax
import jax.numpy as jnp
from loguru import logger

from fdtdx.config import SimulationConfig
from fdtdx.core.misc import validate_symmetric_axis_cells
from fdtdx.core.null import Null
from fdtdx.core.physics.symmetry import (
    field_component_parity,
    mirror_extend_low_side,
    mirror_pairs_on_plane,
)
from fdtdx.fdtd.container import ArrayContainer, ObjectContainer
from fdtdx.objects.boundaries.bloch import BlochBoundary
from fdtdx.objects.boundaries.boundary import BaseBoundary
from fdtdx.objects.boundaries.pec import PerfectElectricConductor
from fdtdx.objects.detectors.diffractive import DiffractiveDetector
from fdtdx.objects.detectors.energy import EnergyDetector
from fdtdx.objects.detectors.field import FieldDetector
from fdtdx.objects.detectors.mode import ModeOverlapDetector
from fdtdx.objects.detectors.phasor import PhasorDetector
from fdtdx.objects.detectors.poynting_flux import PoyntingFluxDetector
from fdtdx.objects.object import SimulationObject
from fdtdx.objects.sources.mode import ModePlaneSource
from fdtdx.typing import SliceTuple3D

#: Canonical order in which the field detectors stack their components (see e.g.
#: ``PhasorDetector.update``): ``Ex, Ey, Ez, Hx, Hy, Hz``. Each entry is ``(field_type, axis)``.
_COMPONENT_NAMES = ("Ex", "Ey", "Ez", "Hx", "Hy", "Hz")
_COMPONENT_SPEC: tuple[tuple[Literal["E", "H"], int], ...] = (
    ("E", 0),
    ("E", 1),
    ("E", 2),
    ("H", 0),
    ("H", 1),
    ("H", 2),
)

_AXIS_NAMES = ("x", "y", "z")


# ──────────────────────────────────────────────────────────────────────────────
# Reduction (used by place_objects)
# ──────────────────────────────────────────────────────────────────────────────


def reduce_resolved_slices(
    resolved_slices: dict[str, SliceTuple3D],
    object_map: dict[str, SimulationObject],
    config: SimulationConfig,
    volume_name: str,
) -> tuple[dict[str, SliceTuple3D], dict[str, SliceTuple3D], set[str], tuple[int, int, int]]:
    """Clip full-domain grid slices onto the symmetric (kept upper) half along each axis.

    Args:
        resolved_slices (dict[str, SliceTuple3D]): Full-domain grid slices, keyed by object name.
        object_map (dict[str, SimulationObject]): Object name → object, used for type-aware warnings.
        config (SimulationConfig): Simulation config; ``config.symmetry`` selects the axes.
        volume_name (str): Name of the simulation volume object.

    Returns:
        tuple: ``(new_slices, unreduced_slices, dropped_names, reduced_volume_shape)`` where
        ``new_slices`` excludes the dropped objects, ``unreduced_slices`` holds each surviving
        object's *unclipped* slice shifted into the reduced coordinate frame (negative start on
        axes where the object crossed the symmetry plane), ``dropped_names`` lists objects entirely
        in the discarded half, and ``reduced_volume_shape`` is the reduced volume grid shape
        ``(Nx', Ny', Nz')``.
    """
    symmetry = config.symmetry
    vol_slice = resolved_slices[volume_name]

    # Per-axis absolute index of the symmetry plane (= shift applied to map full → reduced),
    # and the reduced volume slice.
    mid_abs = [0, 0, 0]
    new_vol: list[tuple[int, int]] = []
    for a in range(3):
        vs0, vs1 = vol_slice[a]
        n = vs1 - vs0
        if symmetry[a] != 0:
            validate_symmetric_axis_cells(n, _AXIS_NAMES[a], subject="simulation volume")
            mid_abs[a] = vs0 + n // 2
            new_vol.append((0, vs1 - mid_abs[a]))
        else:
            mid_abs[a] = vs0
            new_vol.append((vs0, vs1))

    reduced_volume_shape = (
        new_vol[0][1] - new_vol[0][0],
        new_vol[1][1] - new_vol[1][0],
        new_vol[2][1] - new_vol[2][0],
    )

    new_slices: dict[str, SliceTuple3D] = {}
    unreduced_slices: dict[str, SliceTuple3D] = {}
    dropped: set[str] = set()
    for name, sl in resolved_slices.items():
        if name == volume_name:
            new_slices[name] = (new_vol[0], new_vol[1], new_vol[2])
            shifted = [(sl[a][0] - mid_abs[a], sl[a][1] - mid_abs[a]) if symmetry[a] != 0 else sl[a] for a in range(3)]
            unreduced_slices[name] = (shifted[0], shifted[1], shifted[2])
            continue

        obj = object_map[name]
        clipped: list[tuple[int, int]] = []
        unclipped: list[tuple[int, int]] = []
        drop = False
        for a in range(3):
            s0, s1 = sl[a]
            if symmetry[a] != 0:
                m = mid_abs[a]
                ns0 = max(s0, m) - m
                ns1 = min(s1, vol_slice[a][1]) - m
                if ns1 <= ns0:
                    drop = True
                elif s0 < m < s1 and (s0 + s1) != 2 * m:
                    logger.warning(
                        f"Object '{name}' straddles the {_AXIS_NAMES[a]}-symmetry plane "
                        f"asymmetrically (grid [{s0}, {s1}] about center {m}); the reduced "
                        f"simulation assumes mirror symmetry, so the discarded half is replaced by "
                        f"the mirror of the kept half."
                    )
                clipped.append((ns0, ns1))
                unclipped.append((s0 - m, s1 - m))
            else:
                clipped.append((s0, s1))
                unclipped.append((s0, s1))

        if drop:
            dropped.add(name)
            if isinstance(obj, BlochBoundary):
                logger.warning(
                    f"Dropped periodic/Bloch boundary '{name}' on a symmetry axis. Periodic "
                    f"boundaries and mirror symmetry are mutually exclusive on the same axis; use "
                    f"PML (or nothing) on the far side of a symmetric axis."
                )
            elif not isinstance(obj, BaseBoundary):
                logger.warning(
                    f"Object '{name}' lies entirely in the discarded (lower) half of a symmetric "
                    f"axis and was removed. Place it in the upper half or straddling the symmetry "
                    f"plane (the geometric center) so it survives the reduction."
                )
            continue

        new_slices[name] = (clipped[0], clipped[1], clipped[2])
        unreduced_slices[name] = (unclipped[0], unclipped[1], unclipped[2])

        if isinstance(obj, BlochBoundary) and symmetry[obj.axis] != 0:
            # The min-side boundary of a symmetric axis is dropped above; a far-side one survives.
            # Its wrap padding must not reach the min-side halo, which belongs to the mirror (see
            # pad_fields_for_boundaries) - say so rather than let the axis look fully periodic.
            logger.warning(
                f"Periodic/Bloch boundary '{name}' sits on the {_AXIS_NAMES[obj.axis]}-symmetry "
                f"axis. Periodic boundaries and mirror symmetry are mutually exclusive on the same "
                f"axis: the halo behind the symmetry plane is set by the mirror, not by wrapping to "
                f"this boundary. Use PML (or nothing) on the far side of a symmetric axis."
            )

    return new_slices, unreduced_slices, dropped, reduced_volume_shape


def make_symmetry_walls(
    config: SimulationConfig,
    reduced_volume_shape: tuple[int, int, int],
    key: jax.Array,
    existing_names: set[str],
) -> list[SimulationObject]:
    """Create the PEC wall objects sitting on each *electric* symmetry plane (reduced min edge).

    Only an electric (``-1``) plane gets a wall object. It sits on the tangential-``E`` node row at
    the reduced domain's min edge, where the odd symmetry does make those components vanish, so the
    condition is exactly a PEC face.

    A magnetic (``+1``) plane gets **no wall object**. Its plane is half a cell below the reduced
    domain: sources and materials are rasterized per cell, so the discrete problem is mirror
    symmetric about the tangential-``H`` node one cell below the min edge, where tangential ``H``
    vanishes — which is precisely what the zero halo of the field padding already supplies. Placing a
    :class:`~fdtdx.PerfectMagneticConductor` there instead (zeroing tangential ``H`` at the first
    cell, half a cell inside the domain) or filling that halo with ``-H_t[0]`` forces the mirror onto
    the cell edge and displaces the plane by half a cell: a consistent discretization of the *wrong*
    problem, which converges cleanly at first order and so looks plausible while being wrong.

    Args:
        config (SimulationConfig): Simulation config; ``config.symmetry`` selects the axes/types.
        reduced_volume_shape (tuple[int, int, int]): Reduced volume grid shape.
        key (jax.Array): JAX key for placement.
        existing_names (set[str]): Names already in use, so wall names can be made unique.

    Returns:
        list[SimulationObject]: PEC boundary objects for the electric symmetry planes, already
        placed on the reduced grid. Empty if every symmetric axis is magnetic.
    """
    walls: list[SimulationObject] = []
    used = set(existing_names)
    for a in range(3):
        if config.symmetry[a] != -1:
            continue
        name = f"_sym_wall_{_AXIS_NAMES[a]}"
        suffix = 0
        while name in used:
            suffix += 1
            name = f"_sym_wall_{_AXIS_NAMES[a]}_{suffix}"
        used.add(name)

        partial_grid_shape: tuple[int | None, int | None, int | None] = (
            1 if a == 0 else None,
            1 if a == 1 else None,
            1 if a == 2 else None,
        )
        wall = PerfectElectricConductor(
            name=name,
            axis=a,
            partial_grid_shape=partial_grid_shape,
            direction="-",
        )

        grid_slice: list[tuple[int, int]] = [(0, reduced_volume_shape[b]) for b in range(3)]
        grid_slice[a] = (0, 1)

        key, subkey = jax.random.split(key)
        wall = wall.place_on_grid(
            grid_slice_tuple=(grid_slice[0], grid_slice[1], grid_slice[2]),
            config=config,
            key=subkey,
        )
        wall = wall.aset("_is_symmetry_wall", True)
        walls.append(wall)
    return walls


def apply_mode_symmetry(
    placed_objects: list[SimulationObject],
    config: SimulationConfig,
) -> list[SimulationObject]:
    """Warn about explicit mode-solver symmetry tuples that ``config.symmetry`` supersedes.

    Mode objects clipped by a symmetry plane no longer ask the mode solver for a symmetric solve:
    :func:`~fdtdx.core.physics.modes.compute_mode_symmetry_reduced` mirrors the reduced
    cross-section back to the full one, solves there and restricts the result, which reproduces the
    mode the unreduced simulation would use (the solver's own symmetric solve does not, because
    FDTDX hands it one cell-centred permittivity array per component while the solver samples
    materials on its staggered grid). A user-set ``symmetry`` tuple is therefore inert here; say so
    rather than silently ignoring it.

    Args:
        placed_objects (list[SimulationObject]): Objects already placed on the reduced grid.
        config (SimulationConfig): Simulation config.

    Returns:
        list[SimulationObject]: The objects, unchanged.
    """
    del config
    for obj in placed_objects:
        if not isinstance(obj, (ModePlaneSource, ModeOverlapDetector)) or obj.symmetry == (0, 0):
            continue
        shape = obj.grid_shape
        if 1 not in shape:
            continue
        propagation_axis = shape.index(1)
        if obj.symmetry_mirror_axes(exclude_axis=propagation_axis):
            logger.warning(
                f"Mode object '{obj.name}' sets symmetry={obj.symmetry} for the mode solver, but "
                f"config.symmetry already reduces its cross-section. The explicit value is ignored: the "
                f"mode is solved on the mirrored full cross-section and restricted to the kept half. "
                f"Remove it, or drop config.symmetry on those axes and keep your own PEC/PMC boundary."
            )
    return placed_objects


# ──────────────────────────────────────────────────────────────────────────────
# Unfolding (public helpers)
# ──────────────────────────────────────────────────────────────────────────────


def _check_has_symmetry(symmetry: tuple[int, int, int]) -> None:
    if not any(s != 0 for s in symmetry):
        raise ValueError(
            "Nothing to unfold: this simulation has no symmetry (config.symmetry is (0, 0, 0)). "
            "The unfold helpers are only meaningful for symmetry-reduced simulations."
        )


[docs] def unfold_fields( field: jax.Array, symmetry: tuple[int, int, int], field_type: Literal["E", "H"], ) -> jax.Array: """Reconstruct a full-domain ``(3, Nx, Ny, Nz)`` field array from the reduced field. Mirrors the field across each symmetric axis with the correct per-component parity and concatenates the mirror image in front of the kept half, so the symmetry plane ends up at the center of the reconstructed array. The index map is chosen per axis from its wall type (see :func:`~fdtdx.core.physics.symmetry.mirror_pairs_on_plane`). Across an **electric** plane, which sits on the reduced domain's min edge, a component sampled *on* it (tangential ``E``, normal ``H``) has its first sample as its own mirror, so only the remaining ones produce images, while the half-cell-offset components mirror one-to-one. Across a **magnetic** plane, which sits half a cell below the min edge, every component mirrors one-to-one. In the on-plane case one reconstructed cell per axis — the outermost of the mirrored half, whose partner lies outside the kept domain — is filled by repeating its neighbour; it is the far-boundary cell of the reconstructed domain (inside the absorbing layer in a typical setup). Args: field (jax.Array): Reduced field, shape ``(3, Nx, Ny, Nz)`` (component axis first). symmetry (tuple[int, int, int]): Per-axis symmetry ``(x, y, z)``; see module docstring. field_type (Literal["E", "H"]): Whether ``field`` is the electric or magnetic field. Returns: jax.Array: Full-domain field; each symmetric axis is doubled in size. """ if field_type not in ("E", "H"): raise ValueError(f"field_type must be 'E' or 'H', got {field_type!r}") _check_has_symmetry(symmetry) arr = field for a in range(3): if symmetry[a] == 0: continue components = [] for c in range(3): single = arr[c : c + 1] low = mirror_extend_low_side( single, axis=a + 1, parity=field_component_parity(field_type, c, a, symmetry[a]), on_plane=mirror_pairs_on_plane(field_type, c, a, symmetry[a]), ) components.append(jnp.concatenate([low, single], axis=a + 1)) arr = jnp.concatenate(components, axis=0) return arr
[docs] def unfold_source_mode( source, config: SimulationConfig, ) -> tuple[jax.Array, jax.Array]: """Reconstruct the full-domain ``(E, H)`` mode profile a mode source injects. A :class:`ModePlaneSource` solves and stores its mode on the *reduced* cross-section (``source._E`` / ``source._H``, real, shape ``(3, Nx, Ny, Nz)`` with a singleton on the propagation axis). This mirrors that profile back to the full transverse cross-section with the correct per-component parity. Only the two transverse axes are unfolded — the propagation axis is never a symmetry plane for a mode source (its mode plane is one cell thick there). Convenience wrapper so callers don't reach into the private ``_E`` / ``_H`` state; for the fields *recorded during the run*, prefer a detector on the source plane plus :func:`unfold_detector_states`. Args: source: A placed mode source whose ``apply`` has already run (so ``_E`` / ``_H`` are populated), e.g. a :class:`ModePlaneSource`. Run :func:`fdtdx.apply_params` first. config (SimulationConfig): Simulation config; must have nonzero ``symmetry`` on at least one axis transverse to the source's propagation axis. Returns: tuple[jax.Array, jax.Array]: Full-domain ``(E, H)`` mode profiles, shape ``(3, Nx, Ny, Nz)``. Raises: ValueError: If the mode has not been computed yet, the source is not a plane, or there is no transverse symmetry to unfold. """ mode_E = getattr(source, "_E", None) mode_H = getattr(source, "_H", None) if mode_E is None or mode_H is None or isinstance(mode_E, Null) or isinstance(mode_H, Null): raise ValueError( f"Source '{getattr(source, 'name', source)}' has no computed mode profile yet. Run " "fdtdx.apply_params (which invokes the source's mode solver) before unfolding." ) shape = source.grid_shape if 1 not in shape: raise ValueError( f"unfold_source_mode expects a plane source (one grid dimension of size 1); got grid shape {shape}." ) propagation_axis = shape.index(1) # Only the transverse axes are symmetry planes for a mode source; mask the propagation axis. transverse_symmetry: tuple[int, int, int] = ( 0 if propagation_axis == 0 else config.symmetry[0], 0 if propagation_axis == 1 else config.symmetry[1], 0 if propagation_axis == 2 else config.symmetry[2], ) if config.symmetry[propagation_axis] != 0: logger.warning( f"config.symmetry is nonzero on the source's propagation axis " f"{_AXIS_NAMES[propagation_axis]}; ignoring it when unfolding the mode profile (a mode " f"plane is never split along its propagation axis)." ) if not any(s != 0 for s in transverse_symmetry): raise ValueError( "No transverse symmetry to unfold for this source plane: config.symmetry is zero on both " "axes transverse to the propagation axis." ) return unfold_fields(mode_E, transverse_symmetry, "E"), unfold_fields(mode_H, transverse_symmetry, "H")
[docs] def unfold_array( arr: jax.Array, symmetry: tuple[int, int, int], spatial_axes: tuple[int, int, int], signs: dict[int, jax.Array] | None = None, on_plane_axes: tuple[int, ...] = (), ) -> jax.Array: """Mirror-and-concatenate a spatial array along each symmetric axis. Generic building block used by :func:`unfold_detector_states`. For every axis ``a`` with ``symmetry[a] != 0`` the array is mirrored along its corresponding array axis ``spatial_axes[a]`` (optionally multiplied by a broadcastable per-component sign), and the mirror image is concatenated in front of the original. Args: arr (jax.Array): Array to unfold. symmetry (tuple[int, int, int]): Per-axis symmetry ``(x, y, z)``; ``0`` axes are skipped. spatial_axes (tuple[int, int, int]): For each physical axis, the corresponding array axis. signs (dict[int, jax.Array] | None): Optional mapping physical-axis → broadcastable sign array applied to the mirror image (defaults to ``+1``). on_plane_axes (tuple[int, ...]): Physical axes whose samples lie *on* the symmetry plane rather than half a cell off it. There the first sample is its own mirror, so it is not duplicated and the outermost reconstructed sample repeats its neighbour. Only an electric plane can be on-plane at all — a magnetic one sits half a cell below the reduced domain, so everything mirrors one-to-one across it; see :func:`_colocated_on_plane_axes`. Returns: jax.Array: The unfolded array. """ _check_has_symmetry(symmetry) for a in range(3): if symmetry[a] == 0: continue ax = spatial_axes[a] sign = 1.0 if signs is None else signs.get(a, 1.0) if a in on_plane_axes: mirror = mirror_extend_low_side(arr, axis=ax, parity=1, on_plane=True) * sign else: mirror = jnp.flip(arr, axis=ax) * sign arr = jnp.concatenate([mirror, arr], axis=ax) return arr
def _colocated_on_plane_axes(detector, touched: tuple[int, int, int]) -> tuple[int, ...]: """Which physical axes a detector's stored samples sit on, for the mirror index map. With ``exact_interpolation`` the six components are co-located at the ``E_z`` Yee point ``(i, j, k+1/2)``: integer along x and y and half a cell off along z. An integer position lands on an **electric** symmetry plane, which sits on the reduced domain's min edge, so those samples pair as ``m ± j``. A **magnetic** plane sits half a cell below the min edge, so nothing is on-plane there and every sample mirrors one-to-one (see :func:`~fdtdx.core.physics.symmetry.mirror_pairs_on_plane`). Without ``exact_interpolation`` the raw staggered components are stored, each with its own offset, and no single index map fits; those fall back to the plain flip. Args: detector: The detector whose stored output is being unfolded. touched (tuple[int, int, int]): Per-axis wall type of the symmetry planes this detector was clipped by (``0`` where it was not). """ if not getattr(detector, "exact_interpolation", False): return () return tuple(a for a in (0, 1) if touched[a] == -1) def _stored_component_spec(components) -> list[tuple[Literal["E", "H"], int]]: """Stored ``(field_type, component_axis)`` order for a Field/Phasor detector's components.""" return [_COMPONENT_SPEC[i] for i, name in enumerate(_COMPONENT_NAMES) if name in components] def _component_signs( spec: list[tuple[Literal["E", "H"], int]], touched: tuple[int, int, int], ndim: int, component_axis: int, ) -> dict[int, jax.Array]: """Per-axis, per-component mirror signs broadcastable against a detector array.""" out: dict[int, jax.Array] = {} for a in range(3): if touched[a] == 0: continue vals = [field_component_parity(ft, ca, a, touched[a]) for (ft, ca) in spec] shape = [1] * ndim shape[component_axis] = len(vals) out[a] = jnp.asarray(vals).reshape(shape) return out def _reduce_factor( parities: list[list[int]], ndim: int, component_axis: int, mean: bool, ) -> jax.Array: """Per-component scale factor that turns a half-domain reduction into the full-domain one. For a linear reduction over the detector region, the full-domain value relates to the stored half-domain value by ``∏_axes (1 + parity)`` for a *sum* and ``∏_axes (1 + parity) / 2`` for a *mean*: an even component doubles (sum) or is unchanged (mean), an odd component vanishes. Args: parities (list[list[int]]): ``parities[c]`` is the per-touched-axis parity list of component ``c``. ndim (int): Rank of the stored array (so the factor broadcasts against it). component_axis (int): Array axis holding the components. mean (bool): True if the detector reduces by averaging, False if by summing. Returns: jax.Array: Factor of shape ``[1, ..., num_components, ..., 1]`` (component axis populated). """ factors = [] for per_axis in parities: f = 1.0 for p in per_axis: f *= (1 + p) / 2 if mean else (1 + p) factors.append(f) shape = [1] * ndim shape[component_axis] = len(factors) return jnp.asarray(factors).reshape(shape) def _poynting_parity(component: int, axis: int, wall: int) -> int: """Mirror parity of the Poynting component ``S_component = (E x H)_component``. ``S_i`` involves the two field components transverse to ``i``, so its parity is the product of the corresponding ``E`` and ``H`` parities. """ j, k = (x for x in range(3) if x != component) return field_component_parity("E", j, axis, wall) * field_component_parity("H", k, axis, wall) def _unfold_poynting(detector, state: dict[str, jax.Array], touched: tuple[int, int, int]): """Unfold a PoyntingFluxDetector (scalar/vector, summed/spatial) via per-component flux parity.""" arr = state["poynting_flux"] on_plane = _colocated_on_plane_axes(detector, touched) components = (0, 1, 2) if detector.keep_all_components else (detector.propagation_axis,) touched_axes = [a for a in range(3) if touched[a] != 0] if detector.reduce_volume: # summed flux -> per-component factor ∏ (1 + parity) parities = [[_poynting_parity(i, a, touched[a]) for a in touched_axes] for i in components] if detector.keep_all_components: factor = _reduce_factor(parities, ndim=arr.ndim, component_axis=1, mean=False) # (T, 3) return {"poynting_flux": arr * factor} scalar = 1.0 for p in parities[0]: scalar *= 1 + p return {"poynting_flux": arr * scalar} # Spatial flux density -> mirror with the flux-component parity. if detector.keep_all_components: # (T, 3, nx, ny, nz) signs = { a: jnp.asarray([_poynting_parity(i, a, touched[a]) for i in components]).reshape((1, 3, 1, 1, 1)) for a in touched_axes } return {"poynting_flux": unfold_array(arr, touched, (2, 3, 4), signs, on_plane)} # (T, nx, ny, nz): single component along the propagation axis. p = detector.propagation_axis signs = {a: jnp.asarray(float(_poynting_parity(p, a, touched[a]))) for a in touched_axes} return {"poynting_flux": unfold_array(arr, touched, (1, 2, 3), signs, on_plane)} def _unfold_energy_slices( state: dict[str, jax.Array], touched: tuple[int, int, int], on_plane_axes: tuple[int, ...] = (), ): """Unfold an EnergyDetector(as_slices=True): mirror each 2D plane along its in-plane symmetric axes. Energy density is even, so the parity is ``+1``; the collapsed-axis mean is already the full-domain value (energy is symmetric about the plane).""" # Each plane stores (T, axis_a, axis_b); map those two physical axes to array axes 1 and 2. planes = { "XY Plane": (0, 1), "XZ Plane": (0, 2), "YZ Plane": (1, 2), } out: dict[str, jax.Array] = {} for key, phys_axes in planes.items(): arr = state[key] sub = [0, 0, 0] spatial_axes = [0, 0, 0] for arr_axis, phys in enumerate(phys_axes, start=1): sub[phys] = touched[phys] spatial_axes[phys] = arr_axis if any(s != 0 for s in sub): out[key] = unfold_array( arr, (sub[0], sub[1], sub[2]), (spatial_axes[0], spatial_axes[1], spatial_axes[2]), on_plane_axes=tuple(a for a in on_plane_axes if a in phys_axes), ) else: out[key] = arr return out def _unfold_one_detector( detector, state: dict[str, jax.Array], touched: tuple[int, int, int], count: int, ) -> dict[str, jax.Array]: """Unfold a single detector's recorded state to the full domain. See module docstring. Everything except :class:`DiffractiveDetector` is recovered purely from the stored reduced output plus the parity table: spatial outputs are mirrored per component, and ``reduce_volume`` sums/means are rescaled per component (even doubles/keeps, odd vanishes). """ # DiffractiveDetector first: it subclasses Detector (not PhasorDetector) and stores per-order # efficiencies from an FFT over the plane. Halving the domain changes the diffraction-order # basis, so the full-domain orders are not a function of the reduced-domain ones — unrecoverable # post-hoc from the stored output. if isinstance(detector, DiffractiveDetector): raise NotImplementedError( "unfold_detector_states cannot unfold a DiffractiveDetector: its diffraction-order basis " "depends on the (reduced) domain size, so the full-domain orders are not recoverable from " "the stored efficiencies. Unfold the fields with unfold_fields and recompute the orders." ) # PhasorDetector (and ModeOverlapDetector, which subclasses it). Note: for ModeOverlapDetector # the mode-overlap S-parameter is already correct on the reduced domain (source and detector # share the reduced plane); unfolding only affects the raw stored phasor field. if isinstance(detector, PhasorDetector): spec = _stored_component_spec(detector.components) if detector.reduce_volume: # (1, num_freqs, num_components); mean reduction arr = state["phasor"] parities = [ [field_component_parity(ft, ca, a, touched[a]) for a in range(3) if touched[a]] for ft, ca in spec ] factor = _reduce_factor(parities, ndim=arr.ndim, component_axis=2, mean=True) return {"phasor": arr * factor.astype(arr.dtype)} arr = state["phasor"] # (1, num_freqs, num_components, nx, ny, nz) signs = _component_signs(spec, touched, arr.ndim, component_axis=2) return {"phasor": unfold_array(arr, touched, (3, 4, 5), signs, _colocated_on_plane_axes(detector, touched))} if isinstance(detector, FieldDetector): spec = _stored_component_spec(detector.components) if detector.reduce_volume: # (T, num_components); mean reduction arr = state["fields"] parities = [ [field_component_parity(ft, ca, a, touched[a]) for a in range(3) if touched[a]] for ft, ca in spec ] factor = _reduce_factor(parities, ndim=arr.ndim, component_axis=1, mean=True) return {"fields": arr * factor.astype(arr.dtype)} arr = state["fields"] # (T, num_components, nx, ny, nz) signs = _component_signs(spec, touched, arr.ndim, component_axis=1) return {"fields": unfold_array(arr, touched, (2, 3, 4), signs, _colocated_on_plane_axes(detector, touched))} if isinstance(detector, EnergyDetector): if detector.as_slices: return _unfold_energy_slices(state, touched, _colocated_on_plane_axes(detector, touched)) if detector.reduce_volume: # summed energy; energy density is even -> x2 per plane return {"energy": state["energy"] * (2**count)} arr = state["energy"] # (T, nx, ny, nz) return { "energy": unfold_array(arr, touched, (1, 2, 3), on_plane_axes=_colocated_on_plane_axes(detector, touched)) } if isinstance(detector, PoyntingFluxDetector): return _unfold_poynting(detector, state, touched) raise NotImplementedError( f"unfold_detector_states does not know how to unfold detector type " f"{type(detector).__name__!r}. Unfold the fields with unfold_fields instead." )
[docs] def unfold_detector_states( arrays: ArrayContainer, objects: ObjectContainer, config: SimulationConfig, ) -> ArrayContainer: """Reconstruct full-domain detector states from a symmetry-reduced simulation. This is a pure post-processing step: it transforms each detector's stored reduced-domain output into the full-domain result using the parity table, with no work added to the FDTD time loop. Each detector that sits on one or more symmetry planes is unfolded per its type — spatial outputs are mirrored with the correct per-component parity; ``reduce_volume`` sums/means are rescaled per component (even components double/keep, odd components vanish); ``as_slices`` energy planes are mirrored along their in-plane symmetric axes. Detectors that do not touch any symmetry plane are returned unchanged. All detector types are supported except :class:`DiffractiveDetector`, whose diffraction-order basis depends on the domain size and so cannot be recovered from the stored efficiencies; for that, unfold the fields with :func:`unfold_fields` and recompute. Args: arrays (ArrayContainer): Arrays returned by the reduced simulation. objects (ObjectContainer): The placed (reduced) objects. config (SimulationConfig): Simulation config (must have nonzero ``symmetry``). Returns: ArrayContainer: A copy of ``arrays`` with ``detector_states`` reconstructed to full domain. Raises: ValueError: If ``config.symmetry`` is ``(0, 0, 0)`` (nothing to unfold). NotImplementedError: For a :class:`DiffractiveDetector` (see above). """ _check_has_symmetry(config.symmetry) detector_by_name = {d.name: d for d in objects.detectors} new_states: dict[str, dict[str, jax.Array]] = {} for name, state in arrays.detector_states.items(): detector = detector_by_name.get(name) if detector is None: new_states[name] = state continue # Symmetric axes on which the detector was clipped because it crossed the symmetry plane. # A detector that merely begins at the plane but lies entirely inside the kept half was not # clipped, so its stored output is already the full-domain result on that axis. touched: tuple[int, int, int] = ( config.symmetry[0] if detector.straddles_symmetry_plane(0) else 0, config.symmetry[1] if detector.straddles_symmetry_plane(1) else 0, config.symmetry[2] if detector.straddles_symmetry_plane(2) else 0, ) count = sum(1 for a in range(3) if touched[a] != 0) if count == 0: new_states[name] = state continue new_states[name] = _unfold_one_detector(detector, state, touched, count) return arrays.aset("detector_states", new_states)