various

packages/musicomp/README.md

@@ -0,0 +1 @@
+

packages/musicomp/default.nix

@@ -0,0 +1,19 @@
+{
+  lib,
+  python3Packages,
+  opusTools,
+}:
+python3Packages.buildPythonApplication {
+  pname = "musicomp";
+  version = "0.1.0";
+  src = ./.;
+  pyproject = true;
+  doCheck = false;
+  build-system = [python3Packages.hatchling];
+  makeWrapperArgs = [
+    "--prefix"
+    "PATH"
+    ":"
+    (lib.makeBinPath [opusTools])
+  ];
+}

packages/musicomp/pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "musicomp"
+version = "0.1.0"
+description = ""
+authors = [
+    {name = "Lukas Wurzinger", email = "lukas@wrz.one"}
+]
+readme = "README.md"
+requires-python = ">=3.12"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project.scripts]
+musicomp = "musicomp.cli:main"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/musicomp"]

packages/musicomp/src/musicomp/__main__.py

@@ -0,0 +1,3 @@
+from .cli import main
+
+main()

packages/musicomp/src/musicomp/clean.py

@@ -0,0 +1,6 @@
+from os import PathLike
+from pathlib import Path
+
+
+def clean(dst: str | PathLike[str]) -> None:
+    Path(dst).unlink(missing_ok=True)

packages/musicomp/src/musicomp/cli.py

@@ -0,0 +1,118 @@
+import sys
+from multiprocessing import Pool, cpu_count
+from argparse import ArgumentParser, ArgumentTypeError, Namespace
+from pathlib import Path
+from .todo import Todo
+
+
+args = Namespace()
+
+
+def task(todo: Todo) -> None:
+    todo.run()
+    if args.verbose:
+        print("finished", todo, file=sys.stderr)
+
+
+def main():
+    def workers_type_func(value: object) -> int:
+        try:
+            value = int(value)  # pyright: ignore[reportArgumentType]
+            if value <= 0:
+                raise ArgumentTypeError(f"{value} is not a positive integer")
+        except ValueError:
+            raise ArgumentTypeError(f"{value} is not an integer")
+
+        return value
+
+    parser = ArgumentParser()
+
+    _ = parser.add_argument(
+        "-w",
+        "--workers",
+        default=cpu_count(),
+        type=workers_type_func,
+        help="amount of worker processes",
+    )
+    _ = parser.add_argument(
+        "-i",
+        "--interactive",
+        action="store_true",
+        help="prompt before running",
+    )
+    _ = parser.add_argument(
+        "-k",
+        "--keep",
+        action="store_true",
+        help="whether source files should be kept if both directories are the same",
+    )
+    _ = parser.add_argument(
+        "-r",
+        "--redo",
+        action="store_true",
+        help="whether everything should be re-encoded regardless of whether they have already been transcoded",
+    )
+    _ = parser.add_argument(
+        "-v",
+        "--verbose",
+        action="count",
+        default=0,
+        help="verbose output",
+    )
+
+    _ = parser.add_argument(
+        "music",
+        nargs=1,
+        type=Path,
+        help="the source directory",
+    )
+    _ = parser.add_argument(
+        "comp",
+        nargs=1,
+        type=Path,
+        help="the destination directory for compressed files",
+    )
+
+    global args
+    args = parser.parse_args(sys.argv[1:])
+
+    assert isinstance(args.workers, int)  # pyright: ignore[reportAny]
+    assert isinstance(args.interactive, bool)  # pyright: ignore[reportAny]
+    assert isinstance(args.keep, bool)  # pyright: ignore[reportAny]
+    assert isinstance(args.redo, bool)  # pyright: ignore[reportAny]
+    assert isinstance(args.verbose, int)  # pyright: ignore[reportAny]
+    assert isinstance(args.music, list)  # pyright: ignore[reportAny]
+    assert len(args.music) == 1  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
+    assert isinstance(args.music[0], Path)  # pyright: ignore[reportUnknownMemberType]
+    assert isinstance(args.comp, list)  # pyright: ignore[reportAny]
+    assert len(args.comp) == 1  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
+    assert isinstance(args.comp[0], Path)  # pyright: ignore[reportUnknownMemberType]
+
+    src_dir = args.music[0]  # pyright: ignore[reportUnknownMemberType]
+    dst_dir = args.comp[0]  # pyright: ignore[reportUnknownMemberType]
+
+    plan = list(
+        Todo.plan(
+            src_dir,
+            dst_dir,
+            replace=src_dir.samefile(dst_dir) and not args.keep,
+            redo=args.redo,
+        )
+    )
+
+    if len(plan) == 0:
+        print("Nothing to do", file=sys.stderr)
+        sys.exit(0)
+
+    if args.verbose >= 1 or args.interactive:
+        print("Plan:", file=sys.stderr)
+        for todo in plan:
+            print(todo, file=sys.stderr)
+
+    if args.interactive:
+        result = input("Do you want to continue? [Y/n] ")
+        if result.lower() not in ("", "y", "yes"):
+            sys.exit(1)
+
+    with Pool(args.workers) as pool:
+        _ = pool.map(task, plan)

packages/musicomp/src/musicomp/replace.py

@@ -0,0 +1,6 @@
+from os import PathLike
+from pathlib import Path
+
+
+def replace(src: str | PathLike[str]) -> None:
+    Path(src).unlink(missing_ok=True)

packages/musicomp/src/musicomp/todo.py

@@ -0,0 +1,68 @@
+from pathlib import Path
+from os import PathLike
+from enum import StrEnum
+from dataclasses import dataclass
+from typing import Self, override
+from collections.abc import Generator
+from .transcode import transcode as todo_transcode
+from .clean import clean as todo_clean
+from .replace import replace as todo_replace
+
+SRC_SUFFIX = ".flac"
+DST_SUFFIX = ".opus"
+
+
+class TodoAct(StrEnum):
+    TRANSCODE = "transcode"
+    CLEAN = "clean"
+    REPLACE = "replace"
+
+
+@dataclass
+class Todo:
+    act: TodoAct
+    src: str | PathLike[str]
+    dst: str | PathLike[str]
+
+    @classmethod
+    def plan(
+        cls: type[Self],
+        src_dir: str | PathLike[str],
+        dst_dir: str | PathLike[str],
+        replace: bool = False,
+        redo: bool = False,
+    ) -> Generator[Self]:
+        def list_files(dir: str | PathLike[str], suffix: str) -> list[Path]:
+            files: list[Path] = []
+            for f in Path(dir).rglob("*"):
+                if f.is_file() and f.suffix == suffix:
+                    files.append(f)
+            return files
+
+        src_files = list_files(src_dir, SRC_SUFFIX)
+        dst_files = list_files(dst_dir, DST_SUFFIX)
+
+        for f in src_files:
+            e = dst_dir / (f.relative_to(src_dir).with_suffix(DST_SUFFIX))
+            if redo or e not in dst_files:
+                yield cls(TodoAct.TRANSCODE, f, e)
+            if replace:
+                yield cls(TodoAct.REPLACE, f, e)
+
+        for f in dst_files:
+            e = src_dir / (f.relative_to(dst_dir).with_suffix(SRC_SUFFIX))
+            if e not in src_files:
+                yield cls(TodoAct.CLEAN, f, e)
+
+    def run(self) -> None:
+        match self.act:
+            case TodoAct.TRANSCODE:
+                todo_transcode(self.src, self.dst)
+            case TodoAct.CLEAN:
+                todo_clean(self.dst)
+            case TodoAct.REPLACE:
+                todo_replace(self.src)
+
+    @override
+    def __str__(self) -> str:
+        return f"{self.act} {self.src} -> {self.dst}"

packages/musicomp/src/musicomp/transcode.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+from subprocess import run
+from os import PathLike
+
+
+class TranscodingError(Exception):
+    pass
+
+
+def transcode(src: str | PathLike[str], dst: str | PathLike[str]) -> None:
+    dst = Path(dst)
+
+    dst.parent.mkdir(parents=True, exist_ok=True)
+
+    if dst.is_file():
+        dst.unlink()
+
+    opusenc: tuple[str, ...] = (
+        "opusenc",
+        "--quiet",
+        "--bitrate",
+        "96.000",
+        "--music",
+        "--vbr",
+        "--comp",
+        "10",
+        "--",
+        str(src),
+        str(dst),
+    )
+
+    cp = run(opusenc)
+    if cp.returncode != 0:
+        raise TranscodingError(f"opusenc exited with code {cp.returncode}")