vivy: tv: 04

Vivy/04/04.vpy

@@ -0,0 +1,63 @@
+import vapoursynth as vs
+
+from lvsfunc.types import Range
+from lvsfunc.dehardsub import HardsubLine, HardsubSign, HardsubMask, bounded_dehardsub
+from yt_common import SelfRunner
+
+from typing import List
+
+import os
+import sys
+sys.path.append("..")
+
+from vivy_common import (VivyConfig, VivySource, antialias, deband, denoise,  # noqa: E402
+                         finalize, fsrcnnx_rescale, letterbox_edgefix, waka_replace)
+
+core = vs.core
+
+
+EPNUM: int = int(os.path.basename(os.path.splitext(__file__)[0]))
+CONFIG: VivyConfig = VivyConfig(EPNUM)
+SOURCE: VivySource = VivySource(CONFIG)
+SIGNS_RU: List[HardsubMask] = [
+    HardsubLine((1278, 3392), ((275, 918), (1356, 112))),
+    HardsubSign((3452, 3572), ((232, 857), (1077, 114)), refframes=3500),
+    HardsubSign((11454, 11489), ((727, 176), (440, 78))),
+    HardsubSign((16803, 16841), ((135, 549), (479, 221))),
+    HardsubLine((29463, 33374), ((275, 890), (1356, 140))),
+    HardsubSign((33950, 34045), ((232, 857), (1077, 114)), refframes=34045),
+]
+NOSCALE: List[Range] = []
+NOAA: List[Range] = []
+LETTERBOX: List[Range] = [(0, 1151)]
+WAKA_REPLACE: List[List[Range]] = [
+    [(30119, 30969)],
+    [],
+]
+
+
+def filter_basic() -> vs.VideoNode:
+    wakas, ref = SOURCE.source()
+    wakas = [w[:33665] + core.std.BlankClip(w, length=21) + w[33665:] for w in wakas]
+    waka = wakas[0]
+    waka, wakas = waka_replace(waka, wakas[1:], WAKA_REPLACE)
+    src = bounded_dehardsub(waka, ref, SIGNS_RU, wakas)
+    return src
+
+
+def filter() -> vs.VideoNode:
+    src = filter_basic()
+    rescale = fsrcnnx_rescale(src, NOSCALE)
+    den = denoise(rescale)
+    deb = deband(den)
+    aa = antialias(deb, NOAA)
+    edgefix = letterbox_edgefix(aa, LETTERBOX)
+    final = finalize(edgefix)
+    final.set_output()
+    return final
+
+
+if __name__ == "__main__":
+    SelfRunner(CONFIG, filter, filter_basic)
+else:
+    filter()

Vivy/vivy_common/config.py

@@ -0,0 +1,35 @@
+from yt_common import Config, FunimationSource
+
+import os
+
+from typing import List
+
+TITLE: str = "Vivy"
+TITLE_LONG: str = f"{TITLE} - Fluorite Eye's Song"
+RESOLUTION: int = 1080
+SUBGROUP: str = "YameteTomete"
+DATAPATH: str = os.path.dirname(__file__)
+
+WAKA_RU_FILENAME: str = f"{TITLE}_{{epnum:02d}}_RU_HD.mp4"
+WAKA_FR_FILENAME: str = f"{TITLE}_{{epnum:02d}}_FR_HD.mp4"
+WAKA_DE_FILENAME: str = f"{TITLE} - Fluorite Eyes Song E{{epnum:02d}} [1080p][AAC][JapDub][GerSub][Web-DL].mkv"
+
+
+class VivyConfig(Config):
+    def __init__(self, epnum: int) -> None:
+        super().__init__(
+            epnum,
+            TITLE,
+            TITLE_LONG,
+            RESOLUTION,
+            DATAPATH
+        )
+
+
+class VivySource(FunimationSource):
+    def get_waka_filenames(self) -> List[str]:
+        return [self.config.format_filename(f) for f in [
+            WAKA_RU_FILENAME,
+            WAKA_FR_FILENAME,
+            WAKA_DE_FILENAME,
+        ]]

yt_common/mypy.ini

@@ -0,0 +1,43 @@
+[mypy]
+python_version = 3.9
+
+ignore_missing_imports = True
+
+disallow_any_generics = True
+
+disallow_untyped_defs = True
+disallow_incomplete_defs = True
+check_untyped_defs = True
+disallow_untyped_decorators = True
+
+no_implicit_optional = True
+strict_optional = True
+
+warn_redundant_casts = True
+warn_unused_ignores = True
+warn_no_return = True
+warn_return_any = True
+warn_unreachable = True
+
+show_none_errors = True
+ignore_errors = False
+
+allow_untyped_globals = False
+allow_redefinition = False
+implicit_reexport = True
+strict_equality = True
+
+show_error_context = False
+show_column_numbers = True
+show_error_codes = True
+color_output = True
+error_summary = True
+pretty = True
+
+mypy_path = .
+
+[mypy-cytoolz.*]
+ignore_errors = True
+
+[mypy-vsutil.*]
+implicit_reexport = True

yt_common/setup.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+
+import setuptools
+
+name = "yt_common"
+version = "0.0.0"
+release = "0.0.0"
+
+setuptools.setup(
+    name=name,
+    version=release,
+    author="louis",
+    author_email="louis@poweris.moe",
+    description="yametetomete encodes common module",
+    packages=["yt_common"],
+    classifiers=[
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: MIT License",
+        "Operating System :: OS Independent",
+    ],
+    package_data={
+        'yt_common': ['py.typed'],
+    },
+    python_requires='>=3.8',
+)

yt_common/yt_common/__init__.py

@@ -0,0 +1,3 @@
+from .config import Config  # noqa: F401
+from .automation import SelfRunner  # noqa: F401
+from .source import DehardsubFileFinder, FunimationSource  # noqa: F401

yt_common/yt_common/automation.py

@@ -3,71 +3,20 @@ import vapoursynth as vs
 import acsuite
 import argparse
 import os
-import functools
-import glob
-import signal
 import string
 import subprocess
-import vsutil
 
-from typing import Any, BinaryIO, Callable, List, Optional, Sequence, Tuple, Union, cast
+from typing import Any, BinaryIO, Callable, List, Optional, Sequence, Union, cast
+
+from .config import Config
+from .log import status, warn, error, success
+from .source import AMAZON_FILENAME, ER_FILENAME, SUBSPLS_FILENAME, FUNI_INTRO, glob_crc
 
 core = vs.core
 
-TITLE: str = "Vivy"
-TITLE_LONG: str = f"{TITLE} - Fluorite Eye's Song"
-RESOLUTION: int = 1080
-
-SUBGROUP = "YameteTomete"
-
-SUBSPLS_FILENAME: str = f"[SubsPlease] {TITLE_LONG} - {{epnum:02d}} ({RESOLUTION}p) [$CRC].mkv"
-ER_FILENAME: str = f"[Erai-raws] {TITLE_LONG} - {{epnum:02d}} [{RESOLUTION}p][Multiple Subtitle].mkv"
-FUNI_INTRO: int = 289
-WAKA_FILENAME: str = f"{TITLE}_{{epnum:02d}}_RU_HD.mp4"
-AMAZON_FILENAME: str = f"{TITLE_LONG} - {{epnum:02d}} (Amazon Prime CBR {RESOLUTION}p).mkv"
-
 AUDIO_OVERRIDE: str = "audio.mka"
-AUDIO_FMT: List[str] = [".mka", ".aac", ".wav", ".flac", ".mp3", ".ogg", ".opus", ".m4a"]
-AFV_FMT: List[str] = [".mkv", ".mp4", ".webm"]
 AUDIO_CUT: str = "_audiogetter_cut.mka"
 
-STATUS: str = '\033[94m'
-WARNING: str = '\033[93m'
-ERROR: str = '\033[91m'
-SUCCESS: str = '\033[92m'
-RESET: str = '\033[0m'
-
-
-def glob_crc(pattern: str) -> str:
-    res = glob.glob(glob.escape(pattern).replace("$CRC", "*"))
-    if len(res) == 0:
-        raise FileNotFoundError(f"File matching \"{pattern}\" not found!")
-    return res[0]
-
-
-def get_ref(epnum: int) -> vs.VideoNode:
-    if epnum >= 4:
-        if os.path.isfile(AMAZON_FILENAME.format(epnum=epnum)):
-            return core.ffms2.Source(AMAZON_FILENAME.format(epnum=epnum))
-        else:
-            print(f"{WARNING}Amazon video not found, dehardsubbing with new funi encode{RESET}")
-
-    try:
-        return core.ffms2.Source(glob_crc(SUBSPLS_FILENAME.format(epnum=epnum)))[FUNI_INTRO:]
-    except FileNotFoundError:
-        pass
-
-    if not os.path.isfile(ER_FILENAME.format(epnum=epnum)):
-        raise FileNotFoundError("Failed to find valid reference video")
-
-    return core.ffms2.Source(ER_FILENAME.format(epnum=epnum))[FUNI_INTRO:]
-
-
-def source(epnum: int) -> Tuple[vs.VideoNode, vs.VideoNode]:
-    waka = vsutil.depth(core.ffms2.Source(WAKA_FILENAME.format(epnum=epnum)), 16)
-    ref = vsutil.depth(get_ref(epnum), 16)
-    return waka, ref
-
 
 def bin_to_plat(binary: str) -> str:
     if os.name == "nt":
@@ -77,7 +26,7 @@ def bin_to_plat(binary: str) -> str:
 
 
 def forward_signal(signum: int, frame: Any, process: Any) -> None:
-    print(f"{WARNING}Forwarding SIGINT{RESET}")
+    warn("Forwarding SIGINT")
     process.send_signal(signum)
 
 
@@ -106,30 +55,31 @@ class Encoder():
         outfile = self.out_template.format(filename=filename)
 
         if os.path.isfile(outfile) and not self.force:
-            print(f"{WARNING}Existing output detected, skipping encode!{RESET}")
+            warn("Existing output detected, skipping encode!")
             return outfile
 
         params = [p.format(frames=end-start, filename=filename) for p in self.params]
 
-        print(f"{STATUS}--- RUNNING ENCODE ---{RESET}")
+        status("--- RUNNING ENCODE ---")
 
         print("+ " + " ".join([self.binary] + list(params)))
 
         process = subprocess.Popen([self.binary] + list(params), stdin=subprocess.PIPE)
 
         # i want the encoder to handle any ctrl-c so it exits properly
-        forward_to_proc = functools.partial(forward_signal, process=process)
+        # forward_to_proc = functools.partial(forward_signal, process=process)
-        signal.signal(signal.SIGINT, forward_to_proc)
+        # signal.signal(signal.SIGINT, forward_to_proc)
+        # turns out this didn't work out the way i had hoped
 
         clip[start:end].output(cast(BinaryIO, process.stdin), y4m=True)
         process.communicate()
 
         # vapoursynth should handle this itself but just in case
         if process.returncode != 0:
-            print(f"{ERROR}--- ENCODE FAILED ---{RESET}")
+            error("--- ENCODE FAILED ---")
             raise BrokenPipeError(f"Pipe to {self.binary} broken")
 
-        print(f"{SUCCESS}--- ENCODE FINISHED ---{RESET}")
+        success("--- ENCODE FINISHED ---")
         self.cleanup.append(outfile)
         return outfile
 
@@ -153,13 +103,20 @@ class Encoder():
 
 
 class AudioGetter():
+    """
+    TODO: really should modularize this a bit instead of assuming amazon->funi
+    """
+    config: Config
+
     audio_file: str
     audio_start: int
     video_src: Optional[vs.VideoNode]
 
     cleanup: List[str]
 
-    def __init__(self, epnum: int, override: Optional[str] = None) -> None:
+    def __init__(self, config: Config, override: Optional[str] = None) -> None:
+        self.config = config
+
         self.audio_start = 0
         self.video_src = None
         self.cleanup = []
@@ -176,30 +133,29 @@ class AudioGetter():
             return
 
         # look for amazon first
-        if os.path.isfile(AMAZON_FILENAME.format(epnum=epnum)):
+        if os.path.isfile(self.config.format_filename(AMAZON_FILENAME)):
-            self.audio_file = AMAZON_FILENAME.format(epnum=epnum)
+            self.audio_file = self.config.format_filename(AMAZON_FILENAME)
-            self.video_src = core.ffms2.Source(AMAZON_FILENAME.format(epnum=epnum))
+            self.video_src = core.ffms2.Source(self.audio_file)
-            print(f"{SUCCESS}Found Amazon audio{RESET}")
+            success("Found Amazon audio")
             return
 
         # as of Ep4 SubsPlease is using new funi 128kbps aac while erai has 256kbps still
         try:
-            if os.path.isfile(ER_FILENAME.format(epnum=epnum)):
+            if os.path.isfile(self.config.format_filename(ER_FILENAME)):
-                self.audio_file = ER_FILENAME.format(epnum=epnum)
+                self.audio_file = self.config.format_filename(ER_FILENAME)
-                self.video_src = core.ffms2.Source(ER_FILENAME.format(epnum=epnum))
+                self.video_src = core.ffms2.Source(self.audio_file)
-            if os.path.isfile(glob_crc(SUBSPLS_FILENAME.format(epnum=epnum))):
+            elif os.path.isfile(glob_crc(self.config.format_filename(SUBSPLS_FILENAME))):
-                self.audio_file = glob_crc(SUBSPLS_FILENAME.format(epnum=epnum))
+                self.audio_file = glob_crc(self.config.format_filename(SUBSPLS_FILENAME))
-                self.video_src = core.ffms2.Source(glob_crc(SUBSPLS_FILENAME.format(epnum=epnum)))
+                self.video_src = core.ffms2.Source(self.audio_file)
-                if (epnum >= 4):
+                warn("Using SubsPlease, audio may be worse than Erai-Raws")
-                    print(f"{WARNING}Using SubsPlease, audio may be worse than Erai-Raws{RESET}")
             else:
                 raise FileNotFoundError()
         except FileNotFoundError:
-            print(f"{ERROR}Could not find audio{RESET}")
+            error("Could not find audio")
             raise
 
         self.audio_start = FUNI_INTRO
-        print(f"{WARNING}No Amazon audio, falling back to Funi{RESET}")
+        warn("No Amazon audio, falling back to Funi")
 
     def trim_audio(self, src: vs.VideoNode,
                    trims: Union[acsuite.Trim, List[acsuite.Trim], None] = None) -> str:
@@ -234,8 +190,8 @@ class AudioGetter():
 
 
 class SelfRunner():
+    config: Config
     clip: vs.VideoNode
-    epnum: int
 
     workraw: bool
 
@@ -245,13 +201,13 @@ class SelfRunner():
     encoder: Encoder
     audio: AudioGetter
 
-    def __init__(self, epnum: int, final_filter: Callable[[], vs.VideoNode],
+    def __init__(self, config: Config, final_filter: Callable[[], vs.VideoNode],
                  workraw_filter: Optional[Callable[[], vs.VideoNode]] = None) -> None:
-        self.epnum = epnum
+        self.config = config
         self.video_clean = False
         self.audio_clean = False
 
-        parser = argparse.ArgumentParser(description=f"Encode {TITLE} Episode {epnum:02d}")
+        parser = argparse.ArgumentParser(description=f"Encode {self.config.title} Episode {self.config.epnum:02d}")
         if workraw_filter:
             parser.add_argument("-w", "--workraw", help="Encode workraw, fast x264", action="store_true")
         parser.add_argument("-s", "--start", nargs='?', type=int, help="Start encode at frame START")
@@ -260,12 +216,12 @@ class SelfRunner():
         parser.add_argument("-c", "--encoder", type=str, help="Override detected encoder binary")
         parser.add_argument("-f", "--force", help="Overwrite existing intermediaries", action="store_true")
         parser.add_argument("-a", "--audio", type=str, help="Force audio file")
-        parser.add_argument("-x", "--suffix", type=str, default="premux", help="Change the suffix of the mux")
+        parser.add_argument("-x", "--suffix", type=str, help="Change the suffix of the mux")
         parser.add_argument("-d", "--no-metadata", help="No extra metadata in premux", action="store_true")
         args = parser.parse_args()
 
         self.workraw = args.workraw if workraw_filter else False
-        self.suffix = args.suffix if not self.workraw else "workraw"
+        self.suffix = args.suffix if args.suffix is not None else "workraw" if self.workraw else "premux"
 
         self.clip = workraw_filter() if workraw_filter and self.workraw else final_filter()
 
@@ -295,33 +251,34 @@ class SelfRunner():
         if start >= end:
             raise ValueError("Start frame must be before end frame!")
 
-        self.encoder = Encoder(epnum, settings_path, args.encoder, args.force)
+        self.encoder = Encoder(self.config.epnum, settings_path, args.encoder, args.force)
-        self.video_file = self.encoder.encode(self.clip, f"{epnum:02d}_{start}_{end}", start, end)
+        self.video_file = self.encoder.encode(self.clip, f"{self.config.epnum:02d}_{start}_{end}", start, end)
 
-        print(f"{STATUS}--- LOOKING FOR AUDIO ---{RESET}")
+        status("--- LOOKING FOR AUDIO ---")
-        self.audio = AudioGetter(self.epnum, args.audio)
+        self.audio = AudioGetter(self.config, args.audio)
 
-        print(f"{STATUS}--- TRIMMING AUDIO ---{RESET}")
+        status("--- TRIMMING AUDIO ---")
         self.audio_file = self.audio.trim_audio(self.clip, (start, end))
 
         try:
-            print(f"{STATUS}--- MUXING FILE ---{RESET}")
+            status("--- MUXING FILE ---")
-            if self._mux(f"{TITLE.lower()}_{epnum:02d}_{args.suffix}.mkv", not args.no_metadata,
+            if self._mux(f"{self.config.title.lower()}_{self.config.epnum:02d}_{self.suffix}.mkv",
+                         not args.no_metadata,
                          not args.no_metadata and start == 0 and end == self.clip.num_frames) != 0:
                 raise Exception("mkvmerge failed")
         except Exception:
-            print(f"{ERROR}--- MUXING FAILED ---{RESET}")
+            error("--- MUXING FAILED ---")
             self.audio.do_cleanup()
             raise
 
-        print(f"{SUCCESS}--- MUXING SUCCESSFUL ---{RESET}")
+        success("--- MUXING SUCCESSFUL ---")
 
         self.audio.do_cleanup()
 
         if not args.keep:
             self.encoder.do_cleanup()
 
-        print(f"{SUCCESS}--- ENCODE COMPLETE ---{RESET}")
+        success("--- ENCODE COMPLETE ---")
 
     def _mux(self, name: str, metadata: bool = True, chapters: bool = True) -> int:
         mkvtoolnix_args = [
@@ -337,11 +294,11 @@ class SelfRunner():
         ]
         if metadata:
             mkvtoolnix_args += [
-                "--title", f"[{SUBGROUP}] {TITLE_LONG} - {self.epnum:02d}",
+                "--title", f"[{self.config.subgroup}] {self.config.title_long} - {self.config.epnum:02d}",
             ]
 
         if chapters:
-            chap = [f for f in ["{self.epnum:02d}.xml", "chapters.xml"] if os.path.isfile(f)]
+            chap = [f for f in [f"{self.config.epnum:02d}.xml", "chapters.xml"] if os.path.isfile(f)]
             if len(chap) != 0:
                 mkvtoolnix_args += [
                     "--chapters", chap[0],

yt_common/yt_common/config.py

@@ -0,0 +1,18 @@
+class Config():
+    epnum: int
+    title: str
+    title_long: str
+    resolution: int
+    datapath: str
+    subgroup: str
+
+    def __init__(self, epnum: int, title: str, title_long: str, resolution: int, datapath: str) -> None:
+        self.epnum = epnum
+        self.title = title
+        self.title_long = title_long
+        self.resolution = resolution
+        self.datapath = datapath
+
+    def format_filename(self, filename: str) -> str:
+        return filename.format(epnum=self.epnum, title=self.title,
+                               title_long=self.title_long, resolution=self.resolution)

yt_common/yt_common/log.py

@@ -0,0 +1,23 @@
+# TODO: real logging shit not this jank-ass crap
+
+STATUS: str = '\033[94m'
+WARNING: str = '\033[93m'
+ERROR: str = '\033[91m'
+SUCCESS: str = '\033[92m'
+RESET: str = '\033[0m'
+
+
+def status(s: str) -> None:
+    print(f"{STATUS}{s}{RESET}")
+
+
+def warn(s: str) -> None:
+    print(f"{WARNING}{s}{RESET}")
+
+
+def error(s: str) -> None:
+    print(f"{ERROR}{s}{RESET}")
+
+
+def success(s: str) -> None:
+    print(f"{SUCCESS}{s}{RESET}")

yt_common/yt_common/source.py

@@ -0,0 +1,72 @@
+import vapoursynth as vs
+
+import vsutil
+
+import glob
+import os
+
+from abc import ABC, abstractmethod
+from typing import List, Tuple
+
+from .config import Config
+from .log import warn, error, success
+
+core = vs.core
+
+SUBSPLS_FILENAME: str = "[SubsPlease] {title_long} - {epnum:02d} ({resolution}p) [$CRC].mkv"
+ER_FILENAME: str = "[Erai-raws] {title_long} - {epnum:02d} [v0][{resolution}p].mkv"
+FUNI_INTRO: int = 289
+AMAZON_FILENAME: str = "{title_long} - {epnum:02d} (Amazon Prime CBR {resolution}p).mkv"
+
+
+def glob_crc(pattern: str) -> str:
+    res = glob.glob(glob.escape(pattern).replace("$CRC", "*"))
+    if len(res) == 0:
+        raise FileNotFoundError(f"File matching \"{pattern}\" not found!")
+    return res[0]
+
+
+class DehardsubFileFinder(ABC):
+    config: Config
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+
+    @abstractmethod
+    def get_waka_filenames(self) -> List[str]:
+        pass
+
+    @abstractmethod
+    def get_ref(self) -> vs.VideoNode:
+        pass
+
+    def source(self) -> Tuple[List[vs.VideoNode], vs.VideoNode]:
+        wakas = [vsutil.depth(core.ffms2.Source(self.config.format_filename(f)), 16)
+                 for f in self.get_waka_filenames()]
+        ref = vsutil.depth(self.get_ref(), 16)
+        return wakas, ref
+
+
+class FunimationSource(DehardsubFileFinder):
+    def get_amazon(self) -> vs.VideoNode:
+        if not os.path.isfile(self.config.format_filename(AMAZON_FILENAME)):
+            warn("Amazon not found, falling back to Funimation")
+            raise FileNotFoundError()
+        success("Found Amazon video")
+        return core.ffms2.Source(self.config.format_filename(AMAZON_FILENAME))
+
+    def get_funi_filename(self) -> str:
+        if os.path.isfile(self.config.format_filename(ER_FILENAME)):
+            return self.config.format_filename(ER_FILENAME)
+
+        error("Erai-raws not found, falling back to SubsPlease")
+        return glob_crc(self.config.format_filename(SUBSPLS_FILENAME))
+
+    def get_funi(self) -> vs.VideoNode:
+        return core.ffms2.Source(self.get_funi_filename())[FUNI_INTRO:]
+
+    def get_ref(self) -> vs.VideoNode:
+        try:
+            return self.get_amazon()
+        except FileNotFoundError:
+            return self.get_funi()