added get/set_user_scores

Author: Narek Mkhitaryan

docs/source/api_reference/api_team.rst

@@ -12,3 +12,5 @@ Team
 .. automethod:: superannotate.SAClient.list_users
 .. automethod:: superannotate.SAClient.pause_user_activity
 .. automethod:: superannotate.SAClient.resume_user_activity
+.. automethod:: superannotate.SAClient.get_user_scores
+.. automethod:: superannotate.SAClient.set_user_scores

pytest.ini

@@ -3,4 +3,4 @@ minversion = 3.7
 log_cli=true
 python_files = test_*.py
 ;pytest_plugins = ['pytest_profiling']
-addopts = -n 4 --dist loadscope
+;addopts = -n 4 --dist loadscope

src/superannotate/lib/app/interface/sdk_interface.py

@@ -608,6 +608,166 @@ def resume_user_activity(
             f"User with email {user.email} has been successfully unblocked from the specified projects: {projects}."
         )
 
+    def get_user_scores(
+        self,
+        project: Union[NotEmptyStr, Tuple[int, int], Tuple[str, str]],
+        item: Union[NotEmptyStr, int],
+        scored_user: NotEmptyStr,
+        *,
+        score_names: Optional[List[str]] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        Retrieve score metadata for a user for a specific item in a specific project.
+
+        :param project: Project and folder as a tuple, folder is optional.
+        :type project: Union[str, Tuple[int, int], Tuple[str, str]]
+
+        :param item:  The unique ID or name of the item.
+        :type item: Union[str, int]
+
+        :param scored_user:  The email address of the project user.
+        :type scored_user: str
+
+        :param score_names:  A list of score names to filter by. If None, returns all scores.
+        :type score_names: Optional[List[str]]
+
+        :return: A list of dictionaries containing score metadata for the user.
+        :rtype: list of dicts
+
+        Request Example:
+        ::
+
+            client.get_user_scores(
+                project=("my_multimodal", "folder1"),
+                item="item1",
+                scored_user="example@superannotate.com",
+                score_names=["Accuracy Score", "Speed"]
+            )
+
+        Response Example:
+        ::
+
+            [
+                {
+                    "createdAt": "2024-06-01T13:00:00",
+                    "updatedAt": "2024-06-01T13:00:00",
+                    "id": 3217575,
+                    "name": "Accuracy Score",
+                    "value": 98,
+                    "weight": 4,
+                },
+                {
+                    "createdAt": "2024-06-01T13:00:00",
+                    "updatedAt": "2024-06-01T13:00:00",
+                    "id": 5657575,
+                    "name": "Speed",
+                    "value": 9,
+                    "weight": 0.4,
+                },
+            ]
+        """
+
+        if isinstance(project, str):
+            project_name, folder_name = extract_project_folder(project)
+            project, folder = self.controller.get_project_folder(
+                project_name, folder_name
+            )
+        else:
+            project_pk, folder_pk = project
+            if isinstance(project_pk, int) and isinstance(folder_pk, int):
+                project = self.controller.get_project_by_id(project_pk).data
+                folder = self.controller.get_folder_by_id(folder_pk, project.id).data
+            elif isinstance(project_pk, str) and isinstance(folder_pk, str):
+                project = self.controller.get_project(project_pk)
+                folder = self.controller.get_folder(project, folder_pk)
+            else:
+                raise AppException("Provided project param is not valid.")
+        response = BaseSerializer.serialize_iterable(
+            self.controller.work_management.get_user_scores(
+                project=project,
+                folder=folder,
+                item=item,
+                scored_user=scored_user,
+                provided_score_names=score_names,
+            )
+        )
+        return response
+
+    def set_user_scores(
+        self,
+        project: Union[NotEmptyStr, Tuple[int, int], Tuple[str, str]],
+        item: Union[NotEmptyStr, int],
+        scored_user: NotEmptyStr,
+        scores: List[Dict[str, Any]],
+    ):
+        """
+        Assign score metadata for a user in a scoring component.
+
+        :param project: Project and folder as a tuple, folder is optional.
+        :type project: Union[str, Tuple[int, int], Tuple[str, str]]
+
+        :param item:  The unique ID or name of the item.
+        :type item: Union[str, int]
+
+        :param scored_user:  Set the email of the user being scored.
+        :type scored_user: str
+
+        :param scores: A list of dictionaries containing the following key-value pairs:
+                * **name** (*str*): The name of the score (required).
+                * **value** (*Any*): The score value (required).
+                * **weight** (*Union[float, int]*, optional): The weight of the score. Defaults to `1` if not provided.
+
+            **Example**:
+            ::
+
+                scores = [
+                    {
+                        "name": "Speed",  # str (required)
+                        "value": 90,      # Any (required)
+                        "weight": 1       # Union[float, int] (optional, defaults to 1.0 if not provided)
+                    }
+                ]
+        :type scores: List[Dict[str, Any]
+
+        Request Example:
+        ::
+
+            client.set_user_scores(
+                project=("my_multimodal", "folder1"),
+                item_=12345,
+                scored_user="example@superannotate.com",
+                scores=[
+                    {"name": "Speed", "value": 90},
+                    {"name": "Accuracy", "value": 9, "weight": 4.0},
+                    {"name": "Attention to Detail", "value": None, "weight": None},
+                ]
+            )
+
+        """
+        if isinstance(project, str):
+            project_name, folder_name = extract_project_folder(project)
+            project, folder = self.controller.get_project_folder(
+                project_name, folder_name
+            )
+        else:
+            project_pk, folder_pk = project
+            if isinstance(project_pk, int) and isinstance(folder_pk, int):
+                project = self.controller.get_project_by_id(project_pk).data
+                folder = self.controller.get_folder_by_id(folder_pk, project.id).data
+            elif isinstance(project_pk, str) and isinstance(folder_pk, str):
+                project = self.controller.get_project(project_pk)
+                folder = self.controller.get_folder(project, folder_pk)
+            else:
+                raise AppException("Provided project param is not valid.")
+        self.controller.work_management.set_user_scores(
+            project=project,
+            folder=folder,
+            item=item,
+            scored_user=scored_user,
+            scores=scores,
+        )
+        logger.info("Scores successfully set.")
+
     def get_component_config(self, project: Union[NotEmptyStr, int], component_id: str):
         """
         Retrieves the configuration for a given project and component ID.

src/superannotate/lib/core/entities/work_managament.py

@@ -1,13 +1,18 @@
 import datetime
 from enum import auto
 from enum import Enum
+from typing import Any
 from typing import Optional
+from typing import Union
 
 from lib.core.entities.base import TimedBaseModel
 from lib.core.enums import WMUserStateEnum
+from lib.core.exceptions import AppException
+from lib.core.pydantic_v1 import BaseModel
 from lib.core.pydantic_v1 import Extra
 from lib.core.pydantic_v1 import Field
 from lib.core.pydantic_v1 import parse_datetime
+from lib.core.pydantic_v1 import root_validator
 from lib.core.pydantic_v1 import validator
 
 
@@ -149,3 +154,55 @@ def json(self, **kwargs):
         if "exclude" not in kwargs:
             kwargs["exclude"] = {"custom_fields"}
         return super().json(**kwargs)
+
+
+class WMScoreEntity(TimedBaseModel):
+    id: int
+    team_id: int
+    name: str
+    description: Optional[str]
+    type: str
+    payload: Optional[dict]
+
+
+class TelemetryScoreEntity(BaseModel):
+    item_id: int
+    team_id: int
+    project_id: int
+    user_id: str
+    user_role: str
+    score_id: int
+    value: Optional[Any]
+    weight: Optional[float]
+
+
+class ScoreEntity(TimedBaseModel):
+    id: int
+    name: str
+    value: Optional[Any]
+    weight: Optional[float]
+
+
+class ScorePayloadEntity(BaseModel):
+    name: str
+    value: Any
+    weight: Optional[Union[float, int]] = 1.0
+
+    class Config:
+        extra = Extra.forbid
+
+    @validator("weight", pre=True, always=True)
+    def validate_weight(cls, v):
+        if v is not None and (not isinstance(v, (int, float)) or v <= 0):
+            raise AppException("Please provide a valid number greater than 0")
+        return v
+
+    @root_validator(pre=True)
+    def check_weight_and_value(cls, values):
+        value = values.get("value")
+        weight = values.get("weight")
+        if (weight is None and value is not None) or (
+            weight is not None and value is None
+        ):
+            raise AppException("Weight and Value must both be set or both be None.")
+        return values

src/superannotate/lib/core/pydantic_v1.py

@@ -22,6 +22,7 @@
 ROOT_KEY = pydantic.utils.ROOT_KEY  # noqa
 sequence_like = pydantic.utils.sequence_like  # noqa
 validator = pydantic.validator  # noqa
+root_validator = pydantic.root_validator  # noqa
 constr = pydantic.constr  # noqa
 conlist = pydantic.conlist  # noqa
 parse_datetime = pydantic.datetime_parse.parse_datetime  # noqa

src/superannotate/lib/core/service_types.py

@@ -5,7 +5,9 @@
 from typing import Union
 
 from lib.core import entities
+from lib.core.entities.work_managament import TelemetryScoreEntity
 from lib.core.entities.work_managament import WMProjectEntity
+from lib.core.entities.work_managament import WMScoreEntity
 from lib.core.entities.work_managament import WMUserEntity
 from lib.core.enums import ProjectType
 from lib.core.exceptions import AppException
@@ -260,6 +262,14 @@ class SettingsListResponse(ServiceResponse):
     res_data: List[entities.SettingEntity] = None
 
 
+class WMScoreListResponse(ServiceResponse):
+    res_data: List[WMScoreEntity] = None
+
+
+class TelemetryScoreListResponse(ServiceResponse):
+    res_data: List[TelemetryScoreEntity] = None
+
+
 PROJECT_TYPE_RESPONSE_MAP = {
     ProjectType.VECTOR: ImageResponse,
     ProjectType.OTHER: ClassificationResponse,

src/superannotate/lib/core/serviceproviders.py

@@ -29,6 +29,7 @@
 from lib.core.service_types import UserResponse
 from lib.core.service_types import WMCustomFieldResponse
 from lib.core.service_types import WMProjectListResponse
+from lib.core.service_types import WMScoreListResponse
 from lib.core.service_types import WMUserListResponse
 from lib.core.service_types import WorkflowListResponse
 from lib.core.types import Attachment
@@ -197,6 +198,24 @@ def update_user_activity(
     ) -> ServiceResponse:
         raise NotImplementedError
 
+    @abstractmethod
+    def list_scores(self) -> WMScoreListResponse:
+        raise NotImplementedError
+
+    @abstractmethod
+    def create_score(
+        self,
+        name: str,
+        description: Optional[str],
+        score_type: Literal["rating", "number", "radio"],
+        payload: dict,
+    ) -> ServiceResponse:
+        raise NotImplementedError
+
+    @abstractmethod
+    def delete_score(self, score_id: int) -> ServiceResponse:
+        raise NotImplementedError
+
 
 class BaseProjectService(SuperannotateServiceProvider):
     @abstractmethod
@@ -688,6 +707,20 @@ def query_item_count(
         raise NotImplementedError
 
 
+class BaseTelemetryScoringService(SuperannotateServiceProvider):
+    @abstractmethod
+    def get_score_values(self, project_id: int, item_id: int, user_id: str):
+        raise NotImplementedError
+
+    @abstractmethod
+    def set_score_values(
+        self,
+        project_id: int,
+        data: List[dict],
+    ) -> ServiceResponse:
+        raise NotImplementedError
+
+
 class BaseServiceProvider:
     projects: BaseProjectService
     folders: BaseFolderService