PyCS/pycs/database/Result.py

import json
import typing as T

from pycs import db
from pycs.database.base import BaseModel
from pycs.database.util import commit_on_return

class ResultConfirmation(BaseModel):
    """ DB Model for user confirmations of results """

    result_id = db.Column(
        db.Integer,
        db.ForeignKey("result.id", ondelete="CASCADE"),
        nullable=False)

    confirming_user = db.Column(db.String, nullable=False)

    serialize_only = BaseModel.serialize_only + (
        "result_id",
        "confirming_user",
    )

class Result(BaseModel):
    """ DB Model for projects """

    file_id = db.Column(
        db.Integer,
        db.ForeignKey("file.id", ondelete="CASCADE"),
        nullable=False)

    origin = db.Column(db.String, nullable=False)
    origin_user = db.Column(db.String, nullable=True)
    type = db.Column(db.String, nullable=False)

    label_id = db.Column(
        db.Integer,
        db.ForeignKey("label.id", ondelete="SET NULL"),
        nullable=True)

    data_encoded = db.Column(db.String)

    result_confirmations = db.relationship("ResultConfirmation",
        backref="result",
        lazy="dynamic",
        passive_deletes=True,
    )

    serialize_only = BaseModel.serialize_only + (
        "file_id",
        "origin",
        "origin_user",
        "type",
        "label_id",
        "data",
        "confirmations"
    )

    def serialize(self):
        """ extends the default serialize with the decoded data attribute """
        result = super().serialize()
        result["data"] = self.data
        result["confirmations"] = self.confirmations

        return result

    @property
    def data(self) -> T.Optional[dict]:
        """ getter for the decoded data attribute """
        return None if self.data_encoded is None else json.loads(self.data_encoded)

    @data.setter
    def data(self, value):
        """
            setter for the decoded data attribute
            The attribute is encoded property before assigned to the object.
        """
        if isinstance(value, str) or value is None:
            self.data_encoded = value

        elif isinstance(value, (dict, list)):
            self.data_encoded = json.dumps(value)

        else:
            raise ValueError(f"Not supported type: {type(value)}")

    @commit_on_return
    def set_origin(self, origin: str, origin_user: str = None):
        """
        set this results origin

        :param origin: either 'user' or 'pipeline'
        :param origin_user: None if origin is 'pipeline' else name of the user
        :return:
        """
        if origin == "pipeline" and not origin_user is None:
            raise ValueError("If an annotation was made by the pipeline no user"\
                "can be specified!")

        self.origin = origin
        self.origin_user = origin_user
        self.reset_confirmations()

    @commit_on_return
    def set_label(self, label: int):
        """
        set this results origin

        :param label: label ID
        :return:
        """
        if self.label_id != label:
            self.reset_confirmations()
            self.label_id = label

    @property
    def confirmations(self) -> T.List[ResultConfirmation]:
        """
            Returns all confirmations for this results

            :return: list of result confirmations
        """

        confirmations = db.session.query(ResultConfirmation).filter(
                            ResultConfirmation.result.has(Result.id==self.id))
        _confirmations = [c.serialize() for c in confirmations.all()]

        _confirmations = [{k:v for k, v in c.items()
                            if k in ('id', 'confirming_user')}
                            for c in _confirmations]

        return _confirmations

    def reset_confirmations(self) -> T.List[ResultConfirmation]:
        """
        Resets all confirmations

        :return: list of result confirmation objects
        """
        confirmations = ResultConfirmation.query.filter(
                ResultConfirmation.result_id == self.id)

        _confirmations = [c.serialize() for c in confirmations.all()]
        confirmations.delete()

        return _confirmations

    @commit_on_return
    def confirm(self, user: str):
        """
            Result is confirmed by the given user. This sets the origin to "user".
            If no username was specified before, the given username is used.
            A confirmation is only added if it does not already exist. The result
            has be labeled to be confirmed.

            :param user: username
        """
        if user is None:
            raise ValueError("When confirming a result the username has to" \
                            "be specified.")

        if self.origin == "pipeline":
            self.set_origin(origin="user", origin_user=user)

        # Get current confirmations.
        confirmations = self.confirmations

        # Results can only be confirmed if the result is labeled.
        # Also, the original annotator cannot confirm the result and we want
        # to avoid duplicates.
        if self.label_id is not None and self.origin_user != user and not len(confirmations) > 0:
            ResultConfirmation.new(commit=False,
                                   result_id=self.id,
                                   confirming_user=user)