"""PymunkPhysics2D: the optional native (Chipmunk2D / pymunk) 2D backend.
A native :class:`~simvx.core.physics.world2d.Physics2DWorld` implementation and
an OPTIONAL accelerator: installing ``pymunk`` (the ``pymunk`` extra) makes
:class:`~simvx.core.physics.root.PhysicsRoot2D` auto-select it, exactly
mirroring the engine's miniaudio "installed -> used" model. When
``pymunk`` is absent, auto-discovery silently falls back to
:class:`~simvx.core.physics.builtin.world2d.BuiltinPhysics2D` (no crash, no warning
spam).
The contract is parity, not bit-identity: "switching backends never changes how a
game plays". A game using ``CharacterBody2D`` must behave the same (within
tolerance) on pymunk as on Builtin: a character is an ordinary KINEMATIC body here
too, and the shared collide-and-slide policy in
:mod:`simvx.core.physics.slide` drives it through the same
:meth:`~simvx.core.physics.world2d.Physics2DWorld.sweep_body` primitive, even
though Chipmunk2D's solver differs from the pure-Python tier.
What maps cleanly to Chipmunk2D
-------------------------------
- bodies / motion types: ``BodyMode`` STATIC/KINEMATIC/DYNAMIC ->
``pymunk.Body`` STATIC/KINEMATIC/DYNAMIC; mass + per-shape moment.
- shapes: Circle/Box(Poly)/Capsule(rounded poly)/Segment/ConvexPolygon(Poly)/
ConcavePolygon(static segment soup) -> ``pymunk.Circle`` / ``pymunk.Poly`` /
``pymunk.Segment``.
- step: ``space.step(dt)`` at the caller's fixed timestep.
- forces: ``apply_force_at_world_point`` / ``apply_impulse_at_world_point`` /
per-step torque accumulation.
- joints: fixed/pin/hinge/spring/groove -> ``PinJoint`` / ``PivotJoint`` /
``DampedSpring`` / ``GrooveJoint`` (a weld is a PivotJoint + a stiff
``DampedRotarySpring`` to lock the relative angle).
- queries: raycast -> ``segment_query``; shapecast/overlap -> ``shape_query``.
- events: a per-pair ``on_collision`` handler edge-diffs body / sensor pairs into
the engine's two event streams (contacts and sensor overlaps).
What needs care (documented honesty caveats)
--------------------------------------------
- CombineMode: Chipmunk MULTIPLIES both coefficients (``mu_a * mu_b`` for friction,
``e_a * e_b`` for elasticity) and has no per-contact combine-mode switch, so the
engine's per-coefficient :class:`~simvx.core.physics.material.CombineMode` cannot
be honoured natively for every neighbour pair. The raw coefficients are handed to
Chipmunk and its fixed rule decides (see ``_combine_into_shape``).
- swept motion: Chipmunk exposes no swept-shape time-of-impact for an un-added
shape, so :meth:`PymunkPhysics2D.sweep_body` substeps a transient probe body with
``shape_query``, exactly like :class:`BuiltinPhysics2D`. ``distance`` is therefore
the last substep proved non-penetrating, not an exact touch distance.
- KINEMATIC bulk WRITE: ``pymunk.batch`` SET is experimental, so per-body writes
are used on the (cold) set paths; the bulk READ uses ``pymunk.batch`` zero-copy
when present.
- collision layers: Chipmunk's ``cpBitmask`` is 32 bits wide, so only the low 32
bits of a collision layer survive into the native broadphase. A body whose layer
uses only bits at or above bit 32 is invisible to the broadphase and therefore to
queries, while Builtin (full-width Python ints) still sees it.
"""
from __future__ import annotations
import math
import numbers
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
import numpy as np
import pymunk
from ..math import Vec2
from .capability import Capability, contact_impulse_estimate, contact_manifold_payload
from .material import DEFAULT_PHYSICS_MATERIAL, PhysicsMaterial, _combine
from .world import (
DEFAULT_ANGULAR_DAMPING,
DEFAULT_GRAVITY_SCALE,
DEFAULT_LINEAR_DAMPING,
BodyMode,
ContactPhase,
normalise_damping,
normalise_gravity,
normalise_gravity_scale,
)
from .world2d import (
_DEFAULT_UP_2D,
BodyHandle,
ContactEvent2D,
JointHandle,
OverlapEvent2D,
Physics2DWorld,
RaycastHit2D,
ShapeHandle,
SweepHit2D,
body_scale_unchanged,
is_unit_scale_2d,
normalise_body_scale_2d,
)
# Whether the optional zero-copy batch read module is importable. pymunk.batch is
# present in 7.x but a guarded import keeps a future trimmed build from crashing.
try: # pragma: no cover - import guard
import pymunk.batch as _batch
_HAS_BATCH = True
except ImportError: # pragma: no cover
_batch = None # type: ignore[assignment]
_HAS_BATCH = False
def _as_array2(v: object) -> np.ndarray:
"""Coerce a Vec2 / sequence to a float32 ``(2,)`` array (always a copy)."""
return np.array(v, dtype=np.float32).reshape(2)
def _cross_2d(a: np.ndarray, b: np.ndarray) -> float:
"""Scalar 2D cross product ``a.x * b.y - a.y * b.x``."""
return float(a[0] * b[1] - a[1] * b[0])
# One-way platform rule thresholds (see :meth:`Physics2DWorld.set_one_way`). Held
# at the same values as the builtin tier so both backends reject the same contacts.
_ONE_WAY_VEL_EPS = 0.01 # m/s: rel-velocity along +normal above this = passing through
_ONE_WAY_NORMAL_TOL = 0.1 # contact-vs-platform normal disagreement tolerance (1 - cos)
#: Bisection halvings applied to a blocking sweep's reported distance. Held at the
#: builtin tier's value so a character parks the same distance from a wall on either
#: 2D backend. The substep scan brackets the blocker between the last fraction it
#: proved clear and the first that penetrated; each halving halves that bracket, so
#: eight take the reported bound from one substep of error to 1/256 of a substep,
#: below any skin the character policy uses. They cost one space query each and only
#: on a sweep that already found a blocker: a clear sweep pays nothing.
_SWEEP_REFINE_STEPS = 8
def _one_way_rejects(
one_way_normal: np.ndarray, platform_to_other: tuple[float, float], rel_vel: tuple[float, float]
) -> bool:
"""Whether a one-way platform lets a contact pass through (discard it).
``platform_to_other`` is the contact normal oriented from the platform toward
the other body and ``rel_vel`` is the other body's velocity relative to the
platform. A contact is discarded when the other body moves along
``+one_way_normal`` faster than the velocity epsilon (it is passing up through)
OR the contact normal disagrees with the solid side (it is hitting the underside
or a side edge rather than landing on top). Both terms are required: the normal
alone flips as soon as the mover's centre crosses the platform's, which would
let the solver shove a body that is still travelling through.
Scalars rather than arrays: this runs per arbiter per step for every one-way
body in the scene, where a temporary array per dot product would dominate it.
"""
nx = float(one_way_normal[0])
ny = float(one_way_normal[1])
if rel_vel[0] * nx + rel_vel[1] * ny > _ONE_WAY_VEL_EPS:
return True
return platform_to_other[0] * nx + platform_to_other[1] * ny < 1.0 - _ONE_WAY_NORMAL_TOL
# pymunk BodyMode mapping. Chipmunk uses the SAME three motion classes
# :class:`BodyMode` does; this is a 1:1 enum map (note pymunk's integer values differ: DYNAMIC=0,
# KINEMATIC=1, STATIC=2 -- never hardcode the ints, always use the symbols).
_BODY_MODE = {
BodyMode.STATIC: pymunk.Body.STATIC,
BodyMode.KINEMATIC: pymunk.Body.KINEMATIC,
BodyMode.DYNAMIC: pymunk.Body.DYNAMIC,
}
def _motion_mode(mode: BodyMode, *, is_sensor: bool) -> BodyMode:
"""The mode Chipmunk actually holds a body at, which for a sensor is not its own.
A STATIC sensor is held KINEMATIC. Chipmunk keeps static shapes in a spatial
index of their own and queries it only against the MOVING one, so a pair of
static shapes is never generated at all: a static sensor reports nothing
static, and nothing that is itself a sensor over a static body. Moving the
sensor into the moving index is what forms those pairs, and it is the same
promotion both Jolt adapters make for the same kind of reason.
Nothing the seam answers about the body changes: ``BodyMode`` is retained
separately, and a kinematic Chipmunk body carries infinite mass and so ignores
gravity, forces and impulses. What could set such a body moving is a velocity,
which Chipmunk DOES integrate for a kinematic body, so both routes to one are
closed against the mode the CALLER asked for: a write to a body the caller made
STATIC while Chipmunk holds it KINEMATIC is withheld from Chipmunk by
:meth:`PymunkPhysics2D.set_body_velocity` (held for readback, not lost), and a
velocity the body picked up while the caller had it KINEMATIC is cleared by
:meth:`PymunkPhysics2D.set_body_mode` when the mode returns to STATIC. Neither
guard sees a body Chipmunk itself holds static, whose velocity Chipmunk never
integrates but does read as a surface velocity in the friction solve. What the
promotion costs is a place in the moving index, which is reindexed and queried
every step, so the price scales with the number of trigger volumes rather than
with the number of bodies.
"""
return BodyMode.KINEMATIC if is_sensor and mode is BodyMode.STATIC else mode
@dataclass(slots=True)
class _ShapeDef:
"""A backend shape recipe: enough to (re)build a pymunk shape on any body.
pymunk shapes are bound to a body at construction, so a reusable ShapeHandle
cannot be a live ``pymunk.Shape``; it is this recipe, replayed by
:meth:`PymunkPhysics2D._instantiate_shape` onto each body that uses it
(mirroring how the builtin stores a ``_Shape2D`` and re-points it).
``kind`` discriminates the geometry; the remaining fields hold its parameters
(only the relevant ones are set per kind).
"""
kind: str # "circle" | "box" | "capsule" | "segment" | "poly" | "concave"
radius: float = 0.0
half_extents: np.ndarray | None = None # box
half_len: float = 0.0 # capsule central-segment half length
a: np.ndarray | None = None # segment endpoint
b: np.ndarray | None = None # segment endpoint
points: np.ndarray | None = None # poly verts (N,2) or concave segs (N,2,2)
def _scaled_sdef(base: _ShapeDef, scale: np.ndarray) -> _ShapeDef:
"""Return ``base`` resized by ``scale``, or ``base`` itself when unscaled.
Scale lives on the body, so a body materialises its own recipe from the shared
one and Chipmunk is handed geometry that is already the right size. The
analytic fields carry MAGNITUDES and take the absolute factor; points are
scaled signed, so a mirroring component mirrors them, and a polygon mirrored on
exactly one axis has its winding restored.
"""
if is_unit_scale_2d(scale):
return base
mag = np.abs(scale)
kind = base.kind
if kind == "circle":
return _ShapeDef("circle", radius=base.radius * float(mag[0]))
if kind == "box":
assert base.half_extents is not None
return _ShapeDef("box", half_extents=(base.half_extents * mag).astype(np.float32))
if kind == "capsule":
# Radius and central-segment length share the one uniform factor the seam
# enforces for a capsule.
return _ShapeDef("capsule", radius=base.radius * float(mag[0]), half_len=base.half_len * float(mag[0]))
if kind == "segment":
assert base.a is not None and base.b is not None
return _ShapeDef(
"segment",
radius=base.radius * float(mag[0]),
a=(base.a * scale).astype(np.float32),
b=(base.b * scale).astype(np.float32),
)
assert base.points is not None
pts = (base.points * scale).astype(np.float32)
if kind == "poly" and float(scale[0]) * float(scale[1]) < 0.0:
pts = pts[::-1].copy() # a single mirrored axis reverses the winding
return _ShapeDef(kind, points=pts)
@dataclass(slots=True)
class _BodyRec:
"""Bookkeeping for one body: its pymunk body + shapes + engine-side metadata.
The pymunk ``Body`` carries the live pose / velocity; this record keeps the
engine-side metadata pymunk does not (the sensor flag, the unique ``collision_type``
used to route the per-pair event handler, the material combine recipe, and the
one-way platform config the contact filter consults).
``mass`` and ``sdef`` are the create-time values, retained because Chipmunk
destroys a body's mass and moment when it leaves DYNAMIC: the engine contract
keeps mass across a mode flip, so it has to be restored from here (and the
moment recomputed from the shape definition) on the way back.
"""
body: pymunk.Body
shapes: list[pymunk.Shape]
mode: BodyMode
# The mode Chipmunk actually holds the body at, which differs from ``mode``
# only for a promoted sensor: see :func:`_motion_mode`. Everything that
# answers a seam question about the body's mode reads ``mode``; this exists
# so a mutation can tell whether the held type has to change too.
motion: BodyMode
mass: float
# ``sdef`` is the EFFECTIVE recipe the live Chipmunk shapes were built from,
# already resized by ``scale``; ``unscaled_sdef`` is the recipe the world's
# shape table handed out, kept so a later rescale starts from the original
# rather than compounding. None means ``sdef`` IS the unscaled recipe.
sdef: _ShapeDef
unscaled_sdef: _ShapeDef | None = None
scale: np.ndarray = field(default_factory=lambda: np.ones(2, dtype=np.float32))
is_sensor: bool = False
collision_layer: int = 1
collision_mask: int = 0xFFFFFFFF
# Whether this body would be ALLOWED to sleep. Recorded so the seam reads
# back what the caller set; inert here, because this space sleeps nothing at
# all (see ``capabilities``).
can_sleep: bool = True
# Surface coefficients, mirrored off the shapes. A pymunk Shape is bound to
# its body at construction, so set_body_shape has to build fresh ones and
# re-apply the surface; keeping the values here means it restores what the
# engine last set rather than reading them back off the shapes it replaces.
friction: float = 0.5
restitution: float = 0.0
# Per-body dynamics. The pair of damping rates and the gravity multiplier are
# kept here because Chipmunk has no per-body knob for any of them: a body that
# deviates from the space-wide default gets its own ``velocity_func``, held in
# ``velocity_func`` so the closure outlives the call that installed it.
linear_damping: float = DEFAULT_LINEAR_DAMPING
angular_damping: float = DEFAULT_ANGULAR_DAMPING
gravity_scale: float = DEFAULT_GRAVITY_SCALE
velocity_func: object = None
one_way: bool = False
one_way_normal: np.ndarray = field(default_factory=lambda: np.array([0.0, 1.0], dtype=np.float32))
def _inverse_mass(rec: _BodyRec | None) -> float:
"""``1 / mass`` for a body that can be pushed, ``0.0`` for one that cannot.
Read off the seam's retained mass and mode rather than the Chipmunk body,
whose mass is destroyed when it leaves DYNAMIC. A destroyed participant
(``None``) is immovable for the same reason a STATIC one is: nothing is going
to accelerate it.
"""
if rec is None or rec.mode is not BodyMode.DYNAMIC or rec.mass <= 0.0:
return 0.0
return 1.0 / rec.mass
def _velocity_at(rec: _BodyRec | None, point: pymunk.Vec2d | None) -> np.ndarray:
"""World velocity of the material point of a body currently at ``point``.
Chipmunk's own ``velocity_at_world_point``, so a spinning body's surface
speed is in the answer. ``point`` of ``None`` means the centre of mass, where
the value is just the linear velocity; a destroyed participant (``None``
record) contributes nothing, as it does to the inverse mass.
"""
if rec is None:
return np.zeros(2, dtype=np.float32)
v = rec.body.velocity if point is None else rec.body.velocity_at_world_point(point)
return np.array([v.x, v.y], dtype=np.float32)
[docs]
class PymunkPhysics2D(Physics2DWorld):
"""Native Chipmunk2D (pymunk) 2D backend implementing the full 2D world interface.
See :class:`~simvx.core.physics.world2d.Physics2DWorld` for the contract. Every
``@abstractmethod`` is implemented over a single ``pymunk.Space``.
"""
def __init__(self, *, gravity: Vec2 | None = None) -> None:
if gravity is None:
gravity = Vec2(0.0, -9.81)
super().__init__(gravity=gravity)
self._space = pymunk.Space()
self._space.gravity = (float(self._gravity[0]), float(self._gravity[1]))
self._space.iterations = self._solver_iterations
# Chipmunk has no per-body damping: its space-wide ``damping`` is a
# per-second velocity multiplier applied as ``v *= damping ** dt``, while
# the seam states damping as ``v *= max(0, 1 - c * dt)``. ``exp(-c)`` is the
# continuous-time equivalent of the seam rate, so a body running the seam
# DEFAULT pays nothing and stays entirely inside Chipmunk (the two forms
# differ by ~3e-7 per step at 60 Hz, which is below the tier's own noise).
# A body that deviates gets a ``velocity_func`` computing the seam formula
# exactly -- see :meth:`_apply_body_dynamics`.
self._space.damping = math.exp(-DEFAULT_LINEAR_DAMPING)
# Chipmunk's ``collision_slop`` default of 0.1 is a PIXEL-scale number: it
# tolerates a tenth of a unit of overlap, which is nothing at the 50-100 unit
# sprites Chipmunk was written for and ten CENTIMETRES at this engine's
# metre scale. The seam states it instead, at the same 1 mm the builtin 2D
# and 3D solvers use, so a crate rests where the geometry says on every
# backend rather than on three of four. Measured against a geometric 0.500:
# with the library default a landed crate settles at 0.477 on a bare drop
# and as deep as 0.400 under load, and at 0.499 with the seam default. The
# seam's damping is what made this visible rather than what caused it --
# damping removes the residual jitter that used to push the body a few
# centimetres back out of its own slop.
#
# The cost is at the other end of the scale, and it is small but real: a
# game working in PIXELS (gravity 980, 50-unit sprites) run at the metre
# default is using a tolerance far tighter than Chipmunk's own guidance for
# that scale, and a settled four-high stack there shows about 0.06 units of
# residual movement where the library default shows none. Which is why the
# number is the world's ``contact_slop`` rather than a constant: such a game
# raises it to its own scale, and this line then states that instead.
self._space.collision_slop = self._contact_slop
self._shapes: dict[ShapeHandle, _ShapeDef] = {}
self._bodies: dict[BodyHandle, _BodyRec] = {}
self._order: list[BodyHandle] = []
self._next_handle = 0
self._joints: dict[JointHandle, list[pymunk.Constraint]] = {}
self._next_joint = 0
# collision_type -> body handle, so a contact handler can recover handles.
self._type_to_handle: dict[int, BodyHandle] = {}
# Event diffing (edge-detected). pymunk's begin/separate callbacks already
# give edges, but we re-diff into the engine's canonical streams so a single
# default handler covers all pairs without per-type registration churn.
self._touching: set[tuple[BodyHandle, BodyHandle]] = set()
# Directed sensor edges: the ones a listener has been TOLD about, and every
# sensor pair Chipmunk currently has an arbiter for, whether the masks make
# it reportable or not. Chipmunk filters nothing for a sensor (the engine's
# one-directional ``sensor.mask & other.layer`` rule is ours alone), so the
# unfiltered set is what lets a live filter edit open or close an edge for
# an overlap that is already there.
self._overlapping: set[tuple[BodyHandle, BodyHandle]] = set()
self._sensor_pairs: set[tuple[BodyHandle, BodyHandle]] = set()
self._contact_events: list[ContactEvent2D] = []
self._overlap_events: list[OverlapEvent2D] = []
# Pending raw pairs captured by the per-step handler (handle pairs that
# began / separated this step), drained into the diff after step().
self._begin_pairs: list[tuple[BodyHandle, BodyHandle, np.ndarray, np.ndarray, np.ndarray, np.ndarray]] = []
self._separate_pairs: list[tuple[BodyHandle, BodyHandle]] = []
# Separations Chipmunk reported OUTSIDE a step, which is what replacing a
# body's shapes does: they belong to the next step's diff, not to a buffer
# that step() is about to clear. Sensor edges whose masks changed between
# steps ride along for the same reason.
self._pending_separations: list[tuple[BodyHandle, BodyHandle]] = []
self._pending_overlap_sync: list[tuple[BodyHandle, BodyHandle]] = []
# ENTER impulse capture: pairs that began this step, and the summed normal
# impulse the solver applied to each (read back off the live arbiters after
# space.step, so the ENTER event can carry a real impulse).
self._began_keys: set[tuple[BodyHandle, BodyHandle]] = set()
self._enter_impulses: dict[tuple[BodyHandle, BodyHandle], float] = {}
# Live arbiter count per body PAIR. A multi-shape pair (a segment soup, a
# concave floor) produces one arbiter per shape pair, but the engine reports
# ONE contact edge per body pair, so begin / separate are refcounted.
self._pair_arbiters: dict[tuple[BodyHandle, BodyHandle], int] = {}
# Pairs a one-way pre_solve handler rejected this step: a pass-through
# contact never collides, so it must never reach the event stream.
self._rejected_keys: set[tuple[BodyHandle, BodyHandle]] = set()
# Pairs the one-way filter ACCEPTED this step while this world did not yet
# consider them touching, mapped to the begin payload the ENTER needs. A
# Chipmunk arbiter fires ``begin`` once in its life, so a pair that turns
# solid after a pass-through has to have its ENTER synthesised from here.
self._accepted_keys: dict[
tuple[BodyHandle, BodyHandle],
tuple[BodyHandle, BodyHandle, np.ndarray, np.ndarray, np.ndarray, np.ndarray],
] = {}
# Bodies created with collision_mask == 0: Chipmunk's bidirectional filter
# would hide them from every space-level query, so the query paths lift
# their shape masks for the query's duration (see _queryable_filters).
self._mask_zero: set[BodyHandle] = set()
# Velocities written to a body the CALLER made STATIC while Chipmunk holds
# it KINEMATIC (a promoted sensor: see _motion_mode). Chipmunk would
# integrate such a write and walk the trigger volume off across the world,
# so it never reaches the library; it is remembered here instead and
# answered from here, which is what the pure-Python tiers do by storing a
# STATIC body's velocity and simply not integrating it. Empty in every
# scene without a promoted sensor, so the bulk reader pays one falsy check.
self._held_velocities: dict[BodyHandle, tuple[float, float, float]] = {}
self._install_handlers()
# -- gravity override (keep pymunk's space in sync) ---------------------
@Physics2DWorld.gravity.setter # type: ignore[attr-defined]
def gravity(self, value: Vec2) -> None:
gx, gy = normalise_gravity(value, 2)
self._gravity = Vec2(gx, gy)
self._space.gravity = (gx, gy)
# -- capability gate ----------------------------------------------------
[docs]
def capabilities(self) -> frozenset[Capability]:
"""Advertise the measured contact impulse and sensors-see-static.
Chipmunk exposes each arbiter's ``total_impulse``, so the ENTER payload is
a real measurement: the solver's total impulse projected onto the contact
normal and summed over the pair's arbiters (see ``_collect_enter_impulses``).
No cross-platform determinism, vehicles or soft bodies.
``SENSOR_DETECTS_STATIC`` is real, and it is this adapter that makes it
so: Chipmunk keeps static shapes in an index of their own and queries it
only against the moving one, so a static-vs-static pair is never
generated, and a sensor is therefore held KINEMATIC (see
:func:`_motion_mode`) to put it in the index that IS queried. Measured on
a sensor overlapping one box: one ENTER whichever mode the box is in, and
one for another STATIC sensor too, where before the STATIC pairs reported
nothing.
``SLEEP`` is absent, and that is a property of Chipmunk rather than of
this adapter: a ``cpSpace`` sleeps nothing until it is given a finite
sleep-time threshold, and asking a body in such a space to sleep aborts
the process at the C level instead of raising. Nothing here ever falls
asleep, so :meth:`sleeping` is permanently False and the whole sleep
surface is inert; see the capability's own documentation for what a
caller does about it.
"""
return frozenset({Capability.CONTACT_IMPULSE, Capability.SENSOR_DETECTS_STATIC})
[docs]
@property
def body_count(self) -> int:
return len(self._bodies)
[docs]
def clear(self) -> None:
"""Remove every body and joint, emptying the world.
See :meth:`~simvx.core.physics.world2d.Physics2DWorld.clear`. Routes through
the existing ``remove_joint`` / ``destroy_body`` so each is correctly removed
from the live ``pymunk.Space`` (joints first, then bodies). The per-step
edge-diff buffers are reset too. Gravity, shapes, and handle counters are
intentionally NOT reset.
"""
for jh in list(self._joints):
self.remove_joint(jh)
for bh in list(self._bodies):
self.destroy_body(bh)
self._touching.clear()
self._overlapping.clear()
self._sensor_pairs.clear()
self._contact_events.clear()
self._overlap_events.clear()
self._begin_pairs.clear()
self._separate_pairs.clear()
self._pending_separations.clear()
self._pending_overlap_sync.clear()
self._began_keys.clear()
self._enter_impulses.clear()
self._pair_arbiters.clear()
self._rejected_keys.clear()
self._accepted_keys.clear()
def _alloc_handle(self) -> int:
h = self._next_handle
self._next_handle += 1
return h
# -- collision routing ---------------------------------------------------
def _install_handlers(self) -> None:
"""Install a single catch-all collision handler for event diffing.
pymunk 7.x's ``on_collision`` with ``None`` collision types matches every
pair. ``begin`` fires when a pair starts touching, ``separate`` when it
stops; we capture the handle pair + (for begin) the contact geometry, and
``_diff_*`` re-routes them into the engine's contact / overlap streams.
Sensors are reported by pymunk too, so the diff classifies sensor pairs
into the separate overlap stream by the engine's one-directional rule.
Chipmunk works per SHAPE pair, so a body pair with several shapes a side (a
concave floor is one Segment per edge) fires begin / separate once per
arbiter. The engine reports one edge per BODY pair, so the handlers refcount
the pair's live arbiters and report only its first begin and last separate.
"""
def _begin(arbiter: pymunk.Arbiter, space: pymunk.Space, data: object) -> bool:
sa, sb = arbiter.shapes
ha = self._type_to_handle.get(sa.collision_type)
hb = self._type_to_handle.get(sb.collision_type)
if ha is None or hb is None:
return True
key = self._canon(ha, hb)
live = self._pair_arbiters.get(key, 0)
self._pair_arbiters[key] = live + 1
if live:
return True # the pair is already touching: one edge per body pair
cps = arbiter.contact_point_set
n = np.array([cps.normal.x, cps.normal.y], dtype=np.float32)
rec_a, rec_b = self._bodies.get(ha), self._bodies.get(hb)
# The estimate is computed from the LINEAR difference instead, which
# is the same pair of centre velocities on every backend where the
# at-point value is each narrow phase's own choice of point.
linear = (_velocity_at(rec_b, None) - _velocity_at(rec_a, None)).astype(np.float32)
# Chipmunk's ``point_a`` is the contact on the first shape's surface.
# Both bodies' velocities are read THERE, not at their centres, so a
# spinning body's surface speed is in the relative velocity. A
# manifold with no points has nowhere to read at, and the seam answers
# that one way on every backend (see ``contact_manifold_payload``).
at = cps.points[0].point_a if cps.points else None
witness = np.array([at.x, at.y], dtype=np.float32) if at is not None else None
at_rel = (_velocity_at(rec_b, at) - _velocity_at(rec_a, at)).astype(np.float32) if at is not None else None
pt, rel = contact_manifold_payload(witness, at_rel, linear)
self._begin_pairs.append((ha, hb, n, pt, rel, linear))
self._began_keys.add(key)
return True
def _separate(arbiter: pymunk.Arbiter, space: pymunk.Space, data: object) -> None:
sa, sb = arbiter.shapes
ha = self._type_to_handle.get(sa.collision_type)
hb = self._type_to_handle.get(sb.collision_type)
if ha is None or hb is None:
return
key = self._canon(ha, hb)
live = self._pair_arbiters.get(key, 0)
if live > 1:
self._pair_arbiters[key] = live - 1
return # other arbiters of the pair still touch
self._pair_arbiters.pop(key, None)
self._separate_pairs.append((ha, hb))
self._space.on_collision(None, None, begin=_begin, separate=_separate)
@staticmethod
def _canon(a: BodyHandle, b: BodyHandle) -> tuple[BodyHandle, BodyHandle]:
return (a, b) if a <= b else (b, a)
def _collect_enter_impulses(self) -> None:
"""Read the solver's normal impulse for the pairs that began this step.
Chipmunk keeps each live arbiter on BOTH of its bodies' arbiter lists after
the step, so the ENTER payload is read back afterwards rather than from a
catch-all ``post_solve`` handler (which would cross the C -> Python boundary
for every touching pair every step, resting worlds included). Only bodies
that began a pair are walked, and always from the DYNAMIC side: a static
level collider carries every contact in the scene on its list, a dynamic
body only its own few.
"""
walks: dict[BodyHandle, dict[int, tuple[BodyHandle, BodyHandle]]] = {}
for key in self._began_keys:
ra, rb = self._bodies.get(key[0]), self._bodies.get(key[1])
if ra is None or rb is None:
continue
if ra.mode is BodyMode.DYNAMIC:
host, other = key
elif rb.mode is BodyMode.DYNAMIC:
other, host = key
else:
continue # two non-dynamic bodies: Chipmunk never solves the pair
walks.setdefault(host, {})[other + 1] = key # collision_type == handle + 1
impulses = self._enter_impulses
for host, wanted in walks.items():
rec = self._bodies.get(host)
if rec is None:
continue
def visit(arbiter: pymunk.Arbiter, wanted: dict[int, tuple[BodyHandle, BodyHandle]] = wanted) -> None:
sa, sb = arbiter.shapes
key = wanted.get(sa.collision_type) or wanted.get(sb.collision_type)
if key is None:
return
# The engine's ENTER payload is the NORMAL impulse magnitude: project
# the solver's total impulse onto the contact normal (raw
# |total_impulse| would also count friction). Summed over the pair's
# arbiters so a segment soup reports one physical impulse.
j = arbiter.total_impulse
n = arbiter.normal
impulses[key] = impulses.get(key, 0.0) + abs(float(j.x * n.x + j.y * n.y))
rec.body.each_arbiter(visit)
def _resolve_one_way_pairs(self) -> None:
"""Reconcile this step's one-way verdicts with the contact stream.
A rejected contact never collides, so it applies no impulse and fires no
event (Builtin discards the pair before its diff for the same reason): drop
this step's ENTER for the pair, and if this world had already reported the pair
as touching, report the EXIT instead. A pair whose shapes disagree (a soup
edge accepted, another rejected) is solid, so an acceptance outranks a
rejection.
The converse needs synthesising. A pair that turns solid again keeps the SAME
Chipmunk arbiter, and an arbiter fires ``begin`` exactly once in its life, so
without this the collision that starts pushing the body would be invisible to
gameplay and its eventual separation would report an EXIT with no ENTER.
"""
accepted = self._accepted_keys
rejected = self._rejected_keys - accepted.keys()
if rejected:
self._begin_pairs = [t for t in self._begin_pairs if self._canon(t[0], t[1]) not in rejected]
self._began_keys -= rejected
self._separate_pairs.extend(rejected & self._touching)
for key, payload in accepted.items():
if key in self._began_keys:
continue # the arbiter began this step: pymunk already reported it
self._begin_pairs.append(payload)
self._began_keys.add(key)
# -- shapes -------------------------------------------------------------
def _store_shape(self, sdef: _ShapeDef) -> ShapeHandle:
"""File a freshly built shape spec under a new handle and return it."""
h = self._alloc_handle()
self._shapes[h] = sdef
return h
[docs]
def create_circle(self, radius: float) -> ShapeHandle:
if not radius > 0.0:
raise ValueError(f"circle radius must be > 0, got {radius}")
return self._store_shape(_ShapeDef("circle", radius=float(radius)))
[docs]
def create_box(self, half_extents: Vec2) -> ShapeHandle:
he = _as_array2(half_extents)
if not np.all(he > 0.0):
raise ValueError(f"box half_extents must all be > 0, got {tuple(he)}")
return self._store_shape(_ShapeDef("box", half_extents=he.copy()))
[docs]
def create_capsule(self, radius: float, height: float) -> ShapeHandle:
if not (radius > 0.0 and height > 0.0):
raise ValueError(f"capsule radius/height must be > 0, got {radius}, {height}")
half_len = max(0.0, height * 0.5 - radius)
return self._store_shape(_ShapeDef("capsule", radius=float(radius), half_len=half_len))
[docs]
def create_segment(self, a: Vec2, b: Vec2, radius: float = 0.0) -> ShapeHandle:
if not radius >= 0.0:
raise ValueError(f"segment radius must be >= 0, got {radius}")
pa, pb = _as_array2(a), _as_array2(b)
if not float(np.dot(pb - pa, pb - pa)) > 1e-12:
raise ValueError(f"segment endpoints must differ, got {tuple(pa)} and {tuple(pb)}")
return self._store_shape(_ShapeDef("segment", radius=float(radius), a=pa, b=pb))
[docs]
def create_convex_polygon(self, points: np.ndarray) -> ShapeHandle:
pts = np.asarray(points, dtype=np.float32).reshape(-1, 2)
if pts.shape[0] < 3:
raise ValueError(f"convex polygon needs >= 3 points, got {pts.shape[0]}")
return self._store_shape(_ShapeDef("poly", points=pts.copy()))
[docs]
def create_concave_polygon(self, segments: np.ndarray) -> ShapeHandle:
segs = np.asarray(segments, dtype=np.float32).reshape(-1, 2, 2)
if segs.shape[0] < 1:
raise ValueError(f"concave polygon needs >= 1 segment, got {segs.shape[0]}")
return self._store_shape(_ShapeDef("concave", points=segs.copy()))
[docs]
def destroy_shape(self, shape: ShapeHandle) -> None:
"""Release this world's record of a shape handle.
See :meth:`~simvx.core.physics.world2d.Physics2DWorld.destroy_shape`. A
handle here names a ``_ShapeDef`` spec, not a live ``pymunk.Shape``: the
Chipmunk shapes belong to the bodies built from it, which keep them and
keep simulating. Dropping the spec is the whole of it. Unknown handles are
a silent no-op.
"""
self._shapes.pop(shape, None)
def _shape_rec(self, shape: ShapeHandle) -> _ShapeDef:
"""Return the spec for ``shape``, or raise ``KeyError`` if it is not ours.
A shape handle is caller input, so an unknown one raises rather than
asserts: ``assert`` is compiled out by ``python -O``, and a released
handle would then build a body against a stale spec. ``KeyError``
matches what every body-handle lookup on this backend already raises.
"""
rec = self._shapes.get(shape)
if rec is None:
raise KeyError(f"unknown shape handle {shape!r}")
return rec
def _instantiate_shape(self, sdef: _ShapeDef, body: pymunk.Body) -> list[pymunk.Shape]:
"""Build the pymunk shape(s) for ``sdef`` bound to ``body``.
Most kinds yield one shape; a CONCAVE soup yields one Segment per edge
(Chipmunk has no native concave collider -- the 2D analogue of a static
triangle mesh is the matching segment list, STATIC-only).
"""
kind = sdef.kind
if kind == "circle":
return [pymunk.Circle(body, sdef.radius)]
if kind == "box":
he = sdef.half_extents
return [pymunk.Poly.create_box(body, (float(he[0]) * 2.0, float(he[1]) * 2.0))]
if kind == "capsule":
# Capsule == a thick Y-axis segment in Chipmunk (a segment with radius
# IS a capsule). A degenerate (half_len ~ 0) capsule becomes a circle.
r = sdef.radius
hl = sdef.half_len
if hl <= 1e-6:
return [pymunk.Circle(body, r)]
return [pymunk.Segment(body, (0.0, -hl), (0.0, hl), r)]
if kind == "segment":
return [pymunk.Segment(body, tuple(map(float, sdef.a)), tuple(map(float, sdef.b)), sdef.radius)]
if kind == "poly":
verts = [tuple(map(float, p)) for p in sdef.points]
return [pymunk.Poly(body, verts)]
# concave: one Segment per edge (STATIC-only level geometry).
shapes: list[pymunk.Shape] = []
for seg in sdef.points:
shapes.append(pymunk.Segment(body, tuple(map(float, seg[0])), tuple(map(float, seg[1])), 0.0))
return shapes
def _shape_moment(self, sdef: _ShapeDef, mass: float) -> float:
"""Real per-shape moment of inertia via pymunk's ``moment_for_*`` helpers."""
kind = sdef.kind
if kind == "circle":
return float(pymunk.moment_for_circle(mass, 0.0, sdef.radius))
if kind == "box":
he = sdef.half_extents
return float(pymunk.moment_for_box(mass, (float(he[0]) * 2.0, float(he[1]) * 2.0)))
if kind == "capsule":
hl = sdef.half_len
return float(pymunk.moment_for_segment(mass, (0.0, -hl), (0.0, hl), sdef.radius))
if kind == "segment":
a = tuple(map(float, sdef.a))
b = tuple(map(float, sdef.b))
return float(pymunk.moment_for_segment(mass, a, b, sdef.radius))
if kind == "poly":
verts = [tuple(map(float, p)) for p in sdef.points]
return float(pymunk.moment_for_poly(mass, verts))
# concave is STATIC-only: never a DYNAMIC body, so the moment is unused.
return 1.0
# -- bodies -------------------------------------------------------------
@staticmethod
def _unpack_transform(transform: object) -> tuple[np.ndarray, float]:
"""Extract (position float32 (2,), rotation float radians).
Implements the ``transform`` contract (see
:meth:`~simvx.core.physics.world2d.Physics2DWorld.create_body`): a
``Transform2D`` (``.position`` + ``.rotation``), a bare ``Vec2`` /
sequence (position only, zero rotation), or a ``(position, rotation)``
pair. Any real scalar is accepted as the rotation, ``numpy`` floats
included, since the engine's own maths is float32.
"""
pos = getattr(transform, "position", None)
if pos is not None:
return _as_array2(pos), float(getattr(transform, "rotation", 0.0))
if isinstance(transform, tuple) and len(transform) == 2:
p, r = transform
# A 2-tuple is ambiguous: (Vec2-pair == position) vs (position, angle).
# Disambiguate by the second element: a scalar is the rotation angle, a
# 2-vector means the tuple itself is a bare position.
if isinstance(r, numbers.Real):
return _as_array2(p), float(r)
return _as_array2(transform), 0.0
return _as_array2(transform), 0.0
def _combine_into_shape(self, shape: pymunk.Shape, material: PhysicsMaterial) -> None:
"""Set per-shape friction / elasticity from a material.
Honesty caveat: Chipmunk applies a FIXED combine rule per coefficient, the
PRODUCT in both cases (``mu_a * mu_b`` for friction, ``e_a * e_b`` for
elasticity), and has no per-contact mode switch. The engine's per-coefficient
:class:`CombineMode` therefore cannot be honoured natively for every
neighbour pair: the raw coefficients are handed to Chipmunk's fixed rule.
Where a scene needs a specific combined value against a known surface, set
the coefficients so that the product reproduces it. (See the module
docstring.) ``_combine`` is referenced here to keep the engine's combine
semantics visible where pymunk is mapped in.
"""
_ = _combine # documented above: Chipmunk's fixed rule decides, not ours
shape.friction = material.friction
shape.elasticity = material.restitution
[docs]
def create_body(
self,
shape: ShapeHandle,
body_type: BodyMode,
transform: object,
*,
mass: float = 1.0,
scale: Vec2 | None = None,
can_sleep: bool = True,
linear_damping: float = DEFAULT_LINEAR_DAMPING,
angular_damping: float = DEFAULT_ANGULAR_DAMPING,
gravity_scale: float = DEFAULT_GRAVITY_SCALE,
collision_layer: int = 1,
collision_mask: int = 0xFFFFFFFF,
is_sensor: bool = False,
material: PhysicsMaterial | None = None,
continuous: bool = False,
) -> BodyHandle:
sdef = self._shape_rec(shape)
if sdef.kind == "concave" and body_type is not BodyMode.STATIC:
raise ValueError(
f"ConcavePolygonShape2D (segment soup) is a STATIC-only collider; got body_type={body_type}."
)
# Mass is retained across set_body_mode, so it must be real in every mode:
# an immovable body created with a bad mass would carry it into a later
# DYNAMIC flip. ``not mass > 0.0`` also rejects NaN.
if not mass > 0.0:
raise ValueError(f"body mass must be > 0, got {mass}")
# Every knob is validated before anything enters the space: a rejected
# argument must not leave a body behind that the seam has no handle for,
# and so cannot destroy.
lin_damping = normalise_damping(linear_damping, "linear_damping")
ang_damping = normalise_damping(angular_damping, "angular_damping")
grav_scale = normalise_gravity_scale(gravity_scale)
position, rotation = self._unpack_transform(transform)
body_scale = np.ones(2, dtype=np.float32) if scale is None else normalise_body_scale_2d(scale, sdef.kind)
unscaled = sdef
sdef = _scaled_sdef(unscaled, body_scale)
motion = _motion_mode(body_type, is_sensor=bool(is_sensor))
if body_type is BodyMode.DYNAMIC:
moment = self._shape_moment(sdef, mass)
body = pymunk.Body(mass, moment, body_type=pymunk.Body.DYNAMIC)
else:
body = pymunk.Body(body_type=_BODY_MODE[motion])
body.position = (float(position[0]), float(position[1]))
body.angle = float(rotation)
handle = self._alloc_handle()
# Unique collision_type per body so the event handler can recover handles.
# collision_type is a positive int keyed off the handle (offset to avoid 0).
ctype = handle + 1
self._type_to_handle[ctype] = handle
surface = DEFAULT_PHYSICS_MATERIAL if material is None else material
shapes = self._instantiate_shape(sdef, body)
flt = pymunk.ShapeFilter(categories=collision_layer & 0xFFFFFFFF, mask=collision_mask & 0xFFFFFFFF)
for s in shapes:
s.collision_type = ctype
s.filter = flt
s.sensor = is_sensor
self._combine_into_shape(s, surface)
self._space.add(body, *shapes)
self._bodies[handle] = _BodyRec(
body=body,
shapes=shapes,
mode=body_type,
motion=motion,
mass=float(mass),
sdef=sdef,
unscaled_sdef=unscaled,
scale=body_scale,
is_sensor=is_sensor,
collision_layer=collision_layer,
collision_mask=collision_mask,
can_sleep=bool(can_sleep),
friction=surface.friction,
restitution=surface.restitution,
linear_damping=lin_damping,
angular_damping=ang_damping,
gravity_scale=grav_scale,
)
self._apply_body_dynamics(self._bodies[handle])
if (collision_mask & 0xFFFFFFFF) == 0:
self._mask_zero.add(handle)
return handle
[docs]
def destroy_body(self, handle: BodyHandle) -> None:
# Idempotent, as on every other backend: freeing an already-removed body is
# a safe no-op. A body can legitimately be gone already -- ``clear()``
# empties the world and the owning node's ``on_exit_tree`` then destroys its
# (now-stale) handle during teardown. Raising here crashed that ordering.
rec = self._bodies.pop(handle, None)
if rec is None:
return
self._mask_zero.discard(handle)
self._held_velocities.pop(handle, None)
self._space.remove(rec.body, *rec.shapes)
if handle in self._order:
self._order.remove(handle)
self._type_to_handle = {t: h for t, h in self._type_to_handle.items() if h != handle}
self._drop_pairs_for(handle)
# Drop joints touching the destroyed body (parity with builtin's purge): a
# joint must never solve against a freed body. Each constraint carries its
# owning (a, b) handles, recorded in ``_add_joint``.
doomed = [
jh for jh, cons in self._joints.items() if any(handle in getattr(c, "_simvx_bodies", ()) for c in cons)
]
for jh in doomed:
for c in self._joints.pop(jh):
if c in self._space.constraints:
self._space.remove(c)
def _disturb(self, rec: _BodyRec, *, wake: bool) -> None:
"""The one home for what a mutation does to sleep on this backend.
Every setter that can change what the solver reads ends in exactly one
call to this, so the wake is a property of the seam rather than a line
each setter remembers. There is no vacated-volume half here: Chipmunk
wakes the whole sleeping COMPONENT a body belongs to, so activating the
changed body already reaches everything resting on it.
Inert in practice, since this space sleeps nothing (see
:meth:`capabilities`), and kept faithful anyway so the one call that would
matter is already in the right place if a space is ever given a finite
sleep-time threshold.
Args:
rec: The mutated body's record.
wake: The caller's ``wake=`` argument. False suppresses the activate.
"""
if not wake:
return
if rec.mode is BodyMode.DYNAMIC:
rec.body.activate()
[docs]
def set_body_transform(
self, handle: BodyHandle, transform: object, *, scale: Vec2 | None = None, wake: bool = True
) -> None:
rec = self._bodies[handle]
position, rotation = self._unpack_transform(transform)
# A byte compare first: the node states its scale on every pose write, and
# validating an unchanged one again would dominate this hot path.
if scale is not None and not body_scale_unchanged(scale, rec.scale):
base = rec.unscaled_sdef or rec.sdef
new_scale = normalise_body_scale_2d(scale, base.kind)
if not np.array_equal(new_scale, rec.scale):
# A Chipmunk shape cannot be re-geometried in place, so a resize
# rebuilds the body's colliders exactly as a shape swap does.
rec.scale = new_scale
self._rebuild_shapes(handle, rec, _scaled_sdef(base, new_scale))
rec.body.position = (float(position[0]), float(position[1]))
rec.body.angle = float(rotation)
self._space.reindex_shapes_for_body(rec.body)
self._disturb(rec, wake=wake)
[docs]
def set_body_velocity(self, handle: BodyHandle, linear: Vec2, angular: float = 0.0) -> None:
rec = self._bodies[handle]
lin = _as_array2(linear)
# Only where the seam is holding the body at a mode the caller did not ask
# for: a STATIC sensor is KINEMATIC to Chipmunk (see ``_motion_mode``), and
# Chipmunk integrates a kinematic body's velocity, so the write would walk a
# trigger volume across the world -- which it does on no other backend, the
# pure-Python tiers never integrating a STATIC body. Withholding it from
# Chipmunk is not the same as losing it: the value is held (and read back,
# as the pure-Python tiers read theirs back) until the body leaves the mode
# that made it unsafe. A body Chipmunk itself holds STATIC is untouched
# here: the write lands, moves nothing, and drives the surface velocity
# Chipmunk gives a static body's friction.
if rec.mode is BodyMode.STATIC and rec.motion is not BodyMode.STATIC:
self._held_velocities[handle] = (float(lin[0]), float(lin[1]), float(angular))
return
if self._held_velocities:
self._held_velocities.pop(handle, None)
rec.body.velocity = (float(lin[0]), float(lin[1]))
rec.body.angular_velocity = float(angular)
if rec.body.body_type == pymunk.Body.DYNAMIC:
rec.body.activate()
[docs]
def set_body_mode(self, handle: BodyHandle, mode: BodyMode, *, wake: bool = True) -> None:
rec = self._bodies[handle]
# Re-assert the concave static-only contract on the shape KIND the body was
# built from (a one-segment soup is still concave), matching create_body.
if rec.sdef.kind == "concave" and mode is not BodyMode.STATIC:
raise ValueError(
f"ConcavePolygonShape2D (segment soup) is a STATIC-only collider; cannot set body_mode={mode}."
)
# Second entry point to the same invariant as create_body, ahead of
# any mutation so a rejected flip leaves the body exactly as it was.
if mode is BodyMode.DYNAMIC and not rec.mass > 0.0:
raise ValueError(f"body mass must be > 0, got {rec.mass}")
# Order is forced by Chipmunk: a non-dynamic body attached to a Space
# reports mass == inf, and pymunk rejects assigning an infinite mass, so
# the mass cannot be written first. The body_type flip is itself what
# zeroes mass and moment, so the retained mass and a moment recomputed
# from it are restored AFTER the flip.
# Chipmunk zeroes a body's velocity on a flip to any type but DYNAMIC
# (a flip TO dynamic keeps it), where the pure-Python tiers keep whatever
# the body was carrying whichever way it goes: a frozen body resumes at
# the speed it was frozen at, and a conveyor belt keeps its surface
# velocity across a mode flip. Captured before the flip and put back
# after it, except where the body is ARRIVING at caller-STATIC while
# Chipmunk holds it KINEMATIC, which has to stop instead.
carried = (rec.body.velocity.x, rec.body.velocity.y, rec.body.angular_velocity)
rec.motion = _motion_mode(mode, is_sensor=rec.is_sensor)
rec.body.body_type = _BODY_MODE[rec.motion]
rec.mode = mode
if mode is BodyMode.STATIC and rec.motion is not BodyMode.STATIC:
# Chipmunk stops a body it is told to make non-dynamic, but a promoted
# sensor is KINEMATIC on both sides of this flip, so the assignment
# above is a no-op and the velocity the body was given while the caller
# had it KINEMATIC survives -- and keeps being integrated. Becoming
# STATIC has to stop the body whether or not the library agrees it
# moved, or a trigger volume returned to STATIC walks off across the
# world (measured: x = 5.0 one second after the flip back).
rec.body.velocity = (0.0, 0.0)
rec.body.angular_velocity = 0.0
else:
# A velocity held back from Chipmunk (see :meth:`set_body_velocity`)
# outranks what the body was carrying: the body has left the state that
# made the write unsafe, so the write becomes live, exactly as the
# pure-Python tiers carry a velocity written to a STATIC body into the
# mode that starts using it.
held = self._held_velocities.pop(handle, None) if self._held_velocities else None
vx, vy, spin = held if held is not None else carried
rec.body.velocity = (vx, vy)
rec.body.angular_velocity = spin
if mode is BodyMode.DYNAMIC:
rec.body.mass = rec.mass
rec.body.moment = self._shape_moment(rec.sdef, rec.mass)
# The seam's parked hand-over (a body entering DYNAMIC with wake=False stays
# out of the simulation until something disturbs it) is unreachable here:
# this space sleeps nothing, so a freed body starts moving on the next step
# whichever way ``wake`` points. That is the documented degradation for a
# backend without Capability.SLEEP; see :meth:`capabilities`.
self._disturb(rec, wake=wake)
# -- live edits to what create_body was given ---------------------------
[docs]
def set_body_mass(self, handle: BodyHandle, mass: float, *, wake: bool = True) -> None:
# Same guard as create_body / set_body_mode: the mass is retained across
# mode flips, so it must be real whatever the body's current mode.
if not mass > 0.0:
raise ValueError(f"body mass must be > 0, got {mass}")
rec = self._bodies[handle]
rec.mass = float(mass)
# Chipmunk zeroes a non-dynamic body's mass and moment while it is attached
# to a Space and refuses an assignment there, so only a DYNAMIC body takes
# the write now; the others pick the retained value up in set_body_mode.
if rec.mode is BodyMode.DYNAMIC:
rec.body.mass = rec.mass
rec.body.moment = self._shape_moment(rec.sdef, rec.mass)
self._disturb(rec, wake=wake)
[docs]
def set_body_filter(
self, handle: BodyHandle, collision_layer: int, collision_mask: int, *, wake: bool = True
) -> None:
rec = self._bodies[handle]
layer = int(collision_layer) & 0xFFFFFFFF
mask = int(collision_mask) & 0xFFFFFFFF
rec.collision_layer = int(collision_layer)
rec.collision_mask = int(collision_mask)
flt = pymunk.ShapeFilter(categories=layer, mask=mask)
for s in rec.shapes:
s.filter = flt
# The mask-zero set drives the query paths' filter lift (a Chipmunk body
# with mask 0 is invisible to every space query), so it tracks the live
# value rather than only the create-time one.
if mask == 0:
self._mask_zero.add(handle)
else:
self._mask_zero.discard(handle)
# Chipmunk caches nothing about a pair's filter decision beyond the current
# arbiter, which is re-tested next step; waking makes sure a sleeper is
# re-solved against a neighbour it newly matches.
self._disturb(rec, wake=wake)
# Solid pairs re-decide themselves: Chipmunk stops accepting a pair its
# filter rejects and reports the separation on the next step. Sensor edges
# are the engine's own decision (Chipmunk reports every sensor pair whatever
# the masks say), so they are re-derived from the pairs it is already
# tracking -- on the NEXT step, because the event buffers are per-step and
# an event appended between two steps would be cleared by the second.
# Nothing to queue in a world with no sensor overlapping anything.
if self._sensor_pairs:
self._pending_overlap_sync.extend(e for e in self._sensor_pairs if handle in e)
[docs]
def set_body_material(self, handle: BodyHandle, material: PhysicsMaterial | None) -> None:
rec = self._bodies[handle]
surface = DEFAULT_PHYSICS_MATERIAL if material is None else material
rec.friction = surface.friction
rec.restitution = surface.restitution
for s in rec.shapes:
self._combine_into_shape(s, surface)
if rec.mode is BodyMode.DYNAMIC:
rec.body.activate()
[docs]
def set_body_damping(self, handle: BodyHandle, linear: float, angular: float) -> None:
rec = self._bodies[handle]
# Both rates are validated before either is written: the call describes the
# body's damping as a whole, so a rejected one must leave it as it was.
lin = normalise_damping(linear, "linear_damping")
ang = normalise_damping(angular, "angular_damping")
rec.linear_damping = lin
rec.angular_damping = ang
self._apply_body_dynamics(rec)
if rec.mode is BodyMode.DYNAMIC:
rec.body.activate()
[docs]
def set_body_gravity_scale(self, handle: BodyHandle, scale: float) -> None:
rec = self._bodies[handle]
rec.gravity_scale = normalise_gravity_scale(scale)
self._apply_body_dynamics(rec)
if rec.mode is BodyMode.DYNAMIC:
rec.body.activate()
def _apply_body_dynamics(self, rec: _BodyRec) -> None:
"""Install (or drop) the per-body velocity integrator this body needs.
A body running the seam defaults needs nothing: the space-wide ``damping``
set in ``__init__`` already produces them, inside Chipmunk. One that
deviates gets a ``velocity_func`` applying the seam formula exactly --
gravity scaled for this body alone, then
``v *= max(0, 1 - damping * dt)`` on each of the two velocities, which is
also what lets the two damping rates differ (Chipmunk's own integrator
applies one number to both).
"""
default = (
rec.linear_damping == DEFAULT_LINEAR_DAMPING
and rec.angular_damping == DEFAULT_ANGULAR_DAMPING
and rec.gravity_scale == DEFAULT_GRAVITY_SCALE
)
if default:
if rec.velocity_func is not None:
rec.velocity_func = None
rec.body.velocity_func = pymunk.Body.update_velocity
return
def velocity_func(body, gravity, damping, dt, _rec=rec): # noqa: ANN001 - Chipmunk callback
scale = _rec.gravity_scale
# Damping first, on the velocity the body already has, then this step's
# acceleration: the seam's order, and Chipmunk's own. The space-wide
# multiplier is passed as 1.0 on purpose, or this body would be damped
# twice.
body.velocity = body.velocity * max(0.0, 1.0 - _rec.linear_damping * dt)
body.angular_velocity = body.angular_velocity * max(0.0, 1.0 - _rec.angular_damping * dt)
pymunk.Body.update_velocity(body, (gravity[0] * scale, gravity[1] * scale), 1.0, dt)
rec.velocity_func = velocity_func
rec.body.velocity_func = velocity_func
def _on_world_settings_changed(self) -> None:
"""Push ``solver_iterations`` and ``contact_slop``; the other knobs have no home.
Chipmunk's own sleeping is switched off in this backend (its
``sleep_time_threshold`` stays infinite), which is why it does not
advertise :attr:`~simvx.core.physics.capability.Capability.SLEEP`. The
seam's sleep thresholds are therefore stored and read back but change
nothing here, and the capability is how a caller finds that out.
``position_iterations`` is stored and read back for the same kind of
reason and a different one: Chipmunk has no position solver at all, so
there is no pass to run more of. Its constraints converge in the velocity
loop, which is why a chain it hangs sits on its rest length while the
builtin solver's sags (see ``docs/core/physics_backends.md``).
``contact_slop`` is exactly Chipmunk's ``collision_slop``, so this lane
needs no emulation: the seam simply states the number rather than
inheriting the library's pixel-scale default.
"""
self._space.iterations = self._solver_iterations
self._space.collision_slop = self._contact_slop
[docs]
def set_body_continuous(self, handle: BodyHandle, enabled: bool) -> None:
"""Accept the flag and integrate discretely: Chipmunk2D has no CCD.
The one live setter this backend cannot honour, for the same reason it
already ignores ``create_body``'s ``continuous`` argument: Chipmunk has no
swept / speculative integration of any kind, so there is nothing to switch
on. Advertised through the absence of
:attr:`~simvx.core.physics.capability.Capability.CONTINUOUS` rather than
raised, so a scene that merely prefers CCD still runs and a caller that
requires it can check for it by name instead of by backend.
"""
self._bodies[handle] # KeyError on an unknown handle, like every setter
[docs]
def set_body_shape(self, handle: BodyHandle, shape: ShapeHandle, *, wake: bool = True) -> None:
sdef = self._shape_rec(shape)
rec = self._bodies[handle]
# Third entry point to the concave STATIC-only invariant, checked
# before any mutation so a rejected swap leaves the body untouched.
if sdef.kind == "concave" and rec.mode is not BodyMode.STATIC:
raise ValueError(
f"ConcavePolygonShape2D (segment soup) is a STATIC-only collider; cannot place it on a "
f"body whose mode is {rec.mode}."
)
# The body keeps its scale across a swap, so the NEW geometry has to be
# able to carry it. Validated before any mutation, like the concave check
# above, so a rejected swap leaves the body exactly as it was.
rec.scale = normalise_body_scale_2d(rec.scale, sdef.kind)
rec.unscaled_sdef = sdef
self._rebuild_shapes(handle, rec, _scaled_sdef(sdef, rec.scale))
self._disturb(rec, wake=wake)
self._space.reindex_shapes_for_body(rec.body)
def _rebuild_shapes(self, handle: BodyHandle, rec: _BodyRec, sdef: _ShapeDef) -> None:
"""Replace a body's live colliders with the ones ``sdef`` describes.
A pymunk Shape is bound to its body at construction and cannot be
re-geometried, so both routes that change a body's geometry -- a shape swap
and a rescale -- detach the old shapes and instantiate the recipe onto the
SAME body. The body object is what joints and the handle->collision_type
routing refer to, so both survive; only the colliders are replaced.
"""
ctype = handle + 1
old = rec.shapes
new = self._instantiate_shape(sdef, rec.body)
flt = pymunk.ShapeFilter(categories=rec.collision_layer & 0xFFFFFFFF, mask=rec.collision_mask & 0xFFFFFFFF)
for s in new:
s.collision_type = ctype
s.filter = flt
s.sensor = rec.is_sensor
s.friction = rec.friction
s.elasticity = rec.restitution
self._space.remove(*old)
self._space.add(*new)
rec.shapes = new
rec.sdef = sdef
# Removing a shape makes Chipmunk kill its live arbiters and report each as
# a separation, right here rather than inside a step. Those belong to the
# NEXT step's diff, where a pair the new geometry still touches begins
# again and cancels its own separation, and one the new geometry has left
# reports the EXIT it really is.
self._pending_separations.extend(self._separate_pairs)
self._separate_pairs.clear()
# The moment is a function of the shape AND the mass, so a DYNAMIC body's
# rotational response follows the geometry it now has.
if rec.mode is BodyMode.DYNAMIC:
rec.body.moment = self._shape_moment(sdef, rec.mass)
def _drop_pairs_for(self, handle: BodyHandle) -> None:
"""Hand every edge involving ``handle`` to the next step, to be closed there.
A destroyed body's announced edges are contacts that HAVE ended, so each is
owed an EXIT like any other separation. Removing the shapes from the space
also makes Chipmunk kill their arbiters and report each as a separation,
right here rather than inside a step; those land in ``_separate_pairs``,
which the next step clears, so they are banked in ``_pending_separations``
together with the edges this world announced itself. The next step's diff
then reports one EXIT per announced edge and forgets the pair (queueing an
edge twice is harmless: the diff closes it once and skips it thereafter).
Bookkeeping that names no announcement (the arbiter refcounts, the live
sensor-pair set) is simply forgotten: it describes shapes that no longer
exist, and dropping it at once stops a later handle reusing the number from
inheriting a stale edge.
"""
ended = [p for p in self._touching if handle in p]
ended += [p for p in self._overlapping if handle in p]
self._pending_separations.extend(ended)
self._pending_separations.extend(self._separate_pairs)
self._separate_pairs.clear()
if self._sensor_pairs:
self._sensor_pairs = {p for p in self._sensor_pairs if handle not in p}
if self._pair_arbiters:
self._pair_arbiters = {p: n for p, n in self._pair_arbiters.items() if handle not in p}
[docs]
def body_velocity(self, handle: BodyHandle) -> tuple[Vec2, float]:
held = self._held_velocities.get(handle) if self._held_velocities else None
if held is not None:
return Vec2(held[0], held[1]), held[2]
b = self._bodies[handle].body
return Vec2(b.velocity.x, b.velocity.y), float(b.angular_velocity)
[docs]
def body_transform(self, handle: BodyHandle) -> tuple[Vec2, float]:
b = self._bodies[handle].body
return Vec2(b.position.x, b.position.y), float(b.angle)
[docs]
def body_mass(self, handle: BodyHandle) -> float:
# The record's retained mass, not Chipmunk's: a non-dynamic body's mass is
# zeroed in the Space, and an immovable body is infinite-mass by mode.
rec = self._bodies[handle]
return float(rec.mass) if rec.mode is BodyMode.DYNAMIC else math.inf
[docs]
def wake(self, handle: BodyHandle) -> None:
rec = self._bodies[handle]
if rec.mode is BodyMode.DYNAMIC:
rec.body.activate()
[docs]
def sleep(self, handle: BodyHandle) -> None:
"""Accept the call and do nothing: this space sleeps nothing.
The one sleep-surface call that CANNOT be forwarded. Chipmunk asserts
``sleep_time_threshold < INFINITY`` inside ``cpBodySleep`` and aborts the
whole process when it does not hold, and this space leaves the threshold
at its default of infinity, so forwarding would turn a harmless call into
a crash. Advertised through the absence of
:attr:`~simvx.core.physics.capability.Capability.SLEEP` rather than
raised, exactly as :meth:`set_body_continuous` handles the missing CCD.
"""
self._bodies[handle] # KeyError on an unknown handle, like every setter
[docs]
def set_body_can_sleep(self, handle: BodyHandle, enabled: bool) -> None:
"""Record the flag; nothing sleeps here, so nothing changes.
Kept honest rather than dropped: a caller that reads it back gets what it
set, and forbidding sleep is already true of every body in this space.
"""
self._bodies[handle].can_sleep = bool(enabled)
[docs]
def sleeping(self, handle: BodyHandle) -> bool:
rec = self._bodies[handle]
if rec.mode is not BodyMode.DYNAMIC:
return False # STATIC / KINEMATIC are immovable, not "asleep"
return bool(rec.body.is_sleeping)
# -- forces -------------------------------------------------------------
[docs]
def apply_impulse(self, handle: BodyHandle, impulse: Vec2, *, at: Vec2 | None = None, angular: float = 0.0) -> None:
rec = self._bodies[handle]
if rec.mode is not BodyMode.DYNAMIC:
return
rec.body.activate()
lin = _as_array2(impulse)
point = rec.body.position if at is None else pymunk.Vec2d(float(at[0]), float(at[1]))
rec.body.apply_impulse_at_world_point((float(lin[0]), float(lin[1])), point)
if angular != 0.0:
# Explicit scalar angular impulse: omega += J_ang / moment.
moment = rec.body.moment
if moment > 0.0:
rec.body.angular_velocity += float(angular) / moment
[docs]
def apply_force(self, handle: BodyHandle, force: Vec2, *, at: Vec2 | None = None) -> None:
rec = self._bodies[handle]
if rec.mode is not BodyMode.DYNAMIC:
return
rec.body.activate()
f = _as_array2(force)
# pymunk auto-clears force each step, matching the engine's one-step force contract.
point = rec.body.position if at is None else pymunk.Vec2d(float(at[0]), float(at[1]))
rec.body.apply_force_at_world_point((float(f[0]), float(f[1])), point)
[docs]
def apply_torque(self, handle: BodyHandle, torque: float) -> None:
rec = self._bodies[handle]
if rec.mode is not BodyMode.DYNAMIC:
return
rec.body.activate()
# pymunk has no torque accumulator API; add to the body's torque field,
# which Chipmunk clears each step (parity with the engine's one-step torque contract).
rec.body.torque += float(torque)
# -- joints -------------------------------------------------------------
def _alloc_joint(self) -> int:
h = self._next_joint
self._next_joint += 1
return h
def _add_joint(self, a: BodyHandle, b: BodyHandle, constraints: list[pymunk.Constraint]) -> JointHandle:
for c in constraints:
self._space.add(c)
handle = self._alloc_joint()
self._joints[handle] = constraints
# Record the body handles on each constraint object for the destroy purge.
for c in constraints:
c._simvx_bodies = (a, b)
return handle
def _joint_ends(self, a: BodyHandle, b: BodyHandle, where: str) -> tuple[pymunk.Body, pymunk.Body]:
"""Return both pymunk bodies of a joint, or raise ``KeyError`` naming the bad end.
Joint endpoints are caller input, so an unknown handle raises rather than
asserts: ``assert`` is compiled out by ``python -O``, and the constraint
would then be filed against a handle no body answers to. ``KeyError``
matches every other body-handle lookup on this backend.
"""
try:
return self._bodies[a].body, self._bodies[b].body
except KeyError as exc:
raise KeyError(f"{where}: unknown body handle {exc.args[0]!r}") from None
[docs]
def create_fixed_joint(self, a: BodyHandle, b: BodyHandle) -> JointHandle:
ba, bb = self._joint_ends(a, b, "create_fixed_joint")
# Weld = a PivotJoint (lock the shared point) + a stiff DampedRotarySpring
# to lock the relative angle (Chipmunk has no rigid relative-angle lock; a
# stiff rotary spring is the canonical pymunk weld, documented honesty).
anchor = (ba.position + bb.position) * 0.5
pivot = pymunk.PivotJoint(ba, bb, anchor)
rest = float(bb.angle - ba.angle)
rotary = pymunk.DampedRotarySpring(ba, bb, rest, stiffness=1e6, damping=1e4)
return self._add_joint(a, b, [pivot, rotary])
[docs]
def create_pin_joint(self, a: BodyHandle, b: BodyHandle, anchor: Vec2) -> JointHandle:
ba, bb = self._joint_ends(a, b, "create_pin_joint")
anc = pymunk.Vec2d(float(anchor[0]), float(anchor[1]))
pivot = pymunk.PivotJoint(ba, bb, anc)
return self._add_joint(a, b, [pivot])
[docs]
def create_hinge_joint(self, a: BodyHandle, b: BodyHandle, anchor: Vec2) -> JointHandle:
# 2D hinge == pin this tier (1-DOF rotation, nothing off-axis to lock);
# motors / limits via SimpleMotor / RotaryLimitJoint are a follow-on.
return self.create_pin_joint(a, b, anchor)
[docs]
def create_spring_joint(
self, a: BodyHandle, b: BodyHandle, rest_length: float, stiffness: float, damping: float
) -> JointHandle:
ba, bb = self._joint_ends(a, b, "create_spring_joint")
rl = float(rest_length)
if rl < 0.0:
rl = float((bb.position - ba.position).length) # auto-capture current distance
# Spring between the two body centres (anchors at each body's local origin).
spring = pymunk.DampedSpring(ba, bb, (0.0, 0.0), (0.0, 0.0), rl, float(stiffness), float(damping))
return self._add_joint(a, b, [spring])
[docs]
def create_groove_joint(
self, a: BodyHandle, b: BodyHandle, groove_a: Vec2, groove_b: Vec2, anchor_b: Vec2
) -> JointHandle:
ba, bb = self._joint_ends(a, b, "create_groove_joint")
ga = (float(groove_a[0]), float(groove_a[1]))
gb = (float(groove_b[0]), float(groove_b[1]))
anc = (float(anchor_b[0]), float(anchor_b[1]))
groove = pymunk.GrooveJoint(ba, bb, ga, gb, anc)
return self._add_joint(a, b, [groove])
[docs]
def remove_joint(self, handle: JointHandle) -> None:
cons = self._joints.pop(handle, None)
if cons is None:
return # no-op on unknown (parity with builtin's silent-drop contract)
for c in cons:
if c in self._space.constraints:
self._space.remove(c)
# -- one-way platforms --------------------------------------------------
[docs]
def set_one_way(self, handle: BodyHandle, enabled: bool, normal: Vec2 = _DEFAULT_UP_2D) -> None:
rec = self._bodies[handle]
n = _as_array2(normal)
length = float(np.linalg.norm(n))
unit = (n / length).astype(np.float32) if length > 1e-9 else _DEFAULT_UP_2D.copy()
changed = bool(enabled) != rec.one_way or not np.array_equal(unit, rec.one_way_normal)
rec.one_way = bool(enabled)
rec.one_way_normal = unit
# Chipmunk implements one-way via a pre_solve handler that ignores a contact
# when the mover comes from the platform's non-solid side. pymunk 7 IGNORES
# the callback's return value: rejection is ``arbiter.process_collision``,
# and that flag PERSISTS for the arbiter's lifetime, so it must be assigned
# unconditionally every call (a pass-through pair is re-accepted by a True
# assignment once the body comes back down onto the solid side).
ctype = handle + 1
def _one_way_pre(arbiter: pymunk.Arbiter, space: pymunk.Space, data: object) -> bool:
sa, sb = arbiter.shapes
ha = self._type_to_handle.get(sa.collision_type)
hb = self._type_to_handle.get(sb.collision_type)
ra = self._bodies.get(ha)
rb = self._bodies.get(hb)
if ra is None or rb is None:
return True
# Sensors are never one-way filtered: an overlap reports DETECTION, not
# collision, so a sensor sweeping through a platform must still open and
# close its overlap edge (Builtin gates sensors out before its one-way
# filter for the same reason).
if ra.is_sensor or rb.is_sensor:
return True
# Gate on the LIVE one_way flags: the handler stays installed after
# ``set_one_way(handle, False)``, which must restore solid behaviour.
# ``arbiter.normal`` points a -> b, so it is already oriented
# platform -> other for side a and needs negating for side b. Chipmunk
# has integrated velocities by pre_solve, matching the point in the step
# where Builtin evaluates the same rule.
an = arbiter.normal
nx, ny = float(an.x), float(an.y)
va, vb = ra.body.velocity, rb.body.velocity
rvx, rvy = float(vb.x - va.x), float(vb.y - va.y)
keep = not (
(ra.one_way and _one_way_rejects(ra.one_way_normal, (nx, ny), (rvx, rvy)))
or (rb.one_way and _one_way_rejects(rb.one_way_normal, (-nx, -ny), (-rvx, -rvy)))
)
arbiter.process_collision = keep
key = self._canon(ha, hb)
if not keep:
self._rejected_keys.add(key)
elif key not in self._touching:
# Solid, but this world does not yet know the pair is touching: capture
# what an ENTER needs in case no ``begin`` fires for it this step
# (see :meth:`_resolve_one_way_pairs`).
cps = arbiter.contact_point_set
normal = np.array([nx, ny], dtype=np.float32)
# Same two velocities the ``begin`` handler captures, in the same
# order and through the same seam rule for a manifold with no
# point: the at-point pair for ``rel_velocity`` and the centre
# pair (already read above for the one-way test) for the estimate.
at = cps.points[0].point_a if cps.points else None
witness = np.array([at.x, at.y], dtype=np.float32) if at is not None else None
at_rel = (_velocity_at(rb, at) - _velocity_at(ra, at)).astype(np.float32) if at is not None else None
linear = np.array([rvx, rvy], dtype=np.float32)
pt, rel = contact_manifold_payload(witness, at_rel, linear)
self._accepted_keys[key] = (ha, hb, normal, pt, rel, linear)
return keep
if enabled:
self._space.on_collision(ctype, None, pre_solve=_one_way_pre)
# A platform that stops resolving contacts from one side has taken support
# away from whatever was resting on that side, exactly as a filter edit
# does, so a real change goes through the one home for the wake. A write
# that changes neither the flag nor the normal takes nothing away.
if changed:
self._disturb(rec, wake=True)
# -- stepping -----------------------------------------------------------
[docs]
def step(self, dt: float) -> None:
self._contact_events = []
self._overlap_events = []
self._begin_pairs.clear()
self._separate_pairs.clear()
self._began_keys.clear()
self._enter_impulses.clear()
self._rejected_keys.clear()
self._accepted_keys.clear()
if self._pending_separations:
self._separate_pairs.extend(self._pending_separations)
self._pending_separations.clear()
self._space.step(dt)
# One-way reconciliation runs BEFORE the impulse read-back so a synthesised
# ENTER carries the impulse its arbiter actually applied this step.
if self._rejected_keys or self._accepted_keys:
self._resolve_one_way_pairs()
if self._began_keys:
self._collect_enter_impulses()
self._diff_events()
def _diff_events(self) -> None:
"""Route this step's begin / separate pairs into the engine's event streams.
Body-body pairs feed the contact stream (canonical ``a<=b`` order); pairs
where a side is a sensor feed the SEPARATE one-directional overlap stream by
the engine's rule (``sensor.mask & other.layer``: the observer decides).
A contact event needs at least one DYNAMIC participant. Chipmunk generates
arbiters for kinematic-vs-static and kinematic-vs-kinematic pairs, but such a
pair cannot be pushed apart and produces no solver work, so the engine reports
nothing for it: the builtin tier drops it at its double-infinite-mass skip
and Jolt drops it natively. Without this gate a character standing on the
ground would emit a contact event on pymunk and nothing anywhere else.
"""
if self._pending_overlap_sync:
# Filter edits made since the last step: re-derive their sensor edges
# first, so a pair this step also separates is not closed twice.
for edge in self._pending_overlap_sync:
self._sync_overlap(*edge)
self._pending_overlap_sync.clear()
for ha, hb, n, pt, rel, linear in self._begin_pairs:
ra, rb = self._bodies.get(ha), self._bodies.get(hb)
if ra is None or rb is None:
continue
if ra.is_sensor or rb.is_sensor:
if ra.is_sensor:
self._sensor_pairs.add((ha, hb))
self._sync_overlap(ha, hb)
if rb.is_sensor:
self._sensor_pairs.add((hb, ha))
self._sync_overlap(hb, ha)
continue
if ra.mode is not BodyMode.DYNAMIC and rb.mode is not BodyMode.DYNAMIC:
continue
key = self._canon(ha, hb)
if key in self._touching:
# Already reported as touching: replacing a body's shapes builds
# fresh arbiters for pairs that never came apart, and re-announcing
# those would fire a pickup or a landing sound a second time.
continue
# Orient normal a->b in canonical order.
ca, cb = key
normal = n if (ca, cb) == (ha, hb) else (-n).astype(np.float32)
rel_ab = rel if (ca, cb) == (ha, hb) else -rel
linear_ab = linear if (ca, cb) == (ha, hb) else -linear
rec_a, rec_b = (ra, rb) if (ca, cb) == (ha, hb) else (rb, ra)
self._touching.add(key)
self._contact_events.append(
ContactEvent2D(
a=ca,
b=cb,
phase=ContactPhase.ENTER,
point=Vec2(pt),
normal=Vec2(normal),
impulse=float(self._enter_impulses.get(key, 0.0)),
# Published even though Chipmunk hands back its real applied
# impulse: it is the one shared formula over the one shared
# quantity, the LINEAR difference rather than the at-point
# value beside it (see the builtin siblings).
impulse_estimate=contact_impulse_estimate(
linear_ab, normal, _inverse_mass(rec_a), _inverse_mass(rec_b)
),
rel_velocity=Vec2(rel_ab),
)
)
zero = Vec2(0.0, 0.0)
for ha, hb in self._separate_pairs:
key = self._canon(ha, hb)
if key in self._began_keys:
# The pair also began this step, so it never stopped touching: one
# arbiter of a multi-shape pair died while another took over, or a
# shape swap replaced the colliders under a live contact.
continue
ra, rb = self._bodies.get(ha), self._bodies.get(hb)
# Which stream the pair belongs to is normally read off the bodies, but a
# destroyed participant leaves no record to read: fall back to what this
# world ANNOUNCED, since a sensor edge never enters ``_touching`` and a
# body-body pair never enters ``_overlapping``.
watched = (ha, hb) in self._overlapping or (hb, ha) in self._overlapping
if watched or (ra is not None and rb is not None and (ra.is_sensor or rb.is_sensor)):
for sensor_h, other_h, rec in ((ha, hb, ra), (hb, ha, rb)):
if rec is not None and not rec.is_sensor:
continue
self._sensor_pairs.discard((sensor_h, other_h))
self._close_overlap(sensor_h, other_h)
continue
if ra is not None and rb is not None:
if ra.mode is not BodyMode.DYNAMIC and rb.mode is not BodyMode.DYNAMIC:
continue
# Only a pair this world reported as touching may report an EXIT (parity
# with Builtin, which synthesises no EXIT for a pair it never entered).
if key not in self._touching:
continue
self._touching.discard(key)
self._contact_events.append(
ContactEvent2D(
a=key[0],
b=key[1],
phase=ContactPhase.EXIT,
point=zero,
normal=zero,
impulse=0.0,
impulse_estimate=0.0,
rel_velocity=zero,
)
)
def _sync_overlap(self, sensor: BodyHandle, other: BodyHandle) -> None:
"""Open or close one directed overlap edge to match the current masks.
Idempotent, so an edge already in the wanted state produces no event: a
shape swap rebuilds every arbiter of the body it touches, and the pair a
sensor was already watching must not be announced a second time.
"""
srec, orec = self._bodies.get(sensor), self._bodies.get(other)
edge = (sensor, other)
if srec is not None and orec is not None and (srec.collision_mask & orec.collision_layer):
if edge not in self._overlapping:
self._overlapping.add(edge)
self._overlap_events.append(OverlapEvent2D(sensor=sensor, other=other, phase=ContactPhase.ENTER))
else:
self._close_overlap(sensor, other)
def _close_overlap(self, sensor: BodyHandle, other: BodyHandle) -> None:
"""Report EXIT for a directed overlap edge that was open (else nothing).
Closing is decided by what was ANNOUNCED, never by the current masks: a
sensor that stops scanning for a body it is holding must report the body
leaving, and a mask test would suppress exactly that event.
"""
edge = (sensor, other)
if edge not in self._overlapping:
return
self._overlapping.discard(edge)
self._overlap_events.append(OverlapEvent2D(sensor=sensor, other=other, phase=ContactPhase.EXIT))
[docs]
def drain_overlap_events(self) -> list[OverlapEvent2D]:
events = self._overlap_events
self._overlap_events = []
return events
# -- bulk transfer (keystone) -------------------------------------------
[docs]
def register_bodies(self, handles: list[BodyHandle]) -> None:
for h in handles:
if h not in self._bodies:
raise KeyError(f"register_bodies: unknown body handle {h!r}")
self._order = list(handles)
[docs]
def read_velocities(self, out: np.ndarray) -> None:
# (N, 3) = [lx, ly, omega]. Fills in place, allocates nothing.
self._check_velocities_out(out, len(self._order))
held_any = bool(self._held_velocities)
for i, handle in enumerate(self._order):
held = self._held_velocities.get(handle) if held_any else None
if held is not None:
out[i] = held
continue
b = self._bodies[handle].body
out[i, 0] = b.velocity.x
out[i, 1] = b.velocity.y
out[i, 2] = b.angular_velocity
# -- queries ------------------------------------------------------------
@contextmanager
def _queryable_filters(self, mask: int) -> Iterator[None]:
"""Temporarily lift zero-mask body filters so space-level queries see them.
Chipmunk's ShapeFilter reject rule is BIDIRECTIONAL: a query misses any
shape whose own mask excludes the query's categories, so a body created
with ``collision_mask == 0`` is invisible to every space-level query. The
engine's public queries are ONE-DIRECTIONAL (the observer decides via
``mask & body.collision_layer``), so for the query's duration each such
shape's mask is lifted to all-ones (group and categories untouched: a
``collision_layer == 0`` body stays invisible, matching Builtin) and the
observer post-filter decides. Filters are restored before returning; no
``space.step`` ever runs inside the context. No-op when no zero-mask body
exists, so the common path pays one falsy set check.
Only bodies this query could actually report are lifted: one whose layer the
query mask excludes is dropped by the same post-filter anyway, so making it
visible would be pure cost. What remains is O(shapes of the zero-mask bodies
the query mask selects), so a large zero-mask segment soup is only paid for
by queries that were going to look at it.
A character is an ordinary KINEMATIC body with real shapes in the space, so a
zero-mask character is tracked in ``_mask_zero`` and lifted here like any
other body.
"""
if not self._mask_zero:
yield
return
saved: list[tuple[pymunk.Shape, pymunk.ShapeFilter]] = []
try:
for h in self._mask_zero:
rec = self._bodies.get(h)
if rec is None or not (mask & rec.collision_layer):
continue
for s in rec.shapes:
f = s.filter
saved.append((s, f))
s.filter = pymunk.ShapeFilter(group=f.group, categories=f.categories, mask=0xFFFFFFFF)
yield
finally:
for s, f in saved:
s.filter = f
@staticmethod
def _native_query_mask(mask: int) -> int:
"""Chipmunk-side mask for an engine query mask (``cpBitmask`` is 32 bits wide).
With zero-mask bodies lifted (see :meth:`_queryable_filters`) and the query
filter's categories left at all-ones, Chipmunk's bidirectional reject rule
reduces to the engine's one-directional observer rule, so the native mask is a
broadphase accelerator for it (the Python post-filter stays the authority).
An engine mask using bits at or above bit 32 cannot be expressed natively:
those queries go permissive and are decided entirely by the post-filter,
which compares the full-width Python ints.
"""
return 0xFFFFFFFF if mask >> 32 else (mask & 0xFFFFFFFF)
[docs]
def raycast(self, origin: Vec2, direction: Vec2, max_dist: float, *, mask: int = 0xFFFFFFFF) -> RaycastHit2D | None:
hits = self.raycast_all(origin, direction, max_dist, mask=mask)
return hits[0] if hits else None
[docs]
def raycast_all(
self, origin: Vec2, direction: Vec2, max_dist: float, *, mask: int = 0xFFFFFFFF
) -> list[RaycastHit2D]:
o = _as_array2(origin)
d = _as_array2(direction)
length = float(np.linalg.norm(d))
if length < 1e-12:
return []
d = (d / length).astype(np.float32)
# A non-finite max_dist (raycast accepts inf) needs a large finite endpoint.
far = 1e7 if not math.isfinite(max_dist) else float(max_dist)
end = o + d * far
# Native mask as a broadphase accelerator (see _native_query_mask); the
# observer post-filter below remains the semantic authority.
flt = pymunk.ShapeFilter(mask=self._native_query_mask(mask))
with self._queryable_filters(mask):
infos = self._space.segment_query((float(o[0]), float(o[1])), (float(end[0]), float(end[1])), 0.0, flt)
hits: list[RaycastHit2D] = []
for info in infos:
handle = self._type_to_handle.get(info.shape.collision_type)
if handle is None:
continue
rec = self._bodies.get(handle)
if rec is None or not (mask & rec.collision_layer):
continue
dist = info.alpha * far
if dist > (max_dist if math.isfinite(max_dist) else far):
continue
point = Vec2(info.point.x, info.point.y)
normal = Vec2(info.normal.x, info.normal.y)
hits.append(RaycastHit2D(body=handle, point=point, normal=normal, distance=float(dist)))
hits.sort(key=lambda h: h.distance)
return hits
def _probe_body(
self, sdef: _ShapeDef, position: np.ndarray, rotation: float
) -> tuple[pymunk.Body, list[pymunk.Shape]]:
"""Build a transient (un-added) kinematic probe body wrapping ``sdef``."""
body = pymunk.Body(body_type=pymunk.Body.KINEMATIC)
body.position = (float(position[0]), float(position[1]))
body.angle = float(rotation)
shapes = self._instantiate_shape(sdef, body)
return body, shapes
[docs]
def shapecast(
self, shape: ShapeHandle, origin: Vec2, direction: Vec2, max_dist: float, *, mask: int = 0xFFFFFFFF
) -> SweepHit2D | None:
sdef = self._shape_rec(shape)
if not math.isfinite(max_dist):
raise ValueError("shapecast max_dist must be finite (a swept shape needs a bounded sweep length)")
if sdef.kind == "concave":
raise ValueError("concave shapes are STATIC-only and cannot be used as a moving query shape")
o = _as_array2(origin)
d = _as_array2(direction)
length = float(np.linalg.norm(d))
if length < 1e-12 or max_dist <= 0.0:
return None
dir_unit = (d / length).astype(np.float32)
# Substepped sweep using shape_query (Chipmunk has no native swept-shape TOI
# in the public API for an un-added shape; conservative-advancement is a
# follow-on). Sized from a coarse feature estimate, capped at 64. The probe
# carries the native query mask so each substep's broadphase is cheap;
# _first_shape_overlap applies the engine's one-directional observer mask.
feature = max(self._feature_size(sdef), 1e-4)
steps = min(64, max(1, math.ceil(max_dist / (feature * 0.5))))
body, shapes = self._probe_body(sdef, o, 0.0)
probe_filter = pymunk.ShapeFilter(mask=self._native_query_mask(mask))
for s in shapes:
s.filter = probe_filter
prev_frac = 0.0
with self._queryable_filters(mask):
for st in range(1, steps + 1):
frac = st / steps
body.position = (
float(o[0] + dir_unit[0] * max_dist * frac),
float(o[1] + dir_unit[1] * max_dist * frac),
)
hit = self._first_shape_overlap(shapes, mask)
if hit is not None:
ho, n = hit
reached = (o + dir_unit * (max_dist * prev_frac)).astype(np.float32)
travelled = float(max_dist) * prev_frac
point = (reached - n * feature).astype(np.float32)
return SweepHit2D(body=ho, point=Vec2(point), normal=Vec2(n), distance=travelled)
prev_frac = frac
return None
def _first_shape_overlap(self, shapes: list[pymunk.Shape], mask: int) -> tuple[BodyHandle, np.ndarray] | None:
"""Return the first (handle, separating-normal) a probe shape overlaps.
The separating normal points back toward the mover (negated from pymunk's
a->b contact normal), matching the ``SweepHit2D`` convention.
"""
for s in shapes:
for info in self._space.shape_query(s):
handle = self._type_to_handle.get(info.shape.collision_type)
if handle is None:
continue
rec = self._bodies.get(handle)
if rec is None or not (mask & rec.collision_layer):
continue
cps = info.contact_point_set
n = np.array([cps.normal.x, cps.normal.y], dtype=np.float32)
# shape_query normal points from the queried shape toward the other;
# the engine's separating normal points toward the mover: negate.
return handle, (-n).astype(np.float32)
return None
[docs]
def overlap(self, shape: ShapeHandle, transform: object, *, mask: int = 0xFFFFFFFF) -> list[BodyHandle]:
sdef = self._shape_rec(shape)
if sdef.kind == "concave":
raise ValueError("concave shapes are STATIC-only and cannot be used as a moving query shape")
position, rotation = self._unpack_transform(transform)
body, shapes = self._probe_body(sdef, position, rotation)
probe_filter = pymunk.ShapeFilter(mask=self._native_query_mask(mask))
for s in shapes:
s.filter = probe_filter
result: set[BodyHandle] = set()
with self._queryable_filters(mask):
for s in shapes:
for info in self._space.shape_query(s):
handle = self._type_to_handle.get(info.shape.collision_type)
if handle is None:
continue
rec = self._bodies.get(handle)
if rec is None or not (mask & rec.collision_layer):
continue
result.add(handle)
return sorted(result)
@staticmethod
def _feature_size(sdef: _ShapeDef) -> float:
if sdef.kind == "circle" or sdef.kind == "capsule":
return float(sdef.radius)
if sdef.kind == "box":
return float(min(sdef.half_extents))
if sdef.kind == "segment":
return max(float(sdef.radius), 1e-3)
if sdef.points is not None:
pts = sdef.points.reshape(-1, 2)
ext = pts.max(axis=0) - pts.min(axis=0)
return float(min(ext)) * 0.5
return 0.1
def _feature_size_of_body(self, rec: _BodyRec) -> float:
bb = None
for s in rec.shapes:
sbb = s.bb
if bb is None:
bb = sbb
else:
bb = bb.merge(sbb)
if bb is None:
return 0.1
return max(min(bb.right - bb.left, bb.top - bb.bottom) * 0.5, 1e-3)
# -- swept motion (the one non-mutating sweep primitive) ----------------
@staticmethod
def _sweep_one_way_rejects(mover: _BodyRec, other: _BodyRec, toward_mover: np.ndarray, motion: np.ndarray) -> bool:
"""One-way pass-through decision for a sweep contact (either side a platform).
``toward_mover`` is the separating normal pointing at the mover, so the
"platform -> other" normal is ``-toward_mover`` when the MOVER is the platform
and ``+toward_mover`` when the other body is. The mover's approach velocity is
the sweep ``motion``; the blocker's is its stored linear velocity. Discards if
EITHER platform lets the contact pass, matching the builtin tier's rule.
"""
ovx, ovy = float(other.body.velocity.x), float(other.body.velocity.y)
mvx, mvy = float(motion[0]), float(motion[1])
if mover.one_way and _one_way_rejects(
mover.one_way_normal, (-float(toward_mover[0]), -float(toward_mover[1])), (ovx - mvx, ovy - mvy)
):
return True
return bool(
other.one_way
and _one_way_rejects(
other.one_way_normal, (float(toward_mover[0]), float(toward_mover[1])), (mvx - ovx, mvy - ovy)
)
)
[docs]
def sweep_body(
self,
handle: BodyHandle,
motion: Vec2,
*,
from_transform: tuple[Vec2, float] | None = None,
skin: float = 0.0,
) -> SweepHit2D | None:
"""Substepped, non-mutating sweep of a body's shapes against the space.
See :meth:`~simvx.core.physics.world2d.Physics2DWorld.sweep_body`. Chipmunk
exposes no swept-shape time-of-impact for an un-added shape, so this substeps
along ``motion`` with a step count sized from the mover's smallest feature and
capped at 64; the substep that penetrates a blocking body brackets the contact,
and the bracket is BISECTED against that blocker alone, so ``distance`` is a
bound good to ``|motion| / (substeps * 2**_SWEEP_REFINE_STEPS)`` rather than to
one whole substep. It is still a lower bound and may be exactly ``0.0`` for a
sweep that begins in contact.
Structurally non-mutating: the sweep runs on a transient un-added probe body
built by :meth:`_probe_body`, so the space's spatial index is never touched
(the mover's own shapes stay indexed at its real pose, and ``from_transform``
costs nothing). ``skin`` is accepted and IGNORED: the substep quantum already
exceeds any sane skin.
A body blocks only when it is not the mover, is not a sensor, passes the
canonical AND layer/mask rule, is not rejected by the one-way filter, and
presents a contact that OPPOSES the sweep. Fraction ``0`` is never sampled
(the substep loop starts at ``st = 1``), so this backend cannot report the
cast axis as a normal.
"""
rec = self._bodies[handle]
m = _as_array2(motion)
dist = float(np.linalg.norm(m))
if dist < 1e-9:
return None
direction = (m / dist).astype(np.float32)
if from_transform is None:
start = np.array([rec.body.position.x, rec.body.position.y], dtype=np.float32)
rotation = float(rec.body.angle)
else:
start, rotation = self._unpack_transform(from_transform)
body, shapes = self._probe_body(rec.sdef, start, rotation)
flt = pymunk.ShapeFilter(categories=rec.collision_layer & 0xFFFFFFFF, mask=rec.collision_mask & 0xFFFFFFFF)
for s in shapes:
s.filter = flt
feature = max(self._feature_size(rec.sdef), 1e-4)
steps = min(64, max(1, math.ceil(dist / (feature * 0.5))))
prev_frac = 0.0
for st in range(1, steps + 1):
frac = st / steps
body.position = (float(start[0] + m[0] * frac), float(start[1] + m[1] * frac))
for s in shapes:
for info in self._space.shape_query(s):
ho = self._type_to_handle.get(info.shape.collision_type)
if ho is None or ho == handle:
continue
orec = self._bodies.get(ho)
if orec is None or orec.is_sensor:
continue
if not (
(rec.collision_mask & orec.collision_layer) and (orec.collision_mask & rec.collision_layer)
):
continue
cps = info.contact_point_set
# shape_query normal points queried->other; separating normal
# toward the mover is the negation.
n = np.array([-cps.normal.x, -cps.normal.y], dtype=np.float32)
# One-way filter, run BEFORE the opposition test and honouring a
# one-way MOVER as well as a one-way blocker, so a body sweeping
# up through a platform passes and a top landing is blocked. The
# mover's relative approach is this sweep's motion (the stored
# linear velocities do not drive a kinematic sweep).
if (rec.one_way or orec.one_way) and self._sweep_one_way_rejects(rec, orec, n, m):
continue
if float(np.dot(direction, n)) > -1e-4:
continue # non-opposing touch: not a blocker
# Re-bound so the closure's defaults carry the narrowed types the
# two `continue` guards above established; a default expression
# is read at the declared type, not the narrowed one.
blocker: BodyHandle = ho
blocker_rec: _BodyRec = orec
def _blocks_at(f: float, ho: BodyHandle = blocker, orec: _BodyRec = blocker_rec) -> bool:
"""The scan's own acceptance test, re-asked at fraction ``f``.
A closure rather than a copy of the predicate, so the bisect can
only ever converge on the blocker the scan actually found: the
query is re-run at ``f`` and every result for another body is
discarded. The ONE-WAY filter is inside it for the same reason.
Omitting it would treat a contact the scan skipped as blocking
and converge on a shorter bound than the real blocker's.
Chipmunk's pairwise ``cpShapesCollide`` would be the cheaper
question, but pymunk's ``Shape.shapes_collide`` wrapper cannot
express a non-overlapping answer at all, which is exactly the
answer a bisect needs; see BUGS.md
``bug-pymunks-shapes_collide-raises-assertionerror-on-shapes-that-do-not-touch``.
"""
body.position = (float(start[0] + m[0] * f), float(start[1] + m[1] * f))
for probe_shape in shapes:
for probe_info in self._space.shape_query(probe_shape):
if self._type_to_handle.get(probe_info.shape.collision_type) != ho:
continue
pn = probe_info.contact_point_set.normal
nf = np.array([-pn.x, -pn.y], dtype=np.float32)
if (rec.one_way or orec.one_way) and self._sweep_one_way_rejects(rec, orec, nf, m):
continue
if float(np.dot(direction, nf)) <= -1e-4:
return True
return False
# See the builtin 2D twin for the bracket and the refinement count.
lo, hi = prev_frac, frac
for _ in range(_SWEEP_REFINE_STEPS):
mid = 0.5 * (lo + hi)
if _blocks_at(mid):
hi = mid
else:
lo = mid
reached = (start + m * lo).astype(np.float32)
travelled = dist * lo
point = (reached - n * feature).astype(np.float32)
return SweepHit2D(body=ho, point=Vec2(point), normal=Vec2(n), distance=travelled)
prev_frac = frac
return None
def _pymunk_world_factory_2d(gravity: Vec2) -> Physics2DWorld:
"""Build a :class:`PymunkPhysics2D` for the given gravity (the world factory)."""
return PymunkPhysics2D(gravity=gravity)
[docs]
def register() -> None:
"""Self-register the pymunk backend with the backend registry.
Called on import (below) and idempotent (``register_backend`` replaces a same-
named entry), so importing this module installs the ``"pymunk"`` backend as an
auto-discoverable native (mirroring miniaudio's "installed -> used" model).
"""
from .backends import BackendEntry, register_backend
register_backend(
BackendEntry(
name="pymunk",
world_factory=None, # pymunk is 2D-only (Chipmunk2D); 3D falls back to Builtin
world_factory_2d=_pymunk_world_factory_2d,
native=True,
)
)
register()
__all__ = ["PymunkPhysics2D", "register"]