THE KEY EMERGES signal_model.py
Defines two independent latent curves and generates noisy exposure streams.
CURVE CHOICES ARE NOW WORLD-JUSTIFIED:
The world constraint was defined first (world_constraint.py).
C1 and C2 were chosen AFTER, to support recovery of the invariants
the world requires without encoding the final composition.
World requires from arm 1: a stable mean radius near R_lock = 1.0
→ C1 is an affine-distorted circle (ellipse with a != b)
→ Not obviously circular in raw noisy form
→ Recoverable to rotational/radial invariance under lawful transform
World requires from arm 2: a periodic modulation with omega=3, phi=pi/6
→ C2 is a sinusoidal ridge with those exact parameters
→ Sampling distortion hides the regularity in raw form
→ Recoverable to periodic invariance under lawful transform
WHAT THIS IS NOT:
These curves do not encode the key. They encode recoverable invariants
that are compatible with the world. The system must still discover those
invariants under noise and sampling distortion. The composition that
satisfies the world is not present in either curve alone.
INDEPENDENCE REQUIREMENT:
C1 and C2 are genuinely independent sources. They are not two views
of the same hidden object. Their eventual composition compatibility
is discovered by the system, not pre-encoded here.
ARCHITECTURAL NOTE:
The latent curves are the hidden truth the system never directly observes.
The system only ever sees noisy, sampled projections (Exposure objects).
No semantic labels are attached to any exposure.
from **future** import annotations
import numpy as np
from abc import ABC, abstractmethod
from typing import Tuple
from core_types import ArmID, Exposure, ExposureStream
#
# Abstract base
#
class LatentCurve(ABC):
Abstract latent curve. The viewer never sees this directly.
It exists only as the source from which noisy exposures are sampled.
```
No semantic labels are attached to subclasses. The name
(EllipseCurve, SinusoidalRidge) is for developer orientation only
and must not appear in any on-screen text or inference path.
"""
@abstractmethod
def sample(self, s: np.ndarray) -> np.ndarray:
"""
Given parameter values s in [0, 1], return clean 2D points.
Shape: (len(s), 2)
"""
...
@abstractmethod
def describe_hidden_truth(self) -> str:
"""
Human-readable description for logging/debugging only.
Never surfaced to the viewer or used in inference.
"""
...
```
#
# C1: Ellipse rotationally recoverable
#
class AffineDistortedCircle(LatentCurve):
C1(s): affine-distorted circle (ellipse) in polar-parameterized form.
```
WORLD-JUSTIFIED CHOICE:
The world requires a stable mean radius near R_lock = 1.0 (Constraint A)
and a recoverable radial boundary for composition.
This curve is NOT an obvious circle in raw noisy form:
- semi_major != semi_minor (aspect ratio != 1)
- tilt rotates the axes off-canonical orientation
- noise and sampling jitter further obscure the structure
Under lawful radial normalization transforms, rotational invariance
is recoverable. The system must discover this it is not labeled.
Parametric form (Chat's specification):
x = a * cos(2*pi*s)
y = b * sin(2*pi*s)
with a = 1.12, b = 0.88, plus tilt and noise at sampling time.
The mean radius across a full revolution approximates R_lock = 1.0
(mean of sqrt((a*cos)^2 + (b*sin)^2) over s in [0,1]).
This ensures the recovered structure is admissible under Constraint A.
"""
def __init__(self,
semi_major: float = 1.12,
semi_minor: float = 0.88,
tilt_rad: float = 0.3,
center: tuple = (0.0, 0.0)):
self.semi_major = semi_major
self.semi_minor = semi_minor
self.tilt_rad = tilt_rad
self.center = np.array(center, dtype=float)
c, s_ = np.cos(tilt_rad), np.sin(tilt_rad)
self._R = np.array([[c, -s_], [s_, c]])
def sample(self, s: np.ndarray) -> np.ndarray:
theta = 2.0 * np.pi * s
local = np.stack([
self.semi_major * np.cos(theta),
self.semi_minor * np.sin(theta)
], axis=1)
rotated = local @ self._R.T
return rotated + self.center
def describe_hidden_truth(self) -> str:
return (f"AffineDistortedCircle: a={self.semi_major}, "
f"b={self.semi_minor}, tilt={np.degrees(self.tilt_rad):.1f}deg "
f"[world-justified: recoverable to R_lock=1.0]")
```
#
# C2: Sinusoidal ridge periodically recoverable
#
class WorldJustifiedRidge(LatentCurve):
C2(s): sinusoidal ridge with world-required periodic parameters,
obscured by sampling distortion and noise.
```
WORLD-JUSTIFIED CHOICE:
The world requires a periodic modulation with:
omega_lock = 3, phi_lock = pi/6, A_lock = 0.12
(Constraint B in world_constraint.py)
This curve embeds those parameters as the recoverable truth.
It is NOT obviously periodic in raw form:
- non-uniform sampling distortion hides the regularity
- additive noise at sampling time further obscures structure
- the x-axis parameterization is irregular
RAW AMPLITUDE vs WORLD AMPLITUDE:
Raw amplitude = 0.25. World requires A_lock = 0.12.
These are intentionally different.
Raw signal amplitude may exceed world-admissible amplitude.
Surviving transforms must recover periodic structure AND normalize
it into admissible range. This is not target matching; it is
lawful normalization under world constraint.
The arm 2 transform family includes y_scale as a generic lawful
operator. A surviving transform will discover both:
- the periodic structure (frequency and phase invariance)
- an amplitude normalization into the admissible range
y_scale bounds are broad (0.1 to 5.0) generic, not answer-tuned.
The correct normalization (~0.48) must be discovered, not injected.
Under lawful phase/frequency recovery transforms, periodic invariance
with the correct parameters is recoverable. The system must discover
this it is not labeled.
Parametric form (Chat's specification, amplitude updated):
x = s + sampling_distortion
y = A0 * sin(omega * 2*pi*s + phi0)
with A0=0.25 (raw), omega=3, phi0=pi/6 (matching world parameters)
The sampling distortion hides the periodicity in the x-coordinate.
The system sees an irregular point cloud, not a clean sinusoid.
NOTE: omega here refers to cycles over s in [0,1], so the full
argument is omega * 2*pi*s, giving 3 complete cycles over s=[0,1].
This matches omega_lock=3 in world_constraint.py when the profile
is expressed in theta in [0, 2*pi].
"""
def __init__(self,
amplitude: float = 0.25,
omega: float = 3.0,
phase: float = np.pi / 6.0,
distortion_strength: float = 0.05):
"""
distortion_strength must satisfy:
d < 1 / (2*pi*2.3) = 0.0692
Above this bound, dx/ds becomes negative somewhere and the
x-coordinate folds, making arc-length reparameterization
impossible (non-injective mapping). At 0.05, the x-axis
remains monotone and the arc-length remap can recover the
uniform parameterization needed for frequency analysis.
The distortion is still strong enough to hide the periodicity
in raw form the dominant FFT bin shifts away from omega=3
but the arc-length transform can undo it.
"""
self.amplitude = amplitude
self.omega = omega
self.phase = phase
self.distortion_strength = distortion_strength
def sample(self, s: np.ndarray) -> np.ndarray:
"""
Sample N points along the distorted ridge.
The distortion is deterministic given s baked into the curve.
Additive noise is applied separately in generate_exposure_stream.
"""
# Sampling distortion: nonlinear warp of x-axis
x_distorted = s + self.distortion_strength * np.sin(2.0 * np.pi * 2.3 * s)
# Periodic y-displacement with world-required frequency and phase
y = self.amplitude * np.sin(
self.omega * 2.0 * np.pi * s + self.phase
)
return np.stack([x_distorted, y], axis=1)
def describe_hidden_truth(self) -> str:
return (f"WorldJustifiedRidge: A_raw={self.amplitude} "
f"(world requires A_lock=0.12 transform must normalize), "
f"omega={self.omega}, phase={self.phase:.4f} (pi/6), "
f"distortion={self.distortion_strength} "
f"[world-justified: recoverable to omega_lock=3, phi_lock=pi/6]")
```
#
# Exposure generation
#
def generate_exposure_stream(
curve: LatentCurve,
arm_id: ArmID,
N: int = 96,
K: int = 8,
noise_sigma: float = 0.08,
jitter_sigma: float = 0.01,
rng: Optional[np.random.Generator] = None
) -> ExposureStream:
Generate K independent noisy exposures from one latent curve.
```
Each exposure is:
x_i^(k) = C(s_i + delta_s_i^(k)) + epsilon_i^(k)
where:
s_i are base sample parameters, uniform on [0,1]
delta_s_i^(k) is optional per-point sampling jitter ~ N(0, jitter_sigma^2)
epsilon_i^(k) is additive 2D Gaussian noise ~ N(0, sigma^2 * I)
ARCHITECTURAL NOTE:
These exposures are noisy projections of unknown latent structure.
No semantic labels are attached. The receiving modules see only
point clouds. They do not know what curve generated them.
Parameters
----------
curve : LatentCurve instance (hidden from inference path)
arm_id : which arm this stream belongs to
N : points per exposure
K : number of independent exposures
noise_sigma : std dev of additive 2D Gaussian noise
jitter_sigma : std dev of parameter sampling jitter (0 = no jitter)
rng : numpy Generator for reproducibility
Returns
-------
ExposureStream
"""
if rng is None:
rng = np.random.default_rng()
# Base sample parameters uniform on [0, 1]
s_base = np.linspace(0.0, 1.0, N, endpoint=False)
exposures = []
for k in range(K):
# Optional sampling jitter
if jitter_sigma > 0:
delta_s = rng.normal(0.0, jitter_sigma, size=N)
s_k = np.clip(s_base + delta_s, 0.0, 1.0)
else:
s_k = s_base.copy()
# Sample clean points from latent curve
clean_points = curve.sample(s_k) # (N, 2)
# Add independent 2D Gaussian noise
noise = rng.normal(0.0, noise_sigma, size=(N, 2))
noisy_points = clean_points + noise # (N, 2)
exposures.append(Exposure(
arm_id=arm_id,
exposure_index=k,
points=noisy_points,
sample_params=s_k
))
return ExposureStream(
arm_id=arm_id,
exposures=exposures,
noise_sigma=noise_sigma,
n_points=N
)
```
#
# Factory builds both arms together
#
from typing import Optional, Tuple
def build_signal_arms(
n_points: int = 96,
k_exposures: int = 8,
noise_sigma: float = 0.08,
seed: int = 42
) -> Tuple[ExposureStream, ExposureStream, LatentCurve, LatentCurve]:
Build both independent signal arms using world-justified curves.
```
The world constraint was defined first. These curves were chosen
to make the required invariants discoverable not to pre-encode
the final composition.
Returns the two exposure streams AND the latent curves.
The latent curves are returned for logging/debugging only.
They must never be passed into the inference pipeline.
Returns
-------
stream_1, stream_2, curve_1, curve_2
"""
rng = np.random.default_rng(seed)
# C1: affine-distorted circle
# World-justified: recoverable to R_lock=1.0 under radial normalization
# a=1.12, b=0.88 per Chat's specification
curve_1 = AffineDistortedCircle(
semi_major=1.12,
semi_minor=0.88,
tilt_rad=0.3
)
# C2: world-justified sinusoidal ridge
# Raw amplitude 0.25 > world A_lock 0.12 intentional.
# Surviving transforms must recover periodic structure AND
# normalize amplitude into admissible range.
# y_scale bounds in transform_families are broad (0.15.0).
# The correct normalization must be discovered, not injected.
#
# distortion_strength=0.05 kept below monotonicity bound 0.069
# (d must satisfy d < 1/(2*pi*2.3) for x to remain monotone).
# At 0.05 the periodicity is hidden in raw form but recoverable
# via arc-length reparameterization.
curve_2 = WorldJustifiedRidge(
amplitude=0.25,
omega=3.0,
phase=np.pi / 6.0,
distortion_strength=0.05
)
stream_1 = generate_exposure_stream(
curve_1, ArmID.ARM_1,
N=n_points, K=k_exposures,
noise_sigma=noise_sigma,
rng=rng
)
stream_2 = generate_exposure_stream(
curve_2, ArmID.ARM_2,
N=n_points, K=k_exposures,
noise_sigma=noise_sigma,
rng=rng
)
return stream_1, stream_2, curve_1, curve_2
```
#
# Quick sanity check run directly to verify
#
if **name** == **main**:
from typing import Optional
```
stream_1, stream_2, c1, c2 = build_signal_arms(seed=42)
print("=== SIGNAL MODEL SANITY CHECK ===")
print(f"C1 hidden truth: {c1.describe_hidden_truth()}")
print(f"C2 hidden truth: {c2.describe_hidden_truth()}")
print()
print(f"Arm 1 stream: K={stream_1.K} exposures, N={stream_1.n_points} points each")
print(f" exposure[0] shape: {stream_1.exposures[0].points.shape}")
print(f" exposure[0] mean: {stream_1.exposures[0].points.mean(axis=0).round(3)}")
print(f" exposure[0] std: {stream_1.exposures[0].points.std(axis=0).round(3)}")
print()
print(f"Arm 2 stream: K={stream_2.K} exposures, N={stream_2.n_points} points each")
print(f" exposure[0] shape: {stream_2.exposures[0].points.shape}")
print(f" exposure[0] mean: {stream_2.exposures[0].points.mean(axis=0).round(3)}")
print(f" exposure[0] std: {stream_2.exposures[0].points.std(axis=0).round(3)}")
print()
print("Cross-arm independence check:")
corr = np.corrcoef(
stream_1.exposures[0].points.flatten(),
stream_2.exposures[0].points.flatten()
)[0, 1]
print(f" Pearson correlation between arm1[0] and arm2[0]: {corr:.4f}")
print(" (Expected: near 0 genuinely independent sources)")
print()
print("PASS signal_model.py is operational.")
```