Apply formatting

src/machine_admin/boot_image/deb_preseed.py

@@ -6,6 +6,7 @@ from machine_admin.util import run_op
 
 logger = logging.getLogger(__name__)
 
+
 class DebPreseeder:
 
     def __init__(self, iso_path: Path, preseed_path: Path):
@@ -59,16 +60,3 @@ class DebPreseeder:
         self.add_preseed_file()
 
         self.regenerate_checksum()
-
-
-
-
-
-
-
-
-        
-
-
-    
-

src/machine_admin/firewall.py

@@ -4,6 +4,7 @@ from .util import run_op
 
 logger = logging.getLogger(__name__)
 
+
 class UfwInterface:
 
     def __init__(self):
@@ -19,6 +20,7 @@ class UfwInterface:
         logger.info(f"Allowing ufw app: {op}")
         run_op(op)
 
+
 class Firewall:
 
     def __init__(self):

src/machine_admin/machine.py

@@ -3,6 +3,7 @@ from .user import User, UserManager
 from .ssh_config import SshConfig
 from .package_manager import PackageManager
 
+
 class Machine:
 
     def __init__(self, default_user: User):

src/machine_admin/package_manager.py

@@ -5,6 +5,7 @@ from .util import run_op, has_program
 
 logger = logging.getLogger(__name__)
 
+
 class AptInterface:
 
     def __init__(self):
@@ -26,6 +27,7 @@ class AptInterface:
         logger.info(f"Installing packages: {op}")
         run_op(op)
 
+
 class SystemPackage:
 
     def __init__(self, name):

src/machine_admin/platform/distro.py

@@ -1,5 +1,6 @@
 from machine_admin.util import has_program
 
+
 class Distro:
 
     def __init__(self, name):

src/machine_admin/platform/linux.py

@@ -2,5 +2,6 @@ import sys
 
 from .distro import Distro
 
+
 def is_linux():
     return sys.platform == "linux" or sys.platform == "linux2"

src/machine_admin/ssh_config.py

@@ -5,14 +5,17 @@ from .util import run_op
 
 logger = logging.getLogger(__name__)
 
+
 class SshConfig:
 
     def __init__(self):
         self.config_path = Path("/etc/ssh/ssh_config")
-        self.target_values = {"PermitRootLogin": "No",
-                              "PasswordAuthentication": "No",
-                              "ChallengeResponseAuthentication": "No",
-                              "UsePAM": "No"}
+        self.target_values = {
+            "PermitRootLogin": "No",
+            "PasswordAuthentication": "No",
+            "ChallengeResponseAuthentication": "No",
+            "UsePAM": "No",
+        }
 
     def sync_target_values(self):
         logger.info(f"Updating ssh config in: {self.config_path}")
@@ -23,8 +26,7 @@ class SshConfig:
         logger.info(f"Restarting ssh service: {op}")
         run_op(op)
 
-    def copy_ssh_dir_to_user(self, username:str):
+    def copy_ssh_dir_to_user(self, username: str):
         op = f"rsync --archive --chown={username}:{username} ~/.ssh /home/{username}"
         logger.info(f"Copying ssh dir to user: {op}")
         run_op(op)
-                              

src/machine_admin/user.py

@@ -3,11 +3,13 @@ from .util import run_op
 
 logger = logging.getLogger(__name__)
 
+
 class User:
     def __init__(self, name, has_sudo=False):
         self.name = name
         self.has_sudo = has_sudo
 
+
 class UserManager:
 
     def __init__(self):

src/machine_admin/util.py

@@ -5,18 +5,17 @@ logger = logging.getLogger(__name__)
 
 _DRY_RUN = False
 
+
 def set_is_dry_run(is_dry_run: bool):
     _DRY_RUN = is_dry_run
 
-def run_op(op: str, capture_output:bool = False,
-    cwd:str = None):
+
+def run_op(op: str, capture_output: bool = False, cwd: str = None):
     if not _DRY_RUN:
         if capture_output:
-            ret = subprocess.run(op, 
-                shell=True,
-                capture_output = True, 
-                text = True,
-                cwd=cwd)
+            ret = subprocess.run(
+                op, shell=True, capture_output=True, text=True, cwd=cwd
+            )
             ret.check_returncode()
             return ret.stdout
         else:
@@ -25,10 +24,8 @@ def run_op(op: str, capture_output:bool = False,
     else:
         logger.info(f"Dry Run | {op}")
 
+
 def has_program(program_name: str):
     op = f"which {program_name}"
-    ret = subprocess.run(op, 
-        shell=True,
-        capture_output = True, 
-        text = True)
+    ret = subprocess.run(op, shell=True, capture_output=True, text=True)
     return len(ret.stdout)

src/machine_setup.py

@@ -10,14 +10,15 @@ logger = logging.getLogger(__name__)
 if __name__ == "__main__":
 
     parser = argparse.ArgumentParser(
-                        prog='MachineSetup',
-                        description='Scripts for machine provisioning')
+        prog="MachineSetup", description="Scripts for machine provisioning"
+    )
 
-    parser.add_argument('username', 
-        help="Name of the default non-root user")
-    parser.add_argument('--dry_run', 
+    parser.add_argument("username", help="Name of the default non-root user")
+    parser.add_argument(
+        "--dry_run",
         help="If set then don't change the system state - used for testing.",
-        default = False)
+        default=False,
+    )
 
     args = parser.parse_args()
     set_is_dry_run(args.dry_run)
@@ -28,4 +29,3 @@ if __name__ == "__main__":
     user = User(args.username, has_sudo=True)
     machine = Machine(user)
     machine.setup()
-

src/music/convert.py

@@ -3,12 +3,14 @@ import os
 import logging
 import shutil
 from pathlib import Path
+from typing import NamedTuple
 import subprocess
 import uuid
 from multiprocessing import Pool
 
 logger = logging.getLogger(__name__)
 
+
 def convert_audio_file(args):
     cmd, output_tmp, output_path = args
 
@@ -16,46 +18,59 @@ def convert_audio_file(args):
     os.makedirs(output_path.parent, exist_ok=True)
     shutil.move(output_tmp, output_path)
 
-class AudioFileConverter():
 
-    def __init__(self, input_dir: Path, output_dir: Path):
-        self.input_dir = input_dir
-        self.output_dir = output_dir
-        self.output_fmt = "mp3"
-        self.input_fmt = "flac"
+class ConversionConfig(NamedTuple):
+    input_dir: Path
+    output_dir: Path
+    output_fmt: str
+    input_fmt: str
 
-    def get_converted_path(self, input_path: Path):
-        relative_path = input_path.relative_to(self.input_dir)
-        output_filename = str(input_path.stem) + f".{self.output_fmt}"
-        output_filename.replace("'", "")
-        output_path = self.output_dir / relative_path.parent / output_filename
-        return output_path
 
-    def convert(self):
+def _get_converted_path(input_path: Path, config: ConversionConfig):
+    """
+    Return the path you would obtain if you moved the file in the input
+    path to the output directory in the config and changed its extension.
+    """
 
-        logger.info("Converting files in %s", self.input_dir)
-        logger.info("Writing output to: %s", self.output_dir)
+    relative_path = input_path.relative_to(config.input_dir)
+    output_filename = str(input_path.stem) + f".{config.output_fmt}"
+    output_filename.replace("'", "")
+    output_path = config.output_dir / relative_path.parent / output_filename
+    return output_path
 
-        os.makedirs(self.output_dir, exist_ok=True)
 
-        input_files = list(self.input_dir.resolve().rglob(f"*.{self.input_fmt}"))
-        output_files = []
-        for input_file in input_files:
-            candidate_output = self.get_converted_path(input_file)
-            if not candidate_output.exists():
-                output_files.append(candidate_output)
+def get_files_recursive(search_path: Path, extension: str):
+    return list(search_path.rglob(f"*.{extension}"))
 
-        tasks = []
-        for idx, (input_path, output_path) in enumerate(zip(input_files, output_files)):
 
-            logger.info("Converting file %d of %d", idx, len(input_files))
+def convert(config: ConversionConfig):
 
-            output_tmp = self.output_dir / (str(uuid.uuid4()) + f".{self.output_fmt}")
-            cmd = f"ffmpeg -i '{input_path}' -ab 320k -map_metadata 0 -id3v2_version 3 '{output_tmp}'"
-            tasks.append((cmd, output_tmp, output_path))
+    logger.info("Converting files in %s", config.input_dir)
+    logger.info("Writing output to: %s", config.output_dir)
 
-        with Pool(10) as p:
-            p.map(convert_audio_file, tasks)
+    os.makedirs(config.output_dir, exist_ok=True)
+
+    input_files = get_files_recursive(
+        config.input_dir.resolve(), config.input_fmt
+    )
+    output_files = []
+    for input_file in input_files:
+        candidate_output = _get_converted_path(input_file)
+        if not candidate_output.exists():
+            output_files.append(candidate_output)
+    print(output_files)
+
+    tasks = []
+    for idx, (input_path, output_path) in enumerate(zip(input_files, output_files)):
+
+        logger.info("Converting file %d of %d", idx, len(input_files))
+
+        output_tmp = self.output_dir / (str(uuid.uuid4()) + f".{self.output_fmt}")
+        cmd = f"ffmpeg -i '{input_path}' -ab 320k -map_metadata 0 -id3v2_version 3 '{output_tmp}'"
+        tasks.append((cmd, output_tmp, output_path))
+
+    with Pool(10) as p:
+        p.map(convert_audio_file, tasks)
 
 
 def move(input_dir: Path, output_dir: Path):
@@ -71,19 +86,21 @@ def move(input_dir: Path, output_dir: Path):
         os.makedirs(output_path.parent, exist_ok=True)
         shutil.move(path, output_path)
 
+
 if __name__ == "__main__":
 
     parser = argparse.ArgumentParser()
 
-    parser.add_argument('--input_dir',
-                        type=Path,
-                        default=Path(),
-        help="Directory with input files for conversion.")
+    parser.add_argument(
+        "--input_dir",
+        type=Path,
+        default=Path(),
+        help="Directory with input files for conversion.",
+    )
 
-    parser.add_argument('--output_dir',
-                        type=Path,
-                        default=Path(),
-                        help="Directory for converted files")
+    parser.add_argument(
+        "--output_dir", type=Path, default=Path(), help="Directory for converted files"
+    )
 
     args = parser.parse_args()
 
@@ -92,6 +109,4 @@ if __name__ == "__main__":
 
     converter = AudioFileConverter(args.input_dir.resolve(), args.output_dir.resolve())
     converter.convert()
-    #move(args.input_dir.resolve(), args.output_dir.resolve())
-  
-
+    # move(args.input_dir.resolve(), args.output_dir.resolve())

src/network/sync.py

@@ -28,37 +28,43 @@ def sync(source_dir: Path, target_dir: Path, host: str):
     source_files = []
     for dir_path, _, files in os.walk(source_dir):
         for each_file in files:
-            source_files.append(Path(dir_path + "/" + each_file).relative_to(source_dir))
+            source_files.append(
+                Path(dir_path + "/" + each_file).relative_to(source_dir)
+            )
 
     sync_files = []
     for source_file in source_files:
         if source_file not in target_files:
             sync_files.append(source_file)
 
-    sync_dir = Path(os.getcwd())/"sync"
+    sync_dir = Path(os.getcwd()) / "sync"
     os.makedirs(sync_dir, exist_ok=True)
     for sync_file in sync_files:
-        os.makedirs((sync_dir/sync_file).parent, exist_ok=True)
+        os.makedirs((sync_dir / sync_file).parent, exist_ok=True)
         shutil.copy(source_dir / sync_file, sync_dir / sync_file)
 
+
 if __name__ == "__main__":
 
     parser = argparse.ArgumentParser()
 
-    parser.add_argument('--source_dir',
-                        type=Path,
-                        default=Path(),
-        help="Directory with source files to sync from.")
+    parser.add_argument(
+        "--source_dir",
+        type=Path,
+        default=Path(),
+        help="Directory with source files to sync from.",
+    )
 
-    parser.add_argument('--host',
-                        type=str,
-                         default = "",
-                        help="Name of host system to sync with.")
+    parser.add_argument(
+        "--host", type=str, default="", help="Name of host system to sync with."
+    )
 
-    parser.add_argument('--target_dir',
-                        type=Path,
-                        default=Path(),
-                        help="Directory with source files to sync to.")
+    parser.add_argument(
+        "--target_dir",
+        type=Path,
+        default=Path(),
+        help="Directory with source files to sync to.",
+    )
 
     args = parser.parse_args()
 
@@ -66,5 +72,3 @@ if __name__ == "__main__":
     logging.getLogger().setLevel(logging.INFO)
 
     sync(args.source_dir.resolve(), args.target_dir.resolve(), args.host)
- 
-