Skip to content

base

Classes:

DVDExtIndexer 

DVDExtIndexer(
    *,
    bin_path: SPathLike | MissingT = MISSING,
    ext: str | MissingT = MISSING,
    force: bool = True,
    default_out_folder: SPathLike | Literal[False] | None = None,
    **kwargs: Any
)

Bases: ExternalIndexer, DVDIndexer

Methods:

Attributes:

Source code 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def __init__(
    self, *, bin_path: SPathLike | MissingT = MISSING, ext: str | MissingT = MISSING,
    force: bool = True, default_out_folder: SPathLike | Literal[False] | None = None,
    **kwargs: Any
) -> None:
    super().__init__(force=force, **kwargs)

    if bin_path is MISSING:
        bin_path = self._bin_path

    if ext is MISSING:
        ext = self._ext

    self.bin_path = SPath(bin_path)
    self.ext = ext
    self.default_out_folder = default_out_folder

bin_path instance-attribute 

bin_path = SPath(bin_path)

default_out_folder instance-attribute 

default_out_folder = default_out_folder

ext instance-attribute 

ext = ext

force instance-attribute 

force = force

indexer_kwargs instance-attribute 

indexer_kwargs = kwargs

iso_path instance-attribute 

iso_path: SPath

file_corrupted 

file_corrupted(index_path: SPath) -> None
Source code 
202
203
204
205
206
207
208
209
def file_corrupted(self, index_path: SPath) -> None:
    if self.force:
        try:
            index_path.unlink()
        except OSError:
            raise CustomRuntimeError("Index file corrupted, tried to delete it and failed.", self.__class__)
    else:
        raise CustomRuntimeError("Index file corrupted! Delete it and retry.", self.__class__)

get_cmd abstractmethod 

get_cmd(files: list[SPath], output: SPath) -> list[str]

Returns the indexer command

Source code 
142
143
144
145
@abstractmethod
def get_cmd(self, files: list[SPath], output: SPath) -> list[str]:
    """Returns the indexer command"""
    raise NotImplementedError

get_idx_file_path 

get_idx_file_path(path: SPath) -> SPath
Source code 
199
200
def get_idx_file_path(self, path: SPath) -> SPath:
    return path.with_suffix(f'.{self.ext}')

get_info abstractmethod 

get_info(index_path: SPath, file_idx: int = 0) -> IndexFileType

Returns info about the indexing file

Source code 
147
148
149
150
@abstractmethod
def get_info(self, index_path: SPath, file_idx: int = 0) -> IndexFileType:
    """Returns info about the indexing file"""
    raise NotImplementedError

get_joined_names classmethod 

get_joined_names(files: list[SPath]) -> str
Source code 
54
55
56
@classmethod
def get_joined_names(cls, files: list[SPath]) -> str:
    return '_'.join([file.name for file in files])

get_out_folder 

get_out_folder(
    output_folder: SPathLike | Literal[False] | None = None,
    file: SPath | None = None,
) -> SPath
Source code 
188
189
190
191
192
193
194
195
196
197
def get_out_folder(
    self, output_folder: SPathLike | Literal[False] | None = None, file: SPath | None = None
) -> SPath:
    if output_folder is None:
        return SPath(file).get_folder() if file else self.get_out_folder(False)

    if not output_folder:
        return SPath(tempfile.gettempdir())

    return SPath(output_folder)

get_video_idx_path 

get_video_idx_path(
    folder: SPath, file_hash: str, video_name: SPathLike
) -> SPath
Source code 
253
254
255
256
257
258
def get_video_idx_path(self, folder: SPath, file_hash: str, video_name: SPathLike) -> SPath:
    vid_name = SPath(video_name).stem
    current_indxer = os.path.basename(self._bin_path)
    filename = '_'.join([file_hash, vid_name, current_indxer])

    return self.get_idx_file_path(PackageStorage(folder).get_file(filename))

get_videos_hash classmethod 

get_videos_hash(files: list[SPath]) -> str
Source code 
58
59
60
61
62
@classmethod
def get_videos_hash(cls, files: list[SPath]) -> str:
    length = sum(file.stat().st_size for file in files)
    to_hash = length.to_bytes(32, 'little') + cls.get_joined_names(files).encode()
    return md5(to_hash).hexdigest()

index 

index(
    files: Sequence[SPath],
    force: bool = False,
    split_files: bool = False,
    output_folder: SPathLike | Literal[False] | None = None,
    *cmd_args: str
) -> list[SPath]
Source code 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def index(
    self, files: Sequence[SPath], force: bool = False, split_files: bool = False,
    output_folder: SPathLike | Literal[False] | None = None, *cmd_args: str
) -> list[SPath]:

    if len(unique_folders := list(set([f.get_folder().to_str() for f in files]))) > 1:
        return [
            c for s in (
                self.index(
                    [f for f in files if f.get_folder().to_str() == folder],
                    force, split_files, output_folder
                )
                for folder in unique_folders
            ) for c in s
        ]

    dest_folder = self.get_out_folder(output_folder, files[0])

    files = list(sorted(set(files)))

    hash_str = self.get_videos_hash(files)

    def _index(files: list[SPath], output: SPath) -> None:
        if output.is_file():
            if output.stat().st_size == 0 or force:
                output.unlink()
            else:
                return self.update_video_filenames(output, files)
        return self._run_index(files, output, cmd_args)

    if not split_files:
        output = self.get_video_idx_path(dest_folder, hash_str, 'JOINED' if len(files) > 1 else 'SINGLE')
        _index(files, output)
        return [output]

    outputs = [self.get_video_idx_path(dest_folder, hash_str, file.name) for file in files]

    for file, output in zip(files, outputs):
        _index([file], output)

    return outputs

normalize_filenames classmethod 

normalize_filenames(file: SPathLike | Sequence[SPathLike]) -> list[SPath]
Source code 
68
69
70
71
72
73
74
75
76
77
78
@classmethod
def normalize_filenames(cls, file: SPathLike | Sequence[SPathLike]) -> list[SPath]:
    files = list[SPath]()

    for f in to_arr(file):
        if str(f).startswith('file:///'):
            f = str(f)[8::]

        files.append(SPath(f))

    return files

parse_vts 

parse_vts(
    title: IFO0Title,
    disable_rff: bool,
    vobidcellids_to_take: list[tuple[int, int]],
    target_vts: IFOX,
    output_folder: SPath,
    vob_input_files: Sequence[SPath],
) -> tuple[VideoNode, list[int], list[tuple[int, int]], list[int]]
Source code 
283
284
285
286
287
def parse_vts(
    self, title: IFO0Title, disable_rff: bool, vobidcellids_to_take: list[tuple[int, int]],
    target_vts: IFOX, output_folder: SPath, vob_input_files: Sequence[SPath]
) -> tuple[vs.VideoNode, list[int], list[tuple[int, int]], list[int]]:
    raise NotImplementedError

source 

source(
    file: SPathLike | Sequence[SPathLike],
    bits: int | None = None,
    *,
    matrix: MatrixT | None = None,
    transfer: TransferT | None = None,
    primaries: PrimariesT | None = None,
    chroma_location: ChromaLocationT | None = None,
    color_range: ColorRangeT | None = None,
    field_based: FieldBasedT | None = None,
    **kwargs: Any
) -> VideoNode
Source code 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
@inject_self
def source(  # type: ignore
    self, file: SPathLike | Sequence[SPathLike],
    bits: int | None = None, *,
    matrix: MatrixT | None = None,
    transfer: TransferT | None = None,
    primaries: PrimariesT | None = None,
    chroma_location: ChromaLocationT | None = None,
    color_range: ColorRangeT | None = None,
    field_based: FieldBasedT | None = None,
    **kwargs: Any
) -> vs.VideoNode:
    index_files = self.index(self.normalize_filenames(file))

    return self._source(
        (self.source_func(idx_filename.to_str(), **kwargs) for idx_filename in index_files),
        bits, matrix, transfer, primaries, chroma_location, color_range, field_based
    )

source_func classmethod 

source_func(path: DataType | SPathLike, *args: Any, **kwargs: Any) -> VideoNode
Source code 
64
65
66
@classmethod
def source_func(cls, path: DataType | SPathLike, *args: Any, **kwargs: Any) -> vs.VideoNode:
    return cls._source_func(str(path), *args, **kwargs)

update_video_filenames abstractmethod 

update_video_filenames(index_path: SPath, filepaths: list[SPath]) -> None
Source code 
152
153
154
@abstractmethod
def update_video_filenames(self, index_path: SPath, filepaths: list[SPath]) -> None:
    raise NotImplementedError

DVDIndexer

Methods:

Attributes:

iso_path instance-attribute 

iso_path: SPath

parse_vts 

parse_vts(
    title: IFO0Title,
    disable_rff: bool,
    vobidcellids_to_take: list[tuple[int, int]],
    target_vts: IFOX,
    output_folder: SPath,
    vob_input_files: Sequence[SPath],
) -> tuple[VideoNode, list[int], list[tuple[int, int]], list[int]]
Source code 
283
284
285
286
287
def parse_vts(
    self, title: IFO0Title, disable_rff: bool, vobidcellids_to_take: list[tuple[int, int]],
    target_vts: IFOX, output_folder: SPath, vob_input_files: Sequence[SPath]
) -> tuple[vs.VideoNode, list[int], list[tuple[int, int]], list[int]]:
    raise NotImplementedError

ExternalIndexer 

ExternalIndexer(
    *,
    bin_path: SPathLike | MissingT = MISSING,
    ext: str | MissingT = MISSING,
    force: bool = True,
    default_out_folder: SPathLike | Literal[False] | None = None,
    **kwargs: Any
)

Bases: Indexer

Methods:

Attributes:

Source code 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def __init__(
    self, *, bin_path: SPathLike | MissingT = MISSING, ext: str | MissingT = MISSING,
    force: bool = True, default_out_folder: SPathLike | Literal[False] | None = None,
    **kwargs: Any
) -> None:
    super().__init__(force=force, **kwargs)

    if bin_path is MISSING:
        bin_path = self._bin_path

    if ext is MISSING:
        ext = self._ext

    self.bin_path = SPath(bin_path)
    self.ext = ext
    self.default_out_folder = default_out_folder

bin_path instance-attribute 

bin_path = SPath(bin_path)

default_out_folder instance-attribute 

default_out_folder = default_out_folder

ext instance-attribute 

ext = ext

force instance-attribute 

force = force

indexer_kwargs instance-attribute 

indexer_kwargs = kwargs

file_corrupted 

file_corrupted(index_path: SPath) -> None
Source code 
202
203
204
205
206
207
208
209
def file_corrupted(self, index_path: SPath) -> None:
    if self.force:
        try:
            index_path.unlink()
        except OSError:
            raise CustomRuntimeError("Index file corrupted, tried to delete it and failed.", self.__class__)
    else:
        raise CustomRuntimeError("Index file corrupted! Delete it and retry.", self.__class__)

get_cmd abstractmethod 

get_cmd(files: list[SPath], output: SPath) -> list[str]

Returns the indexer command

Source code 
142
143
144
145
@abstractmethod
def get_cmd(self, files: list[SPath], output: SPath) -> list[str]:
    """Returns the indexer command"""
    raise NotImplementedError

get_idx_file_path 

get_idx_file_path(path: SPath) -> SPath
Source code 
199
200
def get_idx_file_path(self, path: SPath) -> SPath:
    return path.with_suffix(f'.{self.ext}')

get_info abstractmethod 

get_info(index_path: SPath, file_idx: int = 0) -> IndexFileType

Returns info about the indexing file

Source code 
147
148
149
150
@abstractmethod
def get_info(self, index_path: SPath, file_idx: int = 0) -> IndexFileType:
    """Returns info about the indexing file"""
    raise NotImplementedError

get_joined_names classmethod 

get_joined_names(files: list[SPath]) -> str
Source code 
54
55
56
@classmethod
def get_joined_names(cls, files: list[SPath]) -> str:
    return '_'.join([file.name for file in files])

get_out_folder 

get_out_folder(
    output_folder: SPathLike | Literal[False] | None = None,
    file: SPath | None = None,
) -> SPath
Source code 
188
189
190
191
192
193
194
195
196
197
def get_out_folder(
    self, output_folder: SPathLike | Literal[False] | None = None, file: SPath | None = None
) -> SPath:
    if output_folder is None:
        return SPath(file).get_folder() if file else self.get_out_folder(False)

    if not output_folder:
        return SPath(tempfile.gettempdir())

    return SPath(output_folder)

get_video_idx_path 

get_video_idx_path(
    folder: SPath, file_hash: str, video_name: SPathLike
) -> SPath
Source code 
253
254
255
256
257
258
def get_video_idx_path(self, folder: SPath, file_hash: str, video_name: SPathLike) -> SPath:
    vid_name = SPath(video_name).stem
    current_indxer = os.path.basename(self._bin_path)
    filename = '_'.join([file_hash, vid_name, current_indxer])

    return self.get_idx_file_path(PackageStorage(folder).get_file(filename))

get_videos_hash classmethod 

get_videos_hash(files: list[SPath]) -> str
Source code 
58
59
60
61
62
@classmethod
def get_videos_hash(cls, files: list[SPath]) -> str:
    length = sum(file.stat().st_size for file in files)
    to_hash = length.to_bytes(32, 'little') + cls.get_joined_names(files).encode()
    return md5(to_hash).hexdigest()

index 

index(
    files: Sequence[SPath],
    force: bool = False,
    split_files: bool = False,
    output_folder: SPathLike | Literal[False] | None = None,
    *cmd_args: str
) -> list[SPath]
Source code 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def index(
    self, files: Sequence[SPath], force: bool = False, split_files: bool = False,
    output_folder: SPathLike | Literal[False] | None = None, *cmd_args: str
) -> list[SPath]:

    if len(unique_folders := list(set([f.get_folder().to_str() for f in files]))) > 1:
        return [
            c for s in (
                self.index(
                    [f for f in files if f.get_folder().to_str() == folder],
                    force, split_files, output_folder
                )
                for folder in unique_folders
            ) for c in s
        ]

    dest_folder = self.get_out_folder(output_folder, files[0])

    files = list(sorted(set(files)))

    hash_str = self.get_videos_hash(files)

    def _index(files: list[SPath], output: SPath) -> None:
        if output.is_file():
            if output.stat().st_size == 0 or force:
                output.unlink()
            else:
                return self.update_video_filenames(output, files)
        return self._run_index(files, output, cmd_args)

    if not split_files:
        output = self.get_video_idx_path(dest_folder, hash_str, 'JOINED' if len(files) > 1 else 'SINGLE')
        _index(files, output)
        return [output]

    outputs = [self.get_video_idx_path(dest_folder, hash_str, file.name) for file in files]

    for file, output in zip(files, outputs):
        _index([file], output)

    return outputs

normalize_filenames classmethod 

normalize_filenames(file: SPathLike | Sequence[SPathLike]) -> list[SPath]
Source code 
68
69
70
71
72
73
74
75
76
77
78
@classmethod
def normalize_filenames(cls, file: SPathLike | Sequence[SPathLike]) -> list[SPath]:
    files = list[SPath]()

    for f in to_arr(file):
        if str(f).startswith('file:///'):
            f = str(f)[8::]

        files.append(SPath(f))

    return files

source 

source(
    file: SPathLike | Sequence[SPathLike],
    bits: int | None = None,
    *,
    matrix: MatrixT | None = None,
    transfer: TransferT | None = None,
    primaries: PrimariesT | None = None,
    chroma_location: ChromaLocationT | None = None,
    color_range: ColorRangeT | None = None,
    field_based: FieldBasedT | None = None,
    **kwargs: Any
) -> VideoNode
Source code 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
@inject_self
def source(  # type: ignore
    self, file: SPathLike | Sequence[SPathLike],
    bits: int | None = None, *,
    matrix: MatrixT | None = None,
    transfer: TransferT | None = None,
    primaries: PrimariesT | None = None,
    chroma_location: ChromaLocationT | None = None,
    color_range: ColorRangeT | None = None,
    field_based: FieldBasedT | None = None,
    **kwargs: Any
) -> vs.VideoNode:
    index_files = self.index(self.normalize_filenames(file))

    return self._source(
        (self.source_func(idx_filename.to_str(), **kwargs) for idx_filename in index_files),
        bits, matrix, transfer, primaries, chroma_location, color_range, field_based
    )

source_func classmethod 

source_func(path: DataType | SPathLike, *args: Any, **kwargs: Any) -> VideoNode
Source code 
64
65
66
@classmethod
def source_func(cls, path: DataType | SPathLike, *args: Any, **kwargs: Any) -> vs.VideoNode:
    return cls._source_func(str(path), *args, **kwargs)

update_video_filenames abstractmethod 

update_video_filenames(index_path: SPath, filepaths: list[SPath]) -> None
Source code 
152
153
154
@abstractmethod
def update_video_filenames(self, index_path: SPath, filepaths: list[SPath]) -> None:
    raise NotImplementedError

Indexer 

Indexer(*, force: bool = True, **kwargs: Any)

Bases: ABC

Abstract indexer interface.

Methods:

Attributes:

Source code 
44
45
46
47
48
def __init__(self, *, force: bool = True, **kwargs: Any) -> None:
    super().__init__()

    self.force = force
    self.indexer_kwargs = kwargs

force instance-attribute 

force = force

indexer_kwargs instance-attribute 

indexer_kwargs = kwargs

get_joined_names classmethod 

get_joined_names(files: list[SPath]) -> str
Source code 
54
55
56
@classmethod
def get_joined_names(cls, files: list[SPath]) -> str:
    return '_'.join([file.name for file in files])

get_videos_hash classmethod 

get_videos_hash(files: list[SPath]) -> str
Source code 
58
59
60
61
62
@classmethod
def get_videos_hash(cls, files: list[SPath]) -> str:
    length = sum(file.stat().st_size for file in files)
    to_hash = length.to_bytes(32, 'little') + cls.get_joined_names(files).encode()
    return md5(to_hash).hexdigest()

normalize_filenames classmethod 

normalize_filenames(file: SPathLike | Sequence[SPathLike]) -> list[SPath]
Source code 
68
69
70
71
72
73
74
75
76
77
78
@classmethod
def normalize_filenames(cls, file: SPathLike | Sequence[SPathLike]) -> list[SPath]:
    files = list[SPath]()

    for f in to_arr(file):
        if str(f).startswith('file:///'):
            f = str(f)[8::]

        files.append(SPath(f))

    return files

source 

source(
    file: SPathLike | Sequence[SPathLike],
    bits: int | None = None,
    *,
    matrix: MatrixT | None = None,
    transfer: TransferT | None = None,
    primaries: PrimariesT | None = None,
    chroma_location: ChromaLocationT | None = None,
    color_range: ColorRangeT | None = None,
    field_based: FieldBasedT | None = None,
    **kwargs: Any
) -> VideoNode
Source code 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@inject_self
def source(
    self, file: SPathLike | Sequence[SPathLike],
    bits: int | None = None, *,
    matrix: MatrixT | None = None,
    transfer: TransferT | None = None,
    primaries: PrimariesT | None = None,
    chroma_location: ChromaLocationT | None = None,
    color_range: ColorRangeT | None = None,
    field_based: FieldBasedT | None = None,
    **kwargs: Any
) -> vs.VideoNode:
    return self._source(
        [self.source_func(f.to_str(), **self.indexer_kwargs | kwargs) for f in self.normalize_filenames(file)],
        bits, matrix, transfer, primaries, chroma_location, color_range, field_based
    )

source_func classmethod 

source_func(path: DataType | SPathLike, *args: Any, **kwargs: Any) -> VideoNode
Source code 
64
65
66
@classmethod
def source_func(cls, path: DataType | SPathLike, *args: Any, **kwargs: Any) -> vs.VideoNode:
    return cls._source_func(str(path), *args, **kwargs)

VSSourceFunc

Bases: Protocol

Methods:

__call__ 

__call__(path: DataType, *args: Any, **kwargs: Any) -> VideoNode
Source code 
35
36
def __call__(self, path: DataType, *args: Any, **kwargs: Any) -> vs.VideoNode:
    ...