D2VWitch ¶

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(), *files, '--output', output]))

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) -> D2VIndexFileInfo:
    with open(index_path, 'r') as f:
        file_content = f.read()

    lines = file_content.split('\n')

    head, lines = lines[:2], lines[2:]

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

    raw_header, lines = self._split_lines(self._split_lines(lines)[1])

    header = D2VIndexHeader()

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

        if key == 'STREAM_TYPE':
            header.stream_type = int(values[0])
        elif key == 'MPEG_TYPE':
            header.MPEG_type = int(values[0])
        elif key == 'IDCT_ALGORITHM':
            header.iDCT_algorithm = int(values[0])
        elif key == 'YUVRGB_SCALE':
            header.YUVRGB_scale = int(values[0])
        elif key == 'LUMINANCE_FILTER':
            header.luminance_filter = tuple(map(int, values))
        elif key == 'CLIPPING':
            header.clipping = list(map(int, values))
        elif key == 'ASPECT_RATIO':
            header.aspect = Fraction(*list(map(int, values[0].split(':'))))
        elif key == 'PICTURE_SIZE':
            header.pic_size = str(values[0])
        elif key == 'FIELD_OPERATION':
            header.field_op = int(values[0])
        elif key == 'FRAME_RATE':
            if matches := re.search(r'.*\((\d+\/\d+)', values[0]):
                header.frame_rate = Fraction(matches.group(1))
        elif key == 'LOCATION':
            header.location = list(map(partial(int, base=16), values))

    frame_data = list[D2VIndexFrameData]()

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

            line = rawline.split(" ", maxsplit=7)

            ffile_idx = int(line[2])

            if ffile_idx < file_idx:
                continue
            elif ffile_idx > file_idx:
                break

            frame_data.append(D2VIndexFrameData(
                int(line[1]), 'I', int(line[5]),
                int(line[6]), int(line[0], 16),
                int(line[4]), int(line[3]),
                list(int(a, 16) for a in line[7:])
            ))
    elif file_idx == -1:
        for rawline in lines:
            if len(rawline) == 0:
                break

            line = rawline.split(" ")

            frame_data.append(D2VIndexFrameData(
                int(line[1]), 'I', int(line[5]),
                int(line[6]), int(line[0], 16),
                int(line[4]), int(line[3]),
                list(int(a, 16) for a in line[7:])
            ))

    return D2VIndexFileInfo(index_path, file_idx, header, frame_data)

@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

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]]:
    dvddd = self._d2v_vobid_frameset(vob_input_files, output_folder)

    if len(dvddd.keys()) == 1 and (0, 0) in dvddd.keys():
        raise CustomValueError(
            'Youre indexer created a d2v file with only zeros for vobid cellid; '
            'This usually means outdated/unpatched D2Vwitch', self.parse_vts
        )

    frameranges = [x for y in [dvddd[a] for a in vobidcellids_to_take] for x in y]

    fflags, vobids, progseq = self._d2v_collect_all_frameflags(vob_input_files, output_folder)

    index_file = self.index(vob_input_files, output_folder=output_folder)[0]
    node = self._source_func(index_file, rff=False)  # type: ignore

    assert len(node) == len(fflags) == len(vobids) == len(progseq)

    progseq = cut_array_on_ranges(progseq, frameranges)
    fflags = cut_array_on_ranges(fflags, frameranges)
    vobids = cut_array_on_ranges(vobids, frameranges)
    node = remap_frames(node, frameranges)

    rff = [(a & 1) for a in fflags]

    if not disable_rff:
        tff = [int((a & 2) >> 1) for a in fflags]
        prog = [int((a & 0b01000000) != 0) for a in fflags]

        node = apply_rff_video(node, rff, tff, prog, progseq)
        vobids = apply_rff_array(vobids, rff, tff, progseq)

    return node, rff, vobids, []

@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:
    with open(index_path, 'r') as file:
        file_content = file.read()

    lines = file_content.split('\n')

    str_filepaths = list(map(str, filepaths))

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

    if not (n_files := int(lines[1])) or n_files != len(str_filepaths):
        self.file_corrupted(index_path)

    end_videos = lines.index('')

    if lines[2:end_videos] == str_filepaths:
        return

    lines[2:end_videos] = str_filepaths

    with open(index_path, 'w') as file:
        file.write('\n'.join(lines))