merge main incl conflict resolution

.env.template

@@ -3,8 +3,3 @@ LLM_API_KEY = "not-relevant"
 LLM_DEFAULT_TEMPERATURE = 1.0
 
 METADATA_PREFIX = "photools|"
-LOG_LEVEL=INFO
-LLM_REQUEST_TIMEOUT_SECONDS=10
-LLM_MAX_COMPLETION_TOKENS=1024
-LLM_MAX_RETRIES=5
-LLM_MAX_KEYWORDS=15

.gitea/workflows/ci.yml

@@ -35,3 +35,5 @@ jobs:
         run: uv run mypy .
       - name: Check formatting (black)
         run: uv run black --check . --extend-exclude .venv
+      - name: Check formatting (black)
+        run: uv run black --check . --extend-exclude .venv

AGENTS.md

@@ -1,42 +1,27 @@
 # Agent Instructions: PhotoTools3
 
 ## Environment Setup
-
 Requires a `.env` file with the following variables:
-
 - `llm_api_key`: API key for the LLM provider.
 - `llm_api_base_url`: Base URL for the LLM API.
 - `llm_default_temperature`: Float value for temperature.
 
 ## System Dependencies
-
 - `exiftool` binary: required by `pyexiftool` at runtime, not installable via pip/uv.
   - Arch: `perl-image-exiftool`
   - Debian/Ubuntu: `libimage-exiftool-perl`
 
 ## Development Toolchain
-
 - **Dependency Management**: Use `uv`.
 - **Python Version**: Requires Python `>=3.14`.
 - **Testing**: Run `pytest` to execute the test suite in `test/`.
 - **Linting & Formatting**: Use `ruff` and `black`.
 
 ## CLI Usage
-
 The primary entry point is the `photools3` command:
-
 - `photools3 keywords <image_path>`: Extracts keywords and captions from a single image.
 
 ## Architecture & Logic
-
 - **`src/photools3/photo.py`**: Core image processing logic, including resizing, RGB conversion, and generating base64 JPEG encodings specifically for LLM vision compatibility.
 - **`src/photools3/llm_session.py`**: Manages the workflow of sending processed images to the LLM and parsing results.
 - **Configuration**: The list of keywords to extract is maintained in `config/keywords.txt`.
-
-## Code style
-
-- Avoid overly verbose docstrings or comments. Be precise!
-
-## Comunication
-
-- When done with code modifications, give a one-liner for the currently unstaged changes that can be used as a commit message.

README.md

@@ -1,101 +0,0 @@
-# PhotoTools3
-
-PhotoTools3 uses an OpenAI-compatible vision API to generate photo captions and
-keywords, then writes the results to photo metadata. It can also assess photo
-quality.
-
-## Setup
-
-Install the Python dependencies with `uv` and make sure ExifTool is installed and
-available on `PATH` (it is a system dependency, not a Python package):
-
-```bash
-uv sync
-```
-
-Create a `.env` file in the project root with the LLM connection details and the
-prefix used to identify metadata written by PhotoTools3:
-
-```dotenv
-llm_api_base_url=http://localhost:1234/v1
-llm_api_key=your-api-key
-llm_default_temperature=0.2
-metadata_prefix=photools3:
-LOG_LEVEL=DEBUG
-LLM_REQUEST_TIMEOUT_SECONDS=10
-LLM_MAX_COMPLETION_TOKENS=1024
-LLM_MAX_RETRIES=5
-LLM_MAX_KEYWORDS=15
-```
-
-Keyword suggestions are restricted to the vocabulary in `config/keywords.txt`.
-`LOG_LEVEL=DEBUG` enables detailed per-request and per-photo timings.
-`LLM_MAX_RETRIES` controls retries for malformed or structurally invalid
-responses; `5` means up to 6 requests for a photo. A valid response with
-unsupported keywords gets at most one targeted correction retry, then remaining
-invalid keywords are discarded.
-Each request is limited by `LLM_REQUEST_TIMEOUT_SECONDS`; a timeout reloads the
-LM Studio model before retrying. `LLM_MAX_COMPLETION_TOKENS` caps the generated
-response length. When all retries are exhausted, the model is reloaded before
-the failed photo is reported. `LLM_MAX_KEYWORDS` sets the maximum number of
-distinct keywords requested and stored per photo; excess keywords are discarded.
-
-## Generate metadata
-
-The `metadata-magic` command accepts one or more image paths and/or directories.
-The default mode is `both`, which generates and writes a caption and keywords:
-
-```bash
-uv run photools3 metadata-magic photo.dng
-```
-
-Choose a mode with `--mode`:
-
-```bash
-# Generate only keywords
-uv run photools3 metadata-magic photo.dng --mode keywords
-
-# Generate only a caption
-uv run photools3 metadata-magic photo.dng --mode captions
-
-# Explicitly request both (the default)
-uv run photools3 metadata-magic photo.dng --mode both
-```
-
-Pass multiple files to process them sequentially as a batch:
-
-```bash
-uv run photools3 metadata-magic photo1.dng photo2.jpg --mode both
-```
-
-Directories are scanned at the top level by default. Add `--recursive` to
-include supported images in subdirectories. You can combine directories and
-individual file paths:
-
-```bash
-uv run photools3 metadata-magic ./photos --recursive --mode keywords
-uv run photools3 metadata-magic ./photos photo-to-include.dng --mode captions
-```
-
-The command reports each processed file and exits with a non-zero status if any
-file fails. In a batch, it continues processing the remaining files after an
-individual failure. Generated metadata is written using ExifTool and XMP
-sidecars.
-
-## Assess photo quality
-
-The `quality` command reports an assessment without changing photo metadata:
-
-```bash
-uv run photools3 quality photo.dng
-```
-
-## Help
-
-Show the available commands and options with:
-
-```bash
-uv run photools3 --help
-uv run photools3 metadata-magic --help
-uv run photools3 quality --help
-```

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "pytest>=9.1.1",
     "ruff>=0.16.9",
     "typer>=0.12.0",
-    "utils>=0.1.3",
+    "utils>=0.1.2",
 ]
 
 [project.scripts]

src/photools3/cli.py

@@ -3,30 +3,16 @@ from pathlib import Path
 from typing import Annotated
 
 import typer
-from PIL import Image
 from utils.logging_config import LoggingSettings
 
-from photools3.llm_session import LlmSessionQuality
+from photools3.llm_session import (
-from photools3.photo import Photo
+    LlmSessionCaptionsAndKeywords,
-from photools3.processing import (
+    LlmSessionQuality,
-    ProcessingMode,
-    process_batch,
-    process_photo,
 )
-from photools3.settings import settings
+from photools3.photo import Photo
 
 app = typer.Typer(help="PhotoTools3: A CLI for image processing.")
-LoggingSettings(LOG_LEVEL=settings.log_level)
+LoggingSettings()
-for name in (
-    "uvicorn",
-    "uvicorn.error",
-    "uvicorn.access",
-    "PIL",
-    "openai",
-    "httpx2",
-    "httpcore2",
-):
-    logging.getLogger(name).setLevel(logging.WARNING)
 logger = logging.getLogger(__name__)
 
 
@@ -35,7 +21,9 @@ def main(
     ctx: typer.Context,
     version: bool | None = typer.Option(None, "--version", is_eager=True),
 ):
-    """Main entry point for PhotoTools3 CLI."""
+    """
+    Main entry point for PhotoTools3 CLI.
+    """
     if version:
         typer.echo("photools3 version 0.1.0")
         raise typer.Exit()
@@ -45,118 +33,36 @@ def main(
         raise typer.Exit()
 
 
-@app.command("metadata-magic")
+@app.command("keywords")
-def metadata_magic(
+def keywords(
-    paths: Annotated[
+    file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")],
-        list[Path], typer.Argument(..., help="Photo file(s) and/or directory paths")
-    ],
-    mode: Annotated[
-        ProcessingMode,
-        typer.Option(
-            "--mode",
-            case_sensitive=False,
-            help="Metadata to generate and write",
-        ),
-    ] = ProcessingMode.BOTH,
-    recursive: Annotated[
-        bool,
-        typer.Option(
-            "--recursive/--no-recursive",
-            help="Recursively discover photos inside input directories",
-        ),
-    ] = False,
 ):
-    """Generate and write keywords, captions, or both for photos."""
+    session = LlmSessionCaptionsAndKeywords()
-    _run_processing(paths, mode, recursive)
+    _run_llm_task(file_path, session)
 
 
 @app.command("quality")
 def quality(
     file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")],
 ):
-    """Assess a photo's quality without changing its metadata."""
+    session = LlmSessionQuality()
+    _run_llm_task(file_path, session)
+
+
+def _run_llm_task(
+    file_path: Path, session: LlmSessionCaptionsAndKeywords | LlmSessionQuality
+):
     try:
-        with LlmSessionQuality() as session, Photo(file_path) as photo:
+        logger.info(f"Processing: {file_path}")
-            result = session.call([photo])
+        photo = Photo(file_path)
-        logger.info("Quality assessment for %s: %s", file_path, result)
+        result = session.call([photo])
+        logger.info(result)
     except FileNotFoundError:
-        logger.error("Photo not found: %s", file_path)
+        logger.error(f"Error: File '{file_path}' does not exist.")
-        raise typer.Exit(code=1) from None
-    except Exception as exc:
-        logger.error("Error assessing photo %s: %s", file_path, exc)
-        raise typer.Exit(code=1) from exc
-
-
-def _run_processing(inputs: list[Path], mode: ProcessingMode, recursive: bool) -> None:
-    photo_paths = _expand_photo_paths(inputs, recursive)
-    if not photo_paths:
-        logger.error("No supported photo files found in the given input paths.")
         raise typer.Exit(code=1)
-
+    except Exception as e:  # noqa: BLE001
-    logger.info("Processing %d photo(s) in %s mode", len(photo_paths), mode.value)
-    if len(photo_paths) == 1:
-        path = photo_paths[0]
-        try:
-            process_photo(path, mode)
-        except FileNotFoundError:
-            logger.error("Photo not found: %s", path)
-            raise typer.Exit(code=1) from None
-        except Exception as exc:
-            logger.error("Failed to process %s: %s", path, exc)
-            raise typer.Exit(code=1) from exc
-        logger.info("Processed: %s", path)
-        return
-
-    try:
-        outcomes = process_batch(photo_paths, mode)
-    except Exception as exc:
-        logger.error("Batch processing failed: %s", exc)
-        raise typer.Exit(code=1) from exc
-
-    failures = 0
-    for outcome in outcomes:
-        if outcome.succeeded:
-            logger.info("Processed: %s", outcome.path)
-        else:
-            failures += 1
-            logger.error("Failed to process %s: %s", outcome.path, outcome.error)
-
-    logger.info(
-        "Batch complete: %d succeeded, %d failed", len(outcomes) - failures, failures
-    )
-    if failures:
         raise typer.Exit(code=1)
-
+        logger.error(f"Error processing image: {e}")
-
-def _expand_photo_paths(inputs: list[Path], recursive: bool) -> list[Path]:
-    """Expand directories to supported image files, preserving input order."""
-    supported_extensions = {
-        ".dng",
-        *(extension.lower() for extension in Image.registered_extensions()),
-    }
-    discovered: list[Path] = []
-
-    for path in inputs:
-        if not path.is_dir():
-            discovered.append(path)
-            continue
-
-        try:
-            candidates = path.rglob("*") if recursive else path.iterdir()
-            discovered.extend(
-                sorted(
-                    candidate
-                    for candidate in candidates
-                    if candidate.is_file()
-                    and candidate.suffix.lower() in supported_extensions
-                )
-            )
-        except OSError as exc:
-            logger.error("Could not read directory %s: %s", path, exc)
-            raise typer.Exit(code=1) from exc
-
-    # Avoid processing the same exact path more than once while retaining order.
-    return list(dict.fromkeys(discovered))
 
 
 if __name__ == "__main__":

src/photools3/llm_session.py

@@ -1,25 +1,20 @@
 import json
 import logging
 import re
-import time
 
 import openai
 from openai import OpenAI
 from pydantic import BaseModel, ValidationError
 
-from photools3.lm_studio import LMStudioModelManager
 from photools3.models import (
     LlmMessageContent,
     LlmMessageRole,
     LlmMessageType,
     LlmRequest,
     LlmRequestMessage,
-    LlmResponseCaption,
     LlmResponseCaptionsAndKeywords,
     LlmResponseGeneral,
-    LlmResponseKeywords,
     LlmResponseQuality,
-    load_valid_keywords,
 )
 from photools3.photo import Photo
 from photools3.settings import settings
@@ -33,23 +28,19 @@ def _clean_llm_response(txt: str):
     return txt
 
 
-def _correction_prompt(instruction: str) -> LlmRequestMessage:
+def _correction_prompt(
-    return LlmRequestMessage(
+    original_request: LlmRequest, exception: Exception, origial_json
-        role=LlmMessageRole.user,
+):
-        content=instruction,
+    correction_prompt = (
+        f"The previous response was invalid due to: {exception}. "
+        f"Please provide the corrected JSON based on this content: "
+        f"{origial_json}"
     )
-
+    original_request.messages.append(
-
+        LlmRequestMessage(
-def _request_with_correction(base_request: LlmRequest, instruction: str) -> LlmRequest:
+            role=LlmMessageRole.user,
-    return base_request.model_copy(
+            content=correction_prompt,
-        update={"messages": [*base_request.messages, _correction_prompt(instruction)]}
+        )
-    )
-
-
-def _validation_summary(exception: ValidationError) -> str:
-    return "; ".join(
-        f"{'.'.join(map(str, error['loc']))}: {error['msg']}"
-        for error in exception.errors(include_input=False)
     )
 
 
@@ -63,7 +54,6 @@ class LlmSession:
         self, system_prompt: str, temperature: float = settings.llm_default_temperature
     ):
         self._models: list = []
-        self._model_id: str | None = None
 
         self.system_prompt = system_prompt
         self.temperature = temperature
@@ -71,30 +61,10 @@ class LlmSession:
         self.client = OpenAI(
             base_url=self.base_url,
             api_key=self.api_key,
-            max_retries=0,
         )
-        self._model_manager = LMStudioModelManager(self.base_url, self.api_key)
-
-    def close(self):
-        self.client.close()
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.close()
-
-    def flush_model(self):
-        """Unload and reload this session's model through LM Studio."""
-        if self._model_id is None:
-            self._model_id = self.models[0].id
-        self._model_manager.reload_model(self._model_id)
 
     def client_chat_completion_create(self, request: LlmRequest, choice_no: int = 0):
-        response = self.client.chat.completions.create(
+        response = self.client.chat.completions.create(**request.model_dump())
-            **request.model_dump(),
-            timeout=settings.llm_request_timeout_seconds,
-        )
         return response.choices[choice_no].message.content
 
     def is_api_reachable(self) -> bool:
@@ -114,140 +84,38 @@ class LlmSession:
         self,
         photos: list[Photo],
         model_id=None,
-        max_retries: int | None = None,
+        max_retries: int = 2,
         prompt: str | None = None,
     ):
-        retry_limit = settings.llm_max_retries if max_retries is None else max_retries
+        if not model_id:
-        if retry_limit < 0:
+            model_id = self.models[0].id
-            raise ValueError("max_retries must be greater than or equal to zero")
 
-        if model_id is None:
+        request = self._build_request(prompt or self.prompt, photos, model_id)
-            if self._model_id is None:
-                self._model_id = self.models[0].id
-            model_id = self._model_id
-        else:
-            self._model_id = model_id
 
-        base_request = self._build_request(prompt or self.prompt, photos, model_id)
+        last_error = None
-        request = base_request
+        for attempt in range(max_retries + 1):
-        tag_correction_used = False
-        valid_keywords = load_valid_keywords()
-        max_keywords = settings.llm_max_keywords
-
-        for attempt in range(retry_limit + 1):
-            request_started = time.perf_counter()
             try:
                 result_raw = self.client_chat_completion_create(request)
-            except openai.APITimeoutError as exc:
+                json_str = _clean_llm_response(result_raw)
-                logger.warning(
+                return self.response_model.model_validate_json(json_str)
-                    "LLM request timed out after %.2f seconds; reloading model",
+
-                    time.perf_counter() - request_started,
+            except (json.JSONDecodeError, ValidationError) as exc:
-                )
+                last_error = exc
-                self.flush_model()
+                if attempt < max_retries:
-                if attempt < retry_limit:
+                    logger.info("Chatbot response is invalid; promting to correct ...")
+                    _correction_prompt(
+                        request,
+                        exc,
+                        json_str if "json_str" in locals() else "origial json",
+                    )
                     continue
-                raise ValueError(
+                else:
-                    f"LLM request timed out after {retry_limit + 1} attempts"
+                    raise ValueError(
-                ) from exc
+                        f"Failed to parse or validate LLM response after {max_retries} retries: {exc}"
+                    )
             except Exception:
-                logger.debug(
-                    "LLM request attempt %d/%d failed after %.2f seconds",
-                    attempt + 1,
-                    retry_limit + 1,
-                    time.perf_counter() - request_started,
-                )
                 raise
 
-            logger.debug(
-                "LLM request attempt %d/%d completed in %.2f seconds",
-                attempt + 1,
-                retry_limit + 1,
-                time.perf_counter() - request_started,
-            )
-            try:
-                response_data = json.loads(_clean_llm_response(result_raw))
-            except (json.JSONDecodeError, TypeError) as exc:
-                if attempt < retry_limit:
-                    logger.info("Chatbot response was not valid JSON; retrying")
-                    request = _request_with_correction(
-                        base_request,
-                        "Return one complete JSON object matching the schema. "
-                        f"The previous response was invalid: {exc}",
-                    )
-                    continue
-
-                logger.warning(
-                    "LLM retry limit reached; reloading model before failing"
-                )
-                self.flush_model()
-                raise ValueError(
-                    f"Failed to parse LLM response after {retry_limit} retries: {exc}"
-                ) from exc
-
-            if (
-                isinstance(response_data, dict)
-                and "tags" in self.response_model.model_fields
-                and isinstance(response_data.get("tags"), list)
-            ):
-                tags = response_data["tags"]
-                considered_tags = tags[:max_keywords]
-                invalid_tags = [
-                    tag
-                    for tag in considered_tags
-                    if not isinstance(tag, str) or tag not in valid_keywords
-                ]
-
-                if invalid_tags and not tag_correction_used and attempt < retry_limit:
-                    tag_correction_used = True
-                    logger.info(
-                        "LLM returned unsupported keywords; requesting one correction"
-                    )
-                    request = _request_with_correction(
-                        base_request,
-                        f"These keywords are not allowed: {invalid_tags!r}. "
-                        "Replace them with exact allowed keywords when sensible; "
-                        f"otherwise omit them. Return at most {max_keywords} "
-                        "distinct keywords and do not list the vocabulary.",
-                    )
-                    continue
-
-                if len(tags) > max_keywords:
-                    logger.info(
-                        "Discarding keywords beyond configured limit of %d",
-                        max_keywords,
-                    )
-                if invalid_tags:
-                    logger.warning("Discarding unsupported keywords: %s", invalid_tags)
-
-                response_data = dict(response_data)
-                valid_tags = [
-                    tag
-                    for tag in tags
-                    if isinstance(tag, str) and tag in valid_keywords
-                ]
-                response_data["tags"] = list(dict.fromkeys(valid_tags))[:max_keywords]
-
-            try:
-                return self.response_model.model_validate(response_data)
-            except ValidationError as exc:
-                if attempt < retry_limit:
-                    logger.info("LLM response did not match the schema; retrying")
-                    request = _request_with_correction(
-                        base_request,
-                        "Return one complete JSON object matching the schema. "
-                        f"Validation issues: {_validation_summary(exc)}",
-                    )
-                    continue
-
-                logger.warning(
-                    "LLM retry limit reached; reloading model before failing"
-                )
-                self.flush_model()
-                raise ValueError(
-                    f"Failed to validate LLM response after {retry_limit} retries: "
-                    f"{_validation_summary(exc)}"
-                ) from exc
-
     def _build_request(
         self,
         prompt: str,
@@ -258,7 +126,6 @@ class LlmSession:
         return LlmRequest(
             model=model_id,
             temperature=self.temperature,
-            max_tokens=settings.llm_max_completion_tokens,
             messages=[
                 # System prompt
                 LlmRequestMessage(
@@ -290,49 +157,13 @@ class LlmSession:
 
 class LlmSessionCaptionsAndKeywords(LlmSession):
     response_model = LlmResponseCaptionsAndKeywords
-    prompt = "Give me a caption and keywords for the following photo"
+    prompt = "Give me a caption and keyword for the following photo(s)"
 
     def __init__(self, temperature: float = settings.llm_default_temperature):
         super().__init__(
             system_prompt=(
                 "You are a helpful photographer's assistant that helps to write **captions** and **keywords** to photos:\n"
                 "Answer in German.\n"
-                f"Choose at most {settings.llm_max_keywords} distinct, relevant keywords. "
-                "Do not repeat or enumerate the allowed-keyword list.\n"
-                "Output must be a JSON of the following schema:\n"
-                f"{self.response_model.model_json_schema()}"
-            ),
-            temperature=temperature,
-        )
-
-
-class LlmSessionCaption(LlmSession):
-    response_model = LlmResponseCaption
-    prompt = "Write a caption for the following photo"
-
-    def __init__(self, temperature: float = settings.llm_default_temperature):
-        super().__init__(
-            system_prompt=(
-                "You are a helpful photographer's assistant that writes captions "
-                "for photos. Answer in German.\n"
-                "Output must be a JSON of the following schema:\n"
-                f"{self.response_model.model_json_schema()}"
-            ),
-            temperature=temperature,
-        )
-
-
-class LlmSessionKeywords(LlmSession):
-    response_model = LlmResponseKeywords
-    prompt = "Choose appropriate keywords for the following photo"
-
-    def __init__(self, temperature: float = settings.llm_default_temperature):
-        super().__init__(
-            system_prompt=(
-                "You are a helpful photographer's assistant that chooses keywords "
-                "for photos. Answer in German.\n"
-                f"Choose at most {settings.llm_max_keywords} distinct, relevant keywords. "
-                "Do not repeat or enumerate the allowed-keyword list.\n"
                 "Output must be a JSON of the following schema:\n"
                 f"{self.response_model.model_json_schema()}"
             ),

src/photools3/lm_studio.py

@@ -1,135 +0,0 @@
-import json
-import logging
-import time
-from urllib.error import HTTPError, URLError
-from urllib.parse import urlsplit, urlunsplit
-from urllib.request import Request, urlopen
-
-logger = logging.getLogger(__name__)
-
-_LOAD_CONFIG_KEYS = (
-    "context_length",
-    "eval_batch_size",
-    "flash_attention",
-    "num_experts",
-    "offload_kv_cache_to_gpu",
-)
-
-
-class LMStudioError(RuntimeError):
-    """Raised when LM Studio model management fails."""
-
-
-class LMStudioModelManager:
-    """Unload and reload a model through LM Studio's native REST API."""
-
-    def __init__(self, openai_base_url: str, api_key: str, timeout: float = 600):
-        parsed_url = urlsplit(openai_base_url)
-        path = parsed_url.path.rstrip("/")
-        path = path.removesuffix("/v1")
-        root_url = urlunsplit(
-            (parsed_url.scheme, parsed_url.netloc, path, "", "")
-        ).rstrip("/")
-        self.api_base_url = f"{root_url}/api/v1"
-        self.api_key = api_key
-        self.timeout = timeout
-
-    def reload_model(self, model_id: str) -> None:
-        """Unload the matching loaded instance and reload it with its config."""
-        started = time.perf_counter()
-        models_response = self._request("GET", "/models")
-        model, loaded_instance = self._find_model(models_response, model_id)
-        model_key = model["key"]
-
-        if loaded_instance is not None:
-            self._request(
-                "POST",
-                "/models/unload",
-                {"instance_id": loaded_instance["id"]},
-            )
-            load_config = loaded_instance.get("config", {})
-        else:
-            load_config = {}
-
-        load_body = {"model": model_key}
-        load_body.update(
-            {
-                key: load_config[key]
-                for key in _LOAD_CONFIG_KEYS
-                if key in load_config and load_config[key] is not None
-            }
-        )
-        self._request("POST", "/models/load", load_body)
-        logger.info(
-            "Reloaded LM Studio model %s in %.1f seconds",
-            model_key,
-            time.perf_counter() - started,
-        )
-
-    @staticmethod
-    def _find_model(models_response: dict, model_id: str) -> tuple[dict, dict | None]:
-        models = models_response.get("models")
-        if not isinstance(models, list):
-            raise LMStudioError("LM Studio model list response has no models array")
-
-        for model in models:
-            loaded_instances = model.get("loaded_instances", [])
-            for instance in loaded_instances:
-                if instance.get("id") == model_id:
-                    return model, instance
-
-        matching_models = [model for model in models if model.get("key") == model_id]
-        if len(matching_models) != 1:
-            raise LMStudioError(
-                f"Could not uniquely identify LM Studio model {model_id!r}"
-            )
-
-        model = matching_models[0]
-        loaded_instances = model.get("loaded_instances", [])
-        if len(loaded_instances) > 1:
-            raise LMStudioError(
-                f"Multiple loaded instances found for LM Studio model {model_id!r}"
-            )
-        loaded_instance = loaded_instances[0] if loaded_instances else None
-        return model, loaded_instance
-
-    def _request(self, method: str, path: str, body: dict | None = None) -> dict:
-        data = None if body is None else json.dumps(body).encode("utf-8")
-        headers = {"Accept": "application/json"}
-        if data is not None:
-            headers["Content-Type"] = "application/json"
-        if self.api_key:
-            headers["Authorization"] = f"Bearer {self.api_key}"
-
-        request = Request(
-            f"{self.api_base_url}{path}",
-            data=data,
-            headers=headers,
-            method=method,
-        )
-        try:
-            with urlopen(request, timeout=self.timeout) as response:
-                response_body = response.read()
-        except HTTPError as exc:
-            detail = exc.read().decode("utf-8", errors="replace")
-            raise LMStudioError(
-                f"LM Studio {method} {path} returned HTTP {exc.code}: {detail}"
-            ) from exc
-        except (URLError, TimeoutError) as exc:
-            raise LMStudioError(
-                f"Could not reach LM Studio at {self.api_base_url}: {exc}"
-            ) from exc
-
-        if not response_body:
-            return {}
-        try:
-            result = json.loads(response_body)
-        except json.JSONDecodeError as exc:
-            raise LMStudioError(
-                f"LM Studio {method} {path} returned invalid JSON"
-            ) from exc
-        if not isinstance(result, dict):
-            raise LMStudioError(
-                f"LM Studio {method} {path} returned an unexpected response"
-            )
-        return result

src/photools3/metadata_writer.py

@@ -87,12 +87,12 @@ class MetadataWriter:
         return metadata
 
     def get_caption(self) -> str:
-        result = (self.metadata or {}).get(XMP_KEYS_CAPTION[0], "")
+        result = self.metadata.get(XMP_KEYS_CAPTION[0], "")
         assert isinstance(result, str)
         return result
 
     def get_keywords(self) -> list[str]:
-        result = (self.metadata or {}).get(XMP_KEYS_KEYWORDS[0], []) or []
+        result = self.metadata.get(XMP_KEYS_KEYWORDS[0], []) or []
         if isinstance(result, str):
             result = [result]
         assert isinstance(result, list)
@@ -108,11 +108,8 @@ class MetadataWriter:
                     tags=metadata_tags,
                     params=["-overwrite_original"],
                 )
-        except Exception as exc:
+        except Exception as exc:  # noqa: BLE001
             log.error(f"Could not write metadata to file {self.sidecar_path}: {exc}")
-            raise RuntimeError(
-                f"Could not write metadata to file {self.sidecar_path}"
-            ) from exc
 
     def write_caption(
         self, caption: Caption, include_prefix: bool = True, override: bool = True
@@ -154,27 +151,9 @@ class MetadataWriter:
         keywords: Keywords | None = None,
         color: Color | None = None,
     ):
-        metadata_tags: XmpMetadata = {}
         if caption is not None:
-            existing_no_prefix, _ = _separate_prefixed_caption(self.get_caption())
+            self.write_caption(caption)
-            if existing_no_prefix:
-                existing_no_prefix = (
-                    f"{existing_no_prefix}."
-                    if not existing_no_prefix.endswith(".")
-                    else existing_no_prefix
-                )
-                existing_no_prefix = f"{existing_no_prefix} "
-            caption = f"{existing_no_prefix}{_prefixed(caption)}"
-            metadata_tags.update({key: caption for key in XMP_KEYS_CAPTION})
-
         if keywords is not None:
-            keywords = list(map(_prefixed, keywords))
+            self.write_keywords(keywords)
-            existing = self.get_keywords()
-            keywords = list(set(existing + keywords))
-            metadata_tags.update({tag: keywords for tag in XMP_KEYS_KEYWORDS})
-
         if color is not None:
-            metadata_tags["XMP-darktable:Colorlabels"] = color
+            self.write_color(color)
-
-        if metadata_tags:
-            self.set_metadata(metadata_tags)

src/photools3/models.py

@@ -1,4 +1,3 @@
-import json
 from enum import StrEnum
 from functools import lru_cache
 from typing import Annotated, Any, Literal
@@ -11,7 +10,7 @@ from pydantic import (
     model_validator,
 )
 
-from .settings import KEYWORDS_LIST_PATH, settings
+from .settings import KEYWORDS_LIST_PATH
 
 
 @lru_cache
@@ -43,7 +42,6 @@ class LlmRequest(BaseModel):
     model: str = "gpt-4o"
     messages: list[LlmRequestMessage]
     temperature: float | None = None
-    max_tokens: int
 
 
 class LlmMessageRole(StrEnum):
@@ -88,7 +86,7 @@ class LlmResponseGeneral(BaseModel):
     text: str
 
 
-class LlmResponseCaption(BaseModel):
+class LlmResponseCaptionsAndKeywords(BaseModel):
     caption: Annotated[
         str,
         Field(
@@ -98,25 +96,17 @@ class LlmResponseCaption(BaseModel):
             )
         ),
     ]
-
+    tags: list[str] = Field(
-
-def _keyword_tags_field():
-    allowed_keywords = json.dumps(
-        sorted(load_valid_keywords()), ensure_ascii=False, separators=(",", ":")
-    )
-    return Field(
-        max_length=settings.llm_max_keywords,
         description=(
-            f"Choose at most {settings.llm_max_keywords} distinct, relevant keywords. "
+            "A list of keywords that properly tag the photo. "
-            "Do not enumerate the vocabulary. Allowed keywords: "
+            "Allowed keywords: "
-            f"{allowed_keywords}"
+            + str(load_valid_keywords())
-        ),
+            .replace("'", '"')
+            .replace("{", "[")
+            .replace("}", "]")
+        )
     )
 
-
-class _LlmResponseWithTags(BaseModel):
-    tags: list[str] = _keyword_tags_field()
-
     @field_validator("tags")
     @classmethod
     def validate_tags(cls, v: list[str]) -> list[str]:
@@ -129,22 +119,6 @@ class _LlmResponseWithTags(BaseModel):
         return v
 
 
-class LlmResponseKeywords(_LlmResponseWithTags):
-    pass
-
-
-class LlmResponseCaptionsAndKeywords(_LlmResponseWithTags):
-    caption: Annotated[
-        str,
-        Field(
-            description=(
-                "A brief but precise description of what the picture shows. "
-                "No interpretation; just what is visible"
-            )
-        ),
-    ]
-
-
 class LlmResponseQuality(BaseModel):
     rating: Annotated[
         Literal["ok", "review", "not_ok"],

src/photools3/photo.py

@@ -1,7 +1,6 @@
 import base64
 from io import BytesIO
 from pathlib import Path
-from typing import Self
 
 from PIL import Image
 
@@ -18,15 +17,6 @@ class Photo:
         self.path = path
         self.image = Image.open(path)
 
-    def close(self):
-        self.image.close()
-
-    def __enter__(self) -> Self:
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.close()
-
     def resize(self, image_max_size=1024) -> Photo:
         """Limiting max dimensions while maintaining aspect ratio"""
         self.image.thumbnail((image_max_size, image_max_size), Image.Resampling.LANCZOS)
@@ -34,9 +24,7 @@ class Photo:
 
     def to_rgb(self):
         if self.image.mode != "RGB":
-            original = self.image
+            self.image = self.image.convert("RGB")
-            self.image = original.convert("RGB")
-            original.close()
         return self
 
     def get_encoded(self):

src/photools3/processing.py

@@ -1,138 +0,0 @@
-import logging
-import time
-from collections.abc import Iterable
-from dataclasses import dataclass
-from enum import StrEnum
-from pathlib import Path
-
-from photools3.llm_session import (
-    LlmSession,
-    LlmSessionCaption,
-    LlmSessionCaptionsAndKeywords,
-    LlmSessionKeywords,
-)
-from photools3.metadata_writer import MetadataWriter
-from photools3.models import (
-    LlmResponseCaption,
-    LlmResponseCaptionsAndKeywords,
-    LlmResponseKeywords,
-)
-from photools3.photo import Photo
-
-logger = logging.getLogger(__name__)
-
-type ProcessingResponse = (
-    LlmResponseCaption | LlmResponseKeywords | LlmResponseCaptionsAndKeywords
-)
-
-
-class ProcessingMode(StrEnum):
-    KEYWORDS = "keywords"
-    CAPTIONS = "captions"
-    BOTH = "both"
-
-
-@dataclass(frozen=True)
-class ProcessingOutcome:
-    path: Path
-    result: ProcessingResponse | None = None
-    error: str | None = None
-
-    @property
-    def succeeded(self) -> bool:
-        return self.error is None and self.result is not None
-
-
-def _create_session(mode: ProcessingMode) -> LlmSession:
-    if mode == ProcessingMode.KEYWORDS:
-        return LlmSessionKeywords()
-    if mode == ProcessingMode.CAPTIONS:
-        return LlmSessionCaption()
-    return LlmSessionCaptionsAndKeywords()
-
-
-def process_photo(
-    path: Path | str,
-    mode: ProcessingMode | str = ProcessingMode.BOTH,
-    *,
-    session: LlmSession | None = None,
-) -> ProcessingResponse:
-    """Process one photo and write the requested LLM result to its metadata.
-
-    If a session is supplied, its lifecycle remains the caller's responsibility.
-    """
-    path = Path(path)
-    mode = ProcessingMode(mode)
-    owns_session = session is None
-    active_session = session if session is not None else _create_session(mode)
-
-    logger.info(f"Processing photo: {path!s}")
-
-    try:
-        with Photo(path) as photo:
-            result = active_session.call([photo])
-
-        caption: str | None = None
-        keywords: list[str] | None = None
-        if mode == ProcessingMode.CAPTIONS:
-            if not isinstance(result, LlmResponseCaption):
-                raise TypeError("Caption mode returned an unexpected response type")
-            caption = result.caption
-        elif mode == ProcessingMode.KEYWORDS:
-            if not isinstance(result, LlmResponseKeywords):
-                raise TypeError("Keywords mode returned an unexpected response type")
-            keywords = result.tags
-        else:
-            if not isinstance(result, LlmResponseCaptionsAndKeywords):
-                raise TypeError("Both mode returned an unexpected response type")
-            caption = result.caption
-            keywords = result.tags
-
-        MetadataWriter(path).write(caption=caption, keywords=keywords)
-        return result
-    finally:
-        if owns_session:
-            active_session.close()
-
-
-def process_batch(
-    paths: Iterable[Path | str],
-    mode: ProcessingMode | str = ProcessingMode.BOTH,
-) -> list[ProcessingOutcome]:
-    """Process photos sequentially, reusing one LLM client for the whole batch.
-
-    A failed photo is recorded in its outcome and does not stop later photos.
-    Only one decoded image is held at a time.
-    """
-    mode = ProcessingMode(mode)
-    session = _create_session(mode)
-    outcomes: list[ProcessingOutcome] = []
-
-    try:
-        for item in paths:
-            path = Path(item)
-            photo_started = time.perf_counter()
-            try:
-                result = process_photo(path, mode, session=session)
-            except Exception as exc:
-                logger.exception("Failed to process %s", path)
-                # Keep only a string, not the exception/traceback, which can
-                # retain references to the just-processed image and stack.
-                outcomes.append(
-                    ProcessingOutcome(
-                        path=path,
-                        error=f"{type(exc).__name__}: {exc}",
-                    )
-                )
-            else:
-                outcomes.append(ProcessingOutcome(path=path, result=result))
-            finally:
-                logger.debug(
-                    "Photo processing for %s took %.2f seconds",
-                    path,
-                    time.perf_counter() - photo_started,
-                )
-    finally:
-        session.close()
-
-    return outcomes

src/photools3/settings.py

@@ -1,6 +1,5 @@
 from pathlib import Path
 
-from pydantic import Field
 from pydantic_core import Url
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -14,16 +13,11 @@ KEYWORDS_LIST_PATH = CONFIG_DIR / "keywords.txt"
 
 
 class Settings(BaseSettings):
-    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+    model_config = SettingsConfigDict(env_file=".env")
 
     llm_api_base_url: Url
     llm_api_key: str
     llm_default_temperature: float
-    llm_request_timeout_seconds: float = Field(default=10, gt=0)
-    llm_max_completion_tokens: int = Field(default=1024, gt=0)
-    llm_max_retries: int = Field(default=5, ge=0)
-    llm_max_keywords: int = Field(default=15, ge=1)
-    log_level: str = "INFO"
 
     metadata_prefix: str
 

test/test_cli.py

@@ -1,197 +1,100 @@
+import logging
 from pathlib import Path
-from types import SimpleNamespace
 from unittest.mock import patch
 
+import pytest
 from typer.testing import CliRunner
 
 from photools3.cli import app
-from photools3.processing import ProcessingMode
 
 runner = CliRunner()
 
-
+CLI_LOGGER = "photools3.cli"
-def _success_outcome(path: Path):
-    return SimpleNamespace(path=path, succeeded=True, error=None)
 
 
-def test_metadata_magic_command_defaults_to_both_for_single_photo(tmp_path):
+@pytest.fixture
-    image_path = tmp_path / "photo.jpg"
+def mock_photo_logic(tmp_path):
-    image_path.write_bytes(b"test image")
+    """Fixture to mock the core logic of keywords command."""
-
     with (
-        patch("photools3.cli.process_photo") as process_photo,
+        patch("photools3.cli.LlmSessionCaptionsAndKeywords") as mocked_session,
-        patch("photools3.cli.process_batch") as process_batch,
+        patch("photools3.cli.Photo") as mocked_photo,
     ):
-        result = runner.invoke(app, ["metadata-magic", str(image_path)])
+
+        # Configure session mock
+        session_instance = mocked_session.return_value
+        session_instance.call.return_value = {
+            "caption": "A beautiful test image",
+            "args": ["test", "mock", "image"],  # Wait, the code uses 'tags'
+        }
+        # Let me re-check the code in cli.py
+        # result.get("tags", [])
+        session_instance.call.return_value = {
+            "caption": "A beautiful test image",
+            "tags": ["test", "mock", "image"],
+        }
+
+        # Configure photo mock
+        photo_instance = mocked_photo.return_value
+        photo_instance.path = tmp_path / "test_image.jpg"
+
+        yield {
+            "session": session_instance,
+            "photo_class": mocked_photo,
+            "photo_instance": photo_instance,
+        }
+
+
+@pytest.fixture
+def dummy_file(tmp_path):
+    """Fixture to create a dummy file for CLI testing."""
+    dummy_path = tmp_path / "test_image.jpg"
+    dummy_path.write_text("dummy content")
+    return dummy_path
+
+
+def test_keywords_command_success(mock_photo_logic, dummy_file, caplog):
+    """Test the 'keywords' command with a valid file path."""
+    # Ensure the mock photo instance points to our dummy file
+    mock_photo_logic["photo_class"].return_value.path = dummy_file
+
+    with caplog.at_level(logging.INFO, logger=CLI_LOGGER):
+        result = runner.invoke(app, ["keywords", str(dummy_file)])
 
     assert result.exit_code == 0
-    process_photo.assert_called_once_with(image_path, ProcessingMode.BOTH)
+    assert "Processing: " in caplog.text
-    process_batch.assert_not_called()
+    # The result is logged as a raw dictionary string
-
+    assert (
-
+        "{'caption': 'A beautiful test image', 'tags': ['test', 'mock', 'image']}"
-def test_metadata_magic_command_selects_caption_mode(tmp_path):
+        in caplog.text
-    image_path = tmp_path / "photo.jpg"
-    image_path.write_bytes(b"test image")
-
-    with patch("photools3.cli.process_photo") as process_photo:
-        result = runner.invoke(
-            app, ["metadata-magic", str(image_path), "--mode", "captions"]
-        )
-
-    assert result.exit_code == 0
-    process_photo.assert_called_once_with(image_path, ProcessingMode.CAPTIONS)
-
-
-def test_metadata_magic_command_batches_multiple_files(tmp_path):
-    paths = [tmp_path / "one.jpg", tmp_path / "two.jpg"]
-    for path in paths:
-        path.write_bytes(b"test image")
-
-    with (
-        patch("photools3.cli.process_photo") as process_photo,
-        patch("photools3.cli.process_batch") as process_batch,
-    ):
-        process_batch.return_value = [_success_outcome(path) for path in paths]
-        result = runner.invoke(
-            app,
-            ["metadata-magic", *(str(path) for path in paths), "--mode", "keywords"],
-        )
-
-    assert result.exit_code == 0
-    process_photo.assert_not_called()
-    process_batch.assert_called_once_with(paths, ProcessingMode.KEYWORDS)
-
-
-def test_metadata_magic_command_expands_directories_recursively(tmp_path):
-    root = tmp_path / "photos"
-    nested = root / "nested"
-    nested.mkdir(parents=True)
-    top_level_photo = root / "one.jpg"
-    nested_photo = nested / "two.png"
-    sidecar = root / "one.jpg.xmp"
-    for path in (top_level_photo, nested_photo, sidecar):
-        path.write_bytes(b"placeholder")
-
-    with (
-        patch("photools3.cli.process_photo") as process_photo,
-        patch("photools3.cli.process_batch") as process_batch,
-    ):
-        process_batch.return_value = [
-            _success_outcome(nested_photo),
-            _success_outcome(top_level_photo),
-        ]
-        result = runner.invoke(
-            app, ["metadata-magic", str(root), "--recursive", "--mode", "both"]
-        )
-
-    assert result.exit_code == 0
-    process_photo.assert_not_called()
-    process_batch.assert_called_once_with(
-        [nested_photo, top_level_photo], ProcessingMode.BOTH
     )
 
 
-def test_metadata_magic_command_does_not_recurse_by_default(tmp_path):
+def test_keywords_command_file_not_found(caplog):
-    root = tmp_path / "photos"
+    """Test the 'keywords' command with a non-existent file path."""
-    nested = root / "nested"
+    non_existent_file = Path("non_existent_file.jpg")
-    nested.mkdir(parents=True)
-    top_level_photo = root / "one.jpg"
-    nested_photo = nested / "two.jpg"
-    top_level_photo.write_bytes(b"placeholder")
-    nested_photo.write_bytes(b"placeholder")
 
-    with (
+    with caplog.at_level(logging.ERROR, logger=CLI_LOGGER):
-        patch("photools3.cli.process_photo") as process_photo,
+        result = runner.invoke(app, ["keywords", str(non_existent_file)])
-        patch("photools3.cli.process_batch") as process_batch,
-    ):
-        result = runner.invoke(app, ["metadata-magic", str(root)])
-
-    assert result.exit_code == 0
-    process_photo.assert_called_once_with(top_level_photo, ProcessingMode.BOTH)
-    process_batch.assert_not_called()
-
-
-def test_metadata_magic_command_returns_failure_if_any_batch_item_fails(tmp_path):
-    paths = [tmp_path / "one.jpg", tmp_path / "two.jpg"]
-    for path in paths:
-        path.write_bytes(b"test image")
-
-    with patch("photools3.cli.process_batch") as process_batch:
-        process_batch.return_value = [
-            _success_outcome(paths[0]),
-            SimpleNamespace(
-                path=paths[1], succeeded=False, error="RuntimeError: write failed"
-            ),
-        ]
-        result = runner.invoke(app, ["metadata-magic", *(str(path) for path in paths)])
 
     assert result.exit_code == 1
+    assert f"Error: File '{non_existent_file}' does not exist." in caplog.text
 
 
-def test_metadata_magic_command_returns_failure_for_single_photo_error(tmp_path):
+def test_keywords_command_error_during_processing(mock_photo_logic, dummy_file):
-    image_path = tmp_path / "photo.jpg"
+    """Test the 'keywords' command when an exception occurs during processing."""
-    image_path.write_bytes(b"test image")
+    # Configure the session mock to raise an exception
+    mock_photo_logic["session"].call.side_effect = Exception("LLM error")
 
-    with patch("photools3.cli.process_photo", side_effect=RuntimeError("LLM error")):
+    result = runner.invoke(app, ["keywords", str(dummy_file)])
-        result = runner.invoke(app, ["metadata-magic", str(image_path)])
-
-    assert result.exit_code == 1
-
-
-def test_keywords_alias_is_not_registered():
-    result = runner.invoke(app, ["keywords", "photo.jpg"])
-
-    assert result.exit_code == 2
-    assert "No such command 'keywords'" in result.output
-
-
-def test_old_process_command_name_is_not_registered():
-    result = runner.invoke(app, ["process", "photo.jpg"])
-
-    assert result.exit_code == 2
-    assert "No such command 'process'" in result.output
-
-
-def test_metadata_magic_command_rejects_empty_directory(tmp_path):
-    with (
-        patch("photools3.cli.process_photo") as process_photo,
-        patch("photools3.cli.process_batch") as process_batch,
-    ):
-        result = runner.invoke(app, ["metadata-magic", str(tmp_path)])
-
-    assert result.exit_code == 1
-    process_photo.assert_not_called()
-    process_batch.assert_not_called()
-
-
-def test_quality_command_closes_photo_and_session(tmp_path):
-    image_path = tmp_path / "photo.jpg"
-    image_path.write_bytes(b"test image")
-
-    with (
-        patch("photools3.cli.LlmSessionQuality") as session_class,
-        patch("photools3.cli.Photo") as photo_class,
-    ):
-        result = runner.invoke(app, ["quality", str(image_path)])
-
-    assert result.exit_code == 0
-    session_class.return_value.__enter__.return_value.call.assert_called_once_with(
-        [photo_class.return_value.__enter__.return_value]
-    )
-    session_class.return_value.__exit__.assert_called_once()
-    photo_class.return_value.__exit__.assert_called_once()
-
-
-def test_quality_command_reports_error_and_exits(tmp_path):
-    image_path = tmp_path / "photo.jpg"
-
-    with patch("photools3.cli.Photo", side_effect=FileNotFoundError(image_path)):
-        result = runner.invoke(app, ["quality", str(image_path)])
 
+    # Note: cli.py raises typer.Exit(code=1) before logging the
+    # "Error processing image: ..." message (the logger.error call is dead
+    # code after the raise), so only the exit code can be asserted here.
     assert result.exit_code == 1
 
 
 def test_version_command():
+    """Test the --version option."""
     result = runner.invoke(app, ["--version"])
 
     assert result.exit_code == 0

test/test_llm_session_mocked.py

@@ -1,10 +1,6 @@
 from unittest.mock import MagicMock, patch
 
-import openai
+from photools3.llm_session import LlmSession
-import pytest
-
-from photools3.llm_session import LlmSession, LlmSessionCaptionsAndKeywords
-from photools3.models import LlmResponseCaptionsAndKeywords
 
 
 def test_llm_session_mocked(test_photo):
@@ -88,216 +84,3 @@ def test_llm_session_captions_and_keywords_mocked(test_photo):
 
         # Verify that the mock was called
         assert mock_client.chat.completions.create.called
-
-
-def test_llm_session_retries_from_original_request_without_accumulating_messages():
-    with patch("photools3.llm_session.OpenAI") as mocked_openai:
-        mock_client = MagicMock()
-        mocked_openai.return_value = mock_client
-        mock_model = MagicMock()
-        mock_model.id = "gpt-4o"
-        mock_client.models.list.return_value.data = [mock_model]
-
-        def response(content):
-            message = MagicMock()
-            message.content = content
-            choice = MagicMock()
-            choice.message = message
-            result = MagicMock()
-            result.choices = [choice]
-            return result
-
-        mock_client.chat.completions.create.side_effect = [
-            response('{"wrong": "first"}'),
-            response('{"wrong": "second"}'),
-            response('{"text": "correct"}'),
-        ]
-
-        session = LlmSession(system_prompt="You are a helper.")
-        result = session.call(photos=[], prompt="Return JSON")
-
-    assert result.text == "correct"
-    requests = mock_client.chat.completions.create.call_args_list
-    messages_per_attempt = [call.kwargs["messages"] for call in requests]
-    assert [len(messages) for messages in messages_per_attempt] == [2, 3, 3]
-    final_correction = messages_per_attempt[2][-1]["content"]
-    assert "Field required" in final_correction
-    assert '"wrong"' not in final_correction
-
-
-def test_llm_session_uses_configured_retry_count(monkeypatch):
-    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 1)
-
-    with patch("photools3.llm_session.OpenAI") as mocked_openai:
-        mock_client = MagicMock()
-        mocked_openai.return_value = mock_client
-        mock_model = MagicMock()
-        mock_model.id = "gpt-4o"
-        mock_client.models.list.return_value.data = [mock_model]
-
-        def response(content):
-            message = MagicMock()
-            message.content = content
-            choice = MagicMock()
-            choice.message = message
-            result = MagicMock()
-            result.choices = [choice]
-            return result
-
-        mock_client.chat.completions.create.side_effect = [
-            response('{"wrong": "first"}'),
-            response('{"wrong": "second"}'),
-            response('{"text": "would be a third call"}'),
-        ]
-        session = LlmSession(system_prompt="You are a helper.")
-
-        with (
-            patch.object(session, "flush_model") as flush_model,
-            pytest.raises(ValueError, match="after 1 retries"),
-        ):
-            session.call(photos=[])
-
-    assert mock_client.chat.completions.create.call_count == 2
-    flush_model.assert_called_once()
-
-
-def test_llm_session_reloads_after_timeout_and_retries(monkeypatch):
-    monkeypatch.setattr(
-        "photools3.llm_session.settings.llm_request_timeout_seconds", 10
-    )
-    monkeypatch.setattr("photools3.llm_session.settings.llm_max_completion_tokens", 768)
-    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 1)
-
-    with patch("photools3.llm_session.OpenAI") as mocked_openai:
-        mock_client = MagicMock()
-        mocked_openai.return_value = mock_client
-        mock_model = MagicMock()
-        mock_model.id = "gpt-4o"
-        mock_client.models.list.return_value.data = [mock_model]
-
-        response_message = MagicMock(content='{"text": "recovered"}')
-        response_choice = MagicMock(message=response_message)
-        success_response = MagicMock(choices=[response_choice])
-        mock_client.chat.completions.create.side_effect = [
-            openai.APITimeoutError(request=MagicMock()),
-            success_response,
-        ]
-
-        session = LlmSession(system_prompt="You are a helper.")
-        with patch.object(session, "flush_model") as flush_model:
-            result = session.call(photos=[])
-
-    assert result.text == "recovered"
-    flush_model.assert_called_once()
-    calls = mock_client.chat.completions.create.call_args_list
-    assert len(calls) == 2
-    assert all(call.kwargs["timeout"] == 10 for call in calls)
-    assert all(call.kwargs["max_tokens"] == 768 for call in calls)
-    mocked_openai.assert_called_once_with(
-        base_url=session.base_url,
-        api_key=session.api_key,
-        max_retries=0,
-    )
-
-
-@pytest.mark.parametrize("failure_kind", ["invalid_response", "timeout"])
-def test_llm_session_reloads_when_retry_limit_is_exhausted(monkeypatch, failure_kind):
-    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 0)
-
-    with patch("photools3.llm_session.OpenAI") as mocked_openai:
-        mock_client = MagicMock()
-        mocked_openai.return_value = mock_client
-        mock_model = MagicMock()
-        mock_model.id = "gpt-4o"
-        mock_client.models.list.return_value.data = [mock_model]
-
-        if failure_kind == "timeout":
-            mock_client.chat.completions.create.side_effect = openai.APITimeoutError(
-                request=MagicMock()
-            )
-        else:
-            response_message = MagicMock(content='{"wrong": "shape"}')
-            response_choice = MagicMock(message=response_message)
-            mock_client.chat.completions.create.return_value = MagicMock(
-                choices=[response_choice]
-            )
-
-        session = LlmSession(system_prompt="You are a helper.")
-        with (
-            patch.object(session, "flush_model") as flush_model,
-            pytest.raises(ValueError),
-        ):
-            session.call(photos=[])
-
-    flush_model.assert_called_once()
-    assert mock_client.chat.completions.create.call_count == 1
-
-
-def test_invalid_keywords_get_one_correction_then_invalid_tags_are_dropped(monkeypatch):
-    monkeypatch.setattr("photools3.llm_session.settings.llm_max_keywords", 3)
-    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 5)
-
-    with patch("photools3.llm_session.OpenAI") as mocked_openai:
-        mock_client = MagicMock()
-        mocked_openai.return_value = mock_client
-        mock_model = MagicMock()
-        mock_model.id = "gpt-4o"
-        mock_client.models.list.return_value.data = [mock_model]
-
-        def response(content):
-            message = MagicMock(content=content)
-            choice = MagicMock(message=message)
-            return MagicMock(choices=[choice])
-
-        mock_client.chat.completions.create.side_effect = [
-            response(
-                '{"caption": "A caption", "tags": ["not_allowed", "architektur"]}'
-            ),
-            response(
-                '{"caption": "A caption", '
-                '"tags": ["still_invalid", "architektur", "personen"]}'
-            ),
-        ]
-        session = LlmSessionCaptionsAndKeywords()
-        with patch.object(session, "flush_model") as flush_model:
-            result = session.call(photos=[])
-
-    assert result.tags == ["architektur", "personen"]
-    assert mock_client.chat.completions.create.call_count == 2
-    flush_model.assert_not_called()
-    correction = mock_client.chat.completions.create.call_args_list[1].kwargs[
-        "messages"
-    ][-1]["content"]
-    assert "not_allowed" in correction
-    assert '"caption"' not in correction
-
-
-def test_excess_keywords_are_truncated_to_configured_limit(monkeypatch):
-    monkeypatch.setattr("photools3.llm_session.settings.llm_max_keywords", 2)
-
-    with patch("photools3.llm_session.OpenAI") as mocked_openai:
-        mock_client = MagicMock()
-        mocked_openai.return_value = mock_client
-        mock_model = MagicMock()
-        mock_model.id = "gpt-4o"
-        mock_client.models.list.return_value.data = [mock_model]
-        message = MagicMock(
-            content=(
-                '{"caption": "A caption", '
-                '"tags": ["architektur", "personen", "landschaft"]}'
-            )
-        )
-        mock_client.chat.completions.create.return_value = MagicMock(
-            choices=[MagicMock(message=message)]
-        )
-
-        result = LlmSessionCaptionsAndKeywords().call(photos=[])
-
-    assert result.tags == ["architektur", "personen"]
-    assert mock_client.chat.completions.create.call_count == 1
-
-
-def test_keyword_response_schema_includes_configured_max_items():
-    schema = LlmResponseCaptionsAndKeywords.model_json_schema()
-
-    assert schema["properties"]["tags"]["maxItems"] == 15

test/test_lm_studio.py

@@ -1,102 +0,0 @@
-import json
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-from photools3.lm_studio import LMStudioError, LMStudioModelManager
-
-
-def _http_response(payload: dict):
-    response = MagicMock()
-    response.__enter__.return_value = response
-    response.read.return_value = json.dumps(payload).encode("utf-8")
-    return response
-
-
-def test_reload_model_unloads_and_reloads_preserving_context_config():
-    model_response = {
-        "models": [
-            {
-                "key": "qwen/qwen3-vl-4b",
-                "loaded_instances": [
-                    {
-                        "id": "qwen/qwen3-vl-4b-instance",
-                        "config": {
-                            "context_length": 16384,
-                            "eval_batch_size": 256,
-                            "flash_attention": True,
-                            "offload_kv_cache_to_gpu": False,
-                            "parallel": 2,
-                        },
-                    }
-                ],
-            }
-        ]
-    }
-    with patch(
-        "photools3.lm_studio.urlopen",
-        side_effect=[
-            _http_response(model_response),
-            _http_response({"instance_id": "qwen/qwen3-vl-4b-instance"}),
-            _http_response({"instance_id": "qwen/qwen3-vl-4b-instance"}),
-        ],
-    ) as mocked_urlopen:
-        manager = LMStudioModelManager(
-            "http://localhost:1234/v1", "test-token", timeout=12
-        )
-        manager.reload_model("qwen/qwen3-vl-4b")
-
-    requests = [call.args[0] for call in mocked_urlopen.call_args_list]
-    assert requests[0].full_url == "http://localhost:1234/api/v1/models"
-    assert requests[0].get_method() == "GET"
-    assert requests[0].get_header("Authorization") == "Bearer test-token"
-
-    assert requests[1].full_url.endswith("/models/unload")
-    assert json.loads(requests[1].data) == {"instance_id": "qwen/qwen3-vl-4b-instance"}
-
-    assert requests[2].full_url.endswith("/models/load")
-    assert json.loads(requests[2].data) == {
-        "model": "qwen/qwen3-vl-4b",
-        "context_length": 16384,
-        "eval_batch_size": 256,
-        "flash_attention": True,
-        "offload_kv_cache_to_gpu": False,
-    }
-
-
-def test_reload_model_loads_by_key_if_no_instance_is_loaded():
-    model_response = {"models": [{"key": "qwen/model", "loaded_instances": []}]}
-    with patch(
-        "photools3.lm_studio.urlopen",
-        side_effect=[
-            _http_response(model_response),
-            _http_response({"status": "loaded"}),
-        ],
-    ) as mocked_urlopen:
-        manager = LMStudioModelManager("http://localhost:1234/v1", "token")
-        manager.reload_model("qwen/model")
-
-    assert mocked_urlopen.call_count == 2
-    load_request = mocked_urlopen.call_args.args[0]
-    assert load_request.full_url.endswith("/models/load")
-    assert json.loads(load_request.data) == {"model": "qwen/model"}
-
-
-def test_reload_model_rejects_ambiguous_model_selection():
-    model_response = {
-        "models": [
-            {
-                "key": "qwen/model",
-                "loaded_instances": [
-                    {"id": "instance-1", "config": {}},
-                    {"id": "instance-2", "config": {}},
-                ],
-            }
-        ]
-    }
-    with patch(
-        "photools3.lm_studio.urlopen", return_value=_http_response(model_response)
-    ):
-        manager = LMStudioModelManager("http://localhost:1234/v1", "token")
-        with pytest.raises(LMStudioError, match="Multiple loaded instances"):
-            manager.reload_model("qwen/model")

test/test_metadata_writer.py

@@ -1,6 +1,5 @@
 import shutil
 from pathlib import Path
-from unittest.mock import patch
 
 import pytest
 
@@ -212,32 +211,3 @@ def test_set_keywords(make_dng_file, pic, expected_keywords):
 def test_set_keywords_on_existing():
     # TODO
     ...
-
-
-def test_write_caption_and_keywords_in_one_metadata_update(tmp_path):
-    writer = MetadataWriter(tmp_path / "photo.jpg")
-    with (
-        patch.object(writer, "get_caption", return_value="Original caption"),
-        patch.object(writer, "get_keywords", return_value=["existing"]),
-        patch.object(writer, "set_metadata") as set_metadata,
-    ):
-        writer.write(caption="Generated caption", keywords=["architektur"])
-
-    set_metadata.assert_called_once()
-    metadata_tags = set_metadata.call_args.args[0]
-    assert metadata_tags["XMP:Description"] == (
-        f"Original caption. {PREFIX}Generated caption"
-    )
-    assert set(metadata_tags["XMP:Subject"]) == {
-        "existing",
-        f"{PREFIX}architektur",
-    }
-
-
-def test_set_metadata_surfaces_write_failures(tmp_path):
-    writer = MetadataWriter(tmp_path / "photo.jpg")
-    with patch("photools3.metadata_writer.exiftool.ExifToolHelper") as exif:
-        exif_instance = exif.return_value.__enter__.return_value
-        exif_instance.execute.side_effect = OSError("ExifTool failed")
-        with pytest.raises(RuntimeError, match="Could not write metadata"):
-            writer.set_metadata({"XMP:Description": "caption"})

test/test_photo.py

@@ -48,15 +48,6 @@ def test_load_photo():
     assert isinstance(photo.image, Image.Image)
 
 
-def test_photo_context_manager_closes_image(tmp_path):
-    path = _make_image_file(tmp_path, "RGB")
-    with Photo(path) as photo:
-        image = photo.image
-        assert image.fp is not None
-
-    assert image.fp is None
-
-
 def test_resize_method_exists():
     """Test that resize method exists and returns a Photo instance"""
     photo = Photo(PATHS_DNG_EXAMPLES[0])

test/test_processing.py

@@ -1,129 +0,0 @@
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-from PIL import Image
-
-from photools3.models import (
-    LlmResponseCaption,
-    LlmResponseCaptionsAndKeywords,
-    LlmResponseKeywords,
-)
-from photools3.processing import (
-    ProcessingMode,
-    _create_session,
-    process_batch,
-    process_photo,
-)
-
-
-@pytest.fixture
-def image_path(tmp_path):
-    path = tmp_path / "photo.png"
-    Image.new("RGB", (20, 20)).save(path)
-    return path
-
-
-@pytest.mark.parametrize(
-    ("mode", "response", "expected_metadata"),
-    [
-        (
-            ProcessingMode.KEYWORDS,
-            LlmResponseKeywords(tags=["architektur"]),
-            {"caption": None, "keywords": ["architektur"]},
-        ),
-        (
-            ProcessingMode.CAPTIONS,
-            LlmResponseCaption(caption="Ein Haus"),
-            {"caption": "Ein Haus", "keywords": None},
-        ),
-        (
-            ProcessingMode.BOTH,
-            LlmResponseCaptionsAndKeywords(caption="Ein Haus", tags=["architektur"]),
-            {"caption": "Ein Haus", "keywords": ["architektur"]},
-        ),
-    ],
-)
-def test_process_photo_writes_only_requested_metadata(
-    image_path, mode, response, expected_metadata
-):
-    session = MagicMock()
-    session.call.return_value = response
-
-    with (
-        patch("photools3.processing._create_session", return_value=session),
-        patch("photools3.processing.MetadataWriter") as writer_class,
-    ):
-        result = process_photo(image_path, mode)
-
-    assert result == response
-    session.call.assert_called_once()
-    session.close.assert_called_once()
-    writer_class.assert_called_once_with(image_path)
-    writer_class.return_value.write.assert_called_once_with(**expected_metadata)
-
-
-def test_process_photo_does_not_close_a_caller_owned_session(image_path):
-    session = MagicMock()
-    session.call.return_value = LlmResponseCaption(caption="Ein Haus")
-
-    with patch("photools3.processing.MetadataWriter"):
-        process_photo(image_path, ProcessingMode.CAPTIONS, session=session)
-
-    session.close.assert_not_called()
-
-
-@pytest.mark.parametrize(
-    ("mode", "constructor"),
-    [
-        (ProcessingMode.KEYWORDS, "LlmSessionKeywords"),
-        (ProcessingMode.CAPTIONS, "LlmSessionCaption"),
-        (ProcessingMode.BOTH, "LlmSessionCaptionsAndKeywords"),
-    ],
-)
-def test_create_session_selects_mode_specific_response_model(mode, constructor):
-    with patch(f"photools3.processing.{constructor}") as session_class:
-        assert _create_session(mode) is session_class.return_value
-
-    session_class.assert_called_once_with()
-
-
-def test_process_batch_is_sequential_reuses_session_and_continues_after_error():
-    active_photos = 0
-    max_active_photos = 0
-
-    class TrackingPhoto:
-        def __init__(self, path):
-            self.path = path
-
-        def __enter__(self):
-            nonlocal active_photos, max_active_photos
-            active_photos += 1
-            max_active_photos = max(max_active_photos, active_photos)
-            return self
-
-        def __exit__(self, exc_type, exc_value, traceback):
-            nonlocal active_photos
-            active_photos -= 1
-
-    session = MagicMock()
-    response = LlmResponseCaptionsAndKeywords(caption="Ein Haus", tags=["architektur"])
-    session.call.side_effect = [response, RuntimeError("temporary error"), response]
-    paths = [Path("one.jpg"), Path("two.jpg"), Path("three.jpg")]
-
-    with (
-        patch("photools3.processing._create_session", return_value=session),
-        patch("photools3.processing.Photo", TrackingPhoto),
-        patch("photools3.processing.MetadataWriter") as writer_class,
-    ):
-        outcomes = process_batch(paths, ProcessingMode.BOTH)
-
-    assert [outcome.succeeded for outcome in outcomes] == [True, False, True]
-    assert [outcome.path for outcome in outcomes] == paths
-    assert outcomes[1].error == "RuntimeError: temporary error"
-    assert max_active_photos == 1
-    assert active_photos == 0
-    assert session.call.call_count == 3
-    assert all(len(call.args[0]) == 1 for call in session.call.call_args_list)
-    assert writer_class.return_value.write.call_count == 2
-    session.close.assert_called_once()

uv.lock

@@ -780,7 +780,7 @@ requires-dist = [
     { name = "pytest", specifier = ">=9.1.1" },
     { name = "ruff", specifier = ">=0.16.9" },
     { name = "typer", specifier = ">=0.12.0" },
-    { name = "utils", specifier = ">=0.1.3", index = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/simple" },
+    { name = "utils", specifier = ">=0.1.2", index = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/simple" },
 ]
 
 [[package]]
@@ -1205,7 +1205,7 @@ wheels = [
 
 [[package]]
 name = "utils"
-version = "0.1.3"
+version = "0.1.2"
 source = { registry = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/simple" }
 dependencies = [
     { name = "black" },
@@ -1217,9 +1217,9 @@ dependencies = [
     { name = "twine" },
     { name = "uvicorn" },
 ]
-sdist = { url = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/files/utils/0.1.3/utils-0.1.3.tar.gz", hash = "sha256:8970a179573a63ab4c546cfd1da8ceca0d915bc4357b49ccffaba9b3d4a0195e" }
+sdist = { url = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/files/utils/0.1.2/utils-0.1.2.tar.gz", hash = "sha256:c711755c3e2fabd9cd380e25e45bc19da9b901868a6eb67028968803c725b118" }
 wheels = [
-    { url = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/files/utils/0.1.3/utils-0.1.3-py3-none-any.whl", hash = "sha256:c24078ba05f9f5b41fc3909598084b93c6df59f4ca60001e1e0b4fa55797e642" },
+    { url = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/files/utils/0.1.2/utils-0.1.2-py3-none-any.whl", hash = "sha256:f3335ba10bd8eb9f5c000287a6ee41987b188372f3dca2783d99e6e70513a5d7" },
 ]
 
 [[package]]