diff --git a/client/ayon_core/pipeline/traits/content.py b/client/ayon_core/pipeline/traits/content.py index 01fe60f0ca..cf31669336 100644 --- a/client/ayon_core/pipeline/traits/content.py +++ b/client/ayon_core/pipeline/traits/content.py @@ -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 diff --git a/client/ayon_core/pipeline/traits/representation.py b/client/ayon_core/pipeline/traits/representation.py index acd78a6ce5..34aadd590a 100644 --- a/client/ayon_core/pipeline/traits/representation.py +++ b/client/ayon_core/pipeline/traits/representation.py @@ -6,6 +6,7 @@ import re import sys import uuid from functools import lru_cache +from types import GenericAlias from typing import ClassVar, Optional, Type, TypeVar, Union from .trait import ( @@ -50,7 +51,8 @@ class Representation: """ _data: dict _module_blacklist: ClassVar[list[str]] = [ - "_", "builtins", "pydantic"] + "_", "builtins", "pydantic", + ] name: str representation_id: str @@ -422,9 +424,19 @@ class Representation: for module in filtered_modules.values(): if not module: continue - for _, klass in inspect.getmembers(module, inspect.isclass): - if inspect.isclass(klass) \ - and issubclass(klass, TraitBase) \ + + for attr_name in dir(module): + klass = getattr(module, attr_name) + if not inspect.isclass(klass): + continue + # this needs to be done because of the bug? in + # python ABCMeta, where ``issubclass`` is not working + # if it hits the GenericAlias (that is in fact + # tuple[int, int]). This is added to the scope by + # ``types`` module. + if type(klass) is GenericAlias: + continue + if issubclass(klass, TraitBase) \ and str(klass.id).startswith(trait_id): trait_candidates.add(klass) return trait_candidates diff --git a/client/ayon_core/pipeline/traits/time.py b/client/ayon_core/pipeline/traits/time.py index 83534aac18..eacaa72235 100644 --- a/client/ayon_core/pipeline/traits/time.py +++ b/client/ayon_core/pipeline/traits/time.py @@ -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(?P0*)\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(?P0*)\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.""" diff --git a/tests/client/ayon_core/pipeline/traits/test_traits.py b/tests/client/ayon_core/pipeline/traits/test_traits.py index 514066d4ef..319401b91e 100644 --- a/tests/client/ayon_core/pipeline/traits/test_traits.py +++ b/tests/client/ayon_core/pipeline/traits/test_traits.py @@ -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")