diff --git a/pyproject.toml b/pyproject.toml index f361a18..5cbb9d3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ requires-python = ">=3.10" dependencies = [ "toml", "alive-progress", + "types-toml", ] authors = [ {name = "Alexander Wainwright", email = "code@figtree.dev"}, @@ -35,7 +36,7 @@ line-length = 80 select = [ "B", "W", - # "ANN", + "ANN", "FIX", "S", "F", # Pyflakes rules diff --git a/src/emulsion/config.py b/src/emulsion/config.py index a0c49de..a7130bf 100644 --- a/src/emulsion/config.py +++ b/src/emulsion/config.py @@ -1,12 +1,13 @@ import copy import os from pathlib import Path +from typing import Any import toml # Default Schema # This defines the "First Class" feel of the app, but is fully overridable. -DEFAULT_CONFIG = { +DEFAULT_CONFIG: dict[str, Any] = { "sidecar": { "extension": ".xmp" }, @@ -55,19 +56,19 @@ DEFAULT_CONFIG = { } -def get_config_path(): +def get_config_path() -> Path: xdg_config_home = os.environ.get( - 'XDG_CONFIG_HOME', Path('~/.config').expanduser() + 'XDG_CONFIG_HOME', str(Path('~/.config').expanduser()) ) return Path(xdg_config_home) / 'emulsion' / 'config.toml' class ConfigLoader: - def __init__(self, path=None): + def __init__(self, path: Path | None = None) -> None: self.path: Path = path or get_config_path() - self.config = copy.deepcopy(DEFAULT_CONFIG) + self.config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG) - def load(self): + def load(self) -> dict[str, Any]: """ Loads the config from disk and merges it into the defaults. Returns the full config dictionary. @@ -85,7 +86,7 @@ class ConfigLoader: return self.config - def _merge(self, base, update): + def _merge(self, base: dict[str, Any], update: dict[str, Any]) -> None: """ Recursively merges 'update' dict into 'base' dict. """ @@ -99,7 +100,7 @@ class ConfigLoader: else: base[key] = value - def save_defaults(self, current_defaults): + def save_defaults(self, current_defaults: dict[str, Any]) -> bool: """ Helpers to write a new config file (for --init-config). This is a bit tricky because we don't want to just dump the massive DEFAULT_CONFIG. We @@ -111,13 +112,13 @@ class ConfigLoader: # doesn't do that. # Ensure directory exists - self.path.parent.mkdir(parents=True) + self.path.parent.mkdir(parents=True, exist_ok=True) # We probably only want to write if it doesn't exist, to avoid # clobbering. - if self.path.exists: + if self.path.exists(): return False - with self.path.open(encoding='utf-8') as f: + with self.path.open('w', encoding='utf-8') as f: toml.dump(current_defaults, f) return True diff --git a/src/emulsion/executor.py b/src/emulsion/executor.py index 468ca1d..945b091 100644 --- a/src/emulsion/executor.py +++ b/src/emulsion/executor.py @@ -1,21 +1,28 @@ import datetime import shlex import subprocess +from argparse import Namespace from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path +from typing import Any from alive_progress import alive_bar class Executor: - def __init__(self, config): + def __init__(self, config: dict[str, Any]) -> None: self.config = config - self.mappings = config.get('mappings', {}) - self.sidecar_ext = config.get('sidecar', {}).get('extension', '.xmp') + self.mappings: dict[str, Any] = config.get('mappings', {}) + self.sidecar_ext: str = config.get( + 'sidecar', {}).get('extension', '.xmp') if not self.sidecar_ext.startswith('.'): self.sidecar_ext = f".{self.sidecar_ext}" - def run_batch(self, files, resolved_values, options): + def run_batch( + self, files: list[str], + resolved_values: dict[str, str], + options: Namespace + ) -> None: """ Main execution entry point. files: List of file paths @@ -23,11 +30,11 @@ class Executor: options: Dictionary/Namespace of process options (dry_run, workers, base_date, etc.) """ - extensions = ['.jpg', '.jpeg', '.tif', '.tiff'] # Filter supported files + extensions = ['.jpg', '.jpeg', '.tif', '.tiff'] valid_files = [ f for f in files - if Path(f).suffix in extensions + if Path(f).suffix.lower() in extensions ] if not valid_files: @@ -55,7 +62,7 @@ class Executor: print(f"Processing {total_files} file(s)...") # Prepare tasks - tasks = [] + tasks: list[tuple[list[str], str, str]] = [] for i, f in enumerate(valid_files): # Calculate timestamp ts_dt = base_dt + datetime.timedelta(seconds=i * time_increment) @@ -63,7 +70,7 @@ class Executor: # Determine file targets (Sidecar logic) target_path, sidecar_source = self._determine_paths( - f, options.embed + Path(f), options.embed ) # Build Command @@ -98,14 +105,16 @@ class Executor: bar() - def _determine_paths(self, original_file, embed): + def _determine_paths( + self, original_file: Path, embed: bool + ) -> tuple[Path, Path | None]: """ Returns (target_path, sidecar_source_if_needed) """ if embed: - return original_file, None + return Path(original_file), None - target_path = f"{original_file}{self.sidecar_ext}" + target_path = Path(f'{original_file}{self.sidecar_ext}') # If sidecar doesn't exist, we need to tell ExifTool to read from source # and write to the new sidecar file. @@ -115,8 +124,12 @@ class Executor: return target_path, None def _build_cmd( - self, file_path, field_values, timestamp_str, sidecar_source=None - ): + self, + file_path: Path, + field_values: dict[str, str], + timestamp_str: str, + sidecar_source: Path | None = None + ) -> list[str]: current_year = datetime.datetime.now().year # Core setup @@ -153,14 +166,14 @@ class Executor: if sidecar_source: # -srcfile SOURCE TARGET cmd.append("-srcfile") - cmd.append(file_path) - cmd.append(sidecar_source) + cmd.append(str(file_path)) + cmd.append(str(sidecar_source)) else: - cmd.append(file_path) + cmd.append(str(file_path)) return cmd - def _run_exiftool(self, cmd, dry_run): + def _run_exiftool(self, cmd: list[str], dry_run: bool) -> tuple[bool, str]: if dry_run: safe_cmd = shlex.join(cmd) return True, f"[DRY RUN] {safe_cmd}" @@ -178,7 +191,7 @@ class Executor: except FileNotFoundError: return False, "Error: 'exiftool' not found. Please install it." - def _parse_date(self, dt_str): + def _parse_date(self, dt_str: str | None) -> datetime.datetime: if not dt_str: # Should be handled by resolver/validator, but safe fallback return datetime.datetime.now() diff --git a/src/emulsion/main.py b/src/emulsion/main.py index b310e8f..01c40a5 100644 --- a/src/emulsion/main.py +++ b/src/emulsion/main.py @@ -3,20 +3,21 @@ import datetime import os import sys from importlib.metadata import PackageNotFoundError, version +from typing import Any from emulsion.config import ConfigLoader from emulsion.executor import Executor from emulsion.resolver import ValueResolver -def get_version(): +def get_version() -> str: try: return version('emulsion') except PackageNotFoundError: return 'unknown' -def parse_args(): +def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description='A tool for updating exif tags' ) @@ -107,7 +108,7 @@ def parse_args(): return parser.parse_args() -def prompt_for_defaults(config): +def prompt_for_defaults(config: dict[str, Any]) -> None: """ Prompts the user for default values to populate the initial config. """ @@ -140,7 +141,7 @@ def prompt_for_defaults(config): print('Invalid number, keeping default.') -def main(): +def main() -> None: try: # 1. Load Config loader = ConfigLoader() @@ -188,7 +189,7 @@ def main(): # 3. Prepare Inputs for Resolver # We need to mash --author and --field author=... into one dict - user_inputs = {} + user_inputs: dict[str, Any] = {} # First-Class args for field in ['author', 'lab', 'make', 'model', 'lens', 'film']: diff --git a/src/emulsion/resolver.py b/src/emulsion/resolver.py index a2f14d0..2f719fd 100644 --- a/src/emulsion/resolver.py +++ b/src/emulsion/resolver.py @@ -1,16 +1,19 @@ import sys +from typing import Any class ValueResolver: - def __init__(self, config): + def __init__(self, config: dict[str, Any]) -> None: """ config: The loaded configuration dictionary containing 'mappings' and 'defaults'. """ - self.mappings = config.get('mappings', {}) - self.defaults = config.get('defaults', {}) + self.mappings: dict[str, Any] = config.get('mappings', {}) + self.defaults: dict[str, Any] = config.get('defaults', {}) - def resolve(self, cli_args, interactive=True): + def resolve( + self, cli_args: dict[str, Any], interactive: bool = True + ) -> dict[str, str]: """ Resolves the final values for all fields. @@ -40,7 +43,7 @@ class ValueResolver: # 3. Identify Prompts # We look at the 'mappings' to see which fields want to be prompted. if interactive: - fields_to_prompt = [] + fields_to_prompt: list[str] = [] for field_name, schema in self.mappings.items(): # Check if prompt is requested if (isinstance(schema, dict) @@ -62,9 +65,12 @@ class ValueResolver: sys.exit(1) # Remove any fields that are still None/Empty (optional, but cleaner) - return {k: v for k, v in resolved.items() if v} + # Also cast values to string + return {k: str(v) for k, v in resolved.items() if v} - def _prompt_user(self, field_name, resolved_dict): + def _prompt_user( + self, field_name: str, resolved_dict: dict[str, Any] + ) -> None: """ Helper to prompt a single field. """