ffx/bin/ffx/model/track.py

# from typing import List
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base, sessionmaker

from .show import Base

from ffx.track_type import TrackType

from ffx.iso_language import IsoLanguage

from ffx.track_disposition import TrackDisposition
from ffx.track_descriptor import TrackDescriptor

from ffx.audio_layout import AudioLayout

import click

class Track(Base):
    """
    relationship(argument, opt1, opt2, ...)
    argument is string of class or Mapped class of the target entity
    backref creates a bi-directional corresponding relationship (back_populates preferred)
    back_populates points to the corresponding relationship (the actual class attribute identifier)
    
    See: https://docs.sqlalchemy.org/en/(14|20)/orm/basic_relationships.html
    """

    __tablename__ = 'tracks'

    # v1.x
    id = Column(Integer, primary_key=True, autoincrement = True)

    # P=pattern_id+sub_index+track_type
    track_type = Column(Integer) # TrackType

    index = Column(Integer)
    source_index = Column(Integer)

    # v1.x
    pattern_id = Column(Integer, ForeignKey('patterns.id', ondelete="CASCADE"))
    pattern = relationship('Pattern', back_populates='tracks')

    track_tags = relationship('TrackTag', back_populates='track', cascade="all, delete", lazy="joined")

    disposition_flags = Column(Integer)
    
    codec_name = Column(String)
    audio_layout = Column(Integer)


    def __init__(self, **kwargs):

        trackType = kwargs.pop('track_type', None)
        if trackType is not None:
            self.track_type = int(trackType)

        dispositionSet = kwargs.pop(TrackDescriptor.DISPOSITION_SET_KEY, set())
        self.disposition_flags = int(TrackDisposition.toFlags(dispositionSet))

        super().__init__(**kwargs) 


    @classmethod
    def fromFfprobeStreamObj(cls, streamObj, patternId):
        """{
                'index': 4,
                'codec_name': 'hdmv_pgs_subtitle',
                'codec_long_name': 'HDMV Presentation Graphic Stream subtitles',
                'codec_type': 'subtitle',
                'codec_tag_string': '[0][0][0][0]',
                'codec_tag': '0x0000',
                'r_frame_rate': '0/0',
                'avg_frame_rate': '0/0',
                'time_base': '1/1000',
                'start_pts': 0,
                'start_time': '0.000000',
                'duration_ts': 1421035,
                'duration': '1421.035000',
                'disposition': {
                    'default': 1,
                    'dub': 0,
                    'original': 0,
                    'comment': 0,
                    'lyrics': 0,
                    'karaoke': 0,
                    'forced': 0,
                    'hearing_impaired': 0,
                    'visual_impaired': 0,
                    'clean_effects': 0,
                    'attached_pic': 0,
                    'timed_thumbnails': 0,
                    'non_diegetic': 0,
                    'captions': 0,
                    'descriptions': 0,
                    'metadata': 0,
                    'dependent': 0,
                    'still_image': 0
                },
                'tags': {
                    'language': 'ger',
                    'title': 'German Full'
                }
        }
        

        # v1.x
        id = Column(Integer, primary_key=True, autoincrement = True)

        # P=pattern_id+sub_index+track_type
        track_type = Column(Integer) # TrackType
        sub_index = Column(Integer)

        # v1.x
        pattern_id = Column(Integer, ForeignKey('patterns.id', ondelete='CASCADE'))
        pattern = relationship('Pattern', back_populates='tracks')


        language = Column(String) # IsoLanguage threeLetter
        title = Column(String)


        track_tags = relationship('TrackTag', back_populates='track', cascade='all, delete')


        disposition_flags = Column(Integer)

    
        """


        trackType = streamObj[TrackDescriptor.FFPROBE_CODEC_TYPE_KEY]

        if trackType in [t.label() for t in TrackType]:

            return cls(pattern_id = patternId,
                        track_type = trackType,
                        codec_name = streamObj[TrackDescriptor.FFPROBE_CODEC_NAME_KEY],
                        disposition_flags = sum([2**t.index() for (k,v) in streamObj[TrackDescriptor.FFPROBE_DISPOSITION_KEY].items()
                                                        if v and (t := TrackDisposition.find(k)) is not None]),
                        audio_layout = AudioLayout.identify(streamObj))

        else:
            return None


    def getId(self):
        return int(self.id)

    def getPatternId(self):
        return int(self.pattern_id)
    
    def getType(self):
        return TrackType.fromIndex(self.track_type)
    
    def getCodec(self):
        return str(self.codec_name)

    def getIndex(self):
        return int(self.index) if self.index is not None else -1

    def getSourceIndex(self):
        return int(self.source_index) if self.source_index is not None else -1

    def getLanguage(self):
        tags = {t.key:t.value for t in self.track_tags}
        return IsoLanguage.findThreeLetter(tags['language']) if 'language' in tags.keys() else IsoLanguage.UNDEFINED
    
    def getTitle(self):
        tags = {t.key:t.value for t in self.track_tags}
        return tags['title'] if 'title' in tags.keys() else ''

    def getDispositionSet(self):
        return TrackDisposition.toSet(self.disposition_flags)
    
    def getAudioLayout(self):
        return AudioLayout.fromIndex(self.audio_layout)

    def getTags(self):
        return {str(t.key):str(t.value) for t in self.track_tags}


    def setDisposition(self, disposition : TrackDisposition):
            self.disposition_flags = self.disposition_flags | int(2**disposition.index())

    def resetDisposition(self, disposition : TrackDisposition):
            self.disposition_flags = self.disposition_flags & sum([2**d.index() for d in TrackDisposition if d != disposition])

    def getDisposition(self, disposition : TrackDisposition):
            return bool(self.disposition_flags & 2**disposition.index())


    def getDescriptor(self, subIndex : int = -1) -> TrackDescriptor:

        kwargs = {}

        kwargs[TrackDescriptor.ID_KEY] = self.getId()
        kwargs[TrackDescriptor.PATTERN_ID_KEY] = self.getPatternId()

        kwargs[TrackDescriptor.INDEX_KEY] = self.getIndex()
        kwargs[TrackDescriptor.SOURCE_INDEX_KEY] = self.getSourceIndex()

        if subIndex > -1:
            kwargs[TrackDescriptor.SUB_INDEX_KEY] = subIndex

        kwargs[TrackDescriptor.TRACK_TYPE_KEY] = self.getType()
        kwargs[TrackDescriptor.CODEC_NAME_KEY] = self.getCodec()
        
        kwargs[TrackDescriptor.DISPOSITION_SET_KEY] = self.getDispositionSet()
        kwargs[TrackDescriptor.TAGS_KEY] = self.getTags()
        
        kwargs[TrackDescriptor.AUDIO_LAYOUT_KEY] = self.getAudioLayout()

        return TrackDescriptor(**kwargs)