Release v0.4.5

This commit is contained in:
Javanaut
2026-07-08 11:10:00 +02:00
26 changed files with 1173 additions and 93 deletions

View File

@@ -1,6 +1,11 @@
Version History
===============
0.4.5
-----
- Release version bumped to 0.4.5.
0.4.4
-----

View File

@@ -1,7 +1,7 @@
[project]
name = "ffx"
description = "FFX recoding and metadata managing tool"
version = "0.4.4"
version = "0.4.5"
license = {file = "LICENSE.md"}
dependencies = [
"requests",

View File

@@ -100,6 +100,19 @@ def normalizeCpuOption(ctx, param, value):
raise click.BadParameter(str(ex)) from ex
def normalizeDeinterlaceOption(ctx, param, value):
if value == 'none':
return value
from ffx.filter.deinterlace_filter import DeinterlaceFilter
try:
DeinterlaceFilter.getModeTokens(value)
except ValueError as ex:
raise click.BadParameter(str(ex)) from ex
return value
def normalizeSubtitleExtension(ctx, param, value):
normalizedExtension = str(value).strip().lower()
if normalizedExtension.startswith('.'):
@@ -1194,7 +1207,21 @@ def checkUniqueDispositions(context, mediaDescriptor: MediaDescriptor):
@click.option("--output-directory", type=str, default='')
@click.option("--deinterlace", is_flag=False, flag_value="default", default="none")
@click.option(
"--deinterlace",
type=str,
metavar="SPEC",
is_flag=False,
flag_value="default",
default="none",
show_default=True,
callback=normalizeDeinterlaceOption,
help=(
"Deinterlace with bwdif followed by ordered upN, downN, and tempHHHH "
"components. Scale factors N range from 2 to 8; H is hexadecimal. "
"For example: up2_temp3366_down2."
),
)
@click.option("--denoise", is_flag=False, flag_value="default", default="none")
@click.option("--denoise-use-hw", is_flag=True, default=False)
@@ -1524,7 +1551,7 @@ def convert(ctx,
NlmeansFilter(**denoiseKwargs)
if deinterlace != 'none' and not copyVideoEffective:
DeinterlaceFilter()
DeinterlaceFilter(mode=deinterlace)
chainYield = list(qf.getChainYield())
@@ -1690,7 +1717,10 @@ def convert(ctx,
# ctx.obj['logger'].debug(f"tmd subindices: {[t.getIndex() for t in targetMediaDescriptor.getAllTrackDescriptors()]} {[t.getSubIndex() for t in targetMediaDescriptor.getAllTrackDescriptors()]} {[t.getDispositionFlag(TrackDisposition.DEFAULT) for t in targetMediaDescriptor.getAllTrackDescriptors()]}")
ctx.obj['logger'].debug(f"tmd subindices: {[t.getIndex() for t in targetMediaDescriptor.getTrackDescriptors()]} {[t.getSubIndex() for t in targetMediaDescriptor.getTrackDescriptors()]} {[t.getDispositionFlag(TrackDisposition.DEFAULT) for t in targetMediaDescriptor.getTrackDescriptors()]}")
ctx.obj['logger'].debug(f"Input mapping tokens (2nd pass): {targetMediaDescriptor.getInputMappingTokens()}")
ctx.obj['logger'].debug(
"Input mapping tokens (2nd pass): "
+ f"{targetMediaDescriptor.getInputMappingTokens(sourceMediaDescriptor=sourceMediaDescriptor)}"
)
fc = FfxController(context, targetMediaDescriptor, sourceMediaDescriptor)

View File

@@ -1,4 +1,4 @@
VERSION='0.4.4'
VERSION='0.4.5'
DATABASE_VERSION = 3
DEFAULT_QUALITY = 32

View File

@@ -108,12 +108,18 @@ def ensureCurrentSchemaCompatibility(databaseContext):
column['name']
for column in inspector.get_columns('shows')
}
trackColumns = {
column['name']
for column in inspector.get_columns('tracks')
}
alterStatements = []
if 'quality' not in showColumns:
alterStatements.append("ALTER TABLE shows ADD COLUMN quality INTEGER DEFAULT 0")
if 'notes' not in showColumns:
alterStatements.append("ALTER TABLE shows ADD COLUMN notes TEXT DEFAULT ''")
if 'dropped' not in trackColumns:
alterStatements.append("ALTER TABLE tracks ADD COLUMN dropped INTEGER DEFAULT 0")
if not alterStatements:
return

View File

@@ -1,9 +1,11 @@
from textual.app import App
from .i18n import set_current_language, t
from .confirm_screen import ConfirmScreen
from .shows_screen import ShowsScreen
from .inspect_details_screen import InspectDetailsScreen
from .media_edit_screen import MediaEditScreen
from .pattern_controller import InvalidPatternCaptureGroupError, PatternController
from .screen_support import configure_screen_log_handler, set_screen_log_pane_enabled
@@ -40,7 +42,20 @@ class FfxApp(App):
self.push_screen(ShowsScreen())
if self.context['command'] == 'inspect':
self.push_screen(InspectDetailsScreen())
try:
self.push_screen(InspectDetailsScreen())
except InvalidPatternCaptureGroupError as ex:
self._invalid_pattern_error = ex
self.push_screen(
ConfirmScreen(
t(
"Pattern #{pattern_id} matches the filename but has no capture group for the season/episode indicator. Delete this pattern and quit?"
).format(pattern_id=ex.pattern_id),
confirm_label=t("Delete"),
cancel_label=t("Quit"),
),
self._handle_invalid_pattern_confirmation,
)
if self.context['command'] == 'edit':
self.push_screen(MediaEditScreen())
@@ -49,3 +64,13 @@ class FfxApp(App):
def getContext(self):
"""Data 'output' method"""
return self.context
def _handle_invalid_pattern_confirmation(self, confirmed: bool):
invalid_pattern_error = getattr(self, "_invalid_pattern_error", None)
if confirmed and invalid_pattern_error is not None:
PatternController(self.context).deletePattern(
invalid_pattern_error.pattern_id
)
self.exit()

View File

@@ -501,12 +501,7 @@ class FfxController():
commandSequence1 = (commandTokens
+ self.__targetMediaDescriptor.getInputMappingTokens(only_video=True))
# Optional tokens
#NOTE: Filters and so needs to run on the first pass as well, as here
# the required bitrate for the second run is determined and recorded
# TODO: Results seems to be slightly better with first pass omitted,
# Confirm or find better filter settings for 2-pass
# commandSequence1 += self.__context['denoiser'].generatefilterTokens()
commandSequence1 += filterTokens
for td in self.__targetMediaDescriptor.getTrackDescriptors(trackType=TrackType.VIDEO):
#HINT: Attached thumbnails are not supported by .webm container format

View File

@@ -1,4 +1,4 @@
import itertools
import re
from .filter import Filter
@@ -7,6 +7,14 @@ class DeinterlaceFilter(Filter):
IDENTIFIER = 'bwdif'
MODE_KEY = 'mode'
DEFAULT_MODE = 'default'
DEFAULT_SCALE_FACTOR = 2
MIN_SCALE_FACTOR = 2
MAX_SCALE_FACTOR = 8
DEFAULT_TEMP_PARAMETERS = '3366'
# DEFAULT_STRENGTH: float = 2.8
# DEFAULT_PATCH_SIZE: int = 13
# DEFAULT_CHROMA_PATCH_SIZE: int = 9
@@ -21,7 +29,12 @@ class DeinterlaceFilter(Filter):
def __init__(self, **kwargs):
self.__mode = kwargs.get(
DeinterlaceFilter.MODE_KEY,
DeinterlaceFilter.DEFAULT_MODE,
)
self.__tokens = DeinterlaceFilter.getModeTokens(self.__mode)
# self.__useHardware = kwargs.get('use_hardware', False)
# self.__strengthList = []
@@ -109,6 +122,59 @@ class DeinterlaceFilter(Filter):
super().__init__(self)
@staticmethod
def getModeTokens(mode):
if mode == DeinterlaceFilter.DEFAULT_MODE:
return ['bwdif=mode=1']
if not isinstance(mode, str) or not mode:
raise ValueError("Deinterlace mode must not be empty")
tokens = ['bwdif=mode=1']
for component in mode.split('_'):
scaleMatch = re.fullmatch(r'(up|down)(\d*)', component)
if scaleMatch:
direction, parameter = scaleMatch.groups()
factor = (
int(parameter)
if parameter
else DeinterlaceFilter.DEFAULT_SCALE_FACTOR
)
if not (
DeinterlaceFilter.MIN_SCALE_FACTOR
<= factor
<= DeinterlaceFilter.MAX_SCALE_FACTOR
):
raise ValueError(
f"Scale factor in '{component}' must be between "
+ f"{DeinterlaceFilter.MIN_SCALE_FACTOR} and "
+ f"{DeinterlaceFilter.MAX_SCALE_FACTOR}"
)
operator = '*' if direction == 'up' else '/'
tokens.append(
f"scale=iw{operator}{factor}:ih{operator}{factor}:flags=lanczos"
)
continue
tempMatch = re.fullmatch(r'temp([0-9a-fA-F]*)', component)
if tempMatch:
parameters = (
tempMatch.group(1)
or DeinterlaceFilter.DEFAULT_TEMP_PARAMETERS
)
if len(parameters) != 4:
raise ValueError(
f"Temporal parameters in '{component}' must contain "
+ "four hexadecimal digits"
)
strengths = [f"{int(digit, 16) / 2:g}" for digit in parameters]
tokens.append(f"hqdn3d={':'.join(strengths)}")
continue
raise ValueError(f"Unknown deinterlace component: '{component}'")
return tokens
def getPayload(self):
@@ -123,10 +189,12 @@ class DeinterlaceFilter(Filter):
# filterName = 'nlmeans_opencl' if self.__useHardware else 'nlmeans'
payload = {'identifier': DeinterlaceFilter.IDENTIFIER,
'parameters': {},
'parameters': {
DeinterlaceFilter.MODE_KEY: self.__mode,
},
'suffices': suffices,
'variant': f"DEINT",
'tokens': ['bwdif=mode=1']}
'tokens': list(self.__tokens)}
return payload

View File

@@ -10,6 +10,7 @@ from textual.widgets._data_table import CellDoesNotExist
from ffx.file_properties import FileProperties
from ffx.helper import DIFF_ADDED_KEY, DIFF_CHANGED_KEY, DIFF_REMOVED_KEY
from ffx.media_descriptor_change_set import MediaDescriptorChangeSet
from ffx.pattern_controller import InvalidPatternCaptureGroupError
from ffx.show_descriptor import ShowDescriptor
from ffx.track_descriptor import TrackDescriptor
@@ -472,11 +473,15 @@ class InspectDetailsScreen(MediaWorkflowScreenBase):
):
mediaTags[tagKey] = tagValue
patternId = self._pc.savePatternSchema(
patternObj,
trackDescriptors=self._sourceMediaDescriptor.getTrackDescriptors(),
mediaTags=mediaTags,
)
try:
patternId = self._pc.savePatternSchema(
patternObj,
trackDescriptors=self._sourceMediaDescriptor.getTrackDescriptors(),
mediaTags=mediaTags,
)
except InvalidPatternCaptureGroupError as ex:
self.notify(str(ex), severity="error")
return
if patternId:
self.reloadProperties(reset_draft=True)
self.updateMediaTags()
@@ -498,10 +503,14 @@ class InspectDetailsScreen(MediaWorkflowScreenBase):
patternObj
and self._currentPattern.getPattern() != patternObj["pattern"]
):
updated = self._pc.updatePattern(
self._currentPattern.getId(),
patternObj,
)
try:
updated = self._pc.updatePattern(
self._currentPattern.getId(),
patternObj,
)
except InvalidPatternCaptureGroupError as ex:
self.notify(str(ex), severity="error")
return False
if updated:
self.reloadProperties(reset_draft=True)
self.updateMediaTags()

View File

@@ -473,45 +473,48 @@ class MediaDescriptor:
td: TrackDescriptor = sortedTrackDescriptors[trackIndex]
#HINT: Attached thumbnails are not supported by .webm container format
if td.getCodec() != TrackCodec.PNG:
trackType = td.getType()
trackCodec = td.getCodec()
if (
trackCodec == TrackCodec.PNG
or trackType == TrackType.ATTACHMENT
or (only_video and trackType != TrackType.VIDEO)
):
continue
sourceTrackDescriptor = sourceTrackDescriptorsByIndex.get(td.getSourceIndex())
if sourceTrackDescriptor is None:
raise ValueError(f"No source track descriptor found for source index {td.getSourceIndex()}")
sourceTrackDescriptor = sourceTrackDescriptorsByIndex.get(td.getSourceIndex())
if sourceTrackDescriptor is None:
raise ValueError(f"No source track descriptor found for source index {td.getSourceIndex()}")
stdi = sourceTrackDescriptor.getIndex()
stdsi = sourceTrackDescriptor.getSubIndex()
stdi = sourceTrackDescriptor.getIndex()
stdsi = sourceTrackDescriptor.getSubIndex()
trackType = td.getType()
trackCodec = td.getCodec()
importedFilePath = td.getExternalSourceFilePath()
if (trackType != TrackType.ATTACHMENT
and (trackType == TrackType.VIDEO or not only_video)):
if use_sub_index:
if importedFilePath:
importedFilePath = td.getExternalSourceFilePath()
inputMappingTokens += [
"-map",
f"{filePointer}:{trackType.indicator()}:0",
]
filePointer += 1
if use_sub_index:
else:
if importedFilePath:
if trackCodec not in [TrackCodec.PGS, TrackCodec.VOBSUB]:
inputMappingTokens += [
"-map",
f"0:{trackType.indicator()}:{stdsi}",
]
inputMappingTokens += [
"-map",
f"{filePointer}:{trackType.indicator()}:0",
]
filePointer += 1
else:
if trackCodec not in [TrackCodec.PGS, TrackCodec.VOBSUB]:
inputMappingTokens += ["-map", f"0:{stdi}"]
else:
if not trackCodec in [TrackCodec.PGS, TrackCodec.VOBSUB]:
inputMappingTokens += [
"-map",
f"0:{trackType.indicator()}:{stdsi}",
]
else:
if not trackCodec in [TrackCodec.PGS, TrackCodec.VOBSUB]:
inputMappingTokens += ["-map", f"0:{stdi}"]
if only_video:
return inputMappingTokens
if sourceMediaDescriptor:
fontDescriptors = [ftd for ftd in sourceMediaDescriptor.getAttachmentTracks()
@@ -522,7 +525,10 @@ class MediaDescriptor:
and ftd.getAttachmentFormat() == AttachmentFormat.TTF]
for ad in sorted(fontDescriptors, key=lambda d: d.getIndex()):
inputMappingTokens += ["-map", f"0:{ad.getIndex()}"]
sourceIndex = ad.getIndex()
if not sourceMediaDescriptor and ad.getSourceIndex() > -1:
sourceIndex = ad.getSourceIndex()
inputMappingTokens += ["-map", f"0:{sourceIndex}"]
return inputMappingTokens

View File

@@ -75,13 +75,17 @@ class Pattern(Base):
# Set ordered subindices
subIndexCounter = {}
outputIndex = 0
for track in self.tracks:
trackType = track.getType()
if trackType == TrackType.ATTACHMENT:
if trackType == TrackType.ATTACHMENT or track.isDropped():
continue
if not trackType in subIndexCounter.keys():
subIndexCounter[trackType] = 0
kwargs[MediaDescriptor.TRACK_DESCRIPTOR_LIST_KEY].append(track.getDescriptor(context, subIndex = subIndexCounter[trackType]))
trackDescriptor = track.getDescriptor(context, subIndex = subIndexCounter[trackType])
trackDescriptor.setIndex(outputIndex)
kwargs[MediaDescriptor.TRACK_DESCRIPTOR_LIST_KEY].append(trackDescriptor)
subIndexCounter[trackType] += 1
outputIndex += 1
return MediaDescriptor(**kwargs)

View File

@@ -36,6 +36,7 @@ class Track(Base):
index = Column(Integer)
source_index = Column(Integer)
dropped = Column(Integer, default=0)
# v1.x
pattern_id = Column(Integer, ForeignKey('patterns.id', ondelete="CASCADE"))
@@ -181,6 +182,9 @@ class Track(Base):
def getSourceIndex(self):
return int(self.source_index) if self.source_index is not None else -1
def isDropped(self) -> bool:
return bool(self.dropped)
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
@@ -221,6 +225,7 @@ class Track(Base):
kwargs[TrackDescriptor.INDEX_KEY] = self.getIndex()
kwargs[TrackDescriptor.SOURCE_INDEX_KEY] = self.getSourceIndex()
kwargs[TrackDescriptor.DROPPED_KEY] = self.isDropped()
if subIndex > -1:
kwargs[TrackDescriptor.SUB_INDEX_KEY] = subIndex

View File

@@ -19,6 +19,21 @@ class InvalidPatternSchemaError(click.ClickException):
pass
class InvalidPatternCaptureGroupError(click.ClickException):
def __init__(self, pattern_id: int, pattern_expression: str):
self.pattern_id = int(pattern_id)
self.pattern_expression = str(pattern_expression)
pattern_label = (
f"Pattern #{self.pattern_id}"
if self.pattern_id >= 0
else "Pattern"
)
super().__init__(
f"{pattern_label} must contain at least one capture group "
f"for the season/episode indicator: {self.pattern_expression!r}"
)
class PatternController:
_compiled_regex_cache: dict[str, re.Pattern] = {}
@@ -66,6 +81,13 @@ class PatternController:
cls._compiled_regex_cache[expression_text] = compiled
return compiled
@classmethod
def _validate_pattern_capture_groups(cls, pattern_id: int, expression: str):
compiled = cls._compile_pattern_expression(pattern_id, expression)
if compiled.groups < 1:
raise InvalidPatternCaptureGroupError(pattern_id, expression)
return compiled
def _coerce_pattern_fields(self, patternObj):
return {
"show_id": int(patternObj["show_id"]),
@@ -103,6 +125,14 @@ class PatternController:
"Patterns must define at least one track before they can be stored."
)
if not any(
not descriptor.isDropped()
for descriptor in normalized_descriptors
):
raise InvalidPatternSchemaError(
"Patterns must define at least one non-dropped track before they can be stored."
)
normalized_descriptors = sorted(
normalized_descriptors, key=lambda descriptor: descriptor.getIndex()
)
@@ -146,6 +176,7 @@ class PatternController:
codec_name=str(trackDescriptor.getFormatDescriptor().identifier()),
index=int(trackDescriptor.getIndex()),
source_index=int(trackDescriptor.getSourceIndex()),
dropped=1 if trackDescriptor.isDropped() else 0,
disposition_flags=int(
TrackDisposition.toFlags(trackDescriptor.getDispositionSet())
),
@@ -179,10 +210,19 @@ class PatternController:
pattern.tracks.append(self._build_track_row(trackDescriptor))
def _validate_persisted_pattern(self, pattern: Pattern):
if not any(track.getType() != TrackType.ATTACHMENT for track in pattern.tracks):
persisted_tracks = [
track
for track in pattern.tracks
if track.getType() != TrackType.ATTACHMENT
]
if not persisted_tracks:
raise InvalidPatternSchemaError(
f"Pattern #{pattern.getId()} ({pattern.getPattern()!r}) is invalid because it has no tracks."
)
if not any(not track.isDropped() for track in persisted_tracks):
raise InvalidPatternSchemaError(
f"Pattern #{pattern.getId()} ({pattern.getPattern()!r}) is invalid because it has no non-dropped tracks."
)
def savePatternSchema(
self,
@@ -194,6 +234,10 @@ class PatternController:
fields = self._coerce_pattern_fields(patternObj)
normalized_tracks = self._normalize_track_descriptors(trackDescriptors)
normalized_tags = self._coerce_media_tags(mediaTags)
self._validate_pattern_capture_groups(
patternId if patternId is not None else -1,
fields["pattern"],
)
session = None
try:
@@ -256,6 +300,7 @@ class PatternController:
def updatePattern(self, patternId, patternObj):
fields = self._coerce_pattern_fields(patternObj)
self._validate_pattern_capture_groups(patternId, fields["pattern"])
session = None
try:
@@ -391,6 +436,11 @@ class PatternController:
patternMatch = compiled.search(str(filename))
if patternMatch is None:
continue
if not patternMatch.groups():
raise InvalidPatternCaptureGroupError(
pattern.getId(),
pattern.getPattern(),
)
self._validate_persisted_pattern(pattern)
matches.append({"match": patternMatch, "pattern": pattern})

View File

@@ -2,14 +2,15 @@ import click, re
from typing import List
from textual import events
from textual.css.query import NoMatches
from textual.screen import Screen
from textual.widgets import Header, Footer, Static, Button, Input, DataTable, TextArea
from textual.containers import Grid
from ffx.model.pattern import Pattern
from ffx.pattern_controller import InvalidPatternCaptureGroupError
from .track_details_screen import TrackDetailsScreen
from .track_delete_screen import TrackDeleteScreen
from .shifted_season_delete_screen import ShiftedSeasonDeleteScreen
from .shifted_season_details_screen import ShiftedSeasonDetailsScreen
@@ -154,6 +155,13 @@ class PatternDetailsScreen(Screen):
def updateTracks(self):
selectedTrack = self.getSelectedTrackDescriptor()
selectedTrackIndex = (
self._track_row_index(selectedTrack)
if selectedTrack is not None
else None
)
selectedCursorColumn = getattr(self.tracksTable.cursor_coordinate, "column", 0)
self.tracksTable.clear()
self.__trackRowData = {}
@@ -161,6 +169,7 @@ class PatternDetailsScreen(Screen):
tracks = self.getCurrentTrackDescriptors()
typeCounter = {}
restoredRowIndex = None
td: TrackDescriptor
for td in tracks:
@@ -185,13 +194,77 @@ class PatternDetailsScreen(Screen):
td.getTitle(),
t('Yes') if TrackDisposition.DEFAULT in dispoSet else t('No'),
t('Yes') if TrackDisposition.FORCED in dispoSet else t('No'),
t('Yes') if td.isDropped() else t('No'),
td.getSourceIndex())
row_key = self.tracksTable.add_row(*map(str, row))
self.__trackRowData[row_key] = td
if (
selectedTrack is not None
and self._descriptor_refs_same_track(td, selectedTrack)
):
restoredRowIndex = len(self.__trackRowData) - 1
typeCounter[trackType] += 1
if restoredRowIndex is None and selectedTrackIndex is not None and self.__trackRowData:
restoredRowIndex = min(selectedTrackIndex, len(self.__trackRowData) - 1)
if restoredRowIndex is not None:
self.tracksTable.move_cursor(
row=restoredRowIndex,
column=selectedCursorColumn,
animate=False,
scroll=False,
)
self.updateTrackDropButtonState()
def _track_row_index(self, trackDescriptor: TrackDescriptor | None):
if trackDescriptor is None:
return None
for rowIndex, descriptor in enumerate(self.__trackRowData.values()):
if self._descriptor_refs_same_track(descriptor, trackDescriptor):
return rowIndex
return None
def _descriptor_refs_same_track(
self,
descriptor: TrackDescriptor,
otherDescriptor: TrackDescriptor,
) -> bool:
if descriptor.getId() != -1 and otherDescriptor.getId() != -1:
return descriptor.getId() == otherDescriptor.getId()
return (
descriptor.getPatternId() == otherDescriptor.getPatternId()
and descriptor.getIndex() == otherDescriptor.getIndex()
and descriptor.getSubIndex() == otherDescriptor.getSubIndex()
and descriptor.getType() == otherDescriptor.getType()
)
def updateTrackDropButtonState(self):
try:
button = self.query_one("#button_toggle_track_drop", Button)
except (NoMatches, AttributeError):
return
selectedTrack = self.getSelectedTrackDescriptor()
button.disabled = selectedTrack is None
button.variant = (
"warning"
if selectedTrack is not None and selectedTrack.isDropped()
else "default"
)
def on_data_table_row_highlighted(
self,
event: DataTable.RowHighlighted,
) -> None:
if event.data_table.id == "tracks_table":
self.updateTrackDropButtonState()
def getCurrentTrackDescriptors(self) -> List[TrackDescriptor]:
if self.__pattern is not None:
@@ -384,6 +457,7 @@ class PatternDetailsScreen(Screen):
self.column_key_track_title = add_auto_table_column(self.tracksTable, t("Title"))
self.column_key_track_default = add_auto_table_column(self.tracksTable, t("Default"))
self.column_key_track_forced = add_auto_table_column(self.tracksTable, t("Forced"))
self.column_key_track_drop = add_auto_table_column(self.tracksTable, t("Drop"))
self.column_key_track_source_index = add_auto_table_column(self.tracksTable, t("SrcIndex"))
self.tracksTable.cursor_type = 'row'
@@ -481,7 +555,7 @@ class PatternDetailsScreen(Screen):
yield Static(t("Streams"))
yield Button(t("Add"), id="button_add_track")
yield Button(t("Edit"), id="button_edit_track")
yield Button(t("Delete"), id="button_delete_track")
yield Button(t("Drop"), id="button_toggle_track_drop")
yield Static(" ")
yield Button(t("Up"), id="button_track_up")
@@ -502,7 +576,7 @@ class PatternDetailsScreen(Screen):
yield Static(" ", classes="five")
# Row 20
yield Static(" ", classes="seven")
yield Static(" ", id="messagestatic", classes="seven")
yield build_screen_log_pane()
yield Footer()
@@ -537,6 +611,12 @@ class PatternDetailsScreen(Screen):
def getNotesFromInput(self):
return str(self.query_one("#notes_textarea", TextArea).text)
def setMessage(self, message: str):
try:
self.query_one("#messagestatic", Static).update(str(message) or " ")
except (NoMatches, AttributeError):
pass
def getSelectedTrackDescriptor(self):
@@ -584,25 +664,26 @@ class PatternDetailsScreen(Screen):
patternDescriptor['quality'] = self.getQualityFromInput()
patternDescriptor['notes'] = self.getNotesFromInput()
if self.__pattern is not None:
try:
if self.__pattern is not None:
if self.__pc.updatePattern(self.__pattern.getId(), patternDescriptor):
self.dismiss(patternDescriptor)
else:
#TODO: Meldung
self.app.pop_screen()
if self.__pc.updatePattern(self.__pattern.getId(), patternDescriptor):
self.dismiss(patternDescriptor)
else:
self.setMessage(t("Unable to update pattern."))
else:
patternId = self.__pc.savePatternSchema(
patternDescriptor,
trackDescriptors=self.__draftTracks,
mediaTags=self.__draftTags,
)
if patternId:
self.dismiss(patternDescriptor)
else:
#TODO: Meldung
self.app.pop_screen()
patternId = self.__pc.savePatternSchema(
patternDescriptor,
trackDescriptors=self.__draftTracks,
mediaTags=self.__draftTags,
)
if patternId:
self.dismiss(patternDescriptor)
else:
self.setMessage(t("Unable to save pattern."))
except InvalidPatternCaptureGroupError as ex:
self.setMessage(str(ex))
if event.button.id == "cancel_button":
@@ -663,11 +744,8 @@ class PatternDetailsScreen(Screen):
),
self.handle_edit_track,
)
if event.button.id == "button_delete_track":
self.app.push_screen(
TrackDeleteScreen(trackDescriptor = selectedTrack),
self.handle_delete_track,
)
if event.button.id == "button_toggle_track_drop":
self.toggleSelectedTrackDropped(selectedTrack)
if event.button.id == "button_add_tag":
@@ -751,6 +829,27 @@ class PatternDetailsScreen(Screen):
self.updateTracks()
def toggleSelectedTrackDropped(self, trackDescriptor: TrackDescriptor):
if trackDescriptor is None:
return
updatedTrack = trackDescriptor.clone(context=self.context)
updatedTrack.setDropped(not trackDescriptor.isDropped())
if self.__pattern is not None:
if not self.__tc.updateTrack(updatedTrack.getId(), updatedTrack):
raise click.ClickException("PatternDetailsScreen.toggleSelectedTrackDropped(): track update failed")
else:
for currentTrack in self.__draftTracks:
if (
currentTrack.getIndex() == trackDescriptor.getIndex()
and currentTrack.getSubIndex() == trackDescriptor.getSubIndex()
):
currentTrack.setDropped(updatedTrack.isDropped())
break
self.updateTracks()
def handle_delete_track(self, trackDescriptor : TrackDescriptor):
if trackDescriptor is None:

View File

@@ -48,6 +48,7 @@ class TrackController():
codec_name = str(trackDescriptor.getFormatDescriptor().identifier()),
index = int(trackDescriptor.getIndex()),
source_index = int(trackDescriptor.getSourceIndex()),
dropped = 1 if trackDescriptor.isDropped() else 0,
disposition_flags = int(TrackDisposition.toFlags(trackDescriptor.getDispositionSet())),
audio_layout = trackDescriptor.getAudioLayout().index())
@@ -83,7 +84,21 @@ class TrackController():
if track is not None:
patternId = int(track.pattern_id)
if trackDescriptor.isDropped() and not track.isDropped():
activeSiblingCount = (
s.query(Track)
.filter(Track.pattern_id == patternId, Track.dropped == 0)
.count()
)
if activeSiblingCount <= 1:
raise click.ClickException(
f"Cannot drop the last active track from pattern #{patternId}. Patterns must define at least one non-dropped track."
)
track.index = int(trackDescriptor.getIndex())
track.source_index = int(trackDescriptor.getSourceIndex())
track.dropped = 1 if trackDescriptor.isDropped() else 0
track.track_type = int(trackDescriptor.getType().index())
track.codec_name = str(trackDescriptor.getFormatDescriptor().identifier())
@@ -255,6 +270,17 @@ class TrackController():
f"Cannot delete the last track from pattern #{patternId}. Patterns must define at least one track."
)
if not track.isDropped():
activeSiblingTracks = [
siblingTrack
for siblingTrack in siblingTracks
if not siblingTrack.isDropped()
]
if len(activeSiblingTracks) <= 1:
raise click.ClickException(
f"Cannot delete the last active track from pattern #{patternId}. Patterns must define at least one non-dropped track."
)
index = 0
for track in siblingTracks:

View File

@@ -21,6 +21,7 @@ class TrackDescriptor:
SUB_INDEX_KEY = "sub_index"
PATTERN_ID_KEY = "pattern_id"
EXTERNAL_SOURCE_FILE_PATH_KEY = "external_source_file"
DROPPED_KEY = "dropped"
DISPOSITION_SET_KEY = "disposition_set"
TAGS_KEY = "tags"
@@ -77,6 +78,8 @@ class TrackDescriptor:
else:
self.__externalSourceFilePath = ''
self.__dropped = bool(kwargs.get(TrackDescriptor.DROPPED_KEY, False))
if TrackDescriptor.INDEX_KEY in kwargs.keys():
if type(kwargs[TrackDescriptor.INDEX_KEY]) is not int:
raise TypeError(
@@ -378,11 +381,18 @@ class TrackDescriptor:
def getExternalSourceFilePath(self):
return self.__externalSourceFilePath
def isDropped(self) -> bool:
return self.__dropped
def setDropped(self, dropped: bool):
self.__dropped = bool(dropped)
def clone(self, context: dict | None = None):
kwargs = {
TrackDescriptor.ID_KEY: int(self.__trackId),
TrackDescriptor.PATTERN_ID_KEY: int(self.__patternId),
TrackDescriptor.EXTERNAL_SOURCE_FILE_PATH_KEY: str(self.__externalSourceFilePath),
TrackDescriptor.DROPPED_KEY: bool(self.__dropped),
TrackDescriptor.INDEX_KEY: int(self.__index),
TrackDescriptor.SOURCE_INDEX_KEY: int(self.__sourceIndex),
TrackDescriptor.SUB_INDEX_KEY: int(self.__subIndex),

View File

@@ -1,7 +1,7 @@
import click
from textual.screen import Screen
from textual.widgets import Header, Footer, Static, Button, SelectionList, Select, DataTable, Input
from textual.widgets import Header, Footer, Static, Button, SelectionList, Select, DataTable, Input, Checkbox
from textual.containers import Grid
from textual.widgets._data_table import CellDoesNotExist
@@ -34,7 +34,7 @@ class TrackDetailsScreen(Screen):
Grid {
grid-size: 5 24;
grid-rows: 2 2 2 2 2 3 3 2 2 3 2 2 2 2 2 6 2 2 6 2 2 2;
grid-rows: 2 2 2 2 2 3 3 3 2 3 2 2 2 2 2 6 2 2 6 2 2 2;
grid-columns: 18 1fr 1fr 1fr 4fr;
height: 100%;
width: 100%;
@@ -146,6 +146,7 @@ class TrackDetailsScreen(Screen):
self.__audioLayout = AudioLayout.LAYOUT_UNDEFINED
self.__index = index
self.__subIndex = subIndex
self.__dropped = False
self.__draftTrackTags = {}
initial_language = IsoLanguage.UNDEFINED
initial_title = ""
@@ -156,6 +157,7 @@ class TrackDetailsScreen(Screen):
self.__audioLayout = trackDescriptor.getAudioLayout()
self.__index = trackDescriptor.getIndex()
self.__subIndex = trackDescriptor.getSubIndex()
self.__dropped = trackDescriptor.isDropped()
self.__draftTrackTags = {
key: value
for key, value in trackDescriptor.getTags().items()
@@ -255,6 +257,8 @@ class TrackDetailsScreen(Screen):
self.query_one("#audio_layout_select", Select).value = self.__audioLayout
self.query_one("#drop_checkbox", Checkbox).value = self.__dropped
for disposition in TrackDisposition:
dispositionIsSet = (
@@ -284,6 +288,8 @@ class TrackDetailsScreen(Screen):
self.query_one("#type_select", Select).disabled = True
self.query_one("#audio_layout_select", Select).disabled = True
self.query_one("#drop_checkbox", Checkbox).disabled = True
def on_select_changed(self, event: Select.Changed) -> None:
if event.select.id == "language_select":
self._handle_language_selection_changed(event.value)
@@ -344,7 +350,9 @@ class TrackDetailsScreen(Screen):
)
# Row 8
yield Static(" ", classes="five")
yield Static(t("Drop"))
yield Checkbox(id="drop_checkbox", classes="four")
# Row 9
yield Static(" ", classes="five")
@@ -462,6 +470,9 @@ class TrackDetailsScreen(Screen):
trackTags["title"] = title
kwargs[TrackDescriptor.TAGS_KEY] = trackTags
kwargs[TrackDescriptor.DROPPED_KEY] = bool(
self.query_one("#drop_checkbox", Checkbox).value
)
dispositionFlags = sum(
[2 ** flag for flag in self.query_one("#dispositions_selection_list", SelectionList).selected]
@@ -504,6 +515,7 @@ class TrackDetailsScreen(Screen):
descriptor
for descriptor in siblingTrackList
if descriptor.getType() == trackDescriptor.getType()
and not descriptor.isDropped()
]
numDefaultTracks = len(
@@ -527,10 +539,12 @@ class TrackDetailsScreen(Screen):
trackDescriptor.setSubIndex(int(self.__subIndex))
if (
TrackDisposition.DEFAULT in trackDescriptor.getDispositionSet()
not trackDescriptor.isDropped()
and TrackDisposition.DEFAULT in trackDescriptor.getDispositionSet()
and numDefaultTracks
) or (
TrackDisposition.FORCED in trackDescriptor.getDispositionSet()
not trackDescriptor.isDropped()
and TrackDisposition.FORCED in trackDescriptor.getDispositionSet()
and numForcedTracks
):

View File

@@ -66,7 +66,7 @@ class PatternManagementCliTests(unittest.TestCase):
[to_track_descriptor(track_descriptors[0])],
)
controller.savePatternSchema(
{"show_id": 2, "pattern": r"^dup_.*$"},
{"show_id": 2, "pattern": r"^dup_(s[0-9]+e[0-9]+).*$"},
[to_track_descriptor(track_descriptors[0])],
)
finally:

View File

@@ -144,6 +144,73 @@ class SubtrackMappingBundleTests(unittest.TestCase):
self.assertIn("first embedded subtitle", extracted_subtitle)
self.assertNotIn("second embedded subtitle", extracted_subtitle)
def test_pattern_dropped_track_is_omitted_like_deleted_track(self):
source_filename = "dropped_s01e01.mkv"
source_path = create_source_fixture(
self.workdir,
source_filename,
[
SourceTrackSpec(TrackType.VIDEO, identity="video-0"),
SourceTrackSpec(TrackType.AUDIO, identity="audio-1", language="eng"),
SourceTrackSpec(
TrackType.SUBTITLE,
identity="subtitle-2",
language="eng",
subtitle_lines=("kept subtitle payload",),
),
SourceTrackSpec(
TrackType.SUBTITLE,
identity="subtitle-3",
language="eng",
subtitle_lines=("dropped subtitle payload",),
),
],
)
prepare_pattern_database(
self.database_path,
r"^dropped_(s[0-9]+e[0-9]+)\.mkv$",
[
PatternTrackSpec(index=0, source_index=0, track_type=TrackType.VIDEO),
PatternTrackSpec(index=1, source_index=1, track_type=TrackType.AUDIO),
PatternTrackSpec(index=2, source_index=2, track_type=TrackType.SUBTITLE),
PatternTrackSpec(
index=3,
source_index=3,
track_type=TrackType.SUBTITLE,
dropped=True,
),
],
)
completed = run_ffx_convert(
self.workdir,
self.home_dir,
self.database_path,
"--video-encoder",
"copy",
"--no-tmdb",
"--no-prompt",
"--no-signature",
str(source_path),
)
self.assertCompleted(completed)
output_path = expected_output_path(self.workdir, source_filename)
streams = ffprobe_json(output_path)["streams"]
self.assertEqual(
[stream["codec_type"] for stream in streams],
["video", "audio", "subtitle"],
)
self.assertNotIn(
"subtitle-3",
[get_tag(stream, "THIS_IS") for stream in streams],
)
extracted_subtitle = extract_first_subtitle_text(self.workdir, output_path)
self.assertIn("kept subtitle payload", extracted_subtitle)
self.assertNotIn("dropped subtitle payload", extracted_subtitle)
def test_cli_rearrange_streams_reorders_tracks_without_database_pattern(self):
source_filename = "cli_s01e01.mkv"
source_path = create_source_fixture(

View File

@@ -56,6 +56,7 @@ class PatternTrackSpec:
track_type: TrackType
tags: Mapping[str, str] = field(default_factory=dict)
dispositions: tuple[TrackDisposition, ...] = ()
dropped: bool = False
audio_layout: AudioLayout = AudioLayout.LAYOUT_STEREO
attachment_format: AttachmentFormat = AttachmentFormat.UNKNOWN
@@ -298,6 +299,7 @@ def prepare_pattern_database(database_path: Path, filename_pattern: str, track_s
TrackDescriptor.TRACK_TYPE_KEY: track.track_type,
TrackDescriptor.TAGS_KEY: dict(track.tags),
TrackDescriptor.DISPOSITION_SET_KEY: set(track.dispositions),
TrackDescriptor.DROPPED_KEY: track.dropped,
}
if track.track_type == TrackType.AUDIO:
kwargs[TrackDescriptor.AUDIO_LAYOUT_KEY] = track.audio_layout

View File

@@ -149,6 +149,49 @@ class DatabaseContextTests(unittest.TestCase):
)
cursor.execute("DROP TABLE shifted_seasons_current")
def rewrite_tracks_table_without_dropped(self, cursor):
cursor.execute("ALTER TABLE tracks RENAME TO tracks_current")
cursor.execute(
"""
CREATE TABLE tracks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
track_type INTEGER,
"index" INTEGER,
source_index INTEGER,
pattern_id INTEGER,
disposition_flags INTEGER,
codec_name VARCHAR,
audio_layout INTEGER,
FOREIGN KEY(pattern_id) REFERENCES patterns(id) ON DELETE CASCADE
)
"""
)
cursor.execute(
"""
INSERT INTO tracks (
id,
track_type,
"index",
source_index,
pattern_id,
disposition_flags,
codec_name,
audio_layout
)
SELECT
id,
track_type,
"index",
source_index,
pattern_id,
disposition_flags,
codec_name,
audio_layout
FROM tracks_current
"""
)
cursor.execute("DROP TABLE tracks_current")
def test_database_context_bootstraps_new_database_with_current_version(self):
with patch("ffx.database.Base.metadata.create_all", wraps=Base.metadata.create_all) as mocked_create_all:
context = databaseContext(str(self.database_path))
@@ -321,6 +364,41 @@ class DatabaseContextTests(unittest.TestCase):
mocked_confirm.assert_not_called()
mocked_echo.assert_not_called()
def test_database_context_repairs_current_track_schema_without_version_bump(self):
context = databaseContext(str(self.database_path))
context["engine"].dispose()
connection = sqlite3.connect(self.database_path)
try:
cursor = connection.cursor()
cursor.execute("PRAGMA foreign_keys=OFF")
self.rewrite_tracks_table_without_dropped(cursor)
connection.commit()
finally:
connection.close()
with patch("ffx.database.click.confirm") as mocked_confirm, patch(
"ffx.database.click.echo"
) as mocked_echo:
reopened_context = databaseContext(str(self.database_path))
try:
self.assertEqual(DATABASE_VERSION, getDatabaseVersion(reopened_context))
connection = sqlite3.connect(self.database_path)
try:
column_names = {
row[1]
for row in connection.execute("PRAGMA table_info(tracks)").fetchall()
}
self.assertIn("dropped", column_names)
finally:
connection.close()
finally:
reopened_context["engine"].dispose()
mocked_confirm.assert_not_called()
mocked_echo.assert_not_called()
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,175 @@
import unittest
import click
from ffx.cli import convert
from ffx.filter.deinterlace_filter import DeinterlaceFilter
from ffx.filter.filter import Filter
class DeinterlaceFilterTest(unittest.TestCase):
def setUp(self):
Filter.filterChain.clear()
def tearDown(self):
Filter.filterChain.clear()
def getPayload(self, mode=DeinterlaceFilter.DEFAULT_MODE):
deinterlaceFilter = DeinterlaceFilter(mode=mode)
payload, = deinterlaceFilter.getYield()
return payload
def test_default_mode_yields_only_bwdif(self):
payload = self.getPayload()
self.assertEqual(['bwdif=mode=1'], payload['tokens'])
def test_up_down_mode_yields_lanczos_scales_after_bwdif(self):
payload = self.getPayload('up_down')
self.assertEqual(
[
'bwdif=mode=1',
'scale=iw*2:ih*2:flags=lanczos',
'scale=iw/2:ih/2:flags=lanczos',
],
payload['tokens'],
)
def test_temp_mode_yields_hqdn3d_after_bwdif(self):
payload = self.getPayload('temp')
self.assertEqual(
['bwdif=mode=1', 'hqdn3d=1.5:1.5:3:3'],
payload['tokens'],
)
def test_up_down_temp_mode_yields_combined_filters_after_bwdif(self):
payload = self.getPayload('up_down_temp')
self.assertEqual(
[
'bwdif=mode=1',
'scale=iw*2:ih*2:flags=lanczos',
'scale=iw/2:ih/2:flags=lanczos',
'hqdn3d=1.5:1.5:3:3',
],
payload['tokens'],
)
def test_explicit_scale_factors_and_filter_order_are_preserved(self):
payload = self.getPayload('up4_temp3366_down4')
self.assertEqual(
[
'bwdif=mode=1',
'scale=iw*4:ih*4:flags=lanczos',
'hqdn3d=1.5:1.5:3:3',
'scale=iw/4:ih/4:flags=lanczos',
],
payload['tokens'],
)
def test_temporal_digits_are_divided_by_two(self):
payload = self.getPayload('temp0044_up2_down2')
self.assertEqual(
[
'bwdif=mode=1',
'hqdn3d=0:0:2:2',
'scale=iw*2:ih*2:flags=lanczos',
'scale=iw/2:ih/2:flags=lanczos',
],
payload['tokens'],
)
def test_temporal_parameters_accept_hexadecimal_digits(self):
lowerPayload = self.getPayload('temp00ff')
mixedPayload = self.getPayload('tempFf33')
self.assertEqual(
['bwdif=mode=1', 'hqdn3d=0:0:7.5:7.5'],
lowerPayload['tokens'],
)
self.assertEqual(
['bwdif=mode=1', 'hqdn3d=7.5:7.5:1.5:1.5'],
mixedPayload['tokens'],
)
def test_scale_factors_accept_boundaries_from_two_to_eight(self):
payload = self.getPayload('up8_down2')
self.assertEqual(
[
'bwdif=mode=1',
'scale=iw*8:ih*8:flags=lanczos',
'scale=iw/2:ih/2:flags=lanczos',
],
payload['tokens'],
)
def test_each_component_uses_its_default_when_parameters_are_omitted(self):
payload = self.getPayload('up_temp_down')
self.assertEqual(
[
'bwdif=mode=1',
'scale=iw*2:ih*2:flags=lanczos',
'hqdn3d=1.5:1.5:3:3',
'scale=iw/2:ih/2:flags=lanczos',
],
payload['tokens'],
)
def test_unknown_mode_is_rejected(self):
with self.assertRaisesRegex(
ValueError,
"Unknown deinterlace component: 'other'",
):
DeinterlaceFilter(mode='other')
def test_invalid_parameters_are_rejected(self):
invalidModes = ('up1', 'down9', 'temp123', 'temp00gg', 'up__down')
for mode in invalidModes:
with self.subTest(mode=mode), self.assertRaises(ValueError):
DeinterlaceFilter(mode=mode)
def test_cli_accepts_named_modes_and_preserves_bare_flag_behavior(self):
upDownContext = convert.make_context(
'convert',
['--deinterlace', 'up_down'],
)
tempContext = convert.make_context(
'convert',
['--deinterlace', 'temp'],
)
combinedContext = convert.make_context(
'convert',
['--deinterlace', 'up_down_temp'],
)
complexContext = convert.make_context(
'convert',
['--deinterlace', 'up4_temp3366_down4'],
)
bareFlagContext = convert.make_context('convert', ['--deinterlace'])
omittedContext = convert.make_context('convert', [])
self.assertEqual('up_down', upDownContext.params['deinterlace'])
self.assertEqual('temp', tempContext.params['deinterlace'])
self.assertEqual('up_down_temp', combinedContext.params['deinterlace'])
self.assertEqual(
'up4_temp3366_down4',
complexContext.params['deinterlace'],
)
self.assertEqual('default', bareFlagContext.params['deinterlace'])
self.assertEqual('none', omittedContext.params['deinterlace'])
def test_cli_rejects_an_invalid_specification(self):
with self.assertRaises(click.BadParameter):
convert.make_context('convert', ['--deinterlace', 'temp123'])
if __name__ == '__main__':
unittest.main()

View File

@@ -161,6 +161,55 @@ class FfxControllerTests(unittest.TestCase):
any(token.startswith("ENCODING_PRESET=") for token in commands[1])
)
def test_vp9_run_job_applies_identical_video_filters_to_both_passes(self):
context = self.make_context(VideoEncoder.VP9)
target_descriptor, source_descriptor = self.make_media_descriptors()
controller = FfxController(context, target_descriptor, source_descriptor)
commands = []
with (
patch.object(
controller,
"executeCommandSequence",
side_effect=lambda command: commands.append(command) or ("", "", 0),
),
patch("ffx.ffx_controller.os.path.exists", return_value=False),
):
controller.runJob(
"input.mkv",
"output.webm",
targetFormat="webm",
chainIteration=[
{
"identifier": "quality",
"parameters": {"quality": 27},
},
{
"identifier": "nlmeans",
"parameters": {},
"tokens": ["nlmeans=s=2.8"],
},
{
"identifier": "bwdif",
"parameters": {},
"tokens": ["bwdif=mode=1"],
},
],
cropArguments={
"output_width": 704,
"output_height": 576,
"x_offset": 8,
"y_offset": 0,
},
)
self.assertEqual(2, len(commands))
expectedFilterChain = (
"crop=704:576:8:0, nlmeans=s=2.8, bwdif=mode=1"
)
self.assert_token_pair(commands[0], "-vf", expectedFilterChain)
self.assert_token_pair(commands[1], "-vf", expectedFilterChain)
def test_av1_run_job_emits_file_level_quality_and_preset_metadata(self):
context = self.make_context(VideoEncoder.AV1)
target_descriptor, source_descriptor = self.make_media_descriptors()

View File

@@ -0,0 +1,146 @@
from __future__ import annotations
from pathlib import Path
import sys
import unittest
SRC_ROOT = Path(__file__).resolve().parents[2] / "src"
if str(SRC_ROOT) not in sys.path:
sys.path.insert(0, str(SRC_ROOT))
from ffx.attachment_format import AttachmentFormat # noqa: E402
from ffx.media_descriptor import MediaDescriptor # noqa: E402
from ffx.track_codec import TrackCodec # noqa: E402
from ffx.track_descriptor import TrackDescriptor # noqa: E402
from ffx.track_type import TrackType # noqa: E402
class MediaDescriptorMappingTests(unittest.TestCase):
def make_target_descriptor(self) -> MediaDescriptor:
return MediaDescriptor(
track_descriptors=[
TrackDescriptor(
index=0,
source_index=0,
sub_index=0,
track_type=TrackType.VIDEO,
codec_name=TrackCodec.H264,
),
TrackDescriptor(
index=1,
source_index=7,
sub_index=0,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.ASS,
),
TrackDescriptor(
index=2,
source_index=8,
sub_index=1,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.ASS,
),
TrackDescriptor(
index=3,
source_index=9,
sub_index=0,
track_type=TrackType.ATTACHMENT,
attachment_format=AttachmentFormat.TTF,
),
]
)
def make_source_descriptor(self) -> MediaDescriptor:
return MediaDescriptor(
track_descriptors=[
TrackDescriptor(
index=0,
source_index=0,
sub_index=0,
track_type=TrackType.VIDEO,
codec_name=TrackCodec.H264,
),
TrackDescriptor(
index=3,
source_index=3,
sub_index=0,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.VOBSUB,
),
TrackDescriptor(
index=4,
source_index=4,
sub_index=1,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.VOBSUB,
),
TrackDescriptor(
index=5,
source_index=5,
sub_index=2,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.PGS,
),
TrackDescriptor(
index=6,
source_index=6,
sub_index=3,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.PGS,
),
TrackDescriptor(
index=7,
source_index=7,
sub_index=4,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.ASS,
),
TrackDescriptor(
index=8,
source_index=8,
sub_index=5,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.ASS,
),
TrackDescriptor(
index=9,
source_index=9,
sub_index=0,
track_type=TrackType.ATTACHMENT,
attachment_format=AttachmentFormat.TTF,
),
]
)
def test_source_aware_mapping_uses_original_subtitle_and_attachment_indices(self):
tokens = self.make_target_descriptor().getInputMappingTokens(
sourceMediaDescriptor=self.make_source_descriptor()
)
self.assertEqual(
[
"-map",
"0:v:0",
"-map",
"0:s:4",
"-map",
"0:s:5",
"-map",
"0:9",
],
tokens,
)
def test_video_only_mapping_does_not_map_font_attachments(self):
tokens = self.make_target_descriptor().getInputMappingTokens(
only_video=True
)
self.assertEqual(["-map", "0:v:0"], tokens)
if __name__ == "__main__":
unittest.main()

View File

@@ -21,6 +21,7 @@ from ffx.file_properties import FileProperties # noqa: E402
from ffx.model.pattern import Pattern # noqa: E402
from ffx.pattern_controller import ( # noqa: E402
DuplicatePatternMatchError,
InvalidPatternCaptureGroupError,
InvalidPatternSchemaError,
PatternController,
)
@@ -65,6 +66,7 @@ def make_track_descriptor(
track_type: TrackType = TrackType.VIDEO,
title: str = "",
dispositions: set[TrackDisposition] | None = None,
dropped: bool = False,
) -> TrackDescriptor:
kwargs = {
TrackDescriptor.INDEX_KEY: index,
@@ -72,6 +74,7 @@ def make_track_descriptor(
TrackDescriptor.TRACK_TYPE_KEY: track_type,
TrackDescriptor.TAGS_KEY: {"title": title} if title else {},
TrackDescriptor.DISPOSITION_SET_KEY: dispositions or set(),
TrackDescriptor.DROPPED_KEY: dropped,
}
if track_type == TrackType.AUDIO:
kwargs[TrackDescriptor.AUDIO_LAYOUT_KEY] = AudioLayout.LAYOUT_STEREO
@@ -142,7 +145,7 @@ class PatternManagementTests(unittest.TestCase):
def test_match_filename_raises_for_duplicate_matches_in_same_show(self):
self.save_pattern(1, r"^same_(s[0-9]+e[0-9]+)\.mkv$")
self.save_pattern(1, r"^same_.*$")
self.save_pattern(1, r"^same_(.*)$")
with self.assertRaises(DuplicatePatternMatchError) as caught:
self.pattern_controller.matchFilename("same_s01e01.mkv")
@@ -152,7 +155,7 @@ class PatternManagementTests(unittest.TestCase):
def test_match_filename_raises_for_duplicate_matches_across_shows(self):
self.save_pattern(1, r"^cross_(s[0-9]+e[0-9]+)\.mkv$")
self.save_pattern(2, r"^cross_.*$")
self.save_pattern(2, r"^cross_(.*)$")
with self.assertRaises(DuplicatePatternMatchError) as caught:
self.pattern_controller.matchFilename("cross_s01e01.mkv")
@@ -179,6 +182,48 @@ class PatternManagementTests(unittest.TestCase):
match = self.pattern_controller.matchFilename("after_s01e01.mkv")
self.assertEqual(pattern_id, match["pattern"].getId())
def test_save_pattern_schema_rejects_pattern_without_capture_group(self):
self.add_show(1, "No Capture Pattern Show")
with self.assertRaises(InvalidPatternCaptureGroupError) as caught:
self.pattern_controller.savePatternSchema(
{
"show_id": 1,
"pattern": r"^nocapture_s[0-9]+e[0-9]+\.mkv$",
},
trackDescriptors=[make_track_descriptor(0)],
)
self.assertIn("capture group", str(caught.exception))
def test_update_pattern_rejects_pattern_without_capture_group(self):
pattern_id = self.save_pattern(1, r"^valid_(s[0-9]+e[0-9]+)\.mkv$")
with self.assertRaises(InvalidPatternCaptureGroupError):
self.pattern_controller.updatePattern(
pattern_id,
{
"show_id": 1,
"pattern": r"^invalid_s[0-9]+e[0-9]+\.mkv$",
"quality": 0,
"notes": "",
},
)
match = self.pattern_controller.matchFilename("valid_s01e01.mkv")
self.assertEqual(pattern_id, match["pattern"].getId())
def test_match_filename_raises_specific_error_for_legacy_pattern_without_capture_group(self):
pattern_id = self.insert_trackless_pattern_row(
1,
r"^legacy_s[0-9]+e[0-9]+\.mkv$",
)
with self.assertRaises(InvalidPatternCaptureGroupError) as caught:
self.pattern_controller.matchFilename("legacy_s01e01.mkv")
self.assertEqual(pattern_id, caught.exception.pattern_id)
def test_save_pattern_schema_rejects_zero_track_patterns(self):
self.add_show(1, "Empty Pattern Show")
@@ -193,6 +238,47 @@ class PatternManagementTests(unittest.TestCase):
self.assertIn("at least one track", str(caught.exception))
def test_save_pattern_schema_rejects_all_dropped_patterns(self):
self.add_show(1, "Dropped Pattern Show")
with self.assertRaises(InvalidPatternSchemaError) as caught:
self.pattern_controller.savePatternSchema(
{
"show_id": 1,
"pattern": r"^dropped_(s[0-9]+e[0-9]+)\.mkv$",
},
trackDescriptors=[make_track_descriptor(0, dropped=True)],
)
self.assertIn("non-dropped track", str(caught.exception))
def test_dropped_pattern_track_is_not_in_conversion_media_descriptor(self):
pattern_id = self.save_pattern(
1,
r"^active_(s[0-9]+e[0-9]+)\.mkv$",
tracks=[
make_track_descriptor(0, track_type=TrackType.VIDEO),
make_track_descriptor(1, track_type=TrackType.AUDIO, dropped=True),
make_track_descriptor(2, track_type=TrackType.SUBTITLE),
],
)
Session = self.context["database"]["session"]
session = Session()
try:
pattern = session.query(Pattern).filter(Pattern.id == pattern_id).first()
media_descriptor = pattern.getMediaDescriptor(self.context)
tracks = media_descriptor.getTrackDescriptors()
finally:
session.close()
self.assertEqual(
[TrackType.VIDEO, TrackType.SUBTITLE],
[track.getType() for track in tracks],
)
self.assertEqual([0, 1], [track.getIndex() for track in tracks])
self.assertEqual([0, 2], [track.getSourceIndex() for track in tracks])
def test_save_pattern_schema_does_not_persist_attachment_tracks(self):
pattern_id = self.save_pattern(
1,
@@ -233,7 +319,7 @@ class PatternManagementTests(unittest.TestCase):
def test_file_properties_skips_pattern_matching_when_disabled(self):
self.save_pattern(1, r"^nopattern_(s[0-9]+e[0-9]+)\.mkv$")
self.save_pattern(2, r"^nopattern_.*$")
self.save_pattern(2, r"^nopattern_(.*)$")
no_pattern_context = dict(self.context)
no_pattern_context["use_pattern"] = False
@@ -257,6 +343,17 @@ class PatternManagementTests(unittest.TestCase):
self.assertIn("last track", str(caught.exception))
def test_track_controller_refuses_to_drop_last_active_track(self):
pattern_id = self.save_pattern(1, r"^drop_(s[0-9]+e[0-9]+)\.mkv$")
track = self.track_controller.getTrack(pattern_id, 0)
descriptor = track.getDescriptor()
descriptor.setDropped(True)
with self.assertRaises(click.ClickException) as caught:
self.track_controller.updateTrack(track.getId(), descriptor)
self.assertIn("last active track", str(caught.exception))
def test_exact_duplicate_pattern_definition_is_rejected(self):
self.save_pattern(1, r"^unique_(s[0-9]+e[0-9]+)\.mkv$")

View File

@@ -76,6 +76,12 @@ class FakeTagTable:
def select_row(self, row_key):
self._selected_row_key = row_key
def move_cursor(self, *, row=None, column=None, animate=False, scroll=True):
if row is not None and self._row_order:
self._selected_row_key = self._row_order[int(row)]
if column is not None:
self.cursor_coordinate = (self.cursor_coordinate[0], int(column))
def get_row_index(self, row_key):
return self._row_order.index(row_key)
@@ -110,6 +116,12 @@ class FakeValueWidget:
self.disabled = False
class FakeButtonWidget:
def __init__(self):
self.disabled = False
self.variant = "default"
class FakeInputWidget:
def __init__(self, value):
self.value = value
@@ -126,18 +138,20 @@ class FakeStaticWidget:
class FakeSelectionListWidget:
def __init__(self, selected):
self.selected = selected
self.disabled = False
def add_option(self, _option):
return None
def make_track_descriptor(index, sub_index, track_type):
def make_track_descriptor(index, sub_index, track_type, *, dropped=False):
return TrackDescriptor(
index=index,
sub_index=sub_index,
track_type=track_type,
codec_name=TrackCodec.UNKNOWN,
audio_layout=AudioLayout.LAYOUT_UNDEFINED,
dropped=dropped,
)
@@ -188,6 +202,7 @@ class TagTableScreenStateTests(unittest.TestCase):
widgets = {
"#type_select": FakeValueWidget(TrackType.AUDIO),
"#audio_layout_select": FakeValueWidget(AudioLayout.LAYOUT_STEREO),
"#drop_checkbox": FakeValueWidget(True),
"#language_select": FakeValueWidget(IsoLanguage.GERMAN),
"#title_input": FakeInputWidget("German Audio"),
"#dispositions_selection_list": FakeSelectionListWidget({0, 6}),
@@ -201,6 +216,7 @@ class TagTableScreenStateTests(unittest.TestCase):
self.assertEqual("deu", descriptor.getTags()["language"])
self.assertEqual("German Audio", descriptor.getTitle())
self.assertEqual("value", descriptor.getTags()["KEEP"])
self.assertTrue(descriptor.isDropped())
def test_track_details_screen_preserves_attachment_format_for_attachment_tracks(self):
screen = object.__new__(TrackDetailsScreen)
@@ -216,6 +232,7 @@ class TagTableScreenStateTests(unittest.TestCase):
widgets = {
"#type_select": FakeValueWidget(TrackType.ATTACHMENT),
"#audio_layout_select": FakeValueWidget(AudioLayout.LAYOUT_UNDEFINED),
"#drop_checkbox": FakeValueWidget(False),
"#language_select": FakeValueWidget(Select.NULL),
"#title_input": FakeInputWidget(""),
"#dispositions_selection_list": FakeSelectionListWidget(set()),
@@ -307,6 +324,7 @@ class TagTableScreenStateTests(unittest.TestCase):
audio_layout=AudioLayout.LAYOUT_STEREO,
tags={"language": "ger"},
)
screen._TrackDetailsScreen__dropped = False
screen._TrackDetailsScreen__metadataOnly = True
screen._TrackDetailsScreen__titleAutoManaged = True
screen._TrackDetailsScreen__suppressTitleChanged = False
@@ -323,6 +341,7 @@ class TagTableScreenStateTests(unittest.TestCase):
"#pattern_label": FakeStaticWidget(),
"#type_select": FakeValueWidget(None),
"#audio_layout_select": FakeValueWidget(None),
"#drop_checkbox": FakeValueWidget(False),
"#dispositions_selection_list": FakeSelectionListWidget(set()),
"#language_select": FakeValueWidget(None),
"#title_input": FakeInputWidget(""),
@@ -355,7 +374,7 @@ class TagTableScreenStateTests(unittest.TestCase):
def test_pattern_details_screen_reads_selected_track_from_row_mapping(self):
first_track = make_track_descriptor(0, 0, TrackType.VIDEO)
second_track = make_track_descriptor(1, 0, TrackType.SUBTITLE)
second_track = make_track_descriptor(1, 0, TrackType.SUBTITLE, dropped=True)
screen = object.__new__(PatternDetailsScreen)
screen.tracksTable = FakeTagTable()
@@ -367,6 +386,101 @@ class TagTableScreenStateTests(unittest.TestCase):
screen.tracksTable.select_row("row-1")
self.assertIs(second_track, screen.getSelectedTrackDescriptor())
self.assertEqual("Ja", screen.tracksTable.rows["row-1"][9])
def test_pattern_details_screen_toggles_draft_track_drop_state(self):
first_track = make_track_descriptor(0, 0, TrackType.VIDEO)
second_track = make_track_descriptor(1, 0, TrackType.SUBTITLE)
drop_button = FakeButtonWidget()
screen = object.__new__(PatternDetailsScreen)
screen.context = {"logger": get_ffx_logger()}
screen.tracksTable = FakeTagTable()
screen._PatternDetailsScreen__draftTracks = [first_track, second_track]
screen._PatternDetailsScreen__pattern = None
screen._PatternDetailsScreen__trackRowData = {}
screen.query_one = lambda selector, _widget_type=None: drop_button
screen.updateTracks()
screen.tracksTable.select_row("row-1")
screen.toggleSelectedTrackDropped(screen.getSelectedTrackDescriptor())
self.assertFalse(first_track.isDropped())
self.assertTrue(second_track.isDropped())
self.assertEqual("Ja", list(screen.tracksTable.rows.values())[1][9])
self.assertIs(second_track, screen.getSelectedTrackDescriptor())
self.assertEqual("warning", drop_button.variant)
def test_pattern_details_screen_updates_drop_button_for_selected_track(self):
first_track = make_track_descriptor(0, 0, TrackType.VIDEO)
second_track = make_track_descriptor(
1,
0,
TrackType.SUBTITLE,
dropped=True,
)
drop_button = FakeButtonWidget()
screen = object.__new__(PatternDetailsScreen)
screen.tracksTable = FakeTagTable()
screen._PatternDetailsScreen__draftTracks = [first_track, second_track]
screen._PatternDetailsScreen__pattern = None
screen._PatternDetailsScreen__trackRowData = {}
screen.query_one = lambda selector, _widget_type=None: drop_button
screen.updateTracks()
self.assertEqual("default", drop_button.variant)
screen.tracksTable.select_row("row-1")
screen.updateTrackDropButtonState()
self.assertEqual("warning", drop_button.variant)
screen.tracksTable.select_row("row-0")
screen.updateTrackDropButtonState()
self.assertEqual("default", drop_button.variant)
def test_pattern_details_screen_toggles_persisted_track_drop_state(self):
class _FakePattern:
def getId(self):
return 5
class _FakeTrackController:
def __init__(self):
self.updated = []
def updateTrack(self, track_id, track_descriptor):
self.updated.append((track_id, track_descriptor))
return True
track = TrackDescriptor(
id=17,
pattern_id=5,
index=1,
source_index=1,
sub_index=0,
track_type=TrackType.SUBTITLE,
codec_name=TrackCodec.UNKNOWN,
dropped=False,
)
track_controller = _FakeTrackController()
screen = object.__new__(PatternDetailsScreen)
screen.context = {"logger": get_ffx_logger()}
screen._PatternDetailsScreen__pattern = _FakePattern()
screen._PatternDetailsScreen__tc = track_controller
screen.updateTracks = lambda: None
screen.toggleSelectedTrackDropped(track)
self.assertEqual(1, len(track_controller.updated))
track_id, updated_track = track_controller.updated[0]
self.assertEqual(17, track_id)
self.assertFalse(track.isDropped())
self.assertTrue(updated_track.isDropped())
def test_pattern_details_screen_reads_selected_tag_from_raw_row_mapping(self):
screen = object.__new__(PatternDetailsScreen)