Compare commits

..
16 Commits
Author SHA1 Message Date
mathias2784 0d6272b4c3 E2e workflow (#21)
Tests / test_and_lint (push) Successful in 49s
Reviewed-on: #21
Co-authored-by: Mathias Scheider <mathias2784@web.de>
2026-10-18 12:02:18 +00:00
mathias2784 ad7dbfdd11 E2e workflow (#20)
Tests / test_and_lint (push) Successful in 34s
Reviewed-on: #20
Co-authored-by: Mathias Scheider <mathias2784@web.de>
2026-10-18 07:24:49 +00:00
mathias2784 ff588f899e Llm session refactor (#19)
Tests / test_and_lint (push) Successful in 34s
Reviewed-on: #19
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-18 06:29:37 +00:00
mathias2784 b6be043426 Mypy integration (#18)
Tests / test_and_lint (push) Successful in 32s
Reviewed-on: #18
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-10 18:37:09 +00:00
mathias2784 b9632d46b2 Metadata writer (#17)
Tests / test_and_lint (push) Successful in 27s
Reviewed-on: #17
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-10 12:22:15 +00:00
mathias2784 6a12144e39 Logging (#16)
Tests / test_and_lint (push) Successful in 12s
Reviewed-on: #16
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-03 08:24:17 +00:00
mathias2784 401488fa7e Llm client quality (#14)
Tests / test_and_lint (push) Successful in 14s
Reviewed-on: #14
Co-authored-by: Mathias Scheider <mathias2784@web.de>
2026-10-02 21:31:21 +00:00
mathias2784 419fb46c3d refactor: simplify CLI command logic by extracting helper functions (#13)
Tests / test_and_lint (push) Successful in 14s
Reviewed-on: #13
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-02 12:03:20 +00:00
mathias2784 ce2da9d953 Add tests for LlmSession and LlmSessionCaptionsAndKeywords in both integration and mocked scenarios (#12)
Tests / test_and_lint (push) Successful in 13s
Reviewed-on: #12
Co-authored-by: Mathias Scheider <mathias2784@web.de>
2026-10-02 09:21:38 +00:00
mathias2784 906707de55 first successful run of photo llm (#11)
Tests / test_and_lint (push) Successful in 11s
Reviewed-on: #11
Co-authored-by: Mathias Scheider <mathias2784@web.de>
2026-10-02 08:16:20 +00:00
mathias2784 0c33153c98 Photo basics (#10)
Tests / test_and_lint (push) Successful in 10s
Reviewed-on: #10
Co-authored-by: Mathias Scheider <mathias2784@web.de>
2026-09-30 21:01:20 +00:00
mathias2784 d9758c598d add ci yaml (#9)
Tests / test_and_lint (push) Successful in 7s
Reviewed-on: #9
Co-authored-by: Mathias Scheider <mathias2784@web.de>
2026-09-30 18:49:48 +00:00
mathias2784 3cd8541b24 first (very trivial) test 2026-09-29 22:22:28 +02:00
mathias2784 2dd82ff803 some more skeleton 2026-09-29 22:22:10 +02:00
mathias2784 d32df3e54a add packages: pydantic pydantic-settings openai 2026-09-29 22:21:35 +02:00
mathias2784 e0c38fc2c6 gitignore .env 2026-09-29 22:20:33 +02:00
33 changed files with 4294 additions and 23 deletions
+10
View File
@@ -0,0 +1,10 @@
LLM_API_BASE_URL = "http://localhost:1234/v1"
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
+37
View File
@@ -0,0 +1,37 @@
---
name: Tests
on: [push]
jobs:
test_and_lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.14"
- name: Install uv
run: pip install uv
- name: Install dependencies
run: uv sync --frozen
- name: copy env
run: cp .env.template .env
- name: Install exiftool (system dependency)
run: |
if command -v exiftool >/dev/null 2>&1; then
echo "exiftool already installed"
elif command -v sudo >/dev/null 2>&1; then
sudo apt-get update && sudo apt-get install -y --no-install-recommends libimage-exiftool-perl
else
apt-get update && apt-get install -y --no-install-recommends libimage-exiftool-perl
fi
- name: Run pytest
run: uv run pytest .
- name: Lint (ruff)
run: uv run ruff check .
- name: Typing (mypy)
run: uv run mypy .
- name: Check formatting (black)
run: uv run black --check . --extend-exclude .venv
+3
View File
@@ -8,3 +8,6 @@ wheels/
# Virtual environments # Virtual environments
.venv .venv
.env
logs/
+42
View File
@@ -0,0 +1,42 @@
# 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.
+101
View File
@@ -0,0 +1,101 @@
# 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
```
+371
View File
@@ -0,0 +1,371 @@
architektur
architektur|gebäude
architektur|bauwerke
architektur|architecture
architektur|brücke
architektur|brücke|bridge
architektur|burg
architektur|burg|castle
architektur|dach
architektur|dach|roof
architektur|dächer
architektur|garten
architektur|garten|garden
architektur|gleis
architektur|gleis|rail
architektur|industrie
architektur|kirche
architektur|kirche|church
architektur|leuchtturm
architektur|leuchtturm|lighhouse
architektur|levada
architektur|modell
architektur|park
architektur|ruinen
architektur|ruinen|ruins
architektur|siedlung
architektur|siedlung|dorf ortschaft
architektur|siedlung|village
architektur|skyline
architektur|stadt
architektur|stadt|city
architektur|strasse
architektur|strasse|street
architektur|fenster
architektur|fenster|window
architektur|tür
architektur|tür|door
architektur|treppe
architektur|treppe|stairs
architektur|turm
architektur|turm|tower
architektur|hafen
architektur|hafen|harbor
architektur|hafen|port
architektur|bahnhof
architektur|bahnhof|train station
architektur|museum
essen&trinken
essen&trinken|Lebensmittel
essen&trinken|food
essen&trinken|braten
essen&trinken|buffett
essen&trinken|burger
essen&trinken|kaffee
essen&trinken|kaffee|coffee
essen&trinken|shrimps
essen&trinken|tapas
essen&trinken|bier
essen&trinken|bier|beer
essen&trinken|wein
essen&trinken|wein|wine
essen&trinken|obst
essen&trinken|obst|fruit
essen&trinken|gemüse
essen&trinken|gemüse|vegetables
essen&trinken|kuchen
essen&trinken|kuchen|cake
essen&trinken|dessert
himmel/wetter
himmel/wetter|weather
himmel/wetter|sky
himmel/wetter|abendrot
himmel/wetter|abendrot|dusk
himmel/wetter|blauer himmel
himmel/wetter|blauer himmel|blue sky
himmel/wetter|diesig
himmel/wetter|diesig|hazy
himmel/wetter|diesig|misty
himmel/wetter|eis
himmel/wetter|eis|ice
himmel/wetter|grau
himmel/wetter|grau|grey
himmel/wetter|mond
himmel/wetter|mond|moon
himmel/wetter|morgengrauen
himmel/wetter|morgengrauen|dawn
himmel/wetter|morgenrot
himmel/wetter|morgenrot|dawn
himmel/wetter|nebel
himmel/wetter|nebel|fog
himmel/wetter|regen
himmel/wetter|regen|rain
himmel/wetter|regenbogen
himmel/wetter|regenbogen|rainbow
himmel/wetter|schnee
himmel/wetter|schnee|snow
himmel/wetter|sonne
himmel/wetter|sonne|sun
himmel/wetter|sonne|sonnenaufgang
himmel/wetter|sonne|sonnenaufgang|sunrise
himmel/wetter|sonne|sonnenstrahlen
himmel/wetter|sonne|sonnenstrahlen|sun beams
himmel/wetter|sonne|sonnenstrahlen|light rays
himmel/wetter|sonne|sonnenuntergang
himmel/wetter|sonne|sonnenuntergang|sunset
himmel/wetter|sterne
himmel/wetter|sterne|stars
himmel/wetter|windig
himmel/wetter|windig|windy
himmel/wetter|wolken
himmel/wetter|wolken|clouds
himmel/wetter|gewitter
himmel/wetter|gewitter|thunderstorm
himmel/wetter|blitz
himmel/wetter|blitz|lightning
himmel/wetter|frost
jahreszeit
jahreszeit|season
jahreszeit|frühling
jahreszeit|frühling|spring
jahreszeit|sommer
jahreszeit|sommer|summer
jahreszeit|herbst
jahreszeit|herbst|autumn
jahreszeit|herbst|fall
jahreszeit|winter
landschaft
landschaft|landscape
landschaft|berge
landschaft|berge|mountains
landschaft|dünen
landschaft|dünen|dunes
landschaft|dünen|sand
landschaft|feld
landschaft|feld|field
landschaft|feld|meadow
landschaft|felsen
landschaft|felsen|rock
landschaft|höhle
landschaft|höhle|cave
landschaft|insel
landschaft|insel|island
landschaft|klippen
landschaft|klippen|cliffs
landschaft|küste
landschaft|küste|coast
landschaft|steinbruch
landschaft|steinbruch|quarry
landschaft|strand
landschaft|strand|beach
landschaft|vulkan
landschaft|vulkan|vulcano
landschaft|wald
landschaft|wald|forest
landschaft|wald|wood
landschaft|wasser
landschaft|wasser|water
landschaft|wasser|bach
landschaft|wasser|bach|creek
landschaft|wasser|brandung
landschaft|wasser|brandung|breakers
landschaft|wasser|brandung|surf
landschaft|wasser|fluss
landschaft|wasser|fluss|river
landschaft|wasser|hochwasser
landschaft|wasser|hochwasser|flood
landschaft|wasser|hochwasser|high water
landschaft|wasser|meer
landschaft|wasser|meer|sea
landschaft|wasser|meer|ocean
landschaft|wasser|see
landschaft|wasser|see|lake
landschaft|wasser|wasserfall
landschaft|wasser|wasserfall|waterfall
landschaft|weinberge
landschaft|weinberge|vineyard
landschaft|wüste
landschaft|wüste|desert
landschaft|hügel
landschaft|hügel|hills
landschaft|tal
landschaft|tal|valley
landschaft|schlucht
landschaft|schlucht|canyon
landschaft|schlucht|gorge
landschaft|gipfel
landschaft|gipfel|summit
landschaft|gipfel|peak
nacht
nacht|night
nacht|langzeitbelichtung
nacht|langzeitbelichtung|long exposure
nacht|lichtspuren
nacht|lichtspuren|light trails
nacht|nachtaufnahme
nacht|nachtaufnahme|night shot
nacht|stadtlichter
nacht|stadtlichter|city lights
personen
personen|auge
personen|auge|eye
personen|geste
personen|geste|gesture
personen|gruppenbild
personen|gruppenbild|group
personen|menschenmenge
personen|menschenmenge|crowd
personen|portrait
personen|pose
personen|quatsch
personen|quatsch|silly
personen|verkleidung
personen|kind
personen|kind|kinder
personen|kind|child
personen|kind|children
personen|paar
personen|paar|couple
personen|silhouette
personen|schatten
personen|schatten|shadow
pflanze
pflanze|plant
pflanze|bäume
pflanze|bäume|trees
pflanze|blumen
pflanze|blumen|flower
pflanze|blüten
pflanze|blüten|flower
pflanze|blüten|bloomy
pflanze|grass
pflanze|grass|gras
pflanze|kaktus
pflanze|kaktus|cactus
pflanze|laub
pflanze|laub|foliage
pflanze|moos
pflanze|moos|moss
pflanze|palmen
pflanze|palmen|palms
pflanze|pilze
pflanze|pilze|mushrooms
pflanze|schilf
pflanze|schilf|reed
pflanze|sonnenblumen
pflanze|sonnenblumen|sunflower
pflanze|sonnentau
pflanze|sonnentau|sundew
pflanze|venusfliegenfalle
pflanze|venusfliegenfalle|venus trap
pflanze|farn
pflanze|farn|fern
pflanze|strauch
pflanze|strauch|bush
pflanze|strauch|shrub
pflanze|efeu
pflanze|efeu|ivy
pflanze|hecke
pflanze|hecke|hedge
sport
sport|fussball
sport|fussball|football
sport|fussball|soccer
sport|schwimmen
sport|schwimmen|swimming
sport|klettern
sport|klettern|climbing
sport|wandern
sport|wandern|hiking
sport|radfahren
sport|radfahren|cycling
sport|ski
sport|ski|skiing
stil
stil|style
stil|schwarzweiss
stil|schwarzweiss|black and white
stil|schwarzweiss|bw
stil|vintage
stil|hdr
stil|panorama
technik
technik|technology
technik|auto
technik|auto|car
technik|eisenbahn
technik|eisenbahn|train
technik|fahrrad
technik|fahrrad|bike
technik|fahrrad|bicycle
technik|flugzeug
technik|flugzeug|plane
technik|handy
technik|handy|smartphone
technik|handy|cell phone
technik|kamera
technik|kamera|camera
technik|musikinstrument
technik|musikinstrument|musical instrument
technik|musikinstrument|Gitarre
technik|musikinstrument|Gitarre|guitar
technik|musikinstrument|Gitarren Effektgerät
technik|musikinstrument|Gitarrenverstärker
technik|musikinstrument|Gitarrenverstärker|guitar amp
technik|objektiv
technik|objektiv|lens
technik|schiff
technik|schiff|ship
technik|seilbahn
technik|seilbahn|teleferico
technik|seilbahn|cable car
technik|motorrad
technik|motorrad|motorcycle
technik|lastwagen
technik|lastwagen|truck
technik|bus
technik|helikopter
technik|helikopter|helicopter
technik|drohne
technik|drohne|drone
technik|windrad
technik|windrad|wind turbine
tier
tier|animal
tier|elefant
tier|elefant|elephant
tier|fisch
tier|fisch|fish
tier|insekten
tier|insekten|insects
tier|insekten|biene
tier|insekten|biene|bee
tier|katze
tier|katze|cat
tier|nager
tier|nager|gnawer
tier|reptil
tier|reptil|reptile
tier|rind
tier|rind|cattle
tier|Rot$DamWild
tier|Rot$DamWild|deer
tier|spinne
tier|spinne|spider
tier|tiger
tier|vogel
tier|vogel|bird
tier|waschbär
tier|waschbär|raccoon
tier|Wildschwein
tier|Wildschwein|boar
tier|ZiegeBockSchaf
tier|ZiegeBockSchaf|goat
tier|hund
tier|hund|dog
tier|pferd
tier|pferd|horse
tier|eichhörnchen
tier|eichhörnchen|squirrel
tier|fuchs
tier|fuchs|fox
tier|seevogel
tier|seevogel|seabird
tier|eidechse
tier|eidechse|lizard
tier|frosch
tier|frosch|frog
tier|schmetterling
tier|schmetterling|butterfly
tier|libelle
tier|libelle|dragonfly
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+25
View File
@@ -8,7 +8,17 @@ authors = [
] ]
requires-python = ">=3.14" requires-python = ">=3.14"
dependencies = [ dependencies = [
"black>=26.5.1",
"mypy>=2.4.0",
"openai>=3.22.0",
"pillow>=12.3.0",
"pydantic>=2.13.5", "pydantic>=2.13.5",
"pydantic-settings>=2.15.0",
"pyexiftool>=0.5.6",
"pytest>=9.1.1",
"ruff>=0.16.9",
"typer>=0.12.0",
"utils>=0.1.3",
] ]
[project.scripts] [project.scripts]
@@ -17,3 +27,18 @@ photools3 = "photools3:main"
[build-system] [build-system]
requires = ["uv_build>=0.12.19,<0.13.0"] requires = ["uv_build>=0.12.19,<0.13.0"]
build-backend = "uv_build" build-backend = "uv_build"
[tool.uv.sources]
utils = { index = "gitea" }
[tool.black]
target-version = [ "py314" ]
[[tool.mypy.overrides]]
module = ["exiftool", "exiftool.*", "utils", "utils.*"]
ignore_missing_imports = true
[[tool.uv.index]]
name = "gitea"
url = "https://gitea.schneiderserver.site/api/packages/mathias2784/pypi/simple"
explicit = true
+5 -2
View File
@@ -1,2 +1,5 @@
def main() -> None: from .cli import app
print("Hello from photools3!")
def main():
app()
+4
View File
@@ -0,0 +1,4 @@
from .cli import app
if __name__ == "__main__":
app()
+163
View File
@@ -0,0 +1,163 @@
import logging
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.photo import Photo
from photools3.processing import (
ProcessingMode,
process_batch,
process_photo,
)
from photools3.settings import settings
app = typer.Typer(help="PhotoTools3: A CLI for image processing.")
LoggingSettings(LOG_LEVEL=settings.log_level)
for name in (
"uvicorn",
"uvicorn.error",
"uvicorn.access",
"PIL",
"openai",
"httpx2",
"httpcore2",
):
logging.getLogger(name).setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
@app.callback(invoke_without_command=True)
def main(
ctx: typer.Context,
version: bool | None = typer.Option(None, "--version", is_eager=True),
):
"""Main entry point for PhotoTools3 CLI."""
if version:
typer.echo("photools3 version 0.1.0")
raise typer.Exit()
if ctx.invoked_subcommand is None:
typer.echo(ctx.get_help())
raise typer.Exit()
@app.command("metadata-magic")
def metadata_magic(
paths: Annotated[
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."""
_run_processing(paths, mode, recursive)
@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."""
try:
with LlmSessionQuality() as session, Photo(file_path) as photo:
result = session.call([photo])
logger.info("Quality assessment for %s: %s", file_path, result)
except FileNotFoundError:
logger.error("Photo not found: %s", file_path)
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)
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)
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__":
app()
+349 -6
View File
@@ -1,13 +1,356 @@
import json
import logging
import re
import time
from photools3.models import LlmRequest, LlmResponse 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
logger = logging.getLogger(__name__)
def _clean_llm_response(txt: str):
# remove ```json ... ``` wrap
txt = re.sub(r"^```json(.*)```$", r"\1", txt, flags=re.DOTALL).strip()
return txt
def _correction_prompt(instruction: str) -> LlmRequestMessage:
return LlmRequestMessage(
role=LlmMessageRole.user,
content=instruction,
)
def _request_with_correction(base_request: LlmRequest, instruction: str) -> LlmRequest:
return base_request.model_copy(
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)
)
class LlmSession: class LlmSession:
def __init__(self): base_url: str = str(settings.llm_api_base_url)
... api_key: str = settings.llm_api_key
response_model: type[BaseModel] = LlmResponseGeneral
prompt: str = ""
def prompt(self, request: LlmRequest) -> LlmResponse: def __init__(
... 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
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): 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(
**request.model_dump(),
timeout=settings.llm_request_timeout_seconds,
)
return response.choices[choice_no].message.content
def is_api_reachable(self) -> bool:
try:
_ = self.client.models.list().data
return True
except openai.APIConnectionError:
return False
@property
def models(self):
if not self._models:
self._models = self.client.models.list().data
return self._models
def call(
self,
photos: list[Photo],
model_id=None,
max_retries: int | None = None,
prompt: str | None = None,
):
retry_limit = settings.llm_max_retries if max_retries is None else max_retries
if retry_limit < 0:
raise ValueError("max_retries must be greater than or equal to zero")
if model_id is None:
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)
request = base_request
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:
logger.warning(
"LLM request timed out after %.2f seconds; reloading model",
time.perf_counter() - request_started,
)
self.flush_model()
if attempt < retry_limit:
continue
raise ValueError(
f"LLM request timed out after {retry_limit + 1} attempts"
) from 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,
photos: list[Photo],
model_id: str,
) -> LlmRequest:
"""Build a valid request that complies with openai API"""
return LlmRequest(
model=model_id,
temperature=self.temperature,
max_tokens=settings.llm_max_completion_tokens,
messages=[
# System prompt
LlmRequestMessage(
role=LlmMessageRole.system,
content=self.system_prompt,
),
# User prompt
LlmRequestMessage(
role=LlmMessageRole.user,
content=[
# Text prompt
LlmMessageContent(
type=LlmMessageType.text,
text=prompt,
),
# Image contents
*[
LlmMessageContent(
type=LlmMessageType.image_url,
image_url=photo.to_llm_image_url(),
)
for photo in photos
],
],
),
],
)
class LlmSessionCaptionsAndKeywords(LlmSession):
response_model = LlmResponseCaptionsAndKeywords
prompt = "Give me a caption and 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 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()}"
),
temperature=temperature,
)
class LlmSessionQuality(LlmSession):
response_model = LlmResponseQuality
prompt = "Assess the quality of the following photos to help me decide which ones to sort out"
def __init__(self, temperature: float = settings.llm_default_temperature):
super().__init__(
system_prompt=(
"You are a helpful photographer's assistant that helps to identify "
"photos that can be sorted out due to qulity issues:\n"
"Output must be a JSON of the following schema:\n"
f"{self.response_model.model_json_schema()}"
),
temperature=temperature,
)
+135
View File
@@ -0,0 +1,135 @@
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
+180
View File
@@ -0,0 +1,180 @@
import logging
import shutil
from pathlib import Path
from typing import Annotated, Any
import exiftool
from pydantic import StringConstraints
from photools3.settings import settings
log = logging.getLogger(__name__)
_EXIFTOOL_MISSING_MSG = (
"exiftool binary not found in PATH. "
"Install the system package: 'perl-image-exiftool' (Arch) or 'libimage-exiftool-perl' (Debian/Ubuntu)."
)
if shutil.which("exiftool") is None:
raise FileNotFoundError(_EXIFTOOL_MISSING_MSG)
XmpKey = Annotated[str, StringConstraints(pattern=r"^\w+\:\w+")]
type Caption = str
type Keywords = list[str]
type Color = int
type XmpMetadata = dict[str, Any]
XMP_KEYS_CAPTION: list[XmpKey] = [
"XMP:Description",
"EXIF:ImageDescription",
"IPTC:Caption-Abstract",
]
XMP_KEYS_KEYWORDS: list[XmpKey] = [
"XMP:HierarchicalSubject",
"XMP:Subject",
"IPTC:Keywords",
]
def _prefixed(txt: str) -> str:
return f"{settings.metadata_prefix}{txt}"
def _is_prefixed(txt: str) -> bool:
return txt.startswith(settings.metadata_prefix)
def _separate_prefixed_caption(txt: str) -> tuple[str, str | None]:
separated = [x.strip() for x in txt.split(settings.metadata_prefix)]
return (separated[0], separated[1] if len(separated) > 1 else None)
class MetadataWriter:
"""
Write metadata (captions and keywords) to image files.
For DNG files, creates/updates XMP sidecar files.
"""
def __init__(
self,
path: Path,
force_sidecar: bool = False,
append: bool = True,
):
self.path: Path = path
self.force_sidecar: bool = force_sidecar
self.append: bool = append
self._metadata: XmpMetadata | None = None
@property
def sidecar_path(self) -> Path:
return Path(str(self.path) + ".xmp")
@property
def metadata(self) -> Any:
path = self.sidecar_path if self.sidecar_path.exists() else self.path
if not path.exists():
log.warning(f"File not found: {path}")
return None
try:
with exiftool.ExifToolHelper() as exif:
metadata = exif.get_metadata([path])[0]
except Exception as exc: # noqa: BLE001
log.error(f"Could not read metadata from {path}: {exc}")
return None
return metadata
def get_caption(self) -> str:
result = (self.metadata or {}).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 []
if isinstance(result, str):
result = [result]
assert isinstance(result, list)
return result
def set_metadata(self, metadata_tags: dict[str, Any]):
try:
with exiftool.ExifToolHelper() as exif:
if not self.sidecar_path.exists():
exif.execute("-o", str(self.sidecar_path), str(self.path))
exif.set_tags(
[str(self.sidecar_path)],
tags=metadata_tags,
params=["-overwrite_original"],
)
except Exception as exc:
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
):
existing_no_prefix, existing_prefix = _separate_prefixed_caption(
self.get_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} "
if include_prefix and existing_prefix and not override:
return
if not include_prefix and existing_no_prefix and not override:
return
if include_prefix:
caption = f"{existing_no_prefix}{_prefixed(caption)}"
metadata_tags = {key: caption for key in XMP_KEYS_CAPTION}
self.set_metadata(metadata_tags)
def write_keywords(self, keywords: Keywords, include_prefix: bool = True):
if include_prefix:
keywords = list(map(_prefixed, keywords))
existing = self.get_keywords()
keywords = list(set(existing + keywords))
metadata_tags = {tag: keywords for tag in XMP_KEYS_KEYWORDS}
self.set_metadata(metadata_tags)
def write_color(self, color: Color):
self.set_metadata({"XMP-darktable:Colorlabels": color})
def write(
self,
caption: Caption | None = None,
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())
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))
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
if metadata_tags:
self.set_metadata(metadata_tags)
+177 -4
View File
@@ -1,8 +1,181 @@
from pydantic import BaseModel import json
from enum import StrEnum
from functools import lru_cache
from typing import Annotated, Any, Literal
from pydantic import (
BaseModel,
BeforeValidator,
Field,
field_validator,
model_validator,
)
from .settings import KEYWORDS_LIST_PATH, settings
@lru_cache
def load_valid_keywords() -> set[str]:
"""Loads all valid keywords from the configuration file."""
if not KEYWORDS_LIST_PATH.exists():
return set()
keywords = set()
with open(KEYWORDS_LIST_PATH, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
# Each line can contain hierarchical tags separated by '|'
# e.g., 'architektur|gebäude' means both 'architektur' and 'gebäude' are valid.
parts = line.split("|")
current_path = ""
for part in parts:
if current_path:
current_path += "|" + part
else:
current_path = part
keywords.add(current_path)
return keywords
class LlmRequest(BaseModel): class LlmRequest(BaseModel):
... model: str = "gpt-4o"
messages: list[LlmRequestMessage]
temperature: float | None = None
max_tokens: int
class LlmResponse(BaseModel):
... class LlmMessageRole(StrEnum):
system = "system"
user = "user"
class LlmRequestMessage(BaseModel):
role: LlmMessageRole
content: list[LlmMessageContent] | str
def transform_to_image_dict(v: Any) -> Any:
if isinstance(v, str):
return {"url": f"data:image/png;base64,{v}"}
return v
Base64ImageUrl = Annotated[dict[str, str], BeforeValidator(transform_to_image_dict)]
class LlmMessageType(StrEnum):
text = "text"
image_url = "image_url"
class LlmMessageContent(BaseModel):
type: LlmMessageType
text: str | None = None
image_url: Base64ImageUrl | None = None
@model_validator(mode="after")
def _type_matches_content(self):
if self.type == "text" and self.text is None:
raise ValueError("text must be set when type is 'text'")
if self.type == "image_url" and self.image_url is None:
raise ValueError("image_url must be set when type is 'image_url'")
return self
class LlmResponseGeneral(BaseModel):
text: str
class LlmResponseCaption(BaseModel):
caption: Annotated[
str,
Field(
description=(
"A brief but precise description of what the picture shows. "
"No interpretation; just what is visible"
)
),
]
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. "
"Do not enumerate the vocabulary. Allowed keywords: "
f"{allowed_keywords}"
),
)
class _LlmResponseWithTags(BaseModel):
tags: list[str] = _keyword_tags_field()
@field_validator("tags")
@classmethod
def validate_tags(cls, v: list[str]) -> list[str]:
valid_keywords = load_valid_keywords()
for tag in v:
if tag not in valid_keywords:
raise ValueError(
f"'{tag}' is not a valid keyword. Must be one from the configuration."
)
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"],
Field(
description=(
"General quality assessment of the photo. "
'Helping to decide whether to keep ("ok"), '
'to consider sorting out ("review"), '
'or to almost certainly sort out ("not_ok")'
)
),
]
reason: (
None
| Annotated[
Literal["blur", "exposure", "style"],
Field(
description=(
'Possible reasons for a lower-than-"ok" assessment. '
'"blur": Obviously mis-focussed or motion-blurred. '
'"exposure": severly under- or over-exposed. '
'"style": photo has severe style issues (beyond personal taste).'
)
),
]
)
@model_validator(mode="after")
def _not_ok_needs_reason(self):
if self.rating != "ok" and self.reason is None:
raise ValueError("lower-than-ok rating needs a reason")
elif self.rating == "ok" and self.reason is not None:
raise ValueError("ok rating needs no reason")
return self
+40 -9
View File
@@ -1,18 +1,49 @@
import base64
from io import BytesIO
from pathlib import Path from pathlib import Path
from typing import Self
from PIL import Image
from photools3.models import transform_to_image_dict
def _exist_guard(path: Path):
pass
class Photo: class Photo:
def __init__(self): def __init__(self, path: Path):
... self.path = path
self.image = Image.open(path)
@classmethod def close(self):
def from_file(cls, path: Path) -> Photo: self.image.close()
# TODO
return cls()
def resize(self) -> Photo: def __enter__(self) -> Self:
# TODO
return 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)
return self
def to_rgb(self):
if self.image.mode != "RGB":
original = self.image
self.image = original.convert("RGB")
original.close()
return self
def get_encoded(self):
buffer = BytesIO()
self.image.save(buffer, format="JPEG", quality=85)
image_data = base64.b64encode(buffer.getvalue()).decode("utf-8")
return image_data
def to_llm_image_url(self):
return transform_to_image_dict(self.resize().to_rgb().get_encoded())
+138
View File
@@ -0,0 +1,138 @@
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
+31
View File
@@ -0,0 +1,31 @@
from pathlib import Path
from pydantic import Field
from pydantic_core import Url
from pydantic_settings import BaseSettings, SettingsConfigDict
PHOTOOLS3_DIR = Path(__file__).parent.resolve()
SRC_DIR = PHOTOOLS3_DIR.parent
ROOT_DIR = SRC_DIR.parent
EXAMPLES_DIR = ROOT_DIR / "examples"
DNG_EXAMPLES_DIR = EXAMPLES_DIR / "dng"
CONFIG_DIR = ROOT_DIR / "config"
KEYWORDS_LIST_PATH = CONFIG_DIR / "keywords.txt"
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
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
settings = Settings() # type: ignore[call-arg]
View File
+22
View File
@@ -0,0 +1,22 @@
import pytest
from photools3.llm_session import LlmSession
from photools3.photo import Photo
from test.test_photo import PATHS_DNG_EXAMPLES
@pytest.fixture
def test_photo():
return Photo(PATHS_DNG_EXAMPLES[0])
@pytest.fixture
def llm_session():
return LlmSession(system_prompt="You are a helpful assistant.")
@pytest.fixture
def llm_session_captions_and_keywords():
from photools3.llm_session import LlmSessionCaptionsAndKeywords
return LlmSessionCaptionsAndKeywords()
+198
View File
@@ -0,0 +1,198 @@
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
from typer.testing import CliRunner
from photools3.cli import app
from photools3.processing import ProcessingMode
runner = CliRunner()
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):
image_path = tmp_path / "photo.jpg"
image_path.write_bytes(b"test image")
with (
patch("photools3.cli.process_photo") as process_photo,
patch("photools3.cli.process_batch") as process_batch,
):
result = runner.invoke(app, ["metadata-magic", str(image_path)])
assert result.exit_code == 0
process_photo.assert_called_once_with(image_path, ProcessingMode.BOTH)
process_batch.assert_not_called()
def test_metadata_magic_command_selects_caption_mode(tmp_path):
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):
root = tmp_path / "photos"
nested = root / "nested"
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 (
patch("photools3.cli.process_photo") as process_photo,
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
def test_metadata_magic_command_returns_failure_for_single_photo_error(tmp_path):
image_path = tmp_path / "photo.jpg"
image_path.write_bytes(b"test image")
with patch("photools3.cli.process_photo", side_effect=RuntimeError("LLM error")):
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)])
assert result.exit_code == 1
def test_version_command():
result = runner.invoke(app, ["--version"])
assert result.exit_code == 0
assert "photools3 version 0.1.0" in result.stdout
+27
View File
@@ -0,0 +1,27 @@
import pytest
from photools3.llm_session import LlmSession
@pytest.mark.skipif(
not LlmSession(system_prompt="").is_api_reachable(), reason="API not available"
)
def test_llm_session(llm_session, test_photo):
llm_session.system_prompt = "Return only JSON format: {'text': '...'}"
result = llm_session.call(
prompt="keywords for the attached picture", photos=[test_photo]
)
assert result is not None
@pytest.mark.skipif(
not LlmSession(system_prompt="").is_api_reachable(), reason="API not available"
)
def test_llm_session_captions_and_keywords(
llm_session_captions_and_keywords, test_photo
):
result = llm_session_captions_and_keywords.call(photos=[test_photo])
assert result.caption
assert result.tags
assert isinstance(result.caption, str)
assert isinstance(result.tags, list)
+303
View File
@@ -0,0 +1,303 @@
from unittest.mock import MagicMock, patch
import openai
import pytest
from photools3.llm_session import LlmSession, LlmSessionCaptionsAndKeywords
from photools3.models import LlmResponseCaptionsAndKeywords
def test_llm_session_mocked(test_photo):
# We want to mock OpenAI client's chat.completions.create
with patch("photools3.llm_session.OpenAI") as mocked_openai:
# Setup the mock response structure
mock_client = MagicMock()
mocked_openai.return_value = mock_client
# Mock the models list to return a proper object with an ID string
# This prevents Pydantic ValidationError when LlmSession tries to use model_id
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_client.models.list.return_value.data = [mock_model]
# Create a response object: response.choices[0].message.content
mock_response = MagicMock()
mock_choice = MagicMock()
mock_message = MagicMock()
mock_message.content = '{"text": "some text"}'
mock_choice.message = mock_message
mock_response.choices = [mock_choice]
# Configure the client to return our mock response
mock_client.chat.completions.create.return_value = mock_response
# Create a new LlmSession instance that will use the mocked OpenAI client
mocked_session = LlmSession(system_prompt="You are a helper.")
# Run the test
result = mocked_session.call(
prompt="keywords for the attached picture", photos=[test_photo]
)
# Verify the results
assert result.text == "some text"
# Verify that the mock was enough
assert mock_client.chat.completions.create.called
def test_llm_session_captions_and_keywords_mocked(test_photo):
# We want to mock OpenAI client's chat.completions.create
with patch("photools3.llm_session.OpenAI") as mocked_openai:
# Setup the mock response structure
mock_client = MagicMock()
mocked_openai.return_value = mock_client
# Mock the models list to return a proper object with an ID string
# This prevents Pydantic ValidationError when LlmSession tries to use model_id
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_client.models.list.return_value.data = [mock_model]
# Create a response object: response.choices[0].message.content
mock_response = MagicMock()
mock_choice = MagicMock()
mock_message = MagicMock()
mock_message.content = (
'{"caption": "A mocked caption", "tags": ["architektur"]}'
)
mock_choice.message = mock_message
mock_response.choices = [mock_choice]
# Configure the client to return our mock response
mock_client.chat.completions.create.return_value = mock_response
# Create a new LlmSessionCaptionsAndKeywords instance that will use the mocked OpenAI client
from photools3.llm_session import LlmSessionCaptionsAndKeywords
mocked_session = LlmSessionCaptionsAndKeywords()
# Run the test
result = mocked_session.call(photos=[test_photo])
# Verify the results
assert result.caption == "A mocked caption"
assert "architektur" in result.tags
# 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
+102
View File
@@ -0,0 +1,102 @@
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")
+243
View File
@@ -0,0 +1,243 @@
import shutil
from pathlib import Path
from unittest.mock import patch
import pytest
from photools3.metadata_writer import (
MetadataWriter,
_is_prefixed,
_prefixed,
)
from photools3.settings import settings
from test.test_photo import PATHS_DNG_EXAMPLES
PREFIX = settings.metadata_prefix
@pytest.fixture
def make_dng_file(tmp_path):
def _make(src: Path) -> Path:
dst = tmp_path / f"copy_{src.name}"
shutil.copy(src, dst)
return dst
return _make
@pytest.fixture
def metadata_writer_on_existing(make_dng_file):
metadata_writer = MetadataWriter(make_dng_file(PATHS_DNG_EXAMPLES[0]))
metadata_writer.write_caption(
"Let's pretend this is the caption of the original image",
include_prefix=False,
)
return metadata_writer
@pytest.mark.parametrize("pic", PATHS_DNG_EXAMPLES)
def test_read_metadata(make_dng_file, pic):
"""primitive test for `MetadataWriter`"""
file_path = make_dng_file(pic)
result = MetadataWriter(file_path)
assert result
assert result.metadata
assert isinstance(result.metadata, dict)
@pytest.mark.parametrize(
"pic, expected_keywords",
[
(PATHS_DNG_EXAMPLES[0], ["architektur", "zeit|abend"]),
(PATHS_DNG_EXAMPLES[1], ["architektur"]),
(PATHS_DNG_EXAMPLES[2], ["morgen", "tiere|vögel"]),
(
PATHS_DNG_EXAMPLES[3],
[
"abend",
"architektur",
"bewegungsunschärfe",
"lichter",
"mainz",
"spiegelung",
"wasser",
],
),
(
PATHS_DNG_EXAMPLES[4],
["grau", "landschaft", "landschaft|weinberge", "personen", "tiefe", "weg"],
),
],
)
def test_get_keywords(make_dng_file, pic, expected_keywords):
path = make_dng_file(pic)
result = MetadataWriter(path).get_keywords()
assert sorted(result) == sorted(expected_keywords)
def test_prefixed():
txt = "bla_bla"
txt_prefixed = _prefixed(txt)
assert txt_prefixed == f"{PREFIX}bla_bla"
@pytest.mark.parametrize(
"txt, expected_result",
[
(f"{PREFIX}some text", True),
("prefix is missing here", False),
("some_perfix|but not the right one", False),
],
)
def test_is_prefixed(txt, expected_result):
assert _is_prefixed(txt) == expected_result
@pytest.mark.parametrize("pic", PATHS_DNG_EXAMPLES)
def test_read_metadata_sidecar_path(make_dng_file, pic):
file_path = make_dng_file(pic)
metadata_writer = MetadataWriter(file_path)
assert isinstance(metadata_writer.sidecar_path, Path)
assert str(metadata_writer.sidecar_path).lower().endswith(".dng.xmp")
# file does not exits yet, only the name
assert not metadata_writer.sidecar_path.exists()
@pytest.mark.parametrize("pic", PATHS_DNG_EXAMPLES)
def test_set_caption(make_dng_file, pic):
metadata_writer = MetadataWriter(make_dng_file(pic))
assert not metadata_writer.sidecar_path.exists() # sidecar does not exist yet ...
metadata_writer.write_caption("This is a test")
assert metadata_writer.sidecar_path.exists() # ... but now!
result = metadata_writer.get_caption()
assert result
assert result == f"{PREFIX}This is a test"
def test_set_caption_on_existing(metadata_writer_on_existing):
assert (
metadata_writer_on_existing.get_caption()
== "Let's pretend this is the caption of the original image"
)
metadata_writer_on_existing.write_caption("Another caption")
assert (
metadata_writer_on_existing.get_caption()
== f"Let's pretend this is the caption of the original image. {PREFIX}Another caption"
)
metadata_writer_on_existing.write_caption("Updated caption")
assert (
metadata_writer_on_existing.get_caption()
== f"Let's pretend this is the caption of the original image. {PREFIX}Updated caption"
)
metadata_writer_on_existing.write_caption("Don't do this!", override=False)
assert (
metadata_writer_on_existing.get_caption()
== f"Let's pretend this is the caption of the original image. {PREFIX}Updated caption"
)
@pytest.mark.parametrize(
"pic, expected_keywords",
[
(
PATHS_DNG_EXAMPLES[0],
[
"architektur",
"zeit|abend",
f"{PREFIX}hund",
f"{PREFIX}katze",
f"{PREFIX}maus",
],
),
(
PATHS_DNG_EXAMPLES[1],
["architektur", f"{PREFIX}hund", f"{PREFIX}katze", f"{PREFIX}maus"],
),
(
PATHS_DNG_EXAMPLES[2],
[
"morgen",
"tiere|vögel",
f"{PREFIX}hund",
f"{PREFIX}katze",
f"{PREFIX}maus",
],
),
(
PATHS_DNG_EXAMPLES[3],
[
"abend",
"architektur",
"bewegungsunschärfe",
"lichter",
"mainz",
"spiegelung",
"wasser",
f"{PREFIX}hund",
f"{PREFIX}katze",
f"{PREFIX}maus",
],
),
(
PATHS_DNG_EXAMPLES[4],
[
"grau",
"landschaft",
"landschaft|weinberge",
"personen",
"tiefe",
"weg",
f"{PREFIX}hund",
f"{PREFIX}katze",
f"{PREFIX}maus",
],
),
],
)
def test_set_keywords(make_dng_file, pic, expected_keywords):
metadata_writer = MetadataWriter(make_dng_file(pic))
metadata_writer.write_keywords(["hund", "katze", "maus"])
result = metadata_writer.get_keywords()
assert result
assert isinstance(result, list)
assert f"{PREFIX}hund" in result
assert f"{PREFIX}katze" in result
assert f"{PREFIX}maus" in result
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"})
+285
View File
@@ -0,0 +1,285 @@
import base64
from io import BytesIO
import pytest
from PIL import Image
from photools3.photo import Photo
from photools3.settings import DNG_EXAMPLES_DIR
PATHS_DNG_EXAMPLES = [
DNG_EXAMPLES_DIR / "example_dng_01.dng",
DNG_EXAMPLES_DIR / "example_dng_02.dng",
DNG_EXAMPLES_DIR / "example_dng_03.dng",
DNG_EXAMPLES_DIR / "example_dng_04.dng",
DNG_EXAMPLES_DIR / "example_dng_05.dng",
]
# (PIL mode) -> file format used to persist the test image
MODE_TO_FORMAT = {
"RGB": "PNG",
"L": "PNG",
"P": "PNG",
"RGBA": "PNG",
"LA": "PNG",
"CMYK": "TIFF",
"1": "PNG",
"I": "TIFF",
"F": "TIFF",
}
ALL_MODES = list(MODE_TO_FORMAT)
JPEG_COMPATIBLE_MODES = ["RGB", "L", "CMYK", "1"]
JPEG_INCOMPATIBLE_MODES = ["P", "RGBA", "LA", "I", "F"]
def _make_image_file(tmp_path, mode: str, size=(64, 48)):
"""Create an image in the given mode, save it to disk and return its path."""
fmt = MODE_TO_FORMAT[mode]
img = Image.new(mode, size)
path = tmp_path / f"test_{mode}.{fmt.lower()}"
img.save(path, format=fmt)
return path
def test_load_photo():
photo = Photo(PATHS_DNG_EXAMPLES[0])
assert photo
assert photo.path == PATHS_DNG_EXAMPLES[0]
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])
resized_photo = photo.resize()
assert isinstance(resized_photo, Photo)
assert resized_photo is photo # Should return self for chaining
def test_resize_with_custom_max_size():
"""Test resize with a custom maximum size"""
photo = Photo(PATHS_DNG_EXAMPLES[0])
original_width, original_height = photo.image.size
assert original_width > 100 or original_height > 100 # Assume it's larger
resized_photo = photo.resize(image_max_size=50)
new_width, new_height = resized_photo.image.size
max_dim = max(new_width, new_height)
assert max_dim <= 50, f"Max dimension {max_dim} exceeds limit of 50"
def test_resize_maintains_aspect_ratio():
"""Test that resize maintains the original aspect ratio"""
photo = Photo(PATHS_DNG_EXAMPLES[0])
original_width, original_height = photo.image.size
resized_photo = photo.resize(image_max_size=100)
new_width, new_height = resized_photo.image.size
original_ratio = original_width / original_height
resized_ratio = new_width / new_height
assert (
abs(original_ratio - resized_ratio) < 0.01
), f"Aspect ratio changed: {original_ratio} -> {resized_ratio}"
def test_resize_image_thumbnail_called():
"""Test that the image was actually resized using thumbnail"""
photo = Photo(PATHS_DNG_EXAMPLES[0])
original_size = photo.image.size
# Resize to a smaller dimension
resized_photo = photo.resize(image_max_size=10)
new_size = resized_photo.image.size
assert (
new_size != original_size
), "Image should be resized when max_size is smaller than original"
def test_resize_chaining():
"""Test that resize returns self for method chaining"""
photo = Photo(PATHS_DNG_EXAMPLES[0])
# Should be able to chain multiple operations
result = photo.resize().resize(image_max_size=50)
assert result is photo
def test_resize_default_parameter():
"""Test resize with default parameter (no arguments)"""
photo = Photo(PATHS_DNG_EXAMPLES[0])
resized_photo = photo.resize()
# Should return a valid Photo instance
assert isinstance(resized_photo, Photo)
assert resized_photo.path == photo.path
# --- to_rgb() ---
@pytest.mark.parametrize("mode", ALL_MODES)
def test_to_rgb_converts_to_rgb_mode(tmp_path, mode):
"""to_rgb() must leave the image in RGB mode regardless of the initial mode"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
result = photo.to_rgb()
assert result.image.mode == "RGB"
@pytest.mark.parametrize("mode", ALL_MODES)
def test_to_rgb_returns_self_for_chaining(tmp_path, mode):
"""to_rgb() returns self so it can be chained"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
assert photo.to_rgb() is photo
def test_to_rgb_keeps_rgb_image_unmodified(tmp_path):
"""RGB images are not re-converted: the same image object is kept"""
path = _make_image_file(tmp_path, "RGB")
photo = Photo(path)
original_image = photo.image
photo.to_rgb()
assert photo.image is original_image
@pytest.mark.parametrize("mode", ["L", "P", "RGBA", "LA", "CMYK", "1", "I", "F"])
def test_to_rgb_replaces_non_rgb_image(tmp_path, mode):
"""Non-RGB images are converted: a new image object is assigned"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
original_image = photo.image
photo.to_rgb()
assert photo.image is not original_image
assert photo.image.mode == "RGB"
@pytest.mark.parametrize("mode", ["RGBA", "LA"])
def test_to_rgb_drops_alpha_channel(tmp_path, mode):
"""Converting alpha-bearing modes to RGB drops the alpha band"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
assert "A" in photo.image.getbands()
photo.to_rgb()
assert photo.image.getbands() == ("R", "G", "B")
def test_to_rgb_on_dng_examples(tmp_path):
"""DNG examples are already RGB; to_rgb() is a safe no-op on every file"""
for path in PATHS_DNG_EXAMPLES:
photo = Photo(path)
original_image = photo.image
assert photo.to_rgb() is photo
assert photo.image is original_image
assert photo.image.mode == "RGB"
# --- get_encoded() ---
@pytest.mark.parametrize("mode", JPEG_COMPATIBLE_MODES)
def test_get_encoded_returns_valid_base64(tmp_path, mode):
"""get_encoded() returns a non-empty, decodable base64 string"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
encoded = photo.get_encoded()
assert isinstance(encoded, str)
assert len(encoded) > 0
raw = base64.b64decode(encoded, validate=True)
assert len(raw) > 0
@pytest.mark.parametrize("mode", JPEG_COMPATIBLE_MODES)
def test_get_encoded_produces_openable_jpeg(tmp_path, mode):
"""The decoded payload must be a valid JPEG with the original dimensions"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
original_size = photo.image.size
encoded = photo.get_encoded()
img = Image.open(BytesIO(base64.b64decode(encoded)))
assert img.format == "JPEG"
assert img.size == original_size
def test_get_encoded_starts_with_jpeg_magic_bytes(tmp_path):
"""JPEG files start with the FFD8FF magic bytes"""
path = _make_image_file(tmp_path, "RGB")
photo = Photo(path)
raw = base64.b64decode(photo.get_encoded())
assert raw[:3] == b"\xff\xd8\xff"
def test_get_encoded_reencodes_instead_of_passing_raw_file_bytes(tmp_path):
"""The output must be a fresh JPEG encoding, not the source file bytes"""
path = _make_image_file(tmp_path, "RGB")
photo = Photo(path)
raw = base64.b64decode(photo.get_encoded())
source_bytes = path.read_bytes()
assert raw != source_bytes
assert source_bytes[:4] == b"\x89PNG" # source is a PNG...
assert raw[:3] == b"\xff\xd8\xff" # ...but the encoding is a JPEG
def test_get_encoded_is_deterministic(tmp_path):
"""Encoding the same image twice yields identical base64"""
path = _make_image_file(tmp_path, "RGB")
assert Photo(path).get_encoded() == Photo(path).get_encoded()
@pytest.mark.parametrize("mode", JPEG_INCOMPATIBLE_MODES)
def test_get_encoded_raises_for_non_jpeg_modes(tmp_path, mode):
"""Modes JPEG cannot store (palette, alpha, integer/float) must raise OSError"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
with pytest.raises(OSError):
photo.get_encoded()
@pytest.mark.parametrize("mode", JPEG_INCOMPATIBLE_MODES)
def test_to_rgb_then_get_encoded_works_around_unsupported_modes(tmp_path, mode):
"""Chaining to_rgb().get_encoded() fixes modes that JPEG cannot store"""
path = _make_image_file(tmp_path, mode)
photo = Photo(path)
with pytest.raises(OSError):
photo.get_encoded()
encoded = photo.to_rgb().get_encoded()
assert isinstance(encoded, str)
img = Image.open(BytesIO(base64.b64decode(encoded)))
assert img.format == "JPEG"
@pytest.mark.parametrize("path", PATHS_DNG_EXAMPLES)
def test_get_encoded_works_on_all_dng_examples(path):
"""Every DNG example (all different files) encodes to a valid JPEG"""
photo = Photo(path)
encoded = photo.get_encoded()
assert isinstance(encoded, str)
raw = base64.b64decode(encoded, validate=True)
assert raw[:3] == b"\xff\xd8\xff"
img = Image.open(BytesIO(raw))
assert img.format == "JPEG"
assert img.size == photo.image.size
def test_full_chain_resize_to_rgb_get_encoded(tmp_path):
"""The intended workflow: resize -> to_rgb -> get_encoded"""
path = _make_image_file(tmp_path, "RGBA")
photo = Photo(path)
photo.resize(32).to_rgb()
assert photo.image.mode == "RGB"
img = Image.open(BytesIO(base64.b64decode(photo.get_encoded())))
assert img.format == "JPEG"
assert max(img.size) <= 32
+129
View File
@@ -0,0 +1,129 @@
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()
+36
View File
@@ -0,0 +1,36 @@
from pathlib import Path
from photools3.settings import (
DNG_EXAMPLES_DIR,
EXAMPLES_DIR,
PHOTOOLS3_DIR,
ROOT_DIR,
SRC_DIR,
settings,
)
def test_photools3_dir():
assert PHOTOOLS3_DIR == Path(__file__).parent.parent.resolve() / "src" / "photools3"
def test_src_dir():
assert SRC_DIR == Path(__file__).parent.parent.resolve() / "src"
def test_root_dir():
assert ROOT_DIR == Path(__file__).parent.parent.resolve()
def test_examples_dir():
assert EXAMPLES_DIR == Path(__file__).parent.parent.resolve() / "examples"
def test_dng_examples_dir():
assert (
DNG_EXAMPLES_DIR == Path(__file__).parent.parent.resolve() / "examples" / "dng"
)
def test_settings_base():
assert settings.llm_api_base_url
Generated
+1138 -2
View File
File diff suppressed because it is too large Load Diff