change how commands work

musicutil/__main__.py

@@ -2,82 +2,23 @@
 
 import argparse
 
-from .commands.process_command import ProcessCommand
+from .commands.process_command import ProcessCommand, add_process_command, get_process_args
-from .commands.copy_command import CopyCommand
+from .commands.copy_command import CopyCommand, add_copy_command, get_copy_args
-from .commands.transcode_command import TranscodeCommand
+from .commands.transcode_command import TranscodeCommand, add_transcode_command, get_transcode_args
 
 parser = argparse.ArgumentParser(
     description="chaos's musicutil")
 subparsers = parser.add_subparsers(dest="subparser_name")
 
-process_parser = subparsers.add_parser('process')
+add_copy_command(subparsers)
-process_parser.add_argument(
+add_process_command(subparsers)
-    'src',
+add_transcode_command(subparsers)
-    type=str,
-    help='src base music directory')
-process_parser.add_argument(
-    '--dry-run', action='store_true')
-
-copy_parser = subparsers.add_parser('copy')
-copy_parser.add_argument(
-    'src',
-    type=str,
-    help='src base music directory')
-copy_parser.add_argument(
-    'dest',
-    type=str,
-    help='dest music directory')
-copy_parser.add_argument(
-    '--transcode-level',
-    type=str,
-    help='transcode level',
-    default="copy")
-copy_parser.add_argument(
-    '--skip-existing',
-    action='store_true')
-copy_parser.add_argument(
-    '--single-directory',
-    action='store_true')
-
-transcode_parser = subparsers.add_parser('transcode')
-transcode_parser.add_argument(
-    'src',
-    type=str,
-    help='src base music directory')
-transcode_parser.add_argument(
-    'dest',
-    type=str,
-    help='dest music directory')
-transcode_parser.add_argument(
-    '--transcode-level',
-    type=str,
-    help='transcode level',
-    default="opus-96k")
-transcode_parser.add_argument(
-    '--ignore-extension',
-    action='store_true')
-transcode_parser.add_argument(
-    '--custom-encoder-config',
-    type=str,
-    help='custom encoder config')
 
 args = parser.parse_args()
 
 if args.subparser_name == "process":
-    ProcessCommand(args.src, args.dry_run).run()
+    ProcessCommand(get_process_args(args)).run()
 elif args.subparser_name == "copy":
-    CopyCommand(
+    CopyCommand(get_copy_args(args)).run()
-        args.src,
-        args.dest,
-        args.transcode_level,
-        args.single_directory,
-        args.skip_existing
-    ).run()
 elif args.subparser_name == "transcode":
-    TranscodeCommand(
+    TranscodeCommand(get_transcode_args(args)).run()
-        args.src,
-        args.dest,
-        args.transcode_level,
-        args.ignore_extension,
-        args.custom_encoder_config
-    ).run()

musicutil/commands/copy_command.py

@@ -1,8 +1,8 @@
 from ..types import File
 from ..utils.scan_for_music import scan_for_music
 from ..utils.load_tag_information import load_tag_information
-from ..transcode_levels import print_transcode_levels, transcode_levels_list
+from ..transcode_presets import print_transcode_presets, transcode_presets_list
-from ..utils.transcoder import transcode, get_transcode_config
+from ..utils.transcoder import TranscodeConfig, transcode, get_transcode_config
 
 from os import makedirs as make_directories
 from os.path import exists as path_exists
@@ -14,38 +14,72 @@ class CopyCommandState:
     files: list[File] = []
     transcoded_files: list[File] = []
 
+class CopyCommandArgs():
+    src: str
+    dest: str
+    transcode_preset: str
+    custom_transcoder_config: str
+    single_directory: bool
+    skip_existing: bool
+
+def add_copy_command(subparsers):
+    copy_parser = subparsers.add_parser('copy')
+    copy_parser.add_argument(
+        'src',
+        type=str,
+        help='src base music directory')
+    copy_parser.add_argument(
+        'dest',
+        type=str,
+        help='dest music directory')
+    copy_parser.add_argument(
+        '--transcode-preset',
+        type=str,
+        help='transcode preset (copy = no transcode)',
+        default="copy")
+    copy_parser.add_argument(
+        '--custom-transcoder-config',
+        type=str,
+        help='custom transcoder config')
+    copy_parser.add_argument(
+        '--skip-existing',
+        action='store_true')
+    copy_parser.add_argument(
+        '--single-directory',
+        action='store_true')
+
+def get_copy_args(args) -> CopyCommandArgs:
+    command_args = CopyCommandArgs()
+    command_args.src = args.src
+    command_args.dest = args.dest
+    command_args.transcode_preset = args.transcode_preset
+    command_args.custom_transcoder_config = args.custom_transcoder_config
+    command_args.single_directory = args.single_directory
+    command_args.skip_existing = args.skip_existing
+    return command_args
+
 class CopyCommand():
-    def __init__(self,
+    def __init__(self, args: CopyCommandArgs):
-                 src: str,
+        self.args = args
-                 dest: str,
-                 transcode_level: str,
-                 single_directory: bool,
-                 skip_existing: bool
-                 ):
-        self.src = src
-        self.dest = dest
-        self.transcode_level = transcode_level
-        self.single_directory = single_directory
-        self.skip_existing = skip_existing
         self.state = CopyCommandState()
 
     def run(self):
-        if self.transcode_level == "list":
+        if self.args.transcode_preset == "list":
-            print_transcode_levels()
+            print_transcode_presets()
             exit()
 
         print("Copying")
         self.scan_for_music()
         self.load_tag_information()
-        if self.single_directory:
+        if self.args.single_directory:
             self.check_for_collisions()
         self.transcode_files()
-        if self.single_directory:
+        if self.args.single_directory:
             self.create_mappings()
 
     def scan_for_music(self):
         print("Scanning For Music")
-        self.state.files = scan_for_music(self.src)
+        self.state.files = scan_for_music(self.args.src)
 
     def load_tag_information(self):
         print("Loading Tag Information")
@@ -74,12 +108,12 @@ class CopyCommand():
     def _transcode_copy(self, file: File):
         src = file.join_path_to()
         dest = file.join_filename(
-        ) if self.single_directory else file.join_path_from_src()
+        ) if self.args.single_directory else file.join_path_from_src()
-        dest = self.dest + "/" + dest
+        dest = self.args.dest + "/" + dest
 
         exists = path_exists(dest)
 
-        if (self.skip_existing and not exists) or not self.skip_existing:
+        if (self.args.skip_existing and not exists) or not self.args.skip_existing:
             print("Copying", src, "to", dest)
             copy_file(
                 src,
@@ -94,19 +128,17 @@ class CopyCommand():
 
         self.state.transcoded_files.append(file)
 
-    def _transcode_with_level(self, file: File, level: str):
+    def _transcode_with_config(self, file: File, trans_config: TranscodeConfig):
-        trans_config = get_transcode_config(file, level)
-
         src = file.join_path_to()
 
         new_file = deep_copy(file)
         new_file.extension = trans_config.file_extension
 
         dest_filepath = new_file.join_filename(
-        ) if self.single_directory else new_file.join_path_from_src()
+        ) if self.args.single_directory else new_file.join_path_from_src()
-        dest_filepath = self.dest + "/" + dest_filepath
+        dest_filepath = self.args.dest + "/" + dest_filepath
 
-        if (self.skip_existing and path_exists(dest_filepath)):
+        if (self.args.skip_existing and path_exists(dest_filepath)):
             print("Skipping transcoding", dest_filepath)
             self.state.transcoded_files.append(new_file)
             return
@@ -120,22 +152,35 @@ class CopyCommand():
     def transcode_files(self):
         print("Transcoding Files")
 
-        if not self.single_directory:
+        if not self.args.single_directory:
             directories = set()
             for file in self.state.files:
                 directories.add(file.path_from_src)
             for dir in directories:
                 make_directories(
-                    self.dest + "/" + dir, exist_ok=True)
+                    self.args.dest + "/" + dir, exist_ok=True)
 
-        if self.transcode_level == "copy":
+        is_transcode_config_set = self.args.custom_transcoder_config is not None
+
+        if self.args.transcode_preset == "copy" and not is_transcode_config_set:
             for file in self.state.files:
                 self._transcode_copy(file)
             return
-        elif self.transcode_level in transcode_levels_list:
+        else:
+            global trans_config
+            trans_config = None
+            if is_transcode_config_set:
+                with open(self.args.custom_transcoder_config, "r+") as f:
+                    trans_config = TranscodeConfig()
+                    trans_config.load_from_file(f)
+            else:
+                trans_config = get_transcode_config(self.args.transcode_preset)
+
+            print(trans_config)
+
             for file in self.state.files:
-                self._transcode_with_level(
+                self._transcode_with_config(
-                    file, self.transcode_level)
+                    file, trans_config)
 
     def create_mappings(self):
         with open(self.dest + "/" + "mappings.txt", "w") as f:

musicutil/commands/process_command.py

@@ -1,4 +1,4 @@
-from ..types import File, Tags
+from ..types import File
 from ..utils.scan_for_music import scan_for_music
 from ..utils.load_tag_information import load_tag_information
 from ..utils.substitutions import reduce_to_ascii_and_substitute
@@ -10,11 +10,28 @@ from os import rename as rename_file
 class ProcessCommandState:
     files: list[File] = []
 
+class ProcessCommandArgs:
+    src: str
+    dry_run: bool
+
+def add_process_command(subparsers):
+    process_parser = subparsers.add_parser('process')
+    process_parser.add_argument(
+        'src',
+        type=str,
+        help='src base music directory')
+    process_parser.add_argument(
+        '--dry-run', action='store_true')
+
+def get_process_args(args) -> ProcessCommandArgs:
+    command_args = ProcessCommandArgs()
+    command_args.src = args.src
+    command_args.dry_run = args.dry_run
+    return command_args
 
 class ProcessCommand():
-    def __init__(self, src: str, dry_run: bool):
+    def __init__(self, args: ProcessCommandArgs):
-        self.src = src
+        self.args = args
-        self.dry_run = dry_run
         self.state = ProcessCommandState()
 
     def run(self):
@@ -25,7 +42,7 @@ class ProcessCommand():
 
     def scan_for_music(self):
         print("Scanning For Music")
-        self.state.files = scan_for_music(self.src)
+        self.state.files = scan_for_music(self.args.src)
 
     def load_tag_information(self):
         print("Loading Tag Information")
@@ -49,7 +66,7 @@ class ProcessCommand():
         new_file = deep_copy(file)
         new_file.filename = proper_filename
 
-        if not self.dry_run:
+        if not self.args.dry_run:
             rename_file(
                 file.join_path_to(),
                 new_file.join_path_to())

musicutil/commands/transcode_command.py

@@ -1,49 +1,76 @@
 from ..utils.transcoder import get_transcode_config, transcode, TranscodeConfig
 from ..utils.scan_for_music import file_from_path
-from ..transcode_levels import print_transcode_levels, transcode_levels
+from ..transcode_presets import print_transcode_presets, transcode_presets
 
 from pathlib import Path
 from json import load as load_json_file
 
+class TranscodeCommandArgs:
+    src: str
+    dest: str
+    transcode_preset: str
+    ignore_extension: bool
+    custom_transcoder_config_path: str
+
+def add_transcode_command(subparsers):
+    transcode_parser = subparsers.add_parser('transcode')
+    transcode_parser.add_argument(
+        'src',
+        type=str,
+        help='src base music directory')
+    transcode_parser.add_argument(
+        'dest',
+        type=str,
+        help='dest music directory')
+    transcode_parser.add_argument(
+        '--transcode-preset',
+        type=str,
+        help='transcode preset',
+        default="opus-96k")
+    transcode_parser.add_argument(
+        '--ignore-extension',
+        action='store_true')
+    transcode_parser.add_argument(
+        '--custom-transcoder-config',
+        type=str,
+        help='custom transcoder config') 
+
+def get_transcode_args(args) -> TranscodeCommandArgs:
+    command_args = TranscodeCommandArgs()
+    command_args.src = args.src
+    command_args.dest = args.dest
+    command_args.transcode_preset = args.transcode_preset
+    command_args.ignore_extension = args.ignore_extension
+    command_args.custom_transcoder_config_path = args.custom_transcoder_config
+    return command_args
+
 class TranscodeCommand:
-    def __init__(
+    def __init__(self, args: TranscodeCommandArgs):
-        self,
+        self.args = args
-        src: str, 
-        dest: str, 
-        transcode_level: str,
-        ignore_extension: bool,
-        custom_encoder_config_path: str,
-    ):
-        self.src = src
-        self.dest = dest
-        self.transcode_level = transcode_level
-        self.ignore_extension = ignore_extension
-        self.custom_encoder_config_path = custom_encoder_config_path
 
     def run(self):
-        if self.transcode_level == "list":
+        if self.args.transcode_preset == "list":
-            print_transcode_levels()
+            print_transcode_presets()
             exit()
 
         print("Transcoding...")
         input_file = file_from_path(Path(self.src), "")
         
-        if self.custom_encoder_config_path is None or len(self.custom_encoder_config_path) == 0:
+        if self.args.custom_transcoder_config_path is None or len(self.args.custom_transcoder_config_path) == 0:
-            trans_config = get_transcode_config(input_file, self.transcode_level)
+            trans_config = get_transcode_config(input_file, self.args.transcode_preset)
         else:
-            with open(self.custom_encoder_config_path, "r+") as f: 
+            with open(self.args.custom_transcoder_config_path, "r+") as file: 
-                trans_config_dict = load_json_file(f)
                 trans_config = TranscodeConfig()
-                trans_config.load_from_dict(trans_config_dict)
+                trans_config.load_from_file(file)
 
-        output_file = file_from_path(Path(self.dest), "")
+        output_file = file_from_path(Path(self.args.dest), "")
 
-        if trans_config.file_extension != output_file.extension and not self.ignore_extension:
+        if trans_config.file_extension != output_file.extension and not self.args.ignore_extension:
             print(
                 f"{output_file.extension} is not the recommended "+
-                f"extension for transcode_level {self.transcode_level} "+
+                f"extension for transcode config "+
                 f"please change it to {trans_config.file_extension} "+
                 f"or run with --ignore-extension"
             )
             exit()
-        transcode(input_file, trans_config, self.dest)
+        transcode(input_file, trans_config, self.args.dest)

musicutil/meta.py

@@ -10,4 +10,5 @@ substitutions = {
 
 # Patch to whatever path ffmpeg is at
 ffmpeg_path = "ffmpeg"
-ffprobe_path = "ffprobe"
+ffprobe_path = "ffprobe"
+r128gain_path = "r128gain"

musicutil/transcode_levels.py

@@ -1,83 +0,0 @@
-from functools import reduce 
-
-def add_to_arr(arr: list[str], items: list[str]) -> list[str]:
-    for item in items:
-        arr.append(item)
-
-# does not include copy
-transcode_levels = {}
-
-transcode_levels["mp3"] = []
-
-# mp3 v0 -> v9
-add_to_arr(transcode_levels["mp3"], [
-    f"mp3-v{quality}" for quality in range(0, 10)
-])
-# mp3 bitrates
-mp3_bitrates = [8, 16, 24, 32, 40, 48, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320]
-add_to_arr(transcode_levels["mp3"], [
-    f"mp3-{bitrate}k" for bitrate in mp3_bitrates
-])
-
-transcode_levels["opus"] = []
-
-opus_bitrates = ["16", "24", "32", "64", "96", "128", "256"]  
-add_to_arr(transcode_levels["opus"], [
-    f"opus-{bitrate}k" for bitrate in opus_bitrates
-])
-
-transcode_levels["vorbis"] = []
-
-add_to_arr(transcode_levels["vorbis"], [
-    f"vorbis-q{quality}" for quality in range(-2, 11)
-])
-
-transcode_levels["speex"] = []
-
-add_to_arr(transcode_levels["speex"], [
-    f"speex-q{quality}" for quality in range(0, 11)
-])
- 
-transcode_levels["g726"] = []
-
-g726_bitrates = ["16", "24", "32", "40"]  
-add_to_arr(transcode_levels["g726"], [
-    f"g726-{bitrate}k" for bitrate in g726_bitrates
-])
-
-# Extra Default Mappings
-preset_transcode_levels = {}
-
-mp3_presets = {
-    "mp3-low": "mp3-v4",
-    "mp3-medium": "mp3-v2",
-    "mp3-high": "mp3-v0",
-}
-
-preset_transcode_levels = preset_transcode_levels | mp3_presets
-add_to_arr(transcode_levels["opus"], mp3_presets.keys())
-
-opus_presets = {
-    "opus-low": "opus-32k",
-    "opus-medium": "opus-64k",
-    "opus-high": "opus-96k",
-    "opus-higher": "opus-128k",
-    "opus-extreme": "opus-256k"
-}
-
-preset_transcode_levels = preset_transcode_levels | opus_presets
-add_to_arr(transcode_levels["opus"], opus_presets.keys())
-
-transcode_levels["mp3"].sort()
-transcode_levels["opus"].sort()
-transcode_levels["vorbis"].sort()
-transcode_levels["speex"].sort()
-transcode_levels["g726"].sort()
-
-def print_transcode_levels():
-    for category in transcode_levels.keys():
-        print(f"Category {category}:")
-        for level in transcode_levels[category]:
-            print(f"- {level}")
-
-transcode_levels_list = reduce(lambda a, b: a+b, transcode_levels.values())

musicutil/transcode_presets.py

@@ -0,0 +1,83 @@
+from functools import reduce 
+
+def add_to_arr(arr: list[str], items: list[str]) -> list[str]:
+    for item in items:
+        arr.append(item)
+
+# does not include copy
+transcode_presets = {}
+
+transcode_presets["mp3"] = []
+
+# mp3 v0 -> v9
+add_to_arr(transcode_presets["mp3"], [
+    f"mp3-v{quality}" for quality in range(0, 10)
+])
+# mp3 bitrates
+mp3_bitrates = [8, 16, 24, 32, 40, 48, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320]
+add_to_arr(transcode_presets["mp3"], [
+    f"mp3-{bitrate}k" for bitrate in mp3_bitrates
+])
+
+transcode_presets["opus"] = []
+
+opus_bitrates = ["16", "24", "32", "64", "96", "128", "256"]  
+add_to_arr(transcode_presets["opus"], [
+    f"opus-{bitrate}k" for bitrate in opus_bitrates
+])
+
+transcode_presets["vorbis"] = []
+
+add_to_arr(transcode_presets["vorbis"], [
+    f"vorbis-q{quality}" for quality in range(-2, 11)
+])
+
+transcode_presets["speex"] = []
+
+add_to_arr(transcode_presets["speex"], [
+    f"speex-q{quality}" for quality in range(0, 11)
+])
+ 
+transcode_presets["g726"] = []
+
+g726_bitrates = ["16", "24", "32", "40"]  
+add_to_arr(transcode_presets["g726"], [
+    f"g726-{bitrate}k" for bitrate in g726_bitrates
+])
+
+# Extra Default Mappings
+preset_transcode_presets = {}
+
+mp3_presets = {
+    "mp3-low": "mp3-v4",
+    "mp3-medium": "mp3-v2",
+    "mp3-high": "mp3-v0",
+}
+
+preset_transcode_presets = preset_transcode_presets | mp3_presets
+add_to_arr(transcode_presets["opus"], mp3_presets.keys())
+
+opus_presets = {
+    "opus-low": "opus-32k",
+    "opus-medium": "opus-64k",
+    "opus-high": "opus-96k",
+    "opus-higher": "opus-128k",
+    "opus-extreme": "opus-256k"
+}
+
+preset_transcode_presets = preset_transcode_presets | opus_presets
+add_to_arr(transcode_presets["opus"], opus_presets.keys())
+
+transcode_presets["mp3"].sort()
+transcode_presets["opus"].sort()
+transcode_presets["vorbis"].sort()
+transcode_presets["speex"].sort()
+transcode_presets["g726"].sort()
+
+def print_transcode_presets():
+    for category in transcode_presets.keys():
+        print(f"Category {category}:")
+        for preset in transcode_presets[category]:
+            print(f"- {preset}")
+
+transcode_presets_list = reduce(lambda a, b: a+b, transcode_presets.values())

musicutil/utils/do_replaygain.py

@@ -0,0 +1,13 @@
+from ..types import File
+from ..meta import r128gain_path, ffmpeg_path
+
+from subprocess import run as run_command
+
+
+def do_replaygain(file: File):
+    run_command([
+        r128gain_path,
+        "-f", ffmpeg_path,
+        "-s",
+        file.join_path_to()
+    ])

musicutil/utils/transcoder.py

@@ -1,7 +1,9 @@
-from ..transcode_levels import transcode_levels, preset_transcode_levels
+from ..transcode_presets import preset_transcode_presets
 from ..types import File
 from ..meta import ffmpeg_path
 
+from yaml import load as load_yaml_file
+from yaml import Loader as YamlLoader
 from subprocess import run as run_command
 
 class TranscodeConfig:
@@ -15,6 +17,9 @@ class TranscodeConfig:
     sample_rate = ""
     channels = ""
 
+    def load_from_file(self, file):
+        self.load_from_dict(load_yaml_file(file, Loader=YamlLoader))
+
     def load_from_dict(self, data):
         if "use_quality" in data:
             self.use_quality = data["use_quality"]
@@ -36,12 +41,12 @@ class TranscodeConfig:
             self.channels = data["channels"]
         return self
 
-def get_transcode_config(file: File, level: str):
+def get_transcode_config(preset: str):
     conf = TranscodeConfig()
-    if level in preset_transcode_levels.keys():
+    if preset in preset_transcode_presets.keys():
-        level = preset_transcode_levels["level"]
+        preset = preset_transcode_presets[preset]
 
-    if level.startswith("g726-") and level.endswith("k"):
+    if preset.startswith("g726-") and preset.endswith("k"):
         conf.load_from_dict({
             "container": "matroska",
             "file_extension": "mka",
@@ -49,53 +54,53 @@ def get_transcode_config(file: File, level: str):
             "sample_rate": "8000",
             "channels": "1",
             "use_bitrate": True,
-            "bitrate": level.replace("g726-", "") 
+            "bitrate": preset.replace("g726-", "") 
         })
         return conf
 
-    if level.startswith("opus-") and level.endswith("k"):
+    if preset.startswith("opus-") and preset.endswith("k"):
         conf.load_from_dict({
             "container": "ogg",
             "file_extension": "opus",
             "encoder": "libopus",
             "use_bitrate": True,
-            "bitrate": level.replace("opus-", "") 
+            "bitrate": preset.replace("opus-", "") 
         })
         return conf
 
-    if level.startswith("mp3-"):
+    if preset.startswith("mp3-"):
         conf.load_from_dict({
             "container": "mp3",
             "file_extension": "mp3",
             "encoder": "libmp3lame",
         })
 
-        if level.startswith("mp3-v"):
+        if preset.startswith("mp3-v"):
             conf.use_quality = True
-            conf.quality = level.replace("mp3-v", "")
+            conf.quality = preset.replace("mp3-v", "")
             return conf
-        elif level.startswith("mp3-") and level.endswith("k"):
+        elif preset.startswith("mp3-") and preset.endswith("k"):
             conf.use_bitrate = True
-            conf.bitrate = level.replace("mp3-", "") 
+            conf.bitrate = preset.replace("mp3-", "") 
             return conf    
 
-    if level.startswith("vorbis-q"):
+    if preset.startswith("vorbis-q"):
         conf.load_from_dict({
             "container": "ogg",
             "file_extension": "ogg",
             "encoder": "libvorbis",
             "use_quality": True,
-            "quality": level.replace("vorbis-q", ""),
+            "quality": preset.replace("vorbis-q", ""),
         })
         return conf
 
-    if level.startswith("speex-q"):
+    if preset.startswith("speex-q"):
         conf.load_from_dict({
             "container": "ogg",
             "file_extension": "ogg",
             "encoder": "libspeex",
             "use_quality": True,
-            "quality": level.replace("speex-q", ""),
+            "quality": preset.replace("speex-q", ""),
         })
         return conf
 

shell.nix

@@ -2,6 +2,6 @@
 let
   fold-to-ascii = (py: py.callPackage ./nix-extra-deps/fold-to-ascii.nix { });
   my_python = pkgs.python39.withPackages
-    (py: with py; [ py.mutagen (fold-to-ascii py) py.autopep8 ]);
+    (py: with py; [ py.mutagen (fold-to-ascii py) py.autopep8 py.pyyaml ]);
 
-in pkgs.mkShell { packages = with pkgs; [ my_python ffmpeg fd ]; }
+in pkgs.mkShell { packages = with pkgs; [ my_python ffmpeg fd r128gain ]; }