DGIndexNV ¶

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__)

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

    return list(
        map(str, [
            self._get_bin_path(), '-i',
            ','.join(str(path.absolute()) for path in files),
            '-h', '-o', output, '-e'
        ])
    )

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

@lru_cache
def get_info(self, index_path: SPath, file_idx: int = -1) -> DGIndexFileInfo:
    with open(index_path, 'r') as file:
        file_content = file.read()

    lines = file_content.split('\n')

    head, lines = self._split_lines(lines)

    if 'DGIndexNV' not in head[0]:
        self.file_corrupted(index_path)

    vid_lines, lines = self._split_lines(lines)
    raw_header, lines = self._split_lines(lines)

    header = DGIndexHeader()

    for rlin in raw_header:
        if split_val := rlin.rstrip().split(' '):
            key: str = split_val[0].upper()
            values: list[str] = split_val[1:]
        else:
            continue

        if key == 'DEVICE':
            header.device = int(values[0])
        elif key == 'DECODE_MODES':
            header.decode_modes = list(map(int, values[0].split(',')))
        elif key == 'STREAM':
            header.stream = tuple(map(int, values))
        elif key == 'RANGE':
            header.ranges = list(map(int, values))
        elif key == 'DEMUX':
            continue
        elif key == 'DEPTH':
            header.depth = int(values[0])
        elif key == 'ASPECT':
            try:
                header.aspect = Fraction(*list(map(int, values)))
            except ZeroDivisionError:
                header.aspect = Fraction(1, 1)
                if os.environ.get('VSSOURCE_DEBUG', False):
                    print(ResourceWarning('Encountered video with 0/0 aspect ratio!'))
        elif key == 'COLORIMETRY':
            header.colorimetry = tuple(map(int, values))
        elif key == 'PKTSIZ':
            header.packet_size = int(values[0])
        elif key == 'VPID':
            header.vpid = int(values[0])

    video_sizes = [int(line[-1]) for line in [line.split(' ') for line in vid_lines]]

    max_sector = sum([0, *video_sizes[:file_idx + 1]])

    idx_file_sector = [max_sector - video_sizes[file_idx], max_sector]

    curr_SEQ, frame_data = 0, []

    for rawline in lines:
        if len(rawline) == 0:
            break

        line: Sequence[str | None] = [*rawline.split(" ", maxsplit=6), *([None] * 6)]

        name = str(line[0])

        if name == 'SEQ':
            curr_SEQ = opt_int(line[1]) or 0

        if curr_SEQ < idx_file_sector[0]:
            continue
        elif curr_SEQ > idx_file_sector[1]:
            break

        try:
            int(name.split(':')[0])
        except ValueError:
            continue

        frame_data.append(DGIndexFrameData(
            int(line[2] or 0) + 2, str(line[1]), *opt_ints(line[4:6])
        ))

    footer = DGIndexFooter()

    for rlin in lines[-10:]:
        if split_val := rlin.rstrip().split(' '):
            values = [split_val[0], ' '.join(split_val[1:])]
        else:
            continue

        for key in footer.__dict__.keys():
            if key.split('_')[-1].upper() in values:
                if key == 'film':
                    try:
                        value = [float(v.replace('%', '')) for v in values if '%' in v][0]
                    except IndexError:
                        value = 0
                else:
                    value = int(values[1])

                footer[key] = value

    return DGIndexFileInfo(index_path, file_idx, header, frame_data, footer)

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

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)

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))

@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()

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

@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

@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
    )

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

def update_video_filenames(self, index_path: SPath, filepaths: list[SPath]) -> None:
    lines = index_path.read_lines()

    str_filepaths = list(map(str, filepaths))

    if 'DGIndexNV' not in lines[0]:
        self.file_corrupted(index_path)

    start_videos = lines.index('') + 1
    end_videos = lines.index('', start_videos)

    if end_videos - start_videos != len(str_filepaths):
        self.file_corrupted(index_path)

    split_lines = [
        line.split(' ') for line in lines[start_videos:end_videos]
    ]

    current_paths = [line[:-1][0] for line in split_lines]

    if current_paths == str_filepaths:
        return

    video_args = [line[-1:] for line in split_lines]

    lines[start_videos:end_videos] = [
        ' '.join([path, *args]) for path, args in zip(str_filepaths, video_args)
    ]

    index_path.write_lines(lines)