feat: improve camera motion and shot transitions

This commit is contained in:
KINDNICK 2026-08-03 06:27:17 +09:00
parent 02d76bc87c
commit 80ff929a1c
22 changed files with 2563 additions and 129 deletions

View File

@ -1,5 +1,18 @@
# Changelog
## 0.1.6
- Updated the separately installable Camera AI package to 0.4.2 and the frozen
Windows worker to 0.1.2.
- Fixed duplicate script GUIDs that could exclude Camera AI package sources from
the Unity editor assembly.
- Added YAMO-calibrated deceleration/acceleration around meaningful direction
changes and rendered end-to-start adjacent-shot transition scoring.
- Added C2-continuous braking seams, single-pass turn processing, and a
source-relative acceleration/jerk guard that demotes unsafe rewrites.
- Preserved generated braking curves in the editable Timeline simplification
preset and added regression coverage for 45-degree and 90-degree turns.
## 0.1.5
- Added the separately installable `Mingle Camera Work AI` package under

View File

@ -1,5 +1,25 @@
# Changelog
## 0.4.2 - 2026-08-03
- Fixed Unity assembly exclusion caused by duplicate legacy/package script GUIDs.
- Added YAMO-calibrated, angle-aware camera translation kinematics: concentrated
direction changes slow before the turn and accelerate afterward, while smooth
arcs and naturally eased motion remain untouched.
- Applied the kinematic pass exactly once per shot: Orbit reversals are measured
in subject-relative space and other motion families in final world space.
- Made the braking window C2-continuous at its outer seams, removed cumulative-
minimum speed kinks, and rejected any rewrite that introduces an acceleration
or jerk regression instead of chasing newly-created seam events.
- Added rendered end-to-start transition scoring against the actually selected
adjacent cameras, including both neighbours during selected-shot regeneration.
- Hard-rejected only perceptible near-duplicate jump cuts and unbridged extreme
axis crossings; screen jumps and opposing motion remain soft ranking signals.
- Preserved the braking envelope around 40-degree-or-greater direction changes
when editable Unity animation curves are simplified.
- Bumped the packaged worker to 0.1.2 and invalidated stale candidate caches for
the new transition and trajectory policies.
## 0.4.1 - 2026-08-03
- Added the UXML/USS camera-generation workflow with a compact narrow-dock

View File

@ -17,7 +17,7 @@ must remain together.
In Unity Package Manager, choose **Add package from git URL** and enter:
```text
https://kindnick-git.duckdns.org/mingle/streamingle-unity-utilities.git?path=/CameraAI~#v0.1.5
https://kindnick-git.duckdns.org/mingle/streamingle-unity-utilities.git?path=/CameraAI~#v0.1.6
```
The initial package download is large because the frozen Windows worker is

View File

@ -1,2 +1,2 @@
fileFormatVersion: 2
guid: f8bdd831f8794ea18ca288f3745462b1
guid: 572fff6d310f4e669243b957117466c1

View File

@ -1,2 +1,2 @@
fileFormatVersion: 2
guid: 3ef4feff8a1c42a0be6affcf5d6429db
guid: c0bd869f0b114cca884580f0d6229b19

View File

@ -41,6 +41,15 @@ namespace Streamingle.Editor
private const string PreviewGenerationProvenanceName =
"AI Camera Preview Generation Provenance";
private const double CutBoundaryGuardSeconds = 0.000001;
private const int DirectionChangeDetectionFlankFrames = 5;
// Matches the worker's 0.35-second braking radius at 60 fps.
private const int DirectionChangePreserveRadiusFrames = 21;
private const float DirectionChangeMinimumSpeedMetersPerSecond = 0.04f;
// Preserve the generated braking envelope for every concentrated turn
// covered by the worker policy (40 degrees and above), not only an
// almost complete reversal. Linear curve simplification may otherwise
// collapse a mild 40--55 or clear 60/90-degree slowdown into a kink.
private const float DirectionChangeMaximumCosine = 0.76604444f;
public const CurveSimplificationPreset DefaultCurveSimplificationPreset =
CurveSimplificationPreset.Balanced;
@ -2451,6 +2460,7 @@ namespace Streamingle.Editor
AddScalarExtrema(values.Count, index => values[index].x, required);
AddScalarExtrema(values.Count, index => values[index].y, required);
AddScalarExtrema(values.Count, index => values[index].z, required);
AddDirectionChangeNeighborhoods(times, values, required);
return SimplifyIndices(
times,
maximumError,
@ -2593,6 +2603,61 @@ namespace Streamingle.Editor
indices.Add(maximumIndex);
}
private static void AddDirectionChangeNeighborhoods(
IReadOnlyList<double> times,
IReadOnlyList<Vector3> values,
ISet<int> indices)
{
var flank = DirectionChangeDetectionFlankFrames;
if (values.Count < flank * 2 + 1)
{
return;
}
for (var center = flank; center + flank < values.Count; center++)
{
var beforeDuration = times[center] - times[center - flank];
var afterDuration = times[center + flank] - times[center];
if (beforeDuration <= double.Epsilon ||
afterDuration <= double.Epsilon)
{
continue;
}
var beforeVelocity =
(values[center] - values[center - flank]) /
(float)beforeDuration;
var afterVelocity =
(values[center + flank] - values[center]) /
(float)afterDuration;
var beforeSpeed = beforeVelocity.magnitude;
var afterSpeed = afterVelocity.magnitude;
if (Mathf.Min(beforeSpeed, afterSpeed) <
DirectionChangeMinimumSpeedMetersPerSecond)
{
continue;
}
var cosine = Vector3.Dot(beforeVelocity, afterVelocity) /
(beforeSpeed * afterSpeed);
if (cosine > DirectionChangeMaximumCosine)
{
continue;
}
var start = Mathf.Max(
0,
center - DirectionChangePreserveRadiusFrames);
var end = Mathf.Min(
values.Count - 1,
center + DirectionChangePreserveRadiusFrames);
for (var index = start; index <= end; index++)
{
indices.Add(index);
}
}
}
private static Quaternion[] MakeQuaternionSequenceContinuous(
IReadOnlyList<Quaternion> rotations)
{

View File

@ -16,7 +16,7 @@ Cinemachine Track이나 기존 카메라 애니메이션은 필요하지 않으
Unity Package Manager의 `Add package from git URL`에는 다음 주소를 사용할 수
있습니다.
`https://kindnick-git.duckdns.org/mingle/streamingle-unity-utilities.git?path=/CameraAI~#v0.1.5`
`https://kindnick-git.duckdns.org/mingle/streamingle-unity-utilities.git?path=/CameraAI~#v0.1.6`
배포 패키지에는 Windows x64용 `CWCameraWorker` 폴더 전체가 포함됩니다.
Python은 따로 설치하지 않아도 되지만, Git 패키지의 대용량 바이너리를 받으려면

View File

@ -1,2 +1,2 @@
fileFormatVersion: 2
guid: 30c854c9afbd493f8b2e660c23584f83
guid: f3fb35740c814042bce3842307136af2

View File

@ -155,6 +155,168 @@ namespace Streamingle.Editor
Is.LessThan(exact.DutchIndices.Length));
}
[Test]
public void EditablePresetPreservesStopThroughDirectionChangeKeys()
{
const int count = 121;
const int center = 60;
const int preserveRadius = 21;
var times = new double[count];
var positions = new Vector3[count];
var rotations = new Quaternion[count];
var fieldOfView = new float[count];
var dutch = new float[count];
for (var index = 0; index < count; index++)
{
times[index] = index / 60.0;
float x;
if (index <= center)
{
var t = index / (float)center;
x = -t * t * t + t * t + t;
}
else
{
var t = (index - center) /
(float)(count - 1 - center);
var progress = -t * t * t + 2f * t * t;
x = 1f - progress;
}
positions[index] = new Vector3(x, 1.6f, -4f);
rotations[index] = Quaternion.identity;
fieldOfView[index] = 40f;
dutch[index] = 0f;
}
var result = Simplify(
new CameraSamples(
times,
positions,
rotations,
fieldOfView,
dutch),
AICameraTimelinePreviewImporter.CurveSimplificationPreset.Editable);
for (var index = center - preserveRadius;
index <= center + preserveRadius;
index++)
{
Assert.That(
result.PositionIndices,
Does.Contain(index),
$"Direction-change key {index} was simplified away.");
}
}
[Test]
public void EditablePresetPreservesBrakedRightAngleTurnKeys()
{
const int count = 121;
const int center = 60;
const int preserveRadius = 21;
var times = new double[count];
var positions = new Vector3[count];
var rotations = new Quaternion[count];
var fieldOfView = new float[count];
var dutch = new float[count];
for (var index = 0; index < count; index++)
{
times[index] = index / 60.0;
if (index <= center)
{
var t = index / (float)center;
var progress = -t * t * t + t * t + t;
positions[index] = new Vector3(progress, 1.6f, -4f);
}
else
{
var t = (index - center) /
(float)(count - 1 - center);
var progress = -t * t * t + 2f * t * t;
positions[index] = new Vector3(1f, 1.6f, -4f + progress);
}
rotations[index] = Quaternion.identity;
fieldOfView[index] = 40f;
dutch[index] = 0f;
}
var result = Simplify(
new CameraSamples(
times,
positions,
rotations,
fieldOfView,
dutch),
AICameraTimelinePreviewImporter.CurveSimplificationPreset.Editable);
for (var index = center - preserveRadius;
index <= center + preserveRadius;
index++)
{
Assert.That(
result.PositionIndices,
Does.Contain(index),
$"Right-angle braking key {index} was simplified away.");
}
}
[Test]
public void EditablePresetPreservesBrakedFortyFiveDegreeTurnKeys()
{
const int count = 121;
const int center = 60;
const int preserveRadius = 21;
var times = new double[count];
var positions = new Vector3[count];
var rotations = new Quaternion[count];
var fieldOfView = new float[count];
var dutch = new float[count];
var outgoing = new Vector3(1f, 0f, 1f).normalized;
for (var index = 0; index < count; index++)
{
times[index] = index / 60.0;
if (index <= center)
{
var t = index / (float)center;
var progress = -t * t * t + t * t + t;
positions[index] = new Vector3(progress, 1.6f, -4f);
}
else
{
var t = (index - center) /
(float)(count - 1 - center);
var progress = -t * t * t + 2f * t * t;
positions[index] =
new Vector3(1f, 1.6f, -4f) + outgoing * progress;
}
rotations[index] = Quaternion.identity;
fieldOfView[index] = 40f;
dutch[index] = 0f;
}
var result = Simplify(
new CameraSamples(
times,
positions,
rotations,
fieldOfView,
dutch),
AICameraTimelinePreviewImporter.CurveSimplificationPreset.Editable);
for (var index = center - preserveRadius;
index <= center + preserveRadius;
index++)
{
Assert.That(
result.PositionIndices,
Does.Contain(index),
$"Forty-five-degree braking key {index} was simplified away.");
}
}
[Test]
public void ExplicitDirectorScopesSeparateSameNamedSongDirectors()
{

View File

@ -1,2 +1,2 @@
fileFormatVersion: 2
guid: 1b88c37fe4bd4e48818660d18810ae43
guid: 8e37a17d5d9e451dad2a8bbc9f1d2f8a

View File

@ -1,2 +1,2 @@
fileFormatVersion: 2
guid: 7a3ea2034f8a4ce1b45f1673472e5e6f
guid: 39679e5d695f47288b75d9429735481a

View File

@ -1,2 +1,2 @@
fileFormatVersion: 2
guid: 74ed53c76a0a4c6c82ea504d79132c63
guid: a959d1021a8c4fce976089fee89d1978

View File

@ -1,3 +1,3 @@
fileFormatVersion: 2
guid: b557e27541794769bd7798589e9c5a9b
guid: af51bc7a0bfd4d4492a7da670bb46a59
timeCreated: 1785484800

Binary file not shown.

View File

@ -1,15 +1,17 @@
{
"schemaVersion": "cw-camera-worker-build-identity-v1",
"workerVersion": "0.1.0",
"createdUtc": "2026-08-02T09:05:09.155787+00:00",
"workerVersion": "0.1.2",
"createdUtc": "2026-08-02T21:19:54.902869+00:00",
"python": "3.10.11",
"sourceRootRelative": "cwai_sources/repository",
"sourceSha256": {
"MachineLearning/CameraDirector/adjacent_transition.py": "6c1b62c2996960af23d62e9c3acd3114600effda03273b006f7076c467c4bdfa",
"MachineLearning/CameraDirector/build_hybrid_preparation_cache.py": "164a3574d5fad05a627bf6d0c6169d42039043bf9a2cd62effbd978359e828f2",
"MachineLearning/CameraDirector/camera_kinematics.py": "c7b85616ef44a0f28f11702459d67d78aad7cf26ed00c2548ecb78b06fc27025",
"MachineLearning/CameraDirector/camera_runtime_data.py": "f787b84257aef4a71905198ab2601673fbe639ca5c37e3a957de12116dceab34",
"MachineLearning/CameraDirector/cw_camera_runtime.py": "c2dff5cf41b8b487272d20e3847b556537f597b185a0f4b0d85ca577b4a51709",
"MachineLearning/CameraDirector/data_driven_cut_planner.py": "b6d48d03f8725aa8327abe81afa483c43edc368e7dee9f9e7228e21982f71ad2",
"MachineLearning/CameraDirector/generate_hybrid.py": "928b20dae7414be577f193319c3816f464286e2ce20366a8e743995e6b6e4dd6",
"MachineLearning/CameraDirector/generate_hybrid.py": "4f3eeb278d9f516c1c4de30b312c2daf0fe33692d912cedd8a3240533e75d467",
"MachineLearning/CameraDirector/hybrid_candidate_cache.py": "a0a5b6a8f612f18cb2875394f17ed89950e1e62e38c1226457d9e35848f6e380",
"MachineLearning/CameraDirector/hybrid_cut_reference.py": "1f136f5514fddf04b21dfb9220d9559a1fff8cd7c17388d81875edbd82a672ac",
"MachineLearning/CameraDirector/hybrid_preparation_cache.py": "1b0df53124109dee2d744777979a0553999cb276cb6f538a17ddce6516e3da8b",
@ -19,9 +21,11 @@
"MachineLearning/CameraDirector/train.py": "4e0fcfd88ec2c241e66497f109420a4dc3cf958d835f78bef4279f58e6d9a8b1"
},
"preparationLogicIdentifier": "123407b6741f05418a4f3237790492d9b74e13b87090acff2b1a07cf9de2912f",
"candidateLogicIdentifier": "6514a8e9f0d25b310200ff62e914f9449814dfc24f2b50cfd1b5e417213e88cf",
"generationCodeIdentifier": "1f50c81bd9ecf865d08cd52d3c1c2ab1b910aa33e51752737e345fdd51856ee8",
"candidateLogicIdentifier": "77793481c02a9b4eda52d009a771444995bd59819717bb98aa46769ec61d4d88",
"generationCodeIdentifier": "95920673d29794c2a7e2866ff8aa3b57089d1cda4f4f0023ec555139efa97bb7",
"generationCodeFiles": [
"adjacent_transition.py",
"camera_kinematics.py",
"camera_runtime_data.py",
"data_driven_cut_planner.py",
"generate_hybrid.py",

View File

@ -0,0 +1,450 @@
"""Auditable pairwise camera-cut transition scoring.
The generator already decides *when* to cut from music and performance data.
This module evaluates the complementary editorial question: whether the end of
one selected camera and the beginning of the next selected camera form a clear,
intentional cut. It deliberately uses only compact rendered boundary state so
the same policy can be applied to a full generation and to both neighbours of a
single-shot regeneration.
"""
from __future__ import annotations
import math
from typing import Mapping
import numpy as np
TRANSITION_STATE_SCHEMA_VERSION = "camera-transition-boundary-state-v1"
TRANSITION_SCORE_SCHEMA_VERSION = "pairwise-camera-transition-score-v1"
TRANSITION_POLICY_VERSION = "rendered-end-start-editorial-continuity-v3"
_MINIMUM_MOVING_SPEED_MPS = 0.12
_OPPOSING_MOTION_START_DEGREES = 110.0
def _finite_array(
value: object,
shape: tuple[int, ...],
label: str,
) -> np.ndarray:
result = np.asarray(value, dtype=np.float64)
if result.shape != shape or not np.isfinite(result).all():
raise ValueError(f"{label} must be finite with shape {shape}.")
return result
def _normalized(value: np.ndarray) -> np.ndarray:
length = float(np.linalg.norm(value))
if length <= 1e-9:
return np.zeros_like(value, dtype=np.float64)
return np.asarray(value, dtype=np.float64) / length
def _vector_angle_degrees(first: np.ndarray, second: np.ndarray) -> float:
first_unit = _normalized(first)
second_unit = _normalized(second)
if not np.any(first_unit) or not np.any(second_unit):
return 0.0
cosine = float(np.clip(np.dot(first_unit, second_unit), -1.0, 1.0))
return float(np.degrees(np.arccos(cosine)))
def _quaternion_angle_degrees(first: np.ndarray, second: np.ndarray) -> float:
first_unit = _normalized(first)
second_unit = _normalized(second)
if not np.any(first_unit) or not np.any(second_unit):
return 0.0
cosine = float(np.clip(abs(np.dot(first_unit, second_unit)), 0.0, 1.0))
return float(np.degrees(2.0 * np.arccos(cosine)))
def _robust_boundary_velocity(
values: np.ndarray,
sample_rate: float,
*,
entering: bool,
window_frames: int,
) -> np.ndarray:
if len(values) < 2:
return np.zeros(3, dtype=np.float64)
steps = np.diff(values, axis=0) * sample_rate
width = min(max(1, int(window_frames) - 1), len(steps))
selected = steps[:width] if entering else steps[-width:]
return np.median(selected, axis=0).astype(np.float64)
def _boundary_summary(
position: np.ndarray,
rotation: np.ndarray,
fov: np.ndarray,
target: np.ndarray,
composition: np.ndarray,
sample_rate: float,
*,
entering: bool,
window_frames: int,
) -> dict[str, object]:
width = min(max(1, int(window_frames)), len(position))
frame_slice = slice(0, width) if entering else slice(len(position) - width, None)
endpoint = 0 if entering else -1
camera_position = np.median(position[frame_slice], axis=0)
target_position = np.median(target[frame_slice], axis=0)
relative_path = position - target
subject_values = composition[frame_slice]
valid_subject = (
np.isfinite(subject_values[:, :3]).all(axis=1)
& (subject_values[:, 3] >= 0.5)
)
if np.any(valid_subject):
subject_uv = np.median(subject_values[valid_subject, :2], axis=0)
subject_height = float(np.median(subject_values[valid_subject, 2]))
else:
subject_uv = np.asarray([0.5, 0.5], dtype=np.float64)
subject_height = 0.0
world_velocity = _robust_boundary_velocity(
position,
sample_rate,
entering=entering,
window_frames=window_frames,
)
relative_velocity = _robust_boundary_velocity(
relative_path,
sample_rate,
entering=entering,
window_frames=window_frames,
)
return {
"cameraPosition": camera_position.tolist(),
"cameraRotation": _normalized(rotation[endpoint]).tolist(),
"targetPosition": target_position.tolist(),
"cameraRelativePosition": (camera_position - target_position).tolist(),
"fovDegrees": float(np.median(fov[frame_slice])),
"subjectUv": subject_uv.tolist(),
"subjectHeightRatio": subject_height,
"worldVelocityMps": world_velocity.tolist(),
"relativeVelocityMps": relative_velocity.tolist(),
"worldSpeedMps": float(np.linalg.norm(world_velocity)),
"relativeSpeedMps": float(np.linalg.norm(relative_velocity)),
}
def build_transition_state(
position: np.ndarray,
rotation: np.ndarray,
fov: np.ndarray,
target: np.ndarray,
composition: np.ndarray,
shot_type: str,
motion_type: str,
*,
sample_rate: float = 60.0,
boundary_window_frames: int = 6,
) -> dict[str, object]:
"""Build JSON-safe rendered boundary state for one camera candidate."""
position = np.asarray(position, dtype=np.float64)
rotation = np.asarray(rotation, dtype=np.float64)
fov = np.asarray(fov, dtype=np.float64)
target = np.asarray(target, dtype=np.float64)
composition = np.asarray(composition, dtype=np.float64)
frame_count = len(position)
if frame_count < 1:
raise ValueError("A transition state requires at least one frame.")
expected = {
"position": (frame_count, 3),
"rotation": (frame_count, 4),
"fov": (frame_count,),
"target": (frame_count, 3),
"composition": (frame_count, 4),
}
observed = {
"position": position.shape,
"rotation": rotation.shape,
"fov": fov.shape,
"target": target.shape,
"composition": composition.shape,
}
if expected != observed:
raise ValueError(
"Transition arrays have incompatible shapes: " + repr(observed)
)
if not all(
np.isfinite(values).all()
for values in (position, rotation, fov, target, composition)
):
raise ValueError("Transition arrays must not contain NaN or Inf.")
if not math.isfinite(sample_rate) or sample_rate <= 0.0:
raise ValueError("sample_rate must be finite and positive.")
if int(boundary_window_frames) < 1:
raise ValueError("boundary_window_frames must be positive.")
return {
"schemaVersion": TRANSITION_STATE_SCHEMA_VERSION,
"shotType": str(shot_type),
"motionType": str(motion_type),
"frameCount": frame_count,
"start": _boundary_summary(
position,
rotation,
fov,
target,
composition,
sample_rate,
entering=True,
window_frames=boundary_window_frames,
),
"end": _boundary_summary(
position,
rotation,
fov,
target,
composition,
sample_rate,
entering=False,
window_frames=boundary_window_frames,
),
}
def _validated_endpoint(state: Mapping[str, object], endpoint: str) -> dict[str, object]:
if state.get("schemaVersion") != TRANSITION_STATE_SCHEMA_VERSION:
raise ValueError("Unsupported camera transition-state schema.")
value = state.get(endpoint)
if not isinstance(value, Mapping):
raise ValueError(f"Transition state is missing {endpoint!r}.")
result = dict(value)
for key, shape in (
("cameraPosition", (3,)),
("cameraRotation", (4,)),
("targetPosition", (3,)),
("cameraRelativePosition", (3,)),
("subjectUv", (2,)),
("worldVelocityMps", (3,)),
("relativeVelocityMps", (3,)),
):
_finite_array(result.get(key), shape, key)
for key in (
"fovDegrees",
"subjectHeightRatio",
"worldSpeedMps",
"relativeSpeedMps",
):
raw = result.get(key)
if isinstance(raw, bool) or not isinstance(raw, (int, float)) or not math.isfinite(raw):
raise ValueError(f"{key} must be finite.")
return result
def score_pairwise_transition(
previous_state: Mapping[str, object],
next_state: Mapping[str, object],
) -> dict[str, object]:
"""Score the rendered END -> START relationship of two hard-cut shots.
A hard cut does not need physically continuous camera velocity. The score
instead penalizes edits that look accidental: same-size near-identical
geometry (jump cut), abrupt subject displacement, an unbridged 180-degree
axis flip, and two fast trajectories that reverse direction at the cut.
Subject scale contrast is measured but intentionally not penalized.
"""
previous = _validated_endpoint(previous_state, "end")
following = _validated_endpoint(next_state, "start")
previous_relative = _finite_array(
previous["cameraRelativePosition"], (3,), "cameraRelativePosition"
)
next_relative = _finite_array(
following["cameraRelativePosition"], (3,), "cameraRelativePosition"
)
geometry_angle = _vector_angle_degrees(previous_relative, next_relative)
rotation_angle = _quaternion_angle_degrees(
_finite_array(previous["cameraRotation"], (4,), "cameraRotation"),
_finite_array(following["cameraRotation"], (4,), "cameraRotation"),
)
fov_delta = abs(float(previous["fovDegrees"]) - float(following["fovDegrees"]))
subject_jump = float(
np.linalg.norm(
_finite_array(previous["subjectUv"], (2,), "subjectUv")
- _finite_array(following["subjectUv"], (2,), "subjectUv")
)
)
previous_height = max(float(previous["subjectHeightRatio"]), 1e-6)
next_height = max(float(following["subjectHeightRatio"]), 1e-6)
scale_ratio = max(previous_height, next_height) / min(
previous_height, next_height
)
previous_velocity = _finite_array(
previous["relativeVelocityMps"], (3,), "relativeVelocityMps"
)
next_velocity = _finite_array(
following["relativeVelocityMps"], (3,), "relativeVelocityMps"
)
previous_speed = float(previous["relativeSpeedMps"])
next_speed = float(following["relativeSpeedMps"])
velocity_angle = _vector_angle_degrees(previous_velocity, next_velocity)
same_shot_type = str(previous_state.get("shotType")) == str(
next_state.get("shotType")
)
visual_change = max(
geometry_angle / 30.0,
rotation_angle / 30.0,
fov_delta / 8.0,
abs(math.log(scale_ratio)) / math.log(1.35),
)
near_duplicate_penalty = max(0.0, 1.0 - visual_change) * (
2.0 if same_shot_type else 0.8
)
# Use what will actually be seen, not the declared W/M/C label. A declared
# medium -> close cut is still a jump cut when the rendered subject scale,
# camera axis, view rotation and screen position barely change. FOV is only
# diagnostic here because distance can compensate for a large lens change.
# A mathematically identical boundary is an invisible continuation, not a
# jump cut. In the YAMO authored audit, most broadly "similar" edges were
# in this sub-perceptual band. Keep them available (with a soft sameness
# cost) and hard-gate only a small-but-visible discontinuity.
invisible_continuation = bool(
geometry_angle < 2.0
and rotation_angle < 2.0
and fov_delta < 1.0
and scale_ratio < 1.08
and subject_jump < 0.03
)
jump_cut_risk = bool(
geometry_angle < 15.0
and rotation_angle < 20.0
and scale_ratio < 1.20
and subject_jump < 0.12
and not invisible_continuation
)
if jump_cut_risk:
near_duplicate_penalty = max(near_duplicate_penalty, 1.75)
screen_position_penalty = max(0.0, subject_jump - 0.14) / 0.20 * 1.25
neither_is_wide = "wide" not in {
str(previous_state.get("shotType", "")).lower(),
str(next_state.get("shotType", "")).lower(),
}
axis_crossing_risk = bool(neither_is_wide and geometry_angle > 150.0)
axis_crossing_penalty = (
max(0.0, geometry_angle - 150.0) / 30.0 * 0.9
if neither_is_wide
else 0.0
)
# Large scale contrast is often the editorial *reason* for a cut. YAMO's
# authored examples deliberately use roughly 2.6x--6.9x changes on the
# same axis, so retain the metric/flag without suppressing that energy.
extreme_scale_penalty = 0.0
both_moving = (
previous_speed >= _MINIMUM_MOVING_SPEED_MPS
and next_speed >= _MINIMUM_MOVING_SPEED_MPS
)
opposing_motion_risk = bool(
both_moving and velocity_angle > _OPPOSING_MOTION_START_DEGREES
)
opposing_motion_penalty = (
max(0.0, velocity_angle - _OPPOSING_MOTION_START_DEGREES)
/ (180.0 - _OPPOSING_MOTION_START_DEGREES)
* min(1.5, math.sqrt(previous_speed * next_speed) / 0.35)
* 1.35
if both_moving
else 0.0
)
orbit_transition = "orbit" in str(
previous_state.get("motionType", "")
).lower() or "orbit" in str(next_state.get("motionType", "")).lower()
if orbit_transition:
opposing_motion_penalty *= 0.35
elif visual_change >= 1.0:
# A clear angle/rotation/FOV/scale contrast makes the edit legible; the
# velocity mismatch remains a soft preference instead of a hard brake.
opposing_motion_penalty *= 0.65
total_penalty = float(
near_duplicate_penalty
+ screen_position_penalty
+ axis_crossing_penalty
+ extreme_scale_penalty
+ opposing_motion_penalty
)
flags = []
if jump_cut_risk:
flags.append("near_duplicate_jump_cut")
if axis_crossing_risk:
flags.append("unbridged_axis_crossing")
if opposing_motion_risk:
flags.append("opposing_high_speed_motion")
if subject_jump > 0.22:
flags.append("large_subject_screen_jump")
if scale_ratio > 2.5:
flags.append("extreme_subject_scale_jump")
return {
"schemaVersion": TRANSITION_SCORE_SCHEMA_VERSION,
"policyVersion": TRANSITION_POLICY_VERSION,
"penalty": total_penalty,
# A neither-wide 180-degree-axis crossing is not made acceptable by a
# low numeric penalty. It needs a bridging wide/neutral shot, so gate
# it exactly like a rendered near-duplicate jump cut. The caller's
# deterministic all-fail fallback still guarantees generation.
# Screen displacement and boundary-motion disagreement remain useful
# ranking preferences, but authored YAMO cuts use both expressively.
# Only the two categorical editorial failures are hard-gated.
"passed": bool(not jump_cut_risk and not axis_crossing_risk),
"flags": flags,
"metrics": {
"cameraGeometryAngleDegrees": geometry_angle,
"cameraRotationAngleDegrees": rotation_angle,
"fovDeltaDegrees": fov_delta,
"subjectUvJump": subject_jump,
"subjectScaleRatio": scale_ratio,
"relativeVelocityAngleDegrees": velocity_angle,
"previousRelativeSpeedMps": previous_speed,
"nextRelativeSpeedMps": next_speed,
"invisibleContinuation": invisible_continuation,
},
"components": {
"nearDuplicate": float(near_duplicate_penalty),
"screenPosition": float(screen_position_penalty),
"axisCrossing": float(axis_crossing_penalty),
"extremeScale": float(extreme_scale_penalty),
"opposingMotion": float(opposing_motion_penalty),
},
}
def score_transition_context(
candidate_state: Mapping[str, object],
*,
previous_state: Mapping[str, object] | None = None,
next_state: Mapping[str, object] | None = None,
) -> dict[str, object]:
"""Score either or both neighbours of one candidate shot."""
previous_score = (
score_pairwise_transition(previous_state, candidate_state)
if previous_state is not None
else None
)
next_score = (
score_pairwise_transition(candidate_state, next_state)
if next_state is not None
else None
)
total = sum(
float(score["penalty"])
for score in (previous_score, next_score)
if score is not None
)
edge_scores = [
score
for score in (previous_score, next_score)
if score is not None
]
return {
"schemaVersion": "camera-transition-context-score-v1",
"policyVersion": TRANSITION_POLICY_VERSION,
"previous": previous_score,
"next": next_score,
"penalty": float(total),
"neighbourCount": len(edge_scores),
"passed": all(bool(score["passed"]) for score in edge_scores),
}

View File

@ -0,0 +1,843 @@
"""Camera-motion kinematics shared by generation and offline evaluation.
The retrieved camera templates are artist-authored curves, but resampling and
retargeting can turn an otherwise intentional direction change into a sharp
per-frame corner. This module rebuilds only a short neighbourhood around that
corner: the camera decelerates into the turn and accelerates away without a
compensating speed spike.
A continuous one-direction orbit remains untouched. Only a persistent orbit
direction reversal (CW to CCW or the reverse) is treated as a turn event.
"""
from __future__ import annotations
from dataclasses import dataclass
import math
import numpy as np
TRANSLATION_KINEMATIC_POLICY_VERSION = (
"yamo-angle-aware-turn-v4-c2-continuity-guard"
)
DEFAULT_REVERSAL_ANGLE_DEGREES = 40.0
DEFAULT_STOP_SPEED_RATIO = 0.12
DEFAULT_REVERSAL_WINDOW_SECONDS = 0.35
DEFAULT_PERSISTENCE_SECONDS = 0.08
DEFAULT_MINIMUM_FLANK_SPEED_MPS = 0.08
DEFAULT_MINIMUM_FLANK_TRAVEL_METERS = 0.006
# With the five-frame persistence window, an ideal constant-curvature path has
# a concentration of 0.20 (one adjacent-frame turn divided by the five-frame
# flank-direction change). Keep a small margin above that value so a fast,
# intentional arc remains untouched while concentrated 40--55 degree corners
# such as the reviewed Shot_020 (0.22) are still eligible.
DEFAULT_MINIMUM_CORNER_CONCENTRATION = 0.21
DEFAULT_MINIMUM_FLANK_DIRECTION_COHERENCE = 0.85
DEFAULT_SPEED_RATIO_TOLERANCE = 0.025
DEFAULT_ANGLE_COMPARISON_TOLERANCE_DEGREES = 1e-3
DEFAULT_MINIMUM_ORBIT_ANGULAR_SPEED_RADIANS_PER_SECOND = math.radians(2.0)
DEFAULT_MAXIMUM_ENDPOINT_ADJUSTMENT_METERS = 0.75
DEFAULT_MAXIMUM_ACCELERATION_REGRESSION_RATIO = 1.25
DEFAULT_MAXIMUM_ACCELERATION_REGRESSION_DELTA_MPS2 = 8.0
DEFAULT_ACCELERATION_GUARD_FLOOR_MPS2 = 25.0
DEFAULT_MAXIMUM_JERK_REGRESSION_RATIO = 1.35
DEFAULT_MAXIMUM_JERK_REGRESSION_DELTA_MPS3 = 350.0
DEFAULT_JERK_GUARD_FLOOR_MPS3 = 350.0
def turn_speed_ratio_limit(angle_degrees: float) -> float:
"""Return an artist-oriented speed target for a sharp direction change.
A 40--55 degree corner receives only a mild slowdown, a right-angle corner
receives a clear slowdown, and turns at or above 120 degrees approach a
stop. The piecewise curve is deliberately monotonic and inspectable in
metadata.
"""
if not math.isfinite(angle_degrees):
raise ValueError("angle_degrees must be finite.")
return float(
np.interp(
float(np.clip(angle_degrees, 0.0, 180.0)),
np.asarray(
[0.0, 40.0, 45.0, 50.0, 55.0, 60.0, 90.0, 120.0, 150.0, 180.0]
),
np.asarray([1.0, 0.90, 0.87, 0.84, 0.80, 0.70, 0.45, 0.12, 0.0, 0.0]),
)
)
def natural_turn_speed_ratio_limit(angle_degrees: float) -> float:
"""Return the preservation threshold for an already eased sharp turn."""
if not math.isfinite(angle_degrees):
raise ValueError("angle_degrees must be finite.")
return float(
np.interp(
float(np.clip(angle_degrees, 0.0, 180.0)),
np.asarray(
[0.0, 40.0, 45.0, 50.0, 55.0, 60.0, 90.0, 120.0, 150.0, 180.0]
),
np.asarray([1.0, 0.90, 0.89, 0.87, 0.86, 0.85, 0.65, 0.40, 0.30, 0.25]),
)
)
@dataclass(frozen=True)
class TranslationReversal:
"""One persistent camera-translation direction-change event.
The legacy type name is retained because existing generated metadata and
downstream analysis use the ``translation*Reversal*`` field family.
"""
frame: int
angle_degrees: float
stop_speed_ratio: float
flank_speed_mps: float
target_speed_ratio: float
corner_concentration: float
natural_easing_detected: bool
@property
def abrupt(self) -> bool:
if self.natural_easing_detected and self.stop_speed_ratio <= (
natural_turn_speed_ratio_limit(self.angle_degrees)
+ DEFAULT_SPEED_RATIO_TOLERANCE
):
return False
return self.stop_speed_ratio > (
self.target_speed_ratio + DEFAULT_SPEED_RATIO_TOLERANCE
)
def motion_family(motion_type: str) -> str:
"""Return the local kinematic family without importing the shot planner."""
normalized = str(motion_type).strip().lower().replace("-", "_")
if normalized in {"static", "fixed", "locked", "stationary", "still"}:
return "static"
if normalized.startswith(("orbit", "arc")):
return "orbit"
if normalized.startswith(("dolly", "push", "pull", "zoom")):
return "dolly"
if normalized.startswith(("truck", "lateral")):
return "truck"
if normalized.startswith(("crane", "pedestal", "boom", "jib")):
return "crane"
if normalized in {"drift", "handheld", "float", "floating"}:
return "drift"
return "unsupported"
def _validated_positions(positions_meters: np.ndarray) -> np.ndarray:
positions = np.asarray(positions_meters, dtype=np.float64)
if positions.ndim != 2 or positions.shape[1] != 3 or len(positions) == 0:
raise ValueError("positions_meters must be a non-empty [frame, 3] array.")
if not np.isfinite(positions).all():
raise ValueError("positions_meters must contain only finite values.")
return positions
def _primary_motion_signal(positions: np.ndarray, family: str) -> np.ndarray:
if family == "dolly":
return np.linalg.norm(positions[:, [0, 2]], axis=1)
if family == "truck":
return positions[:, 0]
if family == "crane":
return positions[:, 1]
# Drift has no declared axis. The principal spatial axis is stable for a
# push-pull or side-to-side drift, including paths whose net delta is near
# zero because they return to their starting point.
centered = positions - np.mean(positions, axis=0, keepdims=True)
if float(np.max(np.linalg.norm(centered, axis=1))) <= 1e-9:
return np.zeros(len(positions), dtype=np.float64)
_, _, axes = np.linalg.svd(centered, full_matrices=False)
return centered @ axes[0]
def _angle_degrees(first: np.ndarray, second: np.ndarray) -> float:
denominator = float(np.linalg.norm(first) * np.linalg.norm(second))
if denominator <= 1e-12:
return 0.0
cosine = float(np.clip(np.dot(first, second) / denominator, -1.0, 1.0))
return math.degrees(math.acos(cosine))
def _group_reversal_candidates(
candidates: list[TranslationReversal],
persistence_frames: int,
) -> list[TranslationReversal]:
if not candidates:
return []
groups: list[list[TranslationReversal]] = [[candidates[0]]]
for candidate in candidates[1:]:
if candidate.frame - groups[-1][-1].frame <= persistence_frames:
groups[-1].append(candidate)
else:
groups.append([candidate])
events: list[TranslationReversal] = []
for group in groups:
low_speed_event = min(
group,
key=lambda event: (event.stop_speed_ratio, event.frame),
)
maximum_angle = max(event.angle_degrees for event in group)
events.append(
TranslationReversal(
frame=low_speed_event.frame,
angle_degrees=maximum_angle,
stop_speed_ratio=low_speed_event.stop_speed_ratio,
flank_speed_mps=low_speed_event.flank_speed_mps,
target_speed_ratio=turn_speed_ratio_limit(maximum_angle),
corner_concentration=max(event.corner_concentration for event in group),
natural_easing_detected=(low_speed_event.natural_easing_detected),
)
)
return events
def analyze_translational_reversals(
positions_meters: np.ndarray,
motion_type: str,
*,
sample_rate: float = 60.0,
reversal_angle_degrees: float = DEFAULT_REVERSAL_ANGLE_DEGREES,
persistence_seconds: float = DEFAULT_PERSISTENCE_SECONDS,
minimum_flank_speed_mps: float = DEFAULT_MINIMUM_FLANK_SPEED_MPS,
minimum_flank_travel_meters: float = (DEFAULT_MINIMUM_FLANK_TRAVEL_METERS),
minimum_corner_concentration: float = (DEFAULT_MINIMUM_CORNER_CONCENTRATION),
minimum_flank_direction_coherence: float = (
DEFAULT_MINIMUM_FLANK_DIRECTION_COHERENCE
),
) -> list[TranslationReversal]:
"""Find persistent, concentrated translation turns in one shot.
The detector requires both sides of a turn to sustain meaningful
travel. Small pose-follow jitter therefore does not become an authored
camera-direction event. Typed motions additionally have to reverse their
declared scalar axis (radius, lateral, or height).
"""
positions = _validated_positions(positions_meters)
if not math.isfinite(sample_rate) or sample_rate <= 0.0:
raise ValueError("sample_rate must be finite and positive.")
if not 0.0 < reversal_angle_degrees <= 180.0:
raise ValueError("reversal_angle_degrees must be in (0, 180].")
if not math.isfinite(persistence_seconds) or persistence_seconds <= 0.0:
raise ValueError("persistence_seconds must be finite and positive.")
if minimum_flank_speed_mps < 0.0 or not math.isfinite(minimum_flank_speed_mps):
raise ValueError("minimum_flank_speed_mps must be finite and non-negative.")
if minimum_flank_travel_meters < 0.0 or not math.isfinite(
minimum_flank_travel_meters
):
raise ValueError("minimum_flank_travel_meters must be finite and non-negative.")
if not 0.0 <= minimum_corner_concentration <= 1.0:
raise ValueError("minimum_corner_concentration must be between zero and one.")
if not 0.0 <= minimum_flank_direction_coherence <= 1.0:
raise ValueError(
"minimum_flank_direction_coherence must be between zero and one."
)
family = motion_family(motion_type)
if family in {"static", "unsupported"} or len(positions) < 7:
return []
persistence_frames = max(2, int(round(persistence_seconds * sample_rate)))
if len(positions) < persistence_frames * 2 + 1:
persistence_frames = max(2, (len(positions) - 1) // 2)
if persistence_frames < 2:
return []
velocity = np.diff(positions, axis=0) * sample_rate
speed = np.linalg.norm(velocity, axis=1)
centers = np.arange(
persistence_frames,
len(positions) - persistence_frames,
dtype=np.int64,
)
velocity_cumulative = np.vstack(
(
np.zeros((1, 3), dtype=np.float64),
np.cumsum(velocity, axis=0),
)
)
before_vectors = (
velocity_cumulative[centers] - velocity_cumulative[centers - persistence_frames]
) / persistence_frames
after_vectors = (
velocity_cumulative[centers + persistence_frames] - velocity_cumulative[centers]
) / persistence_frames
before_speeds = np.linalg.norm(before_vectors, axis=1)
after_speeds = np.linalg.norm(after_vectors, axis=1)
flank_speeds = np.minimum(before_speeds, after_speeds)
speed_cumulative = np.concatenate((np.zeros(1, dtype=np.float64), np.cumsum(speed)))
before_mean_speeds = (
speed_cumulative[centers] - speed_cumulative[centers - persistence_frames]
) / persistence_frames
after_mean_speeds = (
speed_cumulative[centers + persistence_frames] - speed_cumulative[centers]
) / persistence_frames
before_coherence = before_speeds / np.maximum(before_mean_speeds, 1e-12)
after_coherence = after_speeds / np.maximum(after_mean_speeds, 1e-12)
denominator = before_speeds * after_speeds
cosine = np.divide(
np.sum(before_vectors * after_vectors, axis=1),
denominator,
out=np.ones_like(denominator),
where=denominator > 1e-12,
)
angles = np.degrees(np.arccos(np.clip(cosine, -1.0, 1.0)))
unit_velocity = np.divide(
velocity,
speed[:, None],
out=np.zeros_like(velocity),
where=speed[:, None] > 1e-12,
)
adjacent_turn_cosine = np.sum(
unit_velocity[:-1] * unit_velocity[1:],
axis=1,
)
adjacent_turn_angles = np.degrees(
np.arccos(np.clip(adjacent_turn_cosine, -1.0, 1.0))
)
corner_window_size = persistence_frames * 2 - 1
corner_windows = np.lib.stride_tricks.sliding_window_view(
adjacent_turn_angles,
corner_window_size,
)
peak_corner_angles = np.max(
corner_windows[centers - persistence_frames],
axis=1,
)
corner_concentration = peak_corner_angles / np.maximum(angles, 1e-12)
step_distance = speed / sample_rate
distance_cumulative = np.concatenate(
(np.zeros(1, dtype=np.float64), np.cumsum(step_distance))
)
before_travel = (
distance_cumulative[centers] - distance_cumulative[centers - persistence_frames]
)
after_travel = (
distance_cumulative[centers + persistence_frames] - distance_cumulative[centers]
)
valid = (
(flank_speeds >= minimum_flank_speed_mps)
& (
angles
>= reversal_angle_degrees
- DEFAULT_ANGLE_COMPARISON_TOLERANCE_DEGREES
)
& (before_coherence >= minimum_flank_direction_coherence)
& (after_coherence >= minimum_flank_direction_coherence)
& (corner_concentration >= minimum_corner_concentration)
& (np.minimum(before_travel, after_travel) >= minimum_flank_travel_meters)
)
if family == "orbit":
radius = np.linalg.norm(positions[:, [0, 2]], axis=1)
azimuth = np.unwrap(np.arctan2(positions[:, 0], positions[:, 2]))
angular_velocity = np.diff(azimuth) * sample_rate
angular_windows = np.lib.stride_tricks.sliding_window_view(
angular_velocity,
persistence_frames,
)
incoming_angular_velocity = np.median(
angular_windows[centers - persistence_frames],
axis=1,
)
outgoing_angular_velocity = np.median(
angular_windows[centers],
axis=1,
)
valid &= incoming_angular_velocity * outgoing_angular_velocity < 0.0
valid &= (
np.minimum(
np.abs(incoming_angular_velocity),
np.abs(outgoing_angular_velocity),
)
>= DEFAULT_MINIMUM_ORBIT_ANGULAR_SPEED_RADIANS_PER_SECOND
)
valid &= radius[centers] >= 0.1
local_speed_windows = np.lib.stride_tricks.sliding_window_view(speed, 4)
local_minimum_speeds = np.min(local_speed_windows[centers - 2], axis=1)
ratios = local_minimum_speeds / np.maximum(flank_speeds, 1e-12)
persistence_speed_windows = np.lib.stride_tricks.sliding_window_view(
speed,
persistence_frames,
)
before_speed_sequence = persistence_speed_windows[centers - persistence_frames]
after_speed_sequence = persistence_speed_windows[centers]
trend_tolerance = flank_speeds[:, None] * 0.03
before_monotonic_ratio = np.mean(
np.diff(before_speed_sequence, axis=1)
<= trend_tolerance[:, : persistence_frames - 1],
axis=1,
)
after_monotonic_ratio = np.mean(
np.diff(after_speed_sequence, axis=1)
>= -trend_tolerance[:, : persistence_frames - 1],
axis=1,
)
natural_speed_limits = np.asarray(
[natural_turn_speed_ratio_limit(float(angle)) for angle in angles],
dtype=np.float64,
)
natural_easing = (
(before_monotonic_ratio >= 0.75)
& (after_monotonic_ratio >= 0.75)
& (ratios <= natural_speed_limits + DEFAULT_SPEED_RATIO_TOLERANCE)
)
candidates = [
TranslationReversal(
frame=int(frame),
angle_degrees=float(angle),
stop_speed_ratio=float(ratio),
flank_speed_mps=float(flank_speed),
target_speed_ratio=turn_speed_ratio_limit(float(angle)),
corner_concentration=float(concentration),
natural_easing_detected=bool(is_naturally_eased),
)
for (
frame,
angle,
ratio,
flank_speed,
concentration,
is_naturally_eased,
) in zip(
centers[valid],
angles[valid],
ratios[valid],
flank_speeds[valid],
corner_concentration[valid],
natural_easing[valid],
)
]
# Multiple adjacent frame centers describe the same physical turn. Keep
# the center with the clearest low-speed evidence so a naturally eased
# reversal is not falsely reported as abrupt.
return _group_reversal_candidates(candidates, persistence_frames)
def _smootherstep(values: np.ndarray) -> np.ndarray:
values = np.asarray(values, dtype=np.float64)
return values**3 * (10.0 + values * (-15.0 + 6.0 * values))
def _slerp_unit_vectors(
first: np.ndarray,
second: np.ndarray,
progress: np.ndarray,
) -> np.ndarray:
first = np.asarray(first, dtype=np.float64)
second = np.asarray(second, dtype=np.float64)
first /= max(float(np.linalg.norm(first)), 1e-12)
second /= max(float(np.linalg.norm(second)), 1e-12)
progress = np.asarray(progress, dtype=np.float64)
dot = float(np.clip(np.dot(first, second), -1.0, 1.0))
if dot > 0.9995:
values = (1.0 - progress[:, None]) * first + progress[:, None] * second
return values / np.maximum(
np.linalg.norm(values, axis=1, keepdims=True),
1e-12,
)
angle = math.acos(dot)
sine = math.sin(angle)
return (
np.sin((1.0 - progress[:, None]) * angle) / sine * first
+ np.sin(progress[:, None] * angle) / sine * second
)
def _ease_through_stop(
positions: np.ndarray,
center: int,
radius: int,
*,
target_speed_ratio: float = 0.0,
turn_angle_degrees: float = 180.0,
) -> np.ndarray:
"""Reintegrate one turn under a no-overspeed S-curve envelope.
Preserving the window's start, turn and end positions while inserting a
stop is mathematically forced to create compensating overspeed. Instead,
this function preserves the incoming and outgoing velocity at the outer
window edges, reduces travel near the corner, and shifts the remaining tail
by the small positional difference. No reconstructed step can exceed its
source step, and constant-speed input becomes monotonic deceleration into
the turn followed by monotonic acceleration away from it.
"""
if radius < 2:
raise ValueError("radius must be at least two frames.")
if not 0.0 <= target_speed_ratio <= 1.0:
raise ValueError("target_speed_ratio must be between zero and one.")
if not 0.0 <= turn_angle_degrees <= 180.0:
raise ValueError("turn_angle_degrees must be between zero and 180.")
left = center - radius
right = center + radius
if left < 0 or right >= len(positions):
raise ValueError("turn window must remain inside positions.")
source = np.asarray(positions, dtype=np.float64)
source_steps = np.diff(source, axis=0)
source_speeds = np.linalg.norm(source_steps, axis=1)
source_directions = np.divide(
source_steps,
source_speeds[:, None],
out=np.zeros_like(source_steps),
where=source_speeds[:, None] > 1e-12,
)
progress = np.linspace(0.0, 1.0, radius)
easing = _smootherstep(progress)
before_factor = 1.0 - (1.0 - target_speed_ratio) * easing
after_factor = target_speed_ratio + (1.0 - target_speed_ratio) * easing
before_speed = source_speeds[left:center] * before_factor
after_speed = source_speeds[center:right] * after_factor
# Do not force monotonicity with cumulative minima. That operation creates
# a new slope discontinuity whenever an authored speed fluctuation becomes
# the running minimum. The C2 factor is already monotonic for a constant-
# speed corner and preserves the source curve's local dynamics otherwise.
directions = source_directions[left:right].copy()
if turn_angle_degrees < 150.0:
flank = min(radius, max(2, int(round(radius * 0.25))))
incoming = np.mean(source_steps[center - flank : center], axis=0)
outgoing = np.mean(source_steps[center : center + flank], axis=0)
if (
float(np.linalg.norm(incoming)) > 1e-12
and float(np.linalg.norm(outgoing)) > 1e-12
):
smoothed_directions = _slerp_unit_vectors(
incoming,
outgoing,
np.linspace(0.0, 1.0, radius * 2),
)
# Preserve the authored direction at both window boundaries for
# every eligible angle, then hand control to the interpolated turn
# with a C2-continuous envelope. Replacing an entire 60--149 degree
# window created a velocity seam at its outer edge. The previous
# sin-squared mild-turn blend also retained a C2 seam there.
edge_progress = np.linspace(0.0, 1.0, radius * 2)
direction_blend = np.where(
edge_progress <= 0.5,
_smootherstep(edge_progress * 2.0),
_smootherstep((1.0 - edge_progress) * 2.0),
)
directions = (
source_directions[left:right]
* (1.0 - direction_blend[:, None])
+ smoothed_directions * direction_blend[:, None]
)
directions /= np.maximum(
np.linalg.norm(directions, axis=1, keepdims=True),
1e-12,
)
reconstructed_steps = (
directions * np.concatenate((before_speed, after_speed))[:, None]
)
output = source.copy()
output[left + 1 : right + 1] = output[left] + np.cumsum(
reconstructed_steps,
axis=0,
)
tail_offset = output[right] - source[right]
if right + 1 < len(output):
output[right + 1 :] = source[right + 1 :] + tail_offset
return output
def translation_dynamics_metrics(
positions_meters: np.ndarray,
sample_rate: float,
) -> dict[str, float]:
"""Return peak translation speed, acceleration and jerk for auditing."""
positions = _validated_positions(positions_meters)
velocity = np.diff(positions, axis=0) * sample_rate
acceleration = np.diff(velocity, axis=0) * sample_rate
jerk = np.diff(acceleration, axis=0) * sample_rate
def peak(values: np.ndarray) -> float:
return float(np.max(np.linalg.norm(values, axis=1))) if len(values) else 0.0
return {
"speedMetersPerSecondMax": peak(velocity),
"accelerationMetersPerSecondSquaredMax": peak(acceleration),
"jerkMetersPerSecondCubedMax": peak(jerk),
}
def _dynamics_regression_exceeds_guard(
before: dict[str, float],
after: dict[str, float],
) -> bool:
"""Reject a rewrite that introduces an edit-visible dynamics spike."""
acceleration_before = before["accelerationMetersPerSecondSquaredMax"]
acceleration_after = after["accelerationMetersPerSecondSquaredMax"]
acceleration_limit = max(
DEFAULT_ACCELERATION_GUARD_FLOOR_MPS2,
min(
acceleration_before
* DEFAULT_MAXIMUM_ACCELERATION_REGRESSION_RATIO,
acceleration_before
+ DEFAULT_MAXIMUM_ACCELERATION_REGRESSION_DELTA_MPS2,
),
)
jerk_before = before["jerkMetersPerSecondCubedMax"]
jerk_after = after["jerkMetersPerSecondCubedMax"]
jerk_limit = max(
DEFAULT_JERK_GUARD_FLOOR_MPS3,
min(
jerk_before * DEFAULT_MAXIMUM_JERK_REGRESSION_RATIO,
jerk_before + DEFAULT_MAXIMUM_JERK_REGRESSION_DELTA_MPS3,
),
)
return bool(
acceleration_after > acceleration_limit + 1e-9
or jerk_after > jerk_limit + 1e-9
)
def _event_metadata(events: list[TranslationReversal]) -> dict[str, object]:
abrupt = [event for event in events if event.abrupt]
return {
"count": len(events),
"abruptCount": len(abrupt),
"frames": [event.frame for event in events],
"abruptFrames": [event.frame for event in abrupt],
"anglesDegrees": [event.angle_degrees for event in events],
"targetSpeedRatios": [event.target_speed_ratio for event in events],
"cornerConcentrations": [event.corner_concentration for event in events],
"worstStopSpeedRatio": (
float(max(event.stop_speed_ratio for event in events)) if events else 0.0
),
"worstTargetSpeedRatioExcess": (
float(
max(
event.stop_speed_ratio - event.target_speed_ratio
for event in abrupt
)
)
if abrupt
else 0.0
),
}
def regularize_translational_reversals(
positions_meters: np.ndarray,
motion_type: str,
*,
sample_rate: float = 60.0,
reversal_window_seconds: float = DEFAULT_REVERSAL_WINDOW_SECONDS,
) -> tuple[np.ndarray, dict[str, object]]:
"""Apply angle-aware braking to abrupt turns and orbit reversals."""
positions = _validated_positions(positions_meters)
if not math.isfinite(reversal_window_seconds) or reversal_window_seconds <= 0.0:
raise ValueError("reversal_window_seconds must be finite and positive.")
family = motion_family(motion_type)
before = analyze_translational_reversals(
positions,
motion_type,
sample_rate=sample_rate,
)
abrupt_before = [event for event in before if event.abrupt]
output = positions.copy()
applied_frames: list[int] = []
applied_angles: list[float] = []
applied_target_speed_ratios: list[float] = []
pass_count = 0
pending_events = abrupt_before
if family not in {"static", "unsupported"} and pending_events:
requested_radius = max(
6,
int(round(reversal_window_seconds * sample_rate)),
)
# Only abrupt events detected on the untouched input are eligible.
# Re-detecting and rewriting new window-boundary events caused a
# three-pass cascade in real shots, multiplying acceleration and jerk
# even though the final turn counter eventually reached zero.
event_frames = [event.frame for event in pending_events]
for event_index, event in enumerate(pending_events):
neighbor_limit = requested_radius
if event_index > 0:
neighbor_limit = min(
neighbor_limit,
(event.frame - event_frames[event_index - 1]) // 2,
)
if event_index + 1 < len(event_frames):
neighbor_limit = min(
neighbor_limit,
(event_frames[event_index + 1] - event.frame) // 2,
)
radius = min(
neighbor_limit,
event.frame,
len(output) - 1 - event.frame,
)
if radius < 4:
continue
output = _ease_through_stop(
output,
event.frame,
radius,
target_speed_ratio=event.target_speed_ratio,
turn_angle_degrees=event.angle_degrees,
)
applied_frames.append(event.frame)
applied_angles.append(event.angle_degrees)
applied_target_speed_ratios.append(event.target_speed_ratio)
pass_count = int(bool(applied_frames))
endpoint_adjustment = output[-1] - positions[-1]
endpoint_adjustment_length = float(np.linalg.norm(endpoint_adjustment))
endpoint_adjustment_clamped = (
endpoint_adjustment_length > DEFAULT_MAXIMUM_ENDPOINT_ADJUSTMENT_METERS
)
if endpoint_adjustment_clamped:
correction_scale = (
DEFAULT_MAXIMUM_ENDPOINT_ADJUSTMENT_METERS / endpoint_adjustment_length
)
# A convex blend between source and slowed steps cannot exceed the
# source peak speed. Pathological multi-turn candidates are therefore
# bounded spatially and remain eligible for the unresolved-turn penalty.
output = positions + (output - positions) * correction_scale
dynamics_before = translation_dynamics_metrics(positions, sample_rate)
attempted_dynamics_after = translation_dynamics_metrics(output, sample_rate)
attempted_regularized_frames = list(applied_frames)
dynamics_guard_triggered = bool(applied_frames) and (
_dynamics_regression_exceeds_guard(
dynamics_before,
attempted_dynamics_after,
)
)
if dynamics_guard_triggered:
# A rejected rewrite remains an abrupt candidate and is therefore
# demoted by the normal selection penalty. Keeping the untouched curve
# is safer than publishing a numerically "resolved" turn with an
# edit-visible acceleration or jerk seam.
output = positions.copy()
applied_frames.clear()
applied_angles.clear()
applied_target_speed_ratios.clear()
pass_count = 0
endpoint_adjustment_clamped = False
after = analyze_translational_reversals(
output,
motion_type,
sample_rate=sample_rate,
)
before_metadata = _event_metadata(before)
after_metadata = _event_metadata(after)
dynamics_after = translation_dynamics_metrics(output, sample_rate)
path_length_before = float(
np.sum(np.linalg.norm(np.diff(positions, axis=0), axis=1))
)
path_length_after = float(np.sum(np.linalg.norm(np.diff(output, axis=0), axis=1)))
metadata: dict[str, object] = {
"translationKinematicPolicyVersion": (TRANSLATION_KINEMATIC_POLICY_VERSION),
"translationKinematicMotionFamily": family,
"translationKinematicPolicyEligible": family not in {"static", "unsupported"},
"translationKinematicPolicyApplied": bool(applied_frames),
"translationKinematicDynamicsGuardTriggered": dynamics_guard_triggered,
"translationKinematicAttemptedRegularizationCount": len(
attempted_regularized_frames
),
"translationKinematicAttemptedReversalFrames": (
attempted_regularized_frames
),
"translationDirectionReversalCountBefore": before_metadata["count"],
"translationAbruptReversalCountBefore": before_metadata["abruptCount"],
"translationDirectionReversalFramesBefore": before_metadata["frames"],
"translationAbruptReversalFramesBefore": before_metadata["abruptFrames"],
"translationWorstStopSpeedRatioBefore": before_metadata["worstStopSpeedRatio"],
"translationRegularizedReversalCount": len(applied_frames),
"translationKinematicRegularizationPassCount": pass_count,
"translationRegularizedReversalFrames": applied_frames,
"translationRegularizedTurnAnglesDegrees": applied_angles,
"translationRegularizedTargetSpeedRatios": (applied_target_speed_ratios),
"translationDirectionReversalCountAfter": after_metadata["count"],
"translationAbruptReversalCountAfter": after_metadata["abruptCount"],
"translationDirectionReversalFramesAfter": after_metadata["frames"],
"translationAbruptReversalFramesAfter": after_metadata["abruptFrames"],
"translationWorstStopSpeedRatioAfter": after_metadata["worstStopSpeedRatio"],
"translationWorstTargetSpeedRatioExcessAfter": after_metadata[
"worstTargetSpeedRatioExcess"
],
"translationStopSpeedRatioLimit": DEFAULT_STOP_SPEED_RATIO,
"translationReversalAngleMinimumDegrees": (DEFAULT_REVERSAL_ANGLE_DEGREES),
"translationMinimumCornerConcentration": (DEFAULT_MINIMUM_CORNER_CONCENTRATION),
"translationSpeedMetersPerSecondMaxBefore": dynamics_before[
"speedMetersPerSecondMax"
],
"translationSpeedMetersPerSecondMaxAfter": dynamics_after[
"speedMetersPerSecondMax"
],
"translationAccelerationMetersPerSecondSquaredMaxBefore": (
dynamics_before["accelerationMetersPerSecondSquaredMax"]
),
"translationAccelerationMetersPerSecondSquaredMaxAfter": (
dynamics_after["accelerationMetersPerSecondSquaredMax"]
),
"translationJerkMetersPerSecondCubedMaxBefore": dynamics_before[
"jerkMetersPerSecondCubedMax"
],
"translationJerkMetersPerSecondCubedMaxAfter": dynamics_after[
"jerkMetersPerSecondCubedMax"
],
"translationAttemptedAccelerationMetersPerSecondSquaredMax": (
attempted_dynamics_after[
"accelerationMetersPerSecondSquaredMax"
]
),
"translationAttemptedJerkMetersPerSecondCubedMax": (
attempted_dynamics_after["jerkMetersPerSecondCubedMax"]
),
"translationPathLengthRetentionRatio": (
path_length_after / path_length_before
if path_length_before > 1e-12
else 1.0
),
"translationEndpointAdjustmentMeters": float(
np.linalg.norm(output[-1] - positions[-1])
),
"translationEndpointAdjustmentLimitMeters": (
DEFAULT_MAXIMUM_ENDPOINT_ADJUSTMENT_METERS
),
"translationEndpointAdjustmentClamped": endpoint_adjustment_clamped,
}
return output.astype(np.float32), metadata
def translational_reversal_selection_penalty(
metrics: dict[str, object],
) -> float:
"""Softly demote candidates whose short shot cannot fit a safe slowdown."""
abrupt_count = int(metrics.get("translationAbruptReversalCountAfter", 0))
ratio_excess = float(
metrics.get("translationWorstTargetSpeedRatioExcessAfter", 0.0)
)
dynamics_guard = bool(
metrics.get("translationKinematicDynamicsGuardTriggered", False)
)
return (
abrupt_count * 2.5
+ ratio_excess * 2.0
+ float(dynamics_guard) * 2.5
)

View File

@ -1,6 +1,6 @@
{
"name": "com.mingle.cw-ai",
"version": "0.4.1",
"version": "0.4.2",
"displayName": "Mingle Camera Work AI",
"description": "High-quality Python CLI-backed Unity Timeline camera generation, per-shot editable clips, dataset export, and A/B review tools.",
"unity": "6000.0",

View File

@ -37,7 +37,7 @@ Install **Git and Git LFS before opening Unity**, then use **Add package from
git URL** with:
```text
https://kindnick-git.duckdns.org/mingle/streamingle-unity-utilities.git?path=/CameraAI~#v0.1.5
https://kindnick-git.duckdns.org/mingle/streamingle-unity-utilities.git?path=/CameraAI~#v0.1.6
```
The Camera AI package includes the complete Windows x64

View File

@ -1,7 +1,7 @@
{
"name": "com.streamingle.utilities",
"displayName": "Streamingle Utilities",
"version": "0.1.5",
"version": "0.1.6",
"unity": "6000.0",
"description": "Reusable Streamingle runtime components and Unity editor utilities.",
"keywords": [