Initial commit
This commit is contained in:
TechnicalTod
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
Core modules for roto node
|
||||
"""
|
||||
@@ -0,0 +1,112 @@
|
||||
"""
|
||||
Curve data validation
|
||||
"""
|
||||
|
||||
from typing import Dict
|
||||
|
||||
|
||||
def validate_schema(data: Dict) -> bool:
|
||||
"""
|
||||
Validate JSON data structure.
|
||||
|
||||
Args:
|
||||
data: Parsed JSON data
|
||||
|
||||
Returns:
|
||||
True if schema is valid, False otherwise
|
||||
"""
|
||||
# Check for required top-level keys
|
||||
if not isinstance(data, dict):
|
||||
return False
|
||||
|
||||
if "meta" not in data or "shapes" not in data:
|
||||
return False
|
||||
|
||||
# Check meta structure
|
||||
meta = data["meta"]
|
||||
if not isinstance(meta, dict):
|
||||
return False
|
||||
|
||||
required_meta = ["version", "width", "height"]
|
||||
if not all(key in meta for key in required_meta):
|
||||
return False
|
||||
|
||||
# Check shapes structure
|
||||
shapes = data["shapes"]
|
||||
if not isinstance(shapes, dict):
|
||||
return False
|
||||
|
||||
# Validate each shape
|
||||
for shape_id, shape in shapes.items():
|
||||
if not isinstance(shape, dict):
|
||||
return False
|
||||
|
||||
if "name" not in shape or "frames" not in shape:
|
||||
return False
|
||||
|
||||
frames = shape["frames"]
|
||||
if not isinstance(frames, dict):
|
||||
return False
|
||||
|
||||
# Validate curves in each frame
|
||||
for frame_num, curves in frames.items():
|
||||
if not isinstance(curves, list):
|
||||
return False
|
||||
|
||||
for curve in curves:
|
||||
if not validate_curve(curve):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def validate_curve(curve: Dict) -> bool:
|
||||
"""
|
||||
Validate a single curve structure.
|
||||
|
||||
Args:
|
||||
curve: Curve dictionary
|
||||
|
||||
Returns:
|
||||
True if curve is valid, False otherwise
|
||||
"""
|
||||
if not isinstance(curve, dict):
|
||||
return False
|
||||
|
||||
required_keys = ["id", "closed", "points"]
|
||||
if not all(key in curve for key in required_keys):
|
||||
return False
|
||||
|
||||
# Check points array
|
||||
points = curve["points"]
|
||||
if not isinstance(points, list):
|
||||
return False
|
||||
|
||||
# Must have at least 2 points for a valid polygon
|
||||
# For closed shapes, need at least 3 points
|
||||
if len(points) < 2:
|
||||
return False
|
||||
if curve.get("closed", False) and len(points) < 3:
|
||||
return False
|
||||
|
||||
# Each point must be [x, y] pair
|
||||
for point in points:
|
||||
if not isinstance(point, (list, tuple)) or len(point) != 2:
|
||||
return False
|
||||
if not all(isinstance(coord, (int, float)) for coord in point):
|
||||
return False
|
||||
|
||||
# Validate optional pointTypes array
|
||||
if "pointTypes" in curve:
|
||||
point_types = curve["pointTypes"]
|
||||
if not isinstance(point_types, list):
|
||||
return False
|
||||
# pointTypes length must match points length
|
||||
if len(point_types) != len(points):
|
||||
return False
|
||||
# All values must be "smooth" or "hard"
|
||||
valid_types = {"smooth", "hard"}
|
||||
if not all(pt in valid_types for pt in point_types):
|
||||
return False
|
||||
|
||||
return True
|
||||
@@ -0,0 +1,223 @@
|
||||
"""
|
||||
Frame evaluation and interpolation logic
|
||||
"""
|
||||
|
||||
import copy
|
||||
from typing import Dict, List
|
||||
|
||||
|
||||
def get_interpolated_point_types(curve_a: Dict, curve_b: Dict, total_points: int) -> List[str]:
|
||||
"""
|
||||
Get interpolated pointTypes array from two curves.
|
||||
Since pointTypes should be consistent across all keyframes, prefer curve_a.
|
||||
|
||||
Args:
|
||||
curve_a: First curve (earlier keyframe)
|
||||
curve_b: Second curve (later keyframe)
|
||||
total_points: Total point count in the interpolated result
|
||||
|
||||
Returns:
|
||||
List of point types ("smooth" or "hard")
|
||||
"""
|
||||
types_a = curve_a.get("pointTypes", []) if isinstance(curve_a.get("pointTypes"), list) else []
|
||||
types_b = curve_b.get("pointTypes", []) if isinstance(curve_b.get("pointTypes"), list) else []
|
||||
|
||||
result = []
|
||||
for i in range(total_points):
|
||||
if i < len(types_a) and types_a[i] in ("smooth", "hard"):
|
||||
result.append(types_a[i])
|
||||
elif i < len(types_b) and types_b[i] in ("smooth", "hard"):
|
||||
result.append(types_b[i])
|
||||
else:
|
||||
result.append("hard") # Default for any missing entries
|
||||
return result
|
||||
|
||||
|
||||
def evaluate_frame(shape: Dict, frame_index: int) -> List[Dict]:
|
||||
"""
|
||||
Evaluate curves for a given frame.
|
||||
Each curve is interpolated independently based on its own keyframes.
|
||||
|
||||
Args:
|
||||
shape: Shape dictionary
|
||||
frame_index: Frame to evaluate
|
||||
|
||||
Returns:
|
||||
List of curve dictionaries for this frame
|
||||
"""
|
||||
frames = shape.get("frames", {})
|
||||
frame_key = str(frame_index)
|
||||
|
||||
# Collect all unique curve IDs and their keyframes
|
||||
all_curve_ids = set()
|
||||
curve_keyframes: Dict[str, List[int]] = {} # curveId -> sorted list of frame numbers
|
||||
|
||||
for frame_str, curves in frames.items():
|
||||
try:
|
||||
frame_num = int(frame_str)
|
||||
except ValueError:
|
||||
continue
|
||||
for curve in (curves or []):
|
||||
if curve and "id" in curve:
|
||||
curve_id = curve["id"]
|
||||
all_curve_ids.add(curve_id)
|
||||
if curve_id not in curve_keyframes:
|
||||
curve_keyframes[curve_id] = []
|
||||
curve_keyframes[curve_id].append(frame_num)
|
||||
|
||||
# Sort keyframes for each curve
|
||||
for curve_id in curve_keyframes:
|
||||
curve_keyframes[curve_id].sort()
|
||||
|
||||
if not all_curve_ids:
|
||||
return []
|
||||
|
||||
result = []
|
||||
|
||||
# Evaluate each curve independently based on its own keyframes
|
||||
for curve_id in all_curve_ids:
|
||||
keyframes = curve_keyframes.get(curve_id, [])
|
||||
if not keyframes:
|
||||
continue
|
||||
|
||||
# Check for exact keyframe for this curve
|
||||
if frame_index in keyframes:
|
||||
frame_curves = frames.get(frame_key, [])
|
||||
curve = next((c for c in frame_curves if c["id"] == curve_id), None)
|
||||
if curve:
|
||||
result.append(copy.deepcopy(curve))
|
||||
continue
|
||||
|
||||
# Find prev and next keyframes FOR THIS CURVE
|
||||
prev_frame = None
|
||||
next_frame = None
|
||||
|
||||
for fn in keyframes:
|
||||
if fn < frame_index:
|
||||
prev_frame = fn
|
||||
elif fn > frame_index:
|
||||
next_frame = fn
|
||||
break
|
||||
|
||||
# Interpolate or hold based on this curve's keyframes
|
||||
if prev_frame is not None and next_frame is not None:
|
||||
# Interpolate between this curve's keyframes
|
||||
curve_a = next((c for c in frames.get(str(prev_frame), []) if c["id"] == curve_id), None)
|
||||
curve_b = next((c for c in frames.get(str(next_frame), []) if c["id"] == curve_id), None)
|
||||
|
||||
if curve_a and curve_b:
|
||||
t = (frame_index - prev_frame) / (next_frame - prev_frame)
|
||||
interpolated_points = []
|
||||
|
||||
# Handle different point counts - use the smaller count
|
||||
min_points = min(len(curve_a["points"]), len(curve_b["points"]))
|
||||
for i in range(min_points):
|
||||
pa = curve_a["points"][i]
|
||||
pb = curve_b["points"][i]
|
||||
x = pa[0] + (pb[0] - pa[0]) * t
|
||||
y = pa[1] + (pb[1] - pa[1]) * t
|
||||
interpolated_points.append([x, y])
|
||||
|
||||
# If one curve has more points, add them at the end (hold from that curve)
|
||||
if len(curve_a["points"]) > min_points:
|
||||
for i in range(min_points, len(curve_a["points"])):
|
||||
interpolated_points.append([curve_a["points"][i][0], curve_a["points"][i][1]])
|
||||
elif len(curve_b["points"]) > min_points:
|
||||
for i in range(min_points, len(curve_b["points"])):
|
||||
interpolated_points.append([curve_b["points"][i][0], curve_b["points"][i][1]])
|
||||
|
||||
result.append({
|
||||
"id": curve_a["id"],
|
||||
"closed": curve_a["closed"],
|
||||
"points": interpolated_points,
|
||||
"pointTypes": get_interpolated_point_types(curve_a, curve_b, len(interpolated_points))
|
||||
})
|
||||
else:
|
||||
# Hold from nearest keyframe for this curve
|
||||
hold_frame = prev_frame if prev_frame is not None else next_frame
|
||||
if hold_frame is not None:
|
||||
curve = next((c for c in frames.get(str(hold_frame), []) if c["id"] == curve_id), None)
|
||||
if curve:
|
||||
result.append(copy.deepcopy(curve))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def interpolate_curves(
|
||||
curves_a: List[Dict],
|
||||
curves_b: List[Dict],
|
||||
frame_a: int,
|
||||
frame_b: int,
|
||||
target_frame: int
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Linearly interpolate curves between two keyframes.
|
||||
Curves that only exist in one keyframe are held (included as-is).
|
||||
|
||||
Args:
|
||||
curves_a: Curves at earlier keyframe
|
||||
curves_b: Curves at later keyframe
|
||||
frame_a: Earlier frame number
|
||||
frame_b: Later frame number
|
||||
target_frame: Frame to interpolate to
|
||||
|
||||
Returns:
|
||||
List of interpolated curves
|
||||
"""
|
||||
# Calculate interpolation factor
|
||||
t = (target_frame - frame_a) / (frame_b - frame_a)
|
||||
|
||||
# Track which curve IDs we've processed
|
||||
processed_ids = set()
|
||||
result = []
|
||||
|
||||
# Process curves from A - interpolate if in both, hold if only in A
|
||||
for curve_a in curves_a:
|
||||
processed_ids.add(curve_a["id"])
|
||||
|
||||
# Find matching curve in curves_b
|
||||
curve_b = None
|
||||
for cb in curves_b:
|
||||
if cb["id"] == curve_a["id"]:
|
||||
curve_b = cb
|
||||
break
|
||||
|
||||
if curve_b is not None:
|
||||
# Curve exists in both keyframes - interpolate
|
||||
points_a = curve_a["points"]
|
||||
points_b = curve_b["points"]
|
||||
|
||||
# Handle different point counts - use the smaller count
|
||||
min_points = min(len(points_a), len(points_b))
|
||||
interpolated_points = []
|
||||
for i in range(min_points):
|
||||
pa = points_a[i]
|
||||
pb = points_b[i]
|
||||
x = pa[0] + (pb[0] - pa[0]) * t
|
||||
y = pa[1] + (pb[1] - pa[1]) * t
|
||||
interpolated_points.append([x, y])
|
||||
|
||||
# If one curve has more points, add them at the end (hold from that curve)
|
||||
if len(points_a) > min_points:
|
||||
for i in range(min_points, len(points_a)):
|
||||
interpolated_points.append([points_a[i][0], points_a[i][1]])
|
||||
elif len(points_b) > min_points:
|
||||
for i in range(min_points, len(points_b)):
|
||||
interpolated_points.append([points_b[i][0], points_b[i][1]])
|
||||
|
||||
result.append({
|
||||
"id": curve_a["id"],
|
||||
"closed": curve_a["closed"],
|
||||
"points": interpolated_points,
|
||||
"pointTypes": get_interpolated_point_types(curve_a, curve_b, len(interpolated_points))
|
||||
})
|
||||
else:
|
||||
# Curve only in A - hold it
|
||||
result.append(copy.deepcopy(curve_a))
|
||||
|
||||
# Process curves only in B (not in A) - hold them
|
||||
for curve_b in curves_b:
|
||||
if curve_b["id"] not in processed_ids:
|
||||
result.append(copy.deepcopy(curve_b))
|
||||
|
||||
return result
|
||||
Reference in New Issue
Block a user