Compare commits
16
Commits
97795214b1
..
master
| Author | SHA1 | Date | |
|---|---|---|---|
 mathias2784
|
0d6272b4c3 | ||
|
mathias2784
|
ad7dbfdd11 | ||
|
mathias2784
|
ff588f899e | ||
|
mathias2784
|
b6be043426 | ||
|
mathias2784
|
b9632d46b2 | ||
|
mathias2784
|
6a12144e39 | ||
|
mathias2784
|
401488fa7e | ||
|
mathias2784
|
419fb46c3d | ||
|
mathias2784
|
ce2da9d953 | ||
|
mathias2784
|
906707de55 | ||
|
mathias2784
|
0c33153c98 | ||
|
mathias2784
|
d9758c598d | ||
|
mathias2784
|
3cd8541b24 | ||
|
mathias2784
|
2dd82ff803 | ||
|
mathias2784
|
d32df3e54a | ||
|
mathias2784
|
e0c38fc2c6 |
@@ -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
|
||||
@@ -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
|
||||
@@ -8,3 +8,6 @@ wheels/
|
||||
|
||||
# Virtual environments
|
||||
.venv
|
||||
|
||||
.env
|
||||
logs/
|
||||
|
||||
@@ -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.
|
||||
@@ -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
|
||||
```
|
||||
|
||||
@@ -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.
@@ -8,7 +8,17 @@ authors = [
|
||||
]
|
||||
requires-python = ">=3.14"
|
||||
dependencies = [
|
||||
"black>=26.5.1",
|
||||
"mypy>=2.4.0",
|
||||
"openai>=3.22.0",
|
||||
"pillow>=12.3.0",
|
||||
"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]
|
||||
@@ -17,3 +27,18 @@ photools3 = "photools3:main"
|
||||
[build-system]
|
||||
requires = ["uv_build>=0.12.19,<0.13.0"]
|
||||
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
|
||||
|
||||
@@ -1,2 +1,5 @@
|
||||
def main() -> None:
|
||||
print("Hello from photools3!")
|
||||
from .cli import app
|
||||
|
||||
|
||||
def main():
|
||||
app()
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
from .cli import app
|
||||
|
||||
if __name__ == "__main__":
|
||||
app()
|
||||
@@ -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()
|
||||
@@ -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:
|
||||
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):
|
||||
...
|
||||
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,
|
||||
)
|
||||
|
||||
@@ -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
|
||||
@@ -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
@@ -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):
|
||||
...
|
||||
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
@@ -1,18 +1,49 @@
|
||||
|
||||
import base64
|
||||
from io import BytesIO
|
||||
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:
|
||||
|
||||
def __init__(self):
|
||||
...
|
||||
def __init__(self, path: Path):
|
||||
self.path = path
|
||||
self.image = Image.open(path)
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, path: Path) -> Photo:
|
||||
# TODO
|
||||
return cls()
|
||||
def close(self):
|
||||
self.image.close()
|
||||
|
||||
def resize(self) -> Photo:
|
||||
# TODO
|
||||
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)
|
||||
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())
|
||||
|
||||
@@ -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
|
||||
@@ -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]
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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")
|
||||
@@ -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"})
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user