Add type hinting stuff
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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']:
|
||||
|
||||
@@ -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.
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user