PyCS/pycs/database/File.py

from __future__ import annotations

import json
import typing as T

from contextlib import closing
from datetime import datetime

from pycs import db
from pycs.database.Collection import Collection
from pycs.database.Result import Result
from pycs.database.base import NamedBaseModel
from pycs.database.util import commit_on_return


class File(NamedBaseModel):

    # table columns
    uuid = db.Column(db.String, nullable=False)

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

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

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

    created = db.Column(db.DateTime, default=datetime.utcnow,
        index=True, nullable=False)

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

    frames = db.Column(db.Integer)

    fps = db.Column(db.Float)

    project_id = db.Column(
        db.Integer,
        db.ForeignKey("project.id", ondelete="CASCADE"),
        nullable=False)

    collection_id = db.Column(
        db.Integer,
        db.ForeignKey("collection.id", ondelete="SET NULL"))

    # contraints
    __table_args__ = (
        db.UniqueConstraint('project_id', 'path'),
    )


    # relationships to other models
    results = db.relationship("Result", backref="file", lazy=True)


    @commit_on_return
    def set_collection(self, id: T.Optional[int]):
        """
        set this file's collection

        :param id: new collection id
        :return:
        """

        self.collection_id = id

    @commit_on_return
    def set_collection_by_reference(self, collection_reference: T.Optional[str]):
        """
        set this file's collection

        :param collection_reference: collection reference
        :return:
        """
        if self.collection_reference is None:
            self.set_collection(None)

        collection = Collection.query.filter_by(reference=collection_reference).one()
        self.collection = collection

    def _get_another_file(self, *query) -> T.Optional[File]:
        """
        get the first file matching the query ordered by descending id

        :return: another file or None
        """
        return File.query.filter(File.project_id == self.project_id, *query)\
            .order_by(File.id.desc())\
            .first()

    def next(self) -> T.Optional[File]:
        """
        get the successor of this file

        :return: another file or None
        """
        query = File.id > self.id,
        return self._get_another_file(*query)


    def previous(self) -> T.Optional[File]:
        """
        get the predecessor of this file

        :return: another file or None
        """
        query = File.id < self.id,
        return self._get_another_file(*query)


    def next_in_collection(self) -> T.Optional[File]:
        """
        get the predecessor of this file

        :return: another file or None
        """
        query = File.id > self.id, File.collection_id == self.collection_id
        return self._get_another_file(*query)


    def previous_in_collection(self) -> T.Optional[File]:
        """
        get the predecessor of this file

        :return: another file or None
        """
        query = File.id < self.id, File.collection_id == self.collection_id
        return self._get_another_file(*query)


    def result(self, id: int) -> T.Optional[Result]:
        return self.results.get(id)


    def create_result(self, origin, result_type, label, data: T.Optional[dict] = None):
        data = data if data is None else json.dumps(data)

        result = Result.new(commit=True,
                            file=self,
                            origin=origin,
                            type=result_type,
                            label=label,
                            data=data)
        return result


    def remove_results(self, origin='pipeline'):

        results = Result.query.filter(Result.file == self, Result.origin == origin)

        results.remove()

        return results