Skip to content

base

Classes:

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 in vssource/indexers/base.py 
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 in vssource/indexers/base.py 
225
226
227
228
229
230
231
232
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 in vssource/indexers/base.py 
160
161
162
163
164
165
@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 in vssource/indexers/base.py 
222
223
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 in vssource/indexers/base.py 
167
168
169
170
171
172
@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 in vssource/indexers/base.py 
58
59
60
@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 in vssource/indexers/base.py 
211
212
213
214
215
216
217
218
219
220
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(gettempdir())

    return SPath(output_folder)

get_video_idx_path 

get_video_idx_path(
    folder: SPath, file_hash: str, video_name: SPathLike
) -> SPath
Source code in vssource/indexers/base.py 
280
281
282
283
284
285
def get_video_idx_path(self, folder: SPath, file_hash: str, video_name: SPathLike) -> SPath:
    vid_name = SPath(video_name).stem
    current_indxer = SPath(self._bin_path).name
    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 in vssource/indexers/base.py 
62
63
64
65
66
@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 in vssource/indexers/base.py 
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
275
276
277
278
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 in vssource/indexers/base.py 
72
73
74
75
76
77
78
79
80
81
82
@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,
    idx_props: bool = True,
    **kwargs: Any
) -> VideoNode
Source code in vssource/indexers/base.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
@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,
    idx_props: bool = True,
    **kwargs: Any,
) -> vs.VideoNode:
    index_files = self.index(self.normalize_filenames(file))

    return super().source(
        index_files,
        bits,
        matrix=matrix,
        transfer=transfer,
        primaries=primaries,
        chroma_location=chroma_location,
        color_range=color_range,
        field_based=field_based,
        idx_props=idx_props,
        **kwargs,
    )

source_func classmethod 

source_func(path: DataType | SPathLike, *args: Any, **kwargs: Any) -> VideoNode
Source code in vssource/indexers/base.py 
68
69
70
@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 in vssource/indexers/base.py 
174
175
176
@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 in vssource/indexers/base.py 
48
49
50
51
52
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 in vssource/indexers/base.py 
58
59
60
@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 in vssource/indexers/base.py 
62
63
64
65
66
@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 in vssource/indexers/base.py 
72
73
74
75
76
77
78
79
80
81
82
@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,
    idx_props: bool = True,
    **kwargs: Any
) -> VideoNode
Source code in vssource/indexers/base.py 
101
102
103
104
105
106
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,
    idx_props: bool = True,
    **kwargs: Any,
) -> vs.VideoNode:
    nfiles = self.normalize_filenames(file)
    clip = self._source(
        [self.source_func(f.to_str(), **self.indexer_kwargs | kwargs) for f in nfiles],
        bits,
        matrix,
        transfer,
        primaries,
        chroma_location,
        color_range,
        field_based,
    )
    if idx_props:
        clip = clip.std.SetFrameProps(IdxFilePath=[f.to_str() for f in nfiles], Idx=self.__class__.__name__)

    return clip

source_func classmethod 

source_func(path: DataType | SPathLike, *args: Any, **kwargs: Any) -> VideoNode
Source code in vssource/indexers/base.py 
68
69
70
@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 in vssource/indexers/base.py 
38
def __call__(self, path: DataType, *args: Any, **kwargs: Any) -> vs.VideoNode: ...