class Temperature(TensorizedAbsoluteState):
"""
Continuous per-object temperature (°C).
Temperature owns every write into its own tensors. Each step, `_update_values` launches,
in order:
1. `_incoming_heat_kernel` — gathers heat from active HeatSourceOrSink entries (whose
activation gates were computed earlier this step; HeatSourceOrSink is a dependency)
into the private INCOMING_HEAT_RATE scratch, recording who-heats-whom in
INFLUENCE_MASK (served to HeatSourceOrSink.affects_obj via is_influenced_by()).
2. `_temperature_decay_kernel` — integrates ambient decay plus the gathered rate, then
zeroes the scratch for the next step.
3. `_self_heating_clamp_kernel` — objects on fire (heat sources with requires_on_fire)
are held at their fire temperature.
Note the deliberate one-step lag in the fire feedback loop: OnFire (which depends on
Temperature) flips True the step temperature crosses ignition; the object's
HeatSourceOrSink gate reads OnFire's previous-step values, so the fire starts heating
(including the self-clamp) on the following step.
"""
# (S, N) float32 — private per-step rate accumulator; written by _incoming_heat_kernel and
# consumed + zeroed by _temperature_decay_kernel within the same _update_values pass.
INCOMING_HEAT_RATE = None # wp.array (S, N) float32 — GPU-only scratch (single source of truth)
# (S_hss, N_hss, N_temp) — which heat source influenced which object this step. The GPU
# uint8 wp.array is the single source of truth; the CPU mirror keeps a torch bool tensor
# (for .item() reads in is_influenced_by) plus a wp view for the graph-safe wp.copy,
# mirroring how the base class keeps VALUES_CPU + VALUES_CPU_WP.
INFLUENCE_MASK = None # wp.array (S_hss, N_hss, N_temp) uint8 — GPU
INFLUENCE_MASK_CPU = None # torch bool (S_hss, N_hss, N_temp), pinned — CPU mirror
INFLUENCE_MASK_CPU_WP = None # wp.array uint8 view of INFLUENCE_MASK_CPU
# Index maps into other states' N dimensions. Built in initialize_view — safe because all
# referenced states (HeatSourceOrSink, AABB, Inside) initialize before Temperature in
# dependency order.
_hss_self_temp_idx = None # wp.array (N_hss,) int32 — HeatSourceOrSink N → Temperature N
_hss_self_inside_idx = None # wp.array (N_hss,) int32 — HeatSourceOrSink N → Inside N
_temp_to_aabb_idx = None # wp.array (N_temp,) int32 — Temperature N → AABB N
_temp_to_inside_idx = None # wp.array (N_temp,) int32 — Temperature N → Inside N
# CSR table giving each (scene, target) its collision-geometry links, so the point-source
# proximity test can measure to the target's actual mesh rather than to its AABB. For scene s
# and Temperature index n the links are
# _target_link_indices[_target_link_offsets[s * N_temp + n] : _target_link_offsets[... + 1]]
# indexing RigidBodyViewAPI.POSE_MATRICES / LINK_MESH_IDS. Targets with no collision geometry
# (e.g. cloth, which is absent from RigidBodyViewAPI) get an empty range.
_target_link_offsets = None # wp.array (S_temp * N_temp + 1,) int32
_target_link_indices = None # wp.array (K,) int32
# Placeholder wp.array to satisfy the kernel signature when Inside tracks no objects.
_placeholder_inside = None # wp.array (1, 1, 1) uint8
@classmethod
def get_dependencies(cls):
deps = super().get_dependencies()
deps.add(AABB)
return deps
@classmethod
def get_optional_dependencies(cls):
deps = super().get_optional_dependencies()
# Optional because objects without HeatSourceOrSink (e.g. cookable food items) are still
# eligible for a Temperature state — but the topo sort still places HeatSourceOrSink
# before Temperature, so _incoming_heat_kernel reads this step's freshly-computed
# activation gates.
deps.add(HeatSourceOrSink)
return deps
@classmethod
def global_initialize(cls):
super().global_initialize()
cls.INCOMING_HEAT_RATE = None
cls.INFLUENCE_MASK = None
cls.INFLUENCE_MASK_CPU = None
cls.INFLUENCE_MASK_CPU_WP = None
cls._hss_self_temp_idx = None
cls._hss_self_inside_idx = None
cls._temp_to_aabb_idx = None
cls._temp_to_inside_idx = None
cls._placeholder_inside = wp.zeros((1, 1, 1), dtype=wp.uint8, device="cuda")
@classmethod
def initialize_view(cls):
# Snapshot which relative paths existed before the rebuild
prev_rel_paths = set(cls.OBJ_IDXS.keys()) if cls.OBJ_IDXS is not None else set()
# Base class rebuilds OBJ_IDXS, IDX_OBJS, VALUES (with value carry-over for survivors)
super().initialize_view()
# Initialize new VALUE slots (not carried over) to DEFAULT_TEMPERATURE
for rel_path, obj_idx in cls.OBJ_IDXS.items():
if rel_path not in prev_rel_paths:
for s_idx in range(len(cls.IDX_OBJS)):
if cls.IDX_OBJS[s_idx][obj_idx] is not None:
cls.VALUES[s_idx, obj_idx] = m.DEFAULT_TEMPERATURE
cls.VALUES_CPU[s_idx, obj_idx] = m.DEFAULT_TEMPERATURE
# Allocate the per-step heat-rate scratch buffer. No carry-over: a partial step from
# the previous configuration would be applied to the wrong indices.
if cls.VALUES.numel() > 0:
cls.INCOMING_HEAT_RATE = wp.zeros(tuple(cls.VALUES.shape), dtype=wp.float32, device="cuda")
else:
cls.INCOMING_HEAT_RATE = None
# Rebuild the maps into the states the heat kernels read.
cls._rebuild_heat_source_maps()
@classmethod
def _rebuild_heat_source_maps(cls):
"""
Rebuild the index maps into HeatSourceOrSink / AABB / Inside plus INFLUENCE_MASK.
Called from initialize_view — safe because those states are (transitive) dependencies
of Temperature, so their views are rebuilt before this one.
"""
N_temp = len(cls.OBJ_IDXS) if cls.OBJ_IDXS is not None else 0
hss_obj_idxs = HeatSourceOrSink.OBJ_IDXS or {}
N_hss = len(hss_obj_idxs)
S_hss = len(HeatSourceOrSink.IDX_OBJS) if HeatSourceOrSink.IDX_OBJS is not None else 0
inside_map = Inside.OBJ_IDXS or {}
aabb_map = AABB.OBJ_IDXS or {}
if N_temp == 0 or N_hss == 0 or S_hss == 0:
cls._hss_self_temp_idx = None
cls._hss_self_inside_idx = None
cls._temp_to_aabb_idx = None
cls._temp_to_inside_idx = None
cls._target_link_offsets = None
cls._target_link_indices = None
cls.INFLUENCE_MASK = None
cls.INFLUENCE_MASK_CPU = None
cls.INFLUENCE_MASK_CPU_WP = None
return
create_tensor_from_list = lazy.isaacsim.core.utils.warp.tensor.create_tensor_from_list
hss_self_temp_idx = th.full((N_hss,), -1, dtype=th.int32)
hss_self_inside_idx = th.full((N_hss,), -1, dtype=th.int32)
for rel_path, h in hss_obj_idxs.items():
hss_self_temp_idx[h] = cls.OBJ_IDXS.get(rel_path, -1)
hss_self_inside_idx[h] = inside_map.get(rel_path, -1)
cls._hss_self_temp_idx = create_tensor_from_list(hss_self_temp_idx, "int32", device="cuda")
cls._hss_self_inside_idx = create_tensor_from_list(hss_self_inside_idx, "int32", device="cuda")
temp_to_aabb = th.full((N_temp,), -1, dtype=th.int32)
temp_to_inside = th.full((N_temp,), -1, dtype=th.int32)
for rel_path, n in cls.OBJ_IDXS.items():
temp_to_aabb[n] = aabb_map.get(rel_path, -1)
temp_to_inside[n] = inside_map.get(rel_path, -1)
cls._temp_to_aabb_idx = create_tensor_from_list(temp_to_aabb, "int32", device="cuda")
cls._temp_to_inside_idx = create_tensor_from_list(temp_to_inside, "int32", device="cuda")
# CSR table of each (scene, target)'s collision links, for the exact point-source test.
# Walks IDX_OBJS the same way AABB.initialize_view does, and skips links with no collision
# geometry via LINK_VERTEX_COUNTS so the kernel never queries a null mesh id.
S_temp = len(cls.IDX_OBJS)
link_offsets = th.zeros((S_temp * N_temp + 1,), dtype=th.int32)
link_indices = []
for s_idx, scene_row in enumerate(cls.IDX_OBJS):
for n, obj in enumerate(scene_row):
if obj is not None and obj.prim_type != PrimType.CLOTH:
for link in obj.links.values():
flat_idx = RigidBodyViewAPI.get_flat_idx(link.prim_path)
if flat_idx is None:
continue
if RigidBodyViewAPI.LINK_VERTEX_COUNTS[flat_idx].item() == 0:
continue # no collision geometry for this link
link_indices.append(flat_idx)
link_offsets[s_idx * N_temp + n + 1] = len(link_indices)
cls._target_link_offsets = create_tensor_from_list(link_offsets, "int32", device="cuda")
# create_tensor_from_list cannot build a zero-length array; the kernel only reads this when
# some (s, n) has a non-empty range, so a 1-element dummy is safe when nothing has geometry.
cls._target_link_indices = create_tensor_from_list(
th.tensor(link_indices or [0], dtype=th.int32), "int32", device="cuda"
)
cls.INFLUENCE_MASK = wp.zeros((S_hss, N_hss, N_temp), dtype=wp.uint8, device="cuda")
cls.INFLUENCE_MASK_CPU = th.zeros((S_hss, N_hss, N_temp), dtype=th.bool).pin_memory()
cls.INFLUENCE_MASK_CPU_WP = _wp_from_torch(cls.INFLUENCE_MASK_CPU)
@classmethod
def pre_update(cls, dt=0.0):
super().pre_update(dt)
# Zero the influence mask every step so _incoming_heat_kernel only OR-writes hits.
if cls.INFLUENCE_MASK is not None:
cls.INFLUENCE_MASK.zero_()
@classmethod
def _update_values(cls, values):
if cls.VALUES_WP is None or cls.INCOMING_HEAT_RATE is None:
return
S, N = cls.VALUES.shape[:2]
if S == 0 or N == 0:
return
hss = HeatSourceOrSink
# 1) Gather incoming heat from active heat sources / sinks into our scratch + mask.
if (
cls.INFLUENCE_MASK is not None
and cls._hss_self_temp_idx is not None
and hss.VALUES_WP is not None
and AABB.VALUES_WP is not None
):
S_hss, N_hss = hss.VALUES.shape[:2]
# Scenes beyond either state's row count hold no (source, target) pairs.
S_common = min(S, S_hss)
inside_values_wp = Inside.VALUES_WP
n_inside_scenes = Inside.VALUES.shape[0] if inside_values_wp is not None else 0
if inside_values_wp is None:
inside_values_wp = cls._placeholder_inside
if S_common > 0 and N_hss > 0:
wp.launch(
kernel=_incoming_heat_kernel,
dim=(S_common, N_hss, N),
inputs=[
hss.VALUES_WP,
hss._requires_inside,
hss._temperatures,
hss._heating_rates,
hss._distance_thresholds,
cls._hss_self_temp_idx,
cls._hss_self_inside_idx,
hss._link_flat_idx,
hss._link_local_offset,
cls._temp_to_aabb_idx,
cls._temp_to_inside_idx,
cls._target_link_offsets,
cls._target_link_indices,
RigidBodyViewAPI.LINK_MESH_IDS,
wp.int32(N),
RigidBodyViewAPI.POSE_MATRICES,
AABB.VALUES_WP,
inside_values_wp,
cls.VALUES_WP,
wp.int32(n_inside_scenes),
cls.INFLUENCE_MASK,
cls.INCOMING_HEAT_RATE,
],
device="cuda",
)
# Mirror the mask for CPU reads (HeatSourceOrSink.affects_obj).
if cls.INFLUENCE_MASK_CPU_WP is not None:
wp.copy(cls.INFLUENCE_MASK_CPU_WP, cls.INFLUENCE_MASK)
# 2) Decay toward ambient + consume the gathered heat rate (also zeroes the scratch).
# dt is read from cls._dt at kernel-launch time inside the captured graph, so the
# per-frame value written in pre_update is visible without re-capturing the graph.
wp.launch(
kernel=_temperature_decay_kernel,
dim=(S, N),
inputs=[
cls.VALUES_WP,
cls.INCOMING_HEAT_RATE,
wp.float32(m.DEFAULT_TEMPERATURE),
wp.float32(m.TEMPERATURE_DECAY_SPEED),
cls._dt,
],
device="cuda",
)
# 3) Hold burning objects at their fire temperature (see kernel docstring for the
# ignition-threshold gate that lets deliberate cooling extinguish them).
if cls._hss_self_temp_idx is not None and hss.VALUES_WP is not None:
S_hss, N_hss = hss.VALUES.shape[:2]
S_common = min(S, S_hss)
if S_common > 0 and N_hss > 0:
wp.launch(
kernel=_self_heating_clamp_kernel,
dim=(S_common, N_hss),
inputs=[
hss.VALUES_WP,
hss._requires_on_fire,
hss._temperatures,
hss._ignition_temperatures,
cls._hss_self_temp_idx,
cls.VALUES_WP,
],
device="cuda",
)
@classmethod
def is_influenced_by(cls, source_obj, target_obj):
"""
Whether @source_obj's heat source / sink contributed heat to @target_obj's temperature
on the most recent update pass.
Args:
source_obj (StatefulObject): Object with the HeatSourceOrSink state.
target_obj (StatefulObject): Object with the Temperature state.
Returns:
bool
"""
# Lazy refresh so the read sees this-step's state.
TensorizedState.maybe_refresh_caches()
if cls.INFLUENCE_MASK_CPU is None:
return False
if HeatSourceOrSink.OBJ_IDXS is None or source_obj.relative_prim_path not in HeatSourceOrSink.OBJ_IDXS:
return False
if cls.OBJ_IDXS is None or target_obj.relative_prim_path not in cls.OBJ_IDXS:
return False
s = source_obj.scene.idx
if s >= cls.INFLUENCE_MASK_CPU.shape[0]:
return False
h = HeatSourceOrSink.OBJ_IDXS[source_obj.relative_prim_path]
n = cls.OBJ_IDXS[target_obj.relative_prim_path]
return bool(cls.INFLUENCE_MASK_CPU[s, h, n].item())
@classproperty
def value_name(cls):
return "temperature"