PyCS/pycs/database/Project.py

import typing as T

from contextlib import closing
from datetime import datetime
from os.path import join
from typing import Iterator
from typing import List
from typing import Optional
from typing import Tuple

from pycs import db
from pycs.database.base import NamedBaseModel

from pycs.database.Collection import Collection
from pycs.database.File import File
from pycs.database.Label import Label
from pycs.database.util import commit_on_return
from pycs.database.util.TreeNodeLabel import TreeNodeLabel

class Project(NamedBaseModel):
    description = db.Column(db.String)

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

    model_id = db.Column(
        db.Integer,
        db.ForeignKey("model.id", ondelete="SET NULL"))

    label_provider_id = db.Column(
        db.Integer,
        db.ForeignKey("label_provider.id", ondelete="SET NULL"))

    root_folder = db.Column(db.String, nullable=False, unique=True)

    external_data = db.Column(db.Boolean, nullable=False)

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

    # contraints
    __table_args__ = ()

    # relationships to other models
    files = db.relationship("File", backref="project", lazy=True)
    labels = db.relationship("Label", backref="project", lazy=True)
    collections = db.relationship("Collection", backref="project", lazy=True)


    def label(self, id: int) -> T.Optional[Label]:
        """
        get a label using its unique identifier

        :param identifier: unique identifier
        :return: label
        """
        return self.labels.get(id)

    def file(self, id: int) -> T.Optional[Label]:
        """
        get a file using its unique identifier

        :param identifier: unique identifier
        :return: file
        """
        return self.files.get(id)

    def collection(self, id: int) -> T.Optional[Collection]:
        """
        get a collection using its unique identifier

        :param identifier: unique identifier
        :return: collection
        """
        return self.collections.get(id)

    def collection_by_reference(self, reference: str) -> T.Optional[Collection]:
        """
        get a collection using its unique identifier

        :param identifier: unique identifier
        :return: collection
        """
        return self.collections.filter_by(reference=reference).one()

    @commit_on_return
    def create_label(self, name: str, reference: str = None,
                     parent_id: int = None,
                     hierarchy_level: str = None) -> Tuple[Optional[Label], bool]:
        """
        create a label for this project. If there is already a label with the same reference
        in the database its name is updated.

        :param name: label name
        :param reference: label reference
        :param parent_id: parent's identifier
        :param hierarchy_level: hierarchy level name
        :return: created or edited label, insert
        """

        label = Label.query.get(project=self, reference=reference)
        is_new = False

        if label is None:
            label = Label.new(project=self, reference=reference)
            is_new = True

        label.set_name(name, commit=False)
        label.set_parent(parent_id, commit=False)
        label.hierarchy_level = hierarchy_level

        return label, is_new

    def label_tree(self) -> List[TreeNodeLabel]:
        """
        get a list of root labels associated with this project

        :return: list of labels
        """
        raise NotImplementedError
        with closing(self.database.con.cursor()) as cursor:
            cursor.execute('''
                WITH RECURSIVE
                    tree AS (
                        SELECT labels.* FROM labels
                            WHERE project = ? AND parent IS NULL
                        UNION ALL
                        SELECT labels.* FROM labels
                            JOIN tree ON labels.parent = tree.id
                    )
                SELECT * FROM tree
            ''', [self.identifier])

            result = []
            lookup = {}

            for row in cursor.fetchall():
                label = TreeNodeLabel(self.database, row)
                lookup[label.identifier] = label

                if label.parent_id is None:
                    result.append(label)
                else:
                    lookup[label.parent_id].children.append(label)

            return result

    @commit_on_return
    def create_collection(self,
                          reference: str,
                          name: str,
                          description: str,
                          position: int,
                          autoselect: bool) -> Tuple[Collection, bool]:
        """
        create a new collection associated with this project

        :param reference: collection reference string
        :param name: collection name
        :param description: collection description
        :param position: position in menus
        :param autoselect: automatically select this collection on session load

        :return: collection object, insert
        """

        collection = Collection.query.get(project=self, reference=reference)
        is_new = False

        if collection is None:
            collection = Collection.new(project=self,
                                        reference=reference)
            is_new = True

        collection.name = name
        collection.description = description
        collection.position = position
        collection.autoselect = autoselect

        return collection, is_new

    @commit_on_return
    def add_file(self, uuid: str, file_type: str, name: str, extension: str, size: int,
                 filename: str, frames: int = None, fps: float = None) -> T.Tuple[File, bool]:
        """
        add a file to this project

        :param uuid: unique identifier which is used for temporary files
        :param file_type: file type (either image or video)
        :param name: file name
        :param extension: file extension
        :param size: file size
        :param filename: actual name in filesystem
        :param frames: frame count
        :param fps: frames per second
        :return: file
        """
        path = join(self.data_folder, filename + extension)

        file = File.objects.get(project=self, path=path)
        is_new = False

        if file is None:
            file = File.new(uuid=uuid, project=self, path=path)
            is_new = True

        file.type = file_type
        file.name = name
        file.extension = extension
        file.size = size
        file.frames = frames
        file.fps = fps

        return file, is_new


    def set_description(self, description: str):
        """
        set this projects description

        :param description: new description
        :return:
        """
        self.description = description
        self
    def count_files(self) -> int:
        """
        count files associated with this project

        :return: count
        """
        return self.files.count()

    def get_files(self, offset: int = 0, limit: int = -1) -> T.Iterator[File]:
        """
        get an iterator of files associated with this project

        :param offset: file offset
        :param limit: file limit
        :return: iterator of files
        """
        return self.files.order_by(File.id.acs()).offset(offset).limit(limit)

    def count_files_without_results(self) -> int:
        """
        count files without associated results

        :return: count
        """
        raise NotImplementedError

        with closing(self.database.con.cursor()) as cursor:
            cursor.execute('''
                SELECT COUNT(*)
                FROM files
                LEFT JOIN results ON files.id = results.file
                WHERE files.project = ? AND results.id IS NULL
            ''', [self.identifier])
            return cursor.fetchone()[0]

    def files_without_results(self) -> Iterator[File]:
        """
        get an iterator of files without associated results

        :return: list of files
        """
        raise NotImplementedError

        with closing(self.database.con.cursor()) as cursor:
            cursor.execute('''
                SELECT files.*
                FROM files
                LEFT JOIN results ON files.id = results.file
                WHERE files.project = ? AND results.id IS NULL
                ORDER BY id ASC
            ''', [self.identifier])

            for row in cursor:
                yield File(self.database, row)

    def files_without_collection(self, offset: int = 0, limit: int = -1) -> Iterator[File]:
        """
        get an iterator of files without not associated with any collection

        :return: list of files
        """
        return self.get_files(offset, limit).filter(File.collection_id == None)


    def count_files_without_collection(self) -> int:
        """
        count files associated with this project but with no collection

        :return: count
        """
        return self.files_without_collection().count()