File size: 3,816 Bytes
8096125
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
"""High-level Python API / SDK for CV Lab developers."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import numpy as np
from PIL import Image

from batch.dataset_processor import process_dataset
from cv_ops.analysis import histogram_figure, low_resolution_pair, pixel_preview, stats
from cv_ops.morphology import apply_morphology
from cv_ops.transforms import reflect, rotate, scale_image, translate
from filters.builtin import BUILTIN_FILTERS, ensure_rgb
from filters.custom import parse_kernel, parse_pipeline
from filters.registry import apply_definition, apply_step, load_definition, save_filter
from models.face_filters import apply_face_filter, ar_filter_names, save_ar_filter
from models.style_transfer import stylize


def load_image(source: str | Path | np.ndarray) -> np.ndarray:
    """Load an image from file path or return RGB numpy array."""
    if isinstance(source, (str, Path)):
        img = Image.open(source).convert("RGB")
        return np.array(img, dtype=np.uint8)
    return ensure_rgb(source)


def save_image(image: np.ndarray, output_path: str | Path) -> str:
    """Save an RGB numpy image array to file."""
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    Image.fromarray(image).save(path)
    return str(path)


def generate_python_snippet(operation: str, params: dict[str, Any]) -> str:
    """Generate reproducible Python code snippet for developers."""
    params_str = json.dumps(params, indent=2)
    return f"""import numpy as np
from PIL import Image
from filters.registry import apply_step

# Load image
image = np.array(Image.open("input.jpg").convert("RGB"))

# Apply filter
params = {params_str}
result = apply_step(image, "{operation}", params)

# Save result
Image.fromarray(result).save("output.jpg")
"""


def process_image(
    image_input: str | Path | np.ndarray,
    operation_type: str,
    operation_name: str = "Grayscale",
    params: dict[str, Any] | None = None,
    output_path: str | Path | None = None,
) -> tuple[np.ndarray, dict[str, Any]]:
    """Programmatic API to process images with any CV Lab operation."""
    img = load_image(image_input)
    params = params or {}

    if operation_type == "filter":
        result = apply_step(img, operation_name, params)
        meta = {"operation": operation_name, "params": params}
    elif operation_type == "pipeline":
        definition = parse_pipeline(operation_name) if isinstance(operation_name, str) else operation_name
        result = apply_definition(img, definition)
        meta = {"definition": definition}
    elif operation_type == "transform":
        if operation_name == "Translation":
            result = translate(img, **params)[0]
        elif operation_name == "Rotation":
            result = rotate(img, **params)[0]
        elif operation_name == "Scaling":
            result = scale_image(img, **params)[0]
        else:
            result = reflect(img, **params)[0]
        meta = {"operation": operation_name, "params": params}
    elif operation_type == "morphology":
        result, kernel = apply_morphology(img, operation_name, **params)
        meta = {"operation": operation_name, "kernel_shape": kernel.shape}
    elif operation_type == "ar":
        result, msg = apply_face_filter(img, operation_name, custom_def=params.get("custom_def"))
        meta = {"status": msg}
    elif operation_type == "style":
        style_img = load_image(params["style_image"])
        result, time_sec, msg = stylize(img, style_img, max_size=params.get("max_size", 512))
        meta = {"time_seconds": time_sec, "message": msg}
    else:
        raise ValueError(f"Unknown operation type: {operation_type}")

    if output_path:
        save_image(result, output_path)

    return result, meta