Frame13/ayon-core
1
0
Fork 1
mirror of https://github.com/ynput/ayon-core.git synced 2025-12-24 21:04:40 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

🎨 handle frame specs

Browse source
This commit is contained in:
Ondřej Samohel 2024-11-26 23:30:23 +01:00
parent dc26079065
commit 97fe8ac294
No known key found for this signature in database
GPG key ID: 02376E18990A97C6
3 changed files with 243 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

46
client/ayon_core/pipeline/traits/content.py
View file

@ -1,14 +1,14 @@
"""Content traits for the pipeline."""
from __future__ import annotations
# TCH003 is there because Path in TYPECHECKING will fail in tests
from pathlib import Path # noqa: TCH003
# TC003 is there because Path in TYPECHECKING will fail in tests
from pathlib import Path # noqa: TC003
from typing import ClassVar, Optional
from pydantic import Field
from .representation import Representation
from .time import FrameRanged
from .time import FrameRanged, Sequence
from .trait import (
MissingTraitError,
TraitBase,
@ -106,7 +106,7 @@ class FileLocations(TraitBase):
id: ClassVar[str] = "ayon.content.FileLocations.v1"
file_paths: list[FileLocation] = Field(..., title="File Path")
def validate(self, representation: Representation) -> bool:
def validate(self, representation: Representation) -> None:
"""Validate the trait.
This method validates the trait against others in the representation.
@ -120,6 +120,7 @@ class FileLocations(TraitBase):
bool: True if the trait is valid, False otherwise
"""
super().validate(representation)
if len(self.file_paths) == 0:
# If there are no file paths, we can't validate
msg = "No file locations defined (empty list)"
@ -128,6 +129,33 @@ class FileLocations(TraitBase):
tmp_frame_ranged: FrameRanged = get_sequence_from_files(
[f.file_path for f in self.file_paths])
frames_from_spec = None
try:
sequence: Sequence = representation.get_trait(Sequence)
if sequence.frame_spec:
frames_from_spec: list[int] = sequence.get_frame_list(
self, sequence.frame_regex)
except MissingTraitError:
# If there is no sequence trait, we can't validate it
pass
if frames_from_spec:
if len(frames_from_spec) != len(self.file_paths) :
# If the number of file paths does not match the frame range,
# the trait is invalid
msg = (
f"Number of file locations ({len(self.file_paths)}) "
"does not match frame range defined by frame spec "
"on Sequence trait: "
f"({len(frames_from_spec)})"
)
raise TraitValidationError(self.name, msg)
# if there is frame spec on the Sequence trait
# we should not validate the frame range from the files.
# the rest is validated by Sequence validators.
return
if len(self.file_paths) - 1 != \
tmp_frame_ranged.frame_end - tmp_frame_ranged.frame_start:
# If the number of file paths does not match the frame range,
@ -140,17 +168,17 @@ class FileLocations(TraitBase):
raise TraitValidationError(self.name, msg)
try:
sequence: FrameRanged = representation.get_trait(FrameRanged)
frame_ranged: FrameRanged = representation.get_trait(FrameRanged)
if sequence.frame_start != tmp_frame_ranged.frame_start or \
sequence.frame_end != tmp_frame_ranged.frame_end:
if frame_ranged.frame_start != tmp_frame_ranged.frame_start or \
frame_ranged.frame_end != tmp_frame_ranged.frame_end:
# If the frame range does not match the sequence trait, the
# trait is invalid. Note that we don't check the frame rate
# because it is not stored in the file paths and is not
# determined by `get_sequence_from_files`.
msg = (
"Frame range "
f"({sequence.frame_start}-{sequence.frame_end}) "
f"({frame_ranged.frame_start}-{frame_ranged.frame_end}) "
"in sequence trait does not match "
"frame range "
f"({tmp_frame_ranged.frame_start}-{tmp_frame_ranged.frame_end}) " # noqa: E501
@ -159,7 +187,7 @@ class FileLocations(TraitBase):
raise TraitValidationError(self.name, msg)
except MissingTraitError:
# If there is no sequence trait, we can't validate it
# If there is no frame_ranged trait, we can't validate it
pass

154
client/ayon_core/pipeline/traits/time.py
View file

@ -7,9 +7,10 @@ from typing import TYPE_CHECKING, ClassVar, Optional
import clique
from pydantic import Field
from .trait import MissingTraitError, TraitBase
from .trait import MissingTraitError, TraitBase, TraitValidationError
if TYPE_CHECKING:
import re
from pathlib import Path
from .content import FileLocations
@ -117,7 +118,7 @@ class Sequence(TraitBase):
frame_padding (int): Frame padding.
frame_regex (str): Frame regex - regular expression to match
frame numbers.
frame_list (str): Frame list specification of frames. This takes
frame_spec (str): Frame list specification of frames. This takes
string like "1-10,20-30,40-50" etc.
"""
@ -127,13 +128,12 @@ class Sequence(TraitBase):
gaps_policy: GapPolicy = Field(
GapPolicy.forbidden, title="Gaps Policy")
frame_padding: int = Field(..., title="Frame Padding")
frame_regex: str = Field(..., title="Frame Regex")
frame_list: Optional[str] = Field(None, title="Frame List")
frame_regex: Optional[str] = Field(None, title="Frame Regex")
frame_spec: Optional[str] = Field(None, title="Frame Specification")
def validate(self, representation: Representation) -> None:
"""Validate the trait."""
if not super().validate(representation):
return False
super().validate(representation)
# if there is FileLocations trait, run validation
# on it as well
@ -142,24 +142,158 @@ class Sequence(TraitBase):
file_locs: FileLocations = representation.get_trait(
FileLocations)
file_locs.validate(representation)
# validate if file locations on representation
# matches the frame list (if any)
self.validate_frame_list(file_locs)
self.validate_frame_padding(file_locs)
except MissingTraitError:
pass
def validate_frame_list(
self, file_locations: FileLocations) -> None:
"""Validate frame list.
This will take FileLocations trait and validate if the
file locations match the frame list specification.
For example, if frame list is "1-10,20-30,40-50", then
the frame numbers in the file locations should match
these frames.
It will skip the validation if frame list is not provided.
Args:
file_locations (FileLocations): File locations trait.
Raises:
TraitValidationError: If frame list does not match
the expected frames.
"""
if self.frame_spec is None:
return
frames: list[int] = self.get_frame_list(
file_locations, self.frame_regex)
expected_frames = self.list_spec_to_frames(self.frame_spec)
if set(frames) != set(expected_frames):
msg = (
"Frame list does not match the expected frames. "
f"Expected: {expected_frames}, Found: {frames}"
)
raise TraitValidationError(self.name, msg)
def validate_frame_padding(
self, file_locations: FileLocations) -> None:
"""Validate frame padding.
This will take FileLocations trait and validate if the
frame padding matches the expected frame padding.
Args:
file_locations (FileLocations): File locations trait.
Raises:
TraitValidationError: If frame padding does not match
the expected frame padding.
"""
expected_padding = self.get_frame_padding(file_locations)
if self.frame_padding != expected_padding:
msg = (
"Frame padding does not match the expected frame padding. "
f"Expected: {expected_padding}, Found: {self.frame_padding}"
)
raise TraitValidationError(msg)
@staticmethod
def get_frame_padding(file_locations: FileLocations) -> int:
"""Get frame padding."""
def list_spec_to_frames(list_spec: str) -> list[int]:
"""Convert list specification to frames."""
frames = []
segments = list_spec.split(",")
for segment in segments:
ranges = segment.split("-")
if len(ranges) == 1:
if not ranges[0].isdigit():
msg = (
"Invalid frame number "
f"in the list: {ranges[0]}"
)
raise ValueError(msg)
frames.append(int(ranges[0]))
continue
start, end = segment.split("-")
start, end = int(start), int(end)
frames.extend(range(start, end + 1))
return frames
@staticmethod
def _get_collection(
file_locations: FileLocations,
regex: Optional[re.Pattern] = None) -> clique.Collection:
r"""Get collection from file locations.
Args:
file_locations (FileLocations): File locations trait.
regex (Optional[re.Pattern]): Regular expression to match
frame numbers. This is passed to ``clique.assemble()``.
Default clique pattern is::
\.(?P<index>(?P<padding>0*)\d+)\.\D+\d?$
Returns:
clique.Collection: Collection instance.
Raises:
ValueError: If zero or multiple collections found.
"""
patterns = None if not regex else [regex]
files: list[Path] = [
file.file_path.as_posix()
for file in file_locations.file_paths
]
src_collections, _ = clique.assemble(files)
src_collections, _ = clique.assemble(files, patterns=patterns)
if len(src_collections) != 1:
msg = (
f"Zero or multiple collections found: {len(src_collections)} "
"expected 1"
)
raise ValueError(msg)
return src_collections[0]
src_collection = src_collections[0]
@staticmethod
def get_frame_padding(file_locations: FileLocations) -> int:
"""Get frame padding."""
src_collection = Sequence._get_collection(file_locations)
destination_indexes = list(src_collection.indexes)
# Use last frame for minimum padding
# - that should cover both 'udim' and 'frame' minimum padding
return len(str(destination_indexes[-1]))
@staticmethod
def get_frame_list(
file_locations: FileLocations,
regex: Optional[re.Pattern] = None,
) -> list[int]:
r"""Get frame list.
Args:
file_locations (FileLocations): File locations trait.
regex (Optional[re.Pattern]): Regular expression to match
frame numbers. This is passed to ``clique.assemble()``.
Default clique pattern is::
\.(?P<index>(?P<padding>0*)\d+)\.\D+\d?$
Returns:
list[int]: List of frame numbers.
"""
src_collection = Sequence._get_collection(file_locations, regex)
return list(src_collection.indexes)
# Do we need one for drop and non-drop frame?
class SMPTETimecode(TraitBase):
"""SMPTE Timecode trait model."""

62
tests/client/ayon_core/pipeline/traits/test_traits.py
View file

@ -406,3 +406,65 @@ def test_sequence_get_frame_padding() -> None:
assert Sequence.get_frame_padding(
file_locations=representation.get_trait(FileLocations)) == 4
def test_sequence_validations() -> None:
"""Test Sequence trait validation."""
file_locations_list = [
FileLocation(
file_path=Path(f"/path/to/file.{frame}.exr"),
file_size=1024,
file_hash=None,
)
for frame in range(1001, 1010 + 1) # because range is zero based
]
file_locations_list += [
FileLocation(
file_path=Path(f"/path/to/file.{frame}.exr"),
file_size=1024,
file_hash=None,
)
for frame in range(1015, 1020 + 1)
]
file_locations_list += [
FileLocation
(
file_path=Path("/path/to/file.1100.exr"),
file_size=1024,
file_hash=None,
)
]
representation = Representation(name="test", traits=[
FileLocations(file_paths=file_locations_list),
FrameRanged(
frame_start=1001,
frame_end=1100, frames_per_second="25"),
Sequence(
frame_padding=4,
frame_spec="1001-1010,1015-1020,1100")
])
representation.get_trait(Sequence).validate(representation)
def test_list_spec_to_frames() -> None:
"""Test converting list specification to frames."""
assert Sequence.list_spec_to_frames("1-10,20-30,55") == [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10,
20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 55
]
assert Sequence.list_spec_to_frames("1,2,3,4,5") == [
1, 2, 3, 4, 5
]
assert Sequence.list_spec_to_frames("1-10") == [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10
]
assert Sequence.list_spec_to_frames("1") == [1]
with pytest.raises(
ValueError,
match="Invalid frame number in the list: .*"):
Sequence.list_spec_to_frames("a")