Replay paths with the consumer

Discover paths once with the propagation consumer, then re-evaluate them for moved endpoints, several frequencies or moved geometry without a new path search.

Path discovery is the expensive part of ray tracing. When only endpoints move a little, or you need the same paths at several frequencies, you can keep the discovered paths (the topology) and only re-evaluate their geometry and coefficients. The propagation consumer does this. It is also the interface other WiTwin solvers use to get paths from Channel.

Set up#

The consumer works on a CompiledScene and takes endpoints as tensor batches instead of AntennaState objects.

import torch
 
from witwin.core import AntennaState, Mesh, PhysicalMaterial, Scene, Structure
from witwin.channel.propagation import consumer
from witwin.channel.scene import compile
 
FREQUENCY_HZ = 3.0e9
vertices = torch.tensor([[2.5, -3.0, -1.0], [2.5, 3.0, -1.0], [2.5, -3.0, 2.0], [2.5, 3.0, 2.0]])
faces = torch.tensor([[0, 1, 2], [1, 3, 2]])
wall = Structure(
    Mesh(vertices, faces, recenter=False, fill_mode="surface"),
    PhysicalMaterial(eps_r=4.0, sigma_e=0.02),
    name="wall",
)
scene = Scene(
    structures=[wall],
    endpoints=[
        AntennaState(1, "tx", torch.tensor([0.0, -1.0, 0.5]), power_w=1.0),
        AntennaState(2, "rx", torch.tensor([0.0, 1.0, 0.5])),
    ],
)
compiled = compile(scene, reference_frequency_hz=FREQUENCY_HZ)
 
 
def batch(ids, positions, powers=None):
    n = len(ids)
    return consumer.EndpointBatch(
        stable_ids=torch.tensor(ids, dtype=torch.int64, device="cuda"),
        positions_m=torch.tensor(positions, dtype=torch.float32, device="cuda"),
        polarizations=torch.tensor([[0.0, 0.0, 1.0]] * n, device="cuda"),
        powers_w=None if powers is None else torch.tensor(powers, device="cuda"),
    )
 
 
sources = batch([1], [[0.0, -1.0, 0.5]], powers=[1.0])
sinks = batch([2], [[0.0, 1.0, 0.5]])

Sources need powers_w; sinks must not have it.

1. Discover#

found = consumer.evaluate(
    compiled,
    consumer.PropagationRequest(
        sources=sources,
        sinks=sinks,
        reference_frequency_hz=FREQUENCY_HZ,
        components=frozenset({"los", "reflection"}),
        max_depth=1,
        response="scalar_transport",
        topology_mode="discover",
        ad_mode="none",
    ),
)
paths = found.paths
print("rows:", paths.path_count, "pair offsets:", paths.pair_offsets.tolist())
print("delays (ns):", [round(d * 1e9, 3) for d in paths.geometry.delay_s.tolist()])
print("components:", paths.topology.component_id.tolist())
rows: 2 pair offsets: [0, 2]
delays (ns): [6.671, 17.963]
components: [0, 1]

The result has exactly K rows, one per path, with no padding. pair_offsets says which rows belong to which source-sink pair (pair p owns rows pair_offsets[p] to pair_offsets[p + 1] - 1). transport.coefficient is the complex coefficient of each row and includes the source amplitude sqrt(powers_w).

2. Prepare#

prepared = consumer.prepare_fixed_topology(paths.topology)
print("buckets:", [(b.component, b.depth, b.row_count) for b in prepared.buckets])
buckets: [('los', 0, 1), ('reflection', 1, 1)]

Preparation groups the rows by component and depth and synchronizes with the GPU once. Do it once per discovery and reuse the handle.

3. Replay for moved endpoints#

moved = consumer.reevaluate(
    compiled,
    consumer.FixedTopologyRequest(
        sources=sources,
        sinks=batch([2], [[0.0, 1.2, 0.5]]),
        reference_frequency_hz=FREQUENCY_HZ,
        topology=prepared,
        response="scalar_transport",
        ad_mode="none",
    ),
)
print("moved delays (ns):", [round(d * 1e9, 3) for d in moved.paths.geometry.delay_s.tolist()])
print("row_valid:", moved.row_valid.tolist())
moved delays (ns): [7.338, 18.221]
row_valid: [True, True]

The receiver moved 20 cm, and both paths were re-evaluated at the new position: the reflection point slid along the wall. row_valid turns False for a row whose path is no longer valid (for example, blocked). Replay never finds new paths, so rediscover when endpoints move far. Fixed-topology replay supports the "los" and "reflection" components (consumer.capabilities().fixed_topology_components).

4. Several frequencies#

wide = consumer.reevaluate(
    compiled,
    consumer.FixedTopologyRequest(
        sources=sources,
        sinks=sinks,
        reference_frequency_hz=FREQUENCY_HZ,
        topology=prepared,
        response="scalar_transport",
        ad_mode="none",
        frequency_offsets_hz=(-100e6, 0.0, 100e6),
    ),
)
print("wideband:", tuple(wide.paths.transport.coefficient_offsets.shape), wide.paths.transport.frequency_offsets_hz)
wideband: (2, 3) (-100000000.0, 0.0, 100000000.0)

Column j of coefficient_offsets is the coefficient of the same rows evaluated by the native kernels at 3.0 GHz + offset[j], rather than extrapolated from the delay. The 0.0 column equals coefficient exactly. Offsets must be at least consumer.native_frequency_resolution_hz(f) apart (256 Hz at 3 GHz). Scenes with dispersive or rough materials refuse wideband offsets.

5. Detect a stale topology#

A topology belongs to the world it was discovered in. rediscovery_required compares the scene versions:

print("rediscovery needed:", consumer.rediscovery_required(compiled, prepared))   # None
 
moved_wall = Mesh(vertices + torch.tensor([0.5, 0.0, 0.0]), wall.geometry.faces, recenter=False, fill_mode="surface")
moved_scene = scene.with_structure_geometry(wall.structure_id, moved_wall)
compiled_moved = compile(moved_scene, reference_frequency_hz=FREQUENCY_HZ)
print("after moving the wall:", consumer.rediscovery_required(compiled_moved, prepared))   # geometry_version

Moving the wall changes the scene's geometry_version. A plain replay against the new compiled scene now raises ValueError. If the path structure is still right (the same wall, slightly moved), declare it:

replay = consumer.reevaluate(
    compiled_moved,
    consumer.FixedTopologyRequest(
        sources=sources,
        sinks=sinks,
        reference_frequency_hz=FREQUENCY_HZ,
        topology=prepared,
        response="scalar_transport",
        ad_mode="none",
        world_motion="fixed_winner_replay",
    ),
)
print([round(d * 1e9, 3) for d in replay.paths.geometry.delay_s.tolist()])   # [6.671, 21.096]

The reflection now comes from the wall at x = 3.0 m. Reusing the same faces tensor matters: a new faces tensor counts as a topology change, which bumps topology_version, and a changed topology, material or assignment version always requires a new discovery.

When to rediscover#

ChangeWhat to do
Endpoints move a littlereevaluate; check row_valid
Endpoints move far, or into new roomsevaluate again
Structure geometry moves, same facesreevaluate with world_motion="fixed_winner_replay"
Structures added or removed, faces replaced, materials changedCompile and evaluate again
Many time instants of one trajectoryevaluate_time_varying