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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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 
221
222
223
224
225
226
227
228
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 
160
161
162
163
@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 
218
219
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 
165
166
167
168
@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 
64
65
66
@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 
207
208
209
210
211
212
213
214
215
216
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 
276
277
278
279
280
281
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 
68
69
70
71
72
@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 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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({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 = 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 
78
79
80
81
82
83
84
85
86
87
88
@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 
314
315
316
317
318
319
320
321
322
323
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 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
@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:
    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 
74
75
76
@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 
170
171
172
@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 
314
315
316
317
318
319
320
321
322
323
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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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 
221
222
223
224
225
226
227
228
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 
160
161
162
163
@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 
218
219
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 
165
166
167
168
@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 
64
65
66
@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 
207
208
209
210
211
212
213
214
215
216
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 
276
277
278
279
280
281
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 
68
69
70
71
72
@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 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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({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 = 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 
78
79
80
81
82
83
84
85
86
87
88
@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 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
@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:
    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 
74
75
76
@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 
170
171
172
@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 
54
55
56
57
58
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 
64
65
66
@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 
68
69
70
71
72
@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 
78
79
80
81
82
83
84
85
86
87
88
@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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@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 
74
75
76
@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 
46
def __call__(self, path: DataType, *args: Any, **kwargs: Any) -> vs.VideoNode: ...