feat: ED-1658 Video Space + Image Space

encord/constants/enums.py

@@ -63,3 +63,8 @@ def to_upper_case_string(self) -> str:
 
 def is_geometric(data_type: DataType) -> bool:
     return data_type in GEOMETRIC_TYPES
+
+
+class SpaceType(StringEnum):
+    VIDEO = "video"
+    IMAGE = "image"

encord/objects/answers.py

@@ -13,7 +13,7 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Dict, Generic, Iterable, List, NoReturn, Optional, Set, TypeVar, Union
+from typing import Any, Dict, Generic, Iterable, List, NoReturn, Optional, Set, TypeVar, Union, cast
 
 from encord.common.deprecated import deprecated
 from encord.objects.attributes import (
@@ -27,6 +27,8 @@
 from encord.objects.frames import Ranges, ranges_to_list
 from encord.objects.ontology_element import _get_element_by_hash
 from encord.objects.options import FlatOption, NestableOption
+from encord.objects.types import AnswerDict as AnswerDict
+from encord.objects.types import AttributeDict, DynamicAttributeObject
 from encord.objects.utils import _lower_snake_case, short_uuid_str
 
 ValueType = TypeVar("ValueType")
@@ -89,7 +91,9 @@ def get(self) -> ValueType:
         assert self._value is not None, "Value can't be none for the answered Answer object"
         return self._value
 
-    def to_encord_dict(self, ranges: Optional[Ranges] = None) -> Optional[Dict[str, Any]]:
+    def to_encord_dict(
+        self, ranges: Optional[Ranges] = None, spaceId: Optional[str] = None
+    ) -> Optional[AttributeDict | DynamicAttributeObject]:
         """A low level helper to convert to the Encord JSON format.
         For most use cases the `get_answer` function should be used instead.
         """
@@ -100,27 +104,39 @@ def to_encord_dict(self, ranges: Optional[Ranges] = None) -> Optional[Dict[str,
         if self.is_dynamic:
             if ranges is None:
                 raise ValueError("Frame range should be set for dynamic answers")
-
-            ret.update(self._get_encord_dynamic_fields(ranges))
+            dynamic_ret = self._add_dynamic_fields(ret, ranges, spaceId)
+            return dynamic_ret
 
         return ret
 
     @abstractmethod
-    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> Dict[str, Any]:
+    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> AttributeDict:
         pass
 
     @abstractmethod
     def from_dict(self, d: Dict[str, Any]) -> None:
         pass
 
-    def _get_encord_dynamic_fields(self, ranges: Ranges) -> Dict[str, Any]:
-        return {
+    def _add_dynamic_fields(
+        self, base_answer: AttributeDict, ranges: Ranges, spaceId: Optional[str] = None
+    ) -> DynamicAttributeObject:
+        ret: DynamicAttributeObject = {
+            "name": base_answer["name"],
+            "value": base_answer["value"],
+            "featureHash": base_answer["featureHash"],
+            "answers": base_answer["answers"],
+            "manualAnnotation": self._is_manual_annotation,
             "dynamic": True,
             "range": ranges_to_list(ranges),
             "shouldPropagate": self._should_propagate,
             "trackHash": self._track_hash,
         }
 
+        if spaceId is not None:
+            ret["spaceId"] = spaceId
+
+        return ret
+
 
 class TextAnswer(Answer[str, TextAttribute]):
     def __init__(self, ontology_attribute: TextAttribute):
@@ -150,7 +166,7 @@ def copy_from(self, text_answer: TextAnswer):
             other_answer = text_answer.get()
             self.set(other_answer)
 
-    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> Dict[str, Any]:
+    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> AttributeDict:
         return {
             "name": self.ontology_attribute.name,
             "value": _lower_snake_case(self.ontology_attribute.name),
@@ -217,19 +233,22 @@ def copy_from(self, radio_answer: RadioAnswer):
             other_answer = radio_answer.get()
             self.set(other_answer)
 
-    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> Dict[str, Any]:
+    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> AttributeDict:
         nestable_option = self._value
         assert nestable_option is not None  # Check is performed earlier, so just to silence mypy
 
         return {
             "name": self.ontology_attribute.name,
             "value": _lower_snake_case(self.ontology_attribute.name),
             "answers": [
-                {
-                    "name": nestable_option.label,
-                    "value": nestable_option.value,
-                    "featureHash": nestable_option.feature_node_hash,
-                }
+                cast(
+                    AnswerDict,
+                    {
+                        "name": nestable_option.label,
+                        "value": nestable_option.value,
+                        "featureHash": nestable_option.feature_node_hash,
+                    },
+                )
             ],
             "featureHash": self.ontology_attribute.feature_node_hash,
             "manualAnnotation": self.is_manual_annotation,
@@ -379,15 +398,18 @@ def _verify_flat_option(self, value: FlatOption) -> None:
                 f"is associated with this class: `{self._ontology_attribute}`"
             )
 
-    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> Dict[str, Any]:
+    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> AttributeDict:
         ontology_attribute: ChecklistAttribute = self._ontology_attribute
         checked_options = [option for option in ontology_attribute.options if self.get_option_value(option)]
-        answers = [
-            {
-                "name": option.label,
-                "value": option.value,
-                "featureHash": option.feature_node_hash,
-            }
+        answers: List[AnswerDict] = [
+            cast(
+                AnswerDict,
+                {
+                    "name": option.label,
+                    "value": option.value,
+                    "featureHash": option.feature_node_hash,
+                },
+            )
             for option in checked_options
         ]
         return {
@@ -449,7 +471,7 @@ def set(self, value: NumericAnswerValue, manual_annotation: bool = DEFAULT_MANUA
         self._answered = True
         self.is_manual_annotation = manual_annotation
 
-    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> Dict[str, Any]:
+    def _to_encord_dict_impl(self, is_dynamic: bool = False) -> AttributeDict:
         return {
             "name": self.ontology_attribute.name,
             "value": _lower_snake_case(self.ontology_attribute.name),

encord/objects/classification_instance.py

@@ -23,6 +23,7 @@
     List,
     NoReturn,
     Optional,
+    Protocol,
     Sequence,
     Set,
     Tuple,
@@ -50,11 +51,13 @@
     _search_child_attributes,
 )
 from encord.objects.options import Option, _get_option_by_hash
-from encord.objects.types import ClassificationAnswer, ClassificationObject
+from encord.objects.spaces.annotation.base_annotation import ClassificationAnnotation
+from encord.objects.types import AttributeDict, ClassificationAnswer
 from encord.objects.utils import check_email, short_uuid_str
 
 if TYPE_CHECKING:
     from encord.objects import LabelRowV2
+    from encord.objects.spaces.base_space import Space
 
 
 class ClassificationInstance:
@@ -81,6 +84,8 @@ def __init__(
         # Only used for frame entities
         self._frames_to_data: Dict[int, ClassificationInstance.FrameData] = defaultdict(self.FrameData)
 
+        self._spaces: dict[str, Space] = dict()
+
     @property
     def classification_hash(self) -> str:
         """A unique identifier for the classification instance."""
@@ -114,6 +119,12 @@ def _last_frame(self) -> Union[int, float]:
         else:
             return self._parent.number_of_frames
 
+    def _operation_not_allowed_for_classifications_on_space(self, extended_message: Optional[str] = None) -> None:
+        base_message = "This operation is not allowed for classifications that exist on a space."
+        error_message = base_message + extended_message if extended_message is not None else base_message
+        if self._is_assigned_to_space():
+            raise LabelRowError(error_message)
+
     @property
     def range_list(self) -> Ranges:
         return self._range_manager.get_ranges()
@@ -172,7 +183,7 @@ def manual_annotation(self) -> bool:
     def manual_annotation(self, manual_annotation: bool) -> None:
         self._instance_data.manual_annotation = manual_annotation
 
-    def is_on_frame(self, frame: int) -> bool:
+    def is_on_frame(self, frame: Frames) -> bool:
         intersection = self._range_manager.intersection(frame)
         return len(intersection) > 0
 
@@ -183,6 +194,15 @@ def is_range_only(self) -> bool:
     def is_global(self) -> bool:
         return self._ontology_classification.is_global
 
+    def _is_assigned_to_space(self) -> bool:
+        return bool(self._spaces)
+
+    def _add_to_space(self, space: Space) -> None:
+        self._spaces[space.space_id] = space
+
+    def _remove_from_space(self, space_id: str) -> None:
+        self._spaces.pop(space_id)
+
     def is_assigned_to_label_row(self) -> bool:
         return self._parent is not None
 
@@ -234,6 +254,10 @@ def set_for_frames(
             manual_annotation: Optionally specify whether the classification instance on this frame was manually annotated. Defaults to `True`.
             reviews: Should only be set by internal functions.
         """
+        self._operation_not_allowed_for_classifications_on_space(
+            extended_message="For adding the classification to different frames on a space, use Space.place_classification."
+        )
+
         if created_at is None:
             created_at = datetime.now()
 
@@ -292,6 +316,8 @@ def get_annotation(self, frame: Union[int, str] = 0) -> Annotation:
         frame: Either the frame number or the image hash if the data type is an image or image group.
             Defaults to the first frame.
         """
+        self._operation_not_allowed_for_classifications_on_space()
+
         if self.is_global():
             raise LabelRowError("Cannot get annotation for a global classification instance.")
         elif isinstance(frame, str):
@@ -311,6 +337,10 @@ def get_annotation(self, frame: Union[int, str] = 0) -> Annotation:
         return self.Annotation(self, frame_num)
 
     def remove_from_frames(self, frames: Frames) -> None:
+        self._operation_not_allowed_for_classifications_on_space(
+            extended_message="For removing the classification from different frames on a space, use Space.unplace_classification."
+        )
+
         range_manager = RangeManager(frame_class=frames)
         ranges_to_remove = range_manager.get_ranges()
 
@@ -323,11 +353,18 @@ def remove_from_frames(self, frames: Frames) -> None:
         if self._parent:
             self._parent._remove_frames_from_classification(self, frames)
 
-    def get_annotations(self) -> List[Annotation]:
+    # TODO: Need to deprecate this old Annotation
+    def get_annotations(self) -> Union[List[Annotation], List[ClassificationAnnotation]]:
         """Returns:
         A list of `ClassificationInstance.Annotation` in order of available frames.
         """
-        return [self.get_annotation(frame_num) for frame_num in sorted(self._frames_to_data.keys())]
+        if self._is_assigned_to_space():
+            res: List[ClassificationAnnotation] = []
+            for space in self._spaces.values():
+                res.extend(space.get_classification_annotations(filter_classifications=[self.classification_hash]))
+            return res
+        else:
+            return [self.get_annotation(frame_num) for frame_num in sorted(self._frames_to_data.keys())]
 
     def is_valid(self) -> None:
         if not len(self._frames_to_data) > 0 and not self.is_range_only():
@@ -371,7 +408,7 @@ def set_answer(
 
         static_answer.set(answer)
 
-    def set_answer_from_list(self, answers_list: List[ClassificationObject]) -> None:
+    def set_answer_from_list(self, answers_list: List[AttributeDict]) -> None:
         """This is a low level helper function and should not be used directly.
 
         Sets the answer for the classification from a dictionary.
@@ -584,6 +621,7 @@ class FrameData:
         manual_annotation: bool = DEFAULT_MANUAL_ANNOTATION
         last_edited_at: datetime = field(default_factory=datetime.now)
         last_edited_by: Optional[str] = None
+        """ This field is deprecated. It will always be None """
         reviews: Optional[List[dict]] = None
 
         @staticmethod
@@ -612,7 +650,7 @@ def from_dict(d: Union[dict, ClassificationAnswer]) -> ClassificationInstance.Fr
                 manual_annotation=manual_annotation,
                 last_edited_at=last_edited_at,
                 last_edited_by=d.get("lastEditedBy"),
-                reviews=d.get("reviews"),
+                reviews=None,
             )
 
         def update_from_optional_fields(