feat(v3): simplify the project to contain only required tools

This commit is contained in:
2026-07-09 04:02:31 +03:30
parent 20a43d5c9a
commit 2112e00982
19 changed files with 482 additions and 508 deletions

View File

@@ -119,6 +119,7 @@ CELERY_TASK_TIME_LIMIT = env_int("CELERY_TASK_TIME_LIMIT", 600)
IMAGE_SESSION_TTL_HOURS = env_int("IMAGE_SESSION_TTL_HOURS", 6)
MAX_UPLOAD_MB = env_int("MAX_UPLOAD_MB", 20)
IMAGE_WORKSPACE_MAX_DIMENSION = env_int("IMAGE_WORKSPACE_MAX_DIMENSION", 1400)
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
SECURE_SSL_REDIRECT = env_bool("DJANGO_SECURE_SSL_REDIRECT", not DEBUG)

View File

@@ -4,7 +4,6 @@ from io import BytesIO
import cv2
import numpy as np
from numpy.lib.stride_tricks import sliding_window_view
from PIL import Image
@@ -177,52 +176,6 @@ def histogram_equalization(image, params):
return gray_to_rgb(equalized)
def target_cdf_from_params(params):
if "cdf" in params:
cdf = np.array(params["cdf"], dtype=np.float64)
if cdf.shape != (256,) or np.any(np.diff(cdf) < 0):
raise ProcessingError("cdf must contain 256 non-decreasing values.")
if cdf[-1] <= 0:
raise ProcessingError("cdf must end with a positive value.")
return cdf / cdf[-1]
mode = params.get("target", "uniform")
levels = np.arange(256, dtype=np.float64)
if mode == "dark":
pdf = np.exp(-levels / 64.0)
elif mode == "bright":
pdf = np.exp(-(255.0 - levels) / 64.0)
elif mode == "bimodal":
pdf = np.exp(-((levels - 72.0) ** 2) / (2 * 22.0**2)) + np.exp(-((levels - 190.0) ** 2) / (2 * 28.0**2))
else:
pdf = np.ones(256, dtype=np.float64)
cdf = np.cumsum(pdf)
return cdf / cdf[-1]
def histogram_matching(image, params):
gray = to_gray(image)
source_counts = np.bincount(gray.ravel(), minlength=256).astype(np.float64)
source_cdf = np.cumsum(source_counts)
source_cdf /= source_cdf[-1]
target_cdf = target_cdf_from_params(params)
target_levels = np.arange(256)
mapping = np.interp(source_cdf, target_cdf, target_levels).round().clip(0, 255).astype(np.uint8)
return gray_to_rgb(mapping[gray])
def local_equalization(image, params):
size = require_odd(params.get("size", 7), "size")
gray = to_gray(image)
radius = size // 2
padded = np.pad(gray, radius, mode="edge")
windows = sliding_window_view(padded, (size, size))
centers = gray[..., None, None]
ranks = np.count_nonzero(windows <= centers, axis=(-1, -2))
equalized = np.round(ranks * 255.0 / (size * size)).astype(np.uint8)
return gray_to_rgb(equalized)
def apply_kernel(image, kernel, normalize_derivative=False):
source = image.astype(np.float32)
if image.ndim == 2:
@@ -309,93 +262,6 @@ def roberts(image, params):
return gradient_magnitude(image, ROBERTS_GX, ROBERTS_GY)
def rgb_to_hsi(image):
rgb = image.astype(np.float32) / 255.0
r, g, b = rgb[..., 0], rgb[..., 1], rgb[..., 2]
numerator = 0.5 * ((r - g) + (r - b))
denominator = np.sqrt((r - g) ** 2 + (r - b) * (g - b)) + 1e-8
theta = np.arccos(np.clip(numerator / denominator, -1.0, 1.0))
h = np.where(b <= g, theta, 2.0 * np.pi - theta) / (2.0 * np.pi)
total = r + g + b
s = np.where(total <= 1e-8, 0.0, 1.0 - 3.0 * np.minimum(np.minimum(r, g), b) / total)
i = total / 3.0
return np.stack([h, s, i], axis=-1)
def hsi_to_rgb(hsi):
h = (hsi[..., 0] % 1.0) * 2.0 * np.pi
s = np.clip(hsi[..., 1], 0.0, 1.0)
i = np.clip(hsi[..., 2], 0.0, 1.0)
r = np.zeros_like(h)
g = np.zeros_like(h)
b = np.zeros_like(h)
sector0 = h < 2.0 * np.pi / 3.0
sector1 = (h >= 2.0 * np.pi / 3.0) & (h < 4.0 * np.pi / 3.0)
sector2 = ~sector0 & ~sector1
h0 = h[sector0]
b[sector0] = i[sector0] * (1.0 - s[sector0])
r[sector0] = i[sector0] * (1.0 + s[sector0] * np.cos(h0) / (np.cos(np.pi / 3.0 - h0) + 1e-8))
g[sector0] = 3.0 * i[sector0] - (r[sector0] + b[sector0])
h1 = h[sector1] - 2.0 * np.pi / 3.0
r[sector1] = i[sector1] * (1.0 - s[sector1])
g[sector1] = i[sector1] * (1.0 + s[sector1] * np.cos(h1) / (np.cos(np.pi / 3.0 - h1) + 1e-8))
b[sector1] = 3.0 * i[sector1] - (r[sector1] + g[sector1])
h2 = h[sector2] - 4.0 * np.pi / 3.0
g[sector2] = i[sector2] * (1.0 - s[sector2])
b[sector2] = i[sector2] * (1.0 + s[sector2] * np.cos(h2) / (np.cos(np.pi / 3.0 - h2) + 1e-8))
r[sector2] = 3.0 * i[sector2] - (g[sector2] + b[sector2])
return ensure_uint8(np.round(np.clip(np.stack([r, g, b], axis=-1), 0.0, 1.0) * 255.0))
def hsi_intensity_filter(image, params):
method = params.get("method", "smooth")
hsi = rgb_to_hsi(image)
intensity = np.round(hsi[..., 2] * 255.0).astype(np.uint8)
if method == "sharpen":
filtered = laplacian(intensity, {"mode": "sharpen", "sign": params.get("sign", "add")})
else:
filtered = box_filter(intensity, {"size": params.get("size", 3)})
hsi[..., 2] = filtered.astype(np.float32) / 255.0
return hsi_to_rgb(hsi)
def pseudo_color_slices(image, params):
gray = to_gray(image)
slices = params.get(
"slices",
[
{"start": 0, "end": 85, "color": [59, 130, 246]},
{"start": 86, "end": 170, "color": [34, 197, 94]},
{"start": 171, "end": 255, "color": [239, 68, 68]},
],
)
output = np.zeros((*gray.shape, 3), dtype=np.uint8)
for item in slices:
start = int(item["start"])
end = int(item["end"])
color = np.array(item["color"], dtype=np.uint8)
if start < 0 or end > 255 or start > end or color.shape != (3,):
raise ProcessingError("Each pseudo-color slice requires start/end in 0..255 and an RGB color.")
output[(gray >= start) & (gray <= end)] = color
return output
def gray_to_color_sinusoidal(image, params):
gray = to_gray(image).astype(np.float32) / 255.0
hue_frequency = float(params.get("hue_frequency", 1.0))
saturation_frequency = float(params.get("saturation_frequency", 0.5))
intensity_frequency = float(params.get("intensity_frequency", 0.25))
h = (0.5 + 0.5 * np.sin(2.0 * np.pi * hue_frequency * gray)) % 1.0
s = 0.55 + 0.4 * np.sin(2.0 * np.pi * saturation_frequency * gray + np.pi / 3.0)
i = 0.5 + 0.45 * np.sin(2.0 * np.pi * intensity_frequency * gray - np.pi / 2.0)
return hsi_to_rgb(np.stack([h, np.clip(s, 0, 1), np.clip(i, 0, 1)], axis=-1))
OPERATIONS = {
"negative": negative,
"log": logarithmic,
@@ -404,8 +270,6 @@ OPERATIONS = {
"gray_slice": gray_slice,
"bit_plane": bit_plane,
"hist_equalization": histogram_equalization,
"hist_match": histogram_matching,
"local_equalization": local_equalization,
"box_filter": box_filter,
"weighted_average": weighted_average,
"median_filter": median_filter,
@@ -413,9 +277,6 @@ OPERATIONS = {
"high_boost": high_boost,
"sobel": sobel,
"roberts": roberts,
"hsi_intensity_filter": hsi_intensity_filter,
"pseudo_color_slices": pseudo_color_slices,
"gray_to_color_sinusoidal": gray_to_color_sinusoidal,
}

View File

@@ -10,20 +10,11 @@ from .algorithms import (
contrast_stretch,
gamma,
gray_slice,
gray_to_color_sinusoidal,
gray_to_rgb,
histogram_equalization,
histogram_matching,
hsi_intensity_filter,
hsi_to_rgb,
logarithmic,
local_equalization,
negative,
normalize_to_uint8,
pseudo_color_slices,
rgb_to_hsi,
roberts,
sobel,
to_gray,
weighted_average,
median_filter,
@@ -38,71 +29,67 @@ CH4 = "Image Enhancement in the Frequency Domain"
CH6 = "Color Image Processing"
def odd_param(default=3, max_value=35):
return {"type": "int", "default": default, "min": 3, "max": max_value, "step": 2, "odd": True}
def with_meta(schema, *, label=None, description=None, show_when=None):
if label:
schema["label"] = label
if description:
schema["description"] = description
if show_when:
schema["show_when"] = show_when
return schema
def float_param(default, min_value, max_value, step=0.1):
return {"type": "float", "default": default, "min": min_value, "max": max_value, "step": step}
def odd_param(default=3, max_value=35, **meta):
return with_meta({"type": "int", "default": default, "min": 3, "max": max_value, "step": 2, "odd": True}, **meta)
def int_param(default, min_value, max_value, step=1):
return {"type": "int", "default": default, "min": min_value, "max": max_value, "step": step}
def float_param(default, min_value, max_value, step=0.1, **meta):
return with_meta({"type": "float", "default": default, "min": min_value, "max": max_value, "step": step}, **meta)
def select_param(default, choices):
return {"type": "select", "default": default, "choices": choices}
def int_param(default, min_value, max_value, step=1, **meta):
return with_meta({"type": "int", "default": default, "min": min_value, "max": max_value, "step": step}, **meta)
def bool_param(default=False):
return {"type": "bool", "default": default}
def select_param(default, choices, **meta):
return with_meta({"type": "select", "default": default, "choices": choices}, **meta)
def crop(image, params):
x = max(0, int(params.get("x", 0)))
y = max(0, int(params.get("y", 0)))
width = int(params.get("width", image.shape[1] - x))
height = int(params.get("height", image.shape[0] - y))
if width <= 0 or height <= 0:
raise ProcessingError("Crop width and height must be positive.")
x2 = min(image.shape[1], x + width)
y2 = min(image.shape[0], y + height)
if x >= x2 or y >= y2:
raise ProcessingError("Crop rectangle is outside the image.")
return image[y:y2, x:x2]
def bool_param(default=False, **meta):
return with_meta({"type": "bool", "default": default}, **meta)
def identity(image, params):
return image.copy()
def kernel_preview(title, matrix, scale=None):
return {"title": title, "matrix": matrix, "scale": scale}
def inverse_log(image, params):
c = float(params.get("c", 1.0))
normalized = image.astype(np.float32) / 255.0
transformed = np.expm1(normalized / max(c, 1e-8))
transformed /= max(float(np.max(transformed)), 1e-8)
return ensure_uint8(np.round(transformed * 255.0))
def kernel_pair_preview(title, gx, gy):
return {"title": title, "kernels": [{"label": "Gx", "matrix": gx}, {"label": "Gy", "matrix": gy}]}
def threshold(image, params):
level = int(params.get("level", 128))
high = int(params.get("high", 255))
low = int(params.get("low", 0))
gray = to_gray(image)
return gray_to_rgb(np.where(gray >= level, high, low).astype(np.uint8))
def mask_param(default=3, max_value=35, label="Mask size"):
schema = odd_param(default, max_value)
schema["label"] = label
return schema
def histeq(image, params):
mode = params.get("mode", "intensity")
if image.ndim == 2 or mode == "grayscale":
if image.ndim == 2:
return histogram_equalization(image, params)
if mode == "rgb":
channels = [histogram_equalization(image[:, :, idx], params)[:, :, 0] for idx in range(3)]
return np.stack(channels, axis=2).astype(np.uint8)
hsi = rgb_to_hsi(image)
intensity = np.round(hsi[..., 2] * 255).astype(np.uint8)
hsi[..., 2] = histogram_equalization(intensity, params)[:, :, 0].astype(np.float32) / 255.0
return hsi_to_rgb(hsi)
channels = [histogram_equalization(image[:, :, idx], params)[:, :, 0] for idx in range(3)]
return np.stack(channels, axis=2).astype(np.uint8)
def rgb_to_gray_matlab(image, params):
red_weight = float(params.get("red_weight", 0.299))
green_weight = float(params.get("green_weight", 0.587))
blue_weight = float(params.get("blue_weight", 0.114))
total = red_weight + green_weight + blue_weight
if not np.isfinite(total) or math.isclose(total, 0.0):
raise ProcessingError("Grayscale weights must have a non-zero finite sum.")
weights = np.array([red_weight, green_weight, blue_weight], dtype=np.float32) / total
gray = np.tensordot(image.astype(np.float32), weights, axes=([2], [0]))
return gray_to_rgb(ensure_uint8(np.round(gray)))
def gaussian_noise(image, params):
@@ -126,49 +113,70 @@ def salt_pepper_noise(image, params):
return output
def speckle_noise(image, params):
variance = float(params.get("variance", 0.04))
noise = np.random.default_rng().normal(0, math.sqrt(max(variance, 0.0)), size=image.shape)
return ensure_uint8(np.round(image.astype(np.float32) + image.astype(np.float32) * noise))
def noise_filter(image, params):
kind = params.get("kind", "gaussian")
if kind == "salt_pepper":
return salt_pepper_noise(image, params)
return gaussian_noise(image, params)
def gaussian_filter(image, params):
size = int(params.get("size", 3))
variance = float(params.get("variance", 1.0))
size = int(params.get("K", params.get("size", 3)))
variance = float(params.get("Q", params.get("variance", 1.0)))
if size < 3 or size % 2 == 0:
raise ProcessingError("size must be an odd integer >= 3.")
raise ProcessingError("K must be an odd integer >= 3.")
sigma = math.sqrt(max(variance, 1e-8))
return cv2.GaussianBlur(image, (size, size), sigmaX=sigma, sigmaY=sigma, borderType=cv2.BORDER_REFLECT)
def max_filter(image, params):
size = int(params.get("size", 3))
size = int(params.get("mask_size", params.get("N", params.get("size", 3))))
if size < 3 or size % 2 == 0:
raise ProcessingError("size must be an odd integer >= 3.")
raise ProcessingError("Mask size must be an odd integer >= 3.")
return cv2.dilate(image, np.ones((size, size), np.uint8))
def min_filter(image, params):
size = int(params.get("size", 3))
size = int(params.get("mask_size", params.get("N", params.get("size", 3))))
if size < 3 or size % 2 == 0:
raise ProcessingError("size must be an odd integer >= 3.")
raise ProcessingError("Mask size must be an odd integer >= 3.")
return cv2.erode(image, np.ones((size, size), np.uint8))
def box_denoise(image, params):
return box_filter(image, {"size": params.get("K", params.get("mask_size", params.get("size", 3)))})
def weighted_denoise(image, params):
return weighted_average(image, {"size": 3})
def median_denoise(image, params):
return median_filter(image, {"size": params.get("mask_size", params.get("N", params.get("size", 3)))})
def gaussian_denoise(image, params):
size = params.get("K", params.get("mask_size", params.get("size", 3)))
variance = params.get("Q", params.get("variance", 1.0))
return gaussian_filter(image, {"K": size, "Q": variance})
def high_boost_slide(image, params):
return high_boost(image, {"amplification": params.get("A", params.get("amplification", 1.5)), "size": params.get("K", params.get("size", 3))})
def laplacian_slide(image, params):
mask_name = params.get("mask", "cross")
kernels = {
"cross": np.array([[0, 1, 0], [1, -5, 1], [0, 1, 0]], dtype=np.float32),
"diagonal": np.array([[1, 1, 1], [1, -9, 1], [1, 1, 1]], dtype=np.float32),
"zero_sum_cross": np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32),
"zero_sum_diagonal": np.array([[1, 1, 1], [1, -8, 1], [1, 1, 1]], dtype=np.float32),
}
kernel = kernels.get(mask_name)
if kernel is None:
raise ProcessingError("Unknown Laplacian mask.")
channels = [cv2.filter2D(image[:, :, idx], cv2.CV_32F, kernel, borderType=cv2.BORDER_REFLECT) for idx in range(image.shape[2])]
result = np.stack(channels, axis=2)
return normalize_to_uint8(result) if params.get("mode", "sharpen") == "detail" else ensure_uint8(result)
return ensure_uint8(result)
def gradient_abs_sum(image, params):
@@ -191,13 +199,6 @@ def rgb_channel(image, params):
return gray_to_rgb(image[:, :, index])
def hsi_view(image, params):
component = params.get("component", "i")
hsi = rgb_to_hsi(image)
index = {"h": 0, "s": 1, "i": 2}.get(component, 2)
return gray_to_rgb(np.round(hsi[:, :, index] * 255.0).astype(np.uint8))
def fft_spectrum(image, params):
gray = to_gray(image).astype(np.float32)
spectrum = np.fft.fftshift(np.fft.fft2(gray))
@@ -210,72 +211,7 @@ def fft_spectrum(image, params):
return gray_to_rgb(normalize_to_uint8(magnitude))
def distance_grid(shape):
rows, cols = shape
u = np.arange(rows) - rows / 2
v = np.arange(cols) - cols / 2
vv, uu = np.meshgrid(v, u)
return np.sqrt(uu**2 + vv**2)
def frequency_filter(image, params):
gray = to_gray(image).astype(np.float32)
d0 = float(params.get("cutoff", 40))
order = int(params.get("order", 2))
family = params.get("family", "gaussian")
kind = params.get("kind", "lowpass")
d = distance_grid(gray.shape)
if family == "ideal":
mask = (d <= d0).astype(np.float32)
elif family == "butterworth":
mask = 1.0 / (1.0 + (d / max(d0, 1e-8)) ** (2 * max(order, 1)))
else:
mask = np.exp(-(d**2) / (2.0 * max(d0, 1e-8) ** 2))
if kind == "highpass":
mask = 1.0 - mask
if params.get("output", "image") == "mask":
return gray_to_rgb(normalize_to_uint8(mask))
f = np.fft.fftshift(np.fft.fft2(gray))
result = np.real(np.fft.ifft2(np.fft.ifftshift(f * mask)))
return gray_to_rgb(normalize_to_uint8(result))
def frequency_laplacian(image, params):
gray = to_gray(image).astype(np.float32)
rows, cols = gray.shape
u = np.arange(rows) - rows / 2
v = np.arange(cols) - cols / 2
vv, uu = np.meshgrid(v, u)
h = -4.0 * (np.pi**2) * (uu**2 + vv**2)
f = np.fft.fftshift(np.fft.fft2(gray))
result = np.real(np.fft.ifft2(np.fft.ifftshift(f * h)))
return gray_to_rgb(normalize_to_uint8(result))
def correlation(image, params):
kernel = np.array(params.get("kernel", [[1, 1, 1], [1, 1, 1], [1, 1, 1]]), dtype=np.float32)
kernel /= max(float(np.sum(np.abs(kernel))), 1e-8)
gray = to_gray(image).astype(np.float32)
return gray_to_rgb(normalize_to_uint8(cv2.filter2D(gray, cv2.CV_32F, kernel, borderType=cv2.BORDER_REFLECT)))
def convolution(image, params):
kernel = np.array(params.get("kernel", [[1, 1, 1], [1, 1, 1], [1, 1, 1]]), dtype=np.float32)
return correlation(image, {"kernel": np.flipud(np.fliplr(kernel)).tolist()})
def bone_scan_workflow(image, params):
gray_rgb = gray_to_rgb(to_gray(image))
lap_detail = laplacian_slide(gray_rgb, {"mask": "zero_sum_diagonal", "mode": "detail"})
sharpened = ensure_uint8(gray_rgb.astype(np.float32) + lap_detail.astype(np.float32))
sobel_img = gradient_abs_sum(gray_rgb, {"operator": "sobel"})
smooth_sobel = box_filter(sobel_img, {"size": 5})
mask = normalize_to_uint8((sharpened.astype(np.float32) * smooth_sobel.astype(np.float32)) / 255.0)
summed = ensure_uint8(gray_rgb.astype(np.float32) + mask.astype(np.float32))
return gamma(summed, {"gamma": float(params.get("gamma", 0.5)), "c": 1.0})
def operation(id, label, chapter, slide_group, func, params=None, supports="both"):
def operation(id, label, chapter, slide_group, func, params=None, supports="both", matrices=None, formula="", repeatable=True):
return {
"id": id,
"label": label,
@@ -283,49 +219,46 @@ def operation(id, label, chapter, slide_group, func, params=None, supports="both
"slide_group": slide_group,
"params": params or {},
"supports": supports,
"matrices": matrices or [],
"formula": formula,
"repeatable": repeatable,
"func": func,
}
OPERATIONS = [
operation("crop", "Crop", CH_BASIC, "Workspace", crop, {"x": int_param(0, 0, 4000), "y": int_param(0, 0, 4000), "width": int_param(256, 1, 8000), "height": int_param(256, 1, 8000)}),
operation("identity", "Identity", CH3, "Point Processing", identity),
operation("negative", "Negative", CH3, "Point Processing", negative),
operation("log", "Log", CH3, "Point Processing", logarithmic, {"c": float_param(1.44, 0.1, 5, 0.05)}),
operation("inverse_log", "Inverse Log", CH3, "Point Processing", inverse_log, {"c": float_param(1.0, 0.1, 5, 0.05)}),
operation("gamma", "Power-Law / Gamma", CH3, "Point Processing", gamma, {"gamma": float_param(1.0, 0.1, 5, 0.05), "c": float_param(1.0, 0.1, 3, 0.05)}),
operation("threshold", "Thresholding", CH3, "Point Processing", threshold, {"level": int_param(128, 0, 255), "low": int_param(0, 0, 255), "high": int_param(255, 0, 255)}),
operation("contrast_stretch", "Contrast Stretching", CH3, "Piecewise Linear", contrast_stretch, {"low": int_param(30, 0, 254), "high": int_param(220, 1, 255)}),
operation("gray_slice", "Gray-Level Slicing", CH3, "Piecewise Linear", gray_slice, {"start": int_param(96, 0, 255), "end": int_param(160, 0, 255), "preserve_background": bool_param(True)}),
operation("bit_plane", "Bit-Plane Slicing", CH3, "Piecewise Linear", bit_plane, {"bit": int_param(7, 0, 7)}),
operation("histeq", "histeq()", CH3, "Histogram Processing", histeq, {"mode": select_param("intensity", ["intensity", "rgb", "grayscale"])}),
operation("hist_match", "Histogram Specification", CH3, "Histogram Processing", histogram_matching, {"target": select_param("uniform", ["uniform", "dark", "bright", "bimodal"])}),
operation("local_equalization", "Local Enhancement", CH3, "Histogram Processing", local_equalization, {"size": odd_param(7, 31)}),
operation("gaussian_noise", "Add Gaussian Noise", CH3, "Noise and Denoising", gaussian_noise, {"mean": float_param(0, -1, 1, 0.01), "variance": float_param(0.01, 0, 0.2, 0.005)}),
operation("salt_pepper_noise", "Add Salt & Pepper Noise", CH3, "Noise and Denoising", salt_pepper_noise, {"amount": float_param(0.03, 0, 0.5, 0.01), "salt_ratio": float_param(0.5, 0, 1, 0.05)}),
operation("speckle_noise", "Add Speckle Noise", CH3, "Noise and Denoising", speckle_noise, {"variance": float_param(0.04, 0, 0.3, 0.01)}),
operation("box_filter", "Box / Average Filter", CH3, "Smoothing Linear Filters", box_filter, {"size": odd_param(3, 35)}),
operation("weighted_average", "Weighted Average Filter", CH3, "Smoothing Linear Filters", weighted_average, {"size": odd_param(3, 35)}),
operation("gaussian_filter", "Gaussian fspecial Filter", CH3, "Smoothing Linear Filters", gaussian_filter, {"size": odd_param(3, 35), "variance": float_param(1.0, 0.01, 25, 0.1)}),
operation("median_filter", "Median Filter", CH3, "Order-Statistics Filters", median_filter, {"size": odd_param(3, 25)}),
operation("max_filter", "Max Filter", CH3, "Order-Statistics Filters", max_filter, {"size": odd_param(3, 25)}),
operation("min_filter", "Min Filter", CH3, "Order-Statistics Filters", min_filter, {"size": odd_param(3, 25)}),
operation("laplacian_slide", "Laplacian Masks", CH3, "Sharpening Spatial Filters", laplacian_slide, {"mask": select_param("cross", ["cross", "diagonal", "zero_sum_cross", "zero_sum_diagonal"]), "mode": select_param("sharpen", ["sharpen", "detail"])}),
operation("gradient_abs_sum", "Gradient abs(imfilter Gx)+abs(imfilter Gy)", CH3, "Gradient Operator", gradient_abs_sum, {"operator": select_param("sobel", ["sobel", "roberts"])}),
operation("sobel", "Sobel Magnitude", CH3, "Gradient Operator", sobel),
operation("roberts", "Roberts Magnitude", CH3, "Gradient Operator", roberts),
operation("high_boost", "High-Boost Filtering", CH3, "High-Boost Filtering", high_boost, {"amplification": float_param(1.5, 1, 6, 0.1), "size": odd_param(3, 35)}),
operation("bone_scan_workflow", "Bone Scan Workflow Preset", CH3, "Combining Spatial Enhancement Methods", bone_scan_workflow, {"gamma": float_param(0.5, 0.1, 2, 0.05)}),
operation("fft_spectrum", "FFT/DFT Spectrum View", CH4, "DFT and FFT", fft_spectrum, {"mode": select_param("log_magnitude", ["magnitude", "log_magnitude", "phase"])}),
operation("frequency_filter", "Ideal/Butterworth/Gaussian Frequency Filter", CH4, "Frequency Domain Filtering", frequency_filter, {"family": select_param("gaussian", ["ideal", "butterworth", "gaussian"]), "kind": select_param("lowpass", ["lowpass", "highpass"]), "cutoff": float_param(40, 1, 512, 1), "order": int_param(2, 1, 10), "output": select_param("image", ["image", "mask"])}),
operation("frequency_laplacian", "Laplacian in Frequency Domain", CH4, "Sharpening Highpass Filtering", frequency_laplacian),
operation("convolution", "Convolution Utility", CH4, "Convolution", convolution),
operation("correlation", "Correlation Utility", CH4, "Correlation", correlation),
operation("rgb_channel", "RGB Channel View", CH6, "RGB color model", rgb_channel, {"channel": select_param("r", ["r", "g", "b"])}),
operation("hsi_view", "HSI Component View", CH6, "HSI color model", hsi_view, {"component": select_param("i", ["h", "s", "i"])}),
operation("hsi_intensity_filter", "HSI Intensity Processing", CH6, "HSI color model", hsi_intensity_filter, {"method": select_param("smooth", ["smooth", "sharpen"]), "size": odd_param(3, 25)}),
operation("pseudo_color_slices", "Pseudocolor Intensity Slicing", CH6, "Pseudocolor Image Processing", pseudo_color_slices),
operation("gray_to_color_sinusoidal", "Gray-Level to Color Transform", CH6, "Gray level to color transformation", gray_to_color_sinusoidal, {"hue_frequency": float_param(1, 0.2, 4, 0.1), "saturation_frequency": float_param(0.5, 0.1, 4, 0.1), "intensity_frequency": float_param(0.25, 0.1, 4, 0.1)}),
operation("histeq", "Histogram Equalization", CH_BASIC, "Histogram", histeq, formula="Default histeq: map gray levels by the cumulative histogram CDF.", repeatable=False),
operation("negative", "Negative", CH3, "Point Processing", negative, formula="s = 255 - r", repeatable=False),
operation("log", "Log", CH3, "Point Processing", logarithmic, {"c": float_param(1.44, 0.1, 5, 0.05, description="Scale factor in s = c log(1 + r).")}, formula="s = c log(1 + r)", repeatable=False),
operation("gamma", "Power-Law / Gamma", CH3, "Point Processing", gamma, {"gamma": float_param(1.0, 0.1, 5, 0.05, description="Exponent gamma in s = c r^gamma."), "c": float_param(1.0, 0.1, 3, 0.05, description="Scale factor c in s = c r^gamma.")}, formula="s = c r^gamma", repeatable=False),
operation("contrast_stretch", "Gray-Level Dynamic Range", CH3, "Piecewise Linear", contrast_stretch, {"low": int_param(30, 0, 254, description="Input gray level mapped toward 0."), "high": int_param(220, 1, 255, description="Input gray level mapped toward 255.")}, formula="Stretch [low, high] to [0, 255].", repeatable=False),
operation("gray_slice", "Gray-Level Slicing", CH3, "Piecewise Linear", gray_slice, {"start": int_param(96, 0, 255, description="Lower bound A of highlighted range [A, B]."), "end": int_param(160, 0, 255, description="Upper bound B of highlighted range [A, B]."), "preserve_background": bool_param(True, description="Keep original background outside [A, B].")}, formula="Highlight gray range A <= r <= B.", repeatable=False),
operation("bit_plane", "Bit-Plane Slicing", CH3, "Piecewise Linear", bit_plane, {"bit": int_param(7, 0, 7, description="Bit plane index, 0 least significant through 7 most significant.")}, formula="Output bit k of each gray level.", repeatable=False),
operation("noise_filter", "Noise Filter", CH3, "Noise and Denoising", noise_filter, {
"kind": select_param("gaussian", ["gaussian", "salt_pepper"], description="Select the noise model."),
"mean": float_param(0, -1, 1, 0.01, description="Gaussian mean in normalized intensity units.", show_when={"param": "kind", "value": "gaussian"}),
"variance": float_param(0.01, 0, 0.2, 0.005, description="Gaussian variance; sigma = sqrt(variance).", show_when={"param": "kind", "value": "gaussian"}),
"amount": float_param(0.03, 0, 0.5, 0.01, description="Salt-and-pepper probability per pixel.", show_when={"param": "kind", "value": "salt_pepper"}),
"salt_ratio": float_param(0.5, 0, 1, 0.05, description="Fraction of impulse noise assigned to salt.", show_when={"param": "kind", "value": "salt_pepper"}),
}, formula="Gaussian: g=f+n. Salt-pepper: pixels become 0 or 255."),
operation("box_filter", "Average / Box Filter", CH3, "Linear Low-Pass Filters", box_denoise, {"K": odd_param(3, 35, description="Odd mask dimension K for the K x K average mask.")}, matrices=[kernel_preview("1 / K^2 box mask", [["1", "1", "1"], ["1", "1", "1"], ["1", "1", "1"]], "1 / K^2")], formula="g = imfilter(f, ones(K,K)/K^2)"),
operation("weighted_average", "Weighted Average Filter", CH3, "Linear Low-Pass Filters", weighted_denoise, matrices=[kernel_preview("Weighted average mask", [[1, 2, 1], [2, 4, 2], [1, 2, 1]], "1 / 16")], formula="g = imfilter(f, weighted mask)"),
operation("gaussian_filter", "Gaussian Filter", CH3, "Linear Low-Pass Filters", gaussian_denoise, {"K": odd_param(3, 35, description="Odd Gaussian mask dimension K."), "Q": float_param(1.0, 0.01, 25, 0.1, description="Variance Q of the Gaussian mask.")}, formula="Gaussian mask controlled by K and variance Q."),
operation("median_filter", "Median Filter", CH3, "Order-Statistics Filters", median_denoise, {"mask_size": mask_param(3, 25, "Window size")}, formula="g(x,y) = median of neighborhood."),
operation("max_filter", "Max Filter", CH3, "Order-Statistics Filters", max_filter, {"mask_size": mask_param(3, 25, "Window size")}, formula="g(x,y) = max of neighborhood."),
operation("min_filter", "Min Filter", CH3, "Order-Statistics Filters", min_filter, {"mask_size": mask_param(3, 25, "Window size")}, formula="g(x,y) = min of neighborhood."),
operation("laplacian_slide", "Laplacian Sharpening Masks", CH3, "Sharpening Spatial Filters", laplacian_slide, {"mask": select_param("cross", ["cross", "diagonal"], description="Choose one of the taught sharpening masks.")}, matrices=[
kernel_preview("Sharpening cross mask", [[0, 1, 0], [1, -5, 1], [0, 1, 0]]),
kernel_preview("Sharpening diagonal mask", [[1, 1, 1], [1, -9, 1], [1, 1, 1]]),
], formula="Sharpen with selected Laplacian mask."),
operation("gradient_abs_sum", "Gradient Operators", CH3, "Gradient Operator", gradient_abs_sum, {"operator": select_param("sobel", ["sobel", "roberts"], description="Choose Gx/Gy pair.")}, matrices=[
kernel_pair_preview("Roberts Cross-Gradient", [[-1, 0], [0, 1]], [[0, -1], [1, 0]]),
kernel_pair_preview("Sobel", [[-1, -2, -1], [0, 0, 0], [1, 2, 1]], [[-1, 0, 1], [-2, 0, 2], [-1, 0, 1]]),
], formula="Gradient image = abs(imfilter(f,Gx)) + abs(imfilter(f,Gy))."),
operation("high_boost", "High-Boost / Edge Emphasis", CH3, "High-Boost Filtering", high_boost_slide, {"A": float_param(1.5, 1, 6, 0.1, description="Boost factor A, where A >= 1."), "K": odd_param(3, 35, description="Odd averaging mask size used for the blurred image.")}, formula="f_hb = A f - blurred(f)."),
operation("fft_spectrum", "FFT/DFT Spectrum View", CH4, "DFT and FFT", fft_spectrum, {"mode": select_param("log_magnitude", ["magnitude", "log_magnitude", "phase"], description="Choose magnitude, log magnitude, or phase display.")}, formula="F(u,v) = DFT{f(x,y)}", repeatable=False),
operation("rgb_to_gray", "Convert to Grayscale", CH6, "Color Conversion", rgb_to_gray_matlab, {"red_weight": float_param(0.299, 0, 1, 0.001, description="R coefficient in gray = aR + bG + cB."), "green_weight": float_param(0.587, 0, 1, 0.001, description="G coefficient in gray = aR + bG + cB."), "blue_weight": float_param(0.114, 0, 1, 0.001, description="B coefficient in gray = aR + bG + cB.")}, formula="gray = 0.299R + 0.587G + 0.114B by default.", repeatable=False),
operation("rgb_channel", "RGB Channel View", CH6, "RGB color model", rgb_channel, {"channel": select_param("r", ["r", "g", "b"], description="Select the RGB channel to view.")}, formula="Show one RGB channel as grayscale.", repeatable=False),
]
OPERATION_MAP = {item["id"]: item for item in OPERATIONS}
@@ -339,4 +272,11 @@ def apply_registered_operation(image, operation_id, params=None):
item = OPERATION_MAP.get(operation_id)
if item is None:
raise ProcessingError(f"Unsupported operation '{operation_id}'.")
return ensure_uint8(item["func"](ensure_uint8(image), params or {}))
operation_params = dict(params or {})
repeat_count = int(operation_params.pop("_repeat", 1))
if repeat_count < 1 or repeat_count > 20:
raise ProcessingError("N must be between 1 and 20.")
result = ensure_uint8(image)
for _ in range(repeat_count):
result = ensure_uint8(item["func"](result, operation_params))
return result

View File

@@ -1,13 +1,15 @@
import time
import cv2
import numpy as np
from django.conf import settings
from django.db.models import Max
from django.utils import timezone
from .algorithms import ProcessingError, average_images, decode_image, histogram, histogram_payload, normalize_to_uint8, process_image, verify_registration
from .models import ImageSession, ImageState, ProcessingJob
from .registry import apply_registered_operation
from .storage import load_image_array, payload_for_image, save_image_array
from .storage import delete_relative_file, load_image_array, payload_for_image, save_image_array
from .tasks import run_batch_job
@@ -15,7 +17,7 @@ def image_session_create(*, uploaded_file=None, image_base64=None):
if uploaded_file and uploaded_file.size > settings.MAX_UPLOAD_MB * 1024 * 1024:
raise ProcessingError(f"Upload exceeds {settings.MAX_UPLOAD_MB} MB.")
image = decode_image(uploaded_file=uploaded_file, base64_image=image_base64)
image = compact_workspace_image(decode_image(uploaded_file=uploaded_file, base64_image=image_base64))
relative_path = save_image_array(image, "original")
hist = histogram(image)
session = ImageSession.objects.create(
@@ -52,9 +54,23 @@ def image_session_create(*, uploaded_file=None, image_base64=None):
return payload
def compact_workspace_image(image):
max_dimension = int(getattr(settings, "IMAGE_WORKSPACE_MAX_DIMENSION", 1400))
if max_dimension <= 0:
return image
height, width = image.shape[:2]
longest = max(width, height)
if longest <= max_dimension:
return image
scale = max_dimension / float(longest)
next_size = (max(1, int(round(width * scale))), max(1, int(round(height * scale))))
return cv2.resize(image, next_size, interpolation=cv2.INTER_AREA).astype(np.uint8)
def image_state_create(*, session, parent, image, operation, params, label=None, prefix="state"):
relative_path = save_image_array(image, prefix)
sequence = session.states.count()
max_sequence = session.states.aggregate(value=Max("sequence"))["value"]
sequence = 0 if max_sequence is None else max_sequence + 1
state = ImageState.objects.create(
session=session,
parent=parent,
@@ -101,6 +117,15 @@ def image_states_payload(*, states):
return [image_state_payload(state=state, include_image=True) for state in states]
def image_state_delete(*, state):
if state.sequence == 0 or state.operation == "upload":
raise ProcessingError("The original S0 upload state cannot be deleted.")
image_path = state.image
state.children.update(parent=None)
state.delete()
delete_relative_file(image_path)
def image_state_apply_operation(*, state, operation, params):
if state.session.expired:
raise ProcessingError("Image session has expired.")

View File

@@ -50,6 +50,140 @@ class ApiTests(TestCase):
)
self.assertEqual(processed.status_code, 400)
def test_operations_are_core_slide_set(self):
response = self.client.get("/api/operations/")
self.assertEqual(response.status_code, 200)
operation_ids = {item["id"] for item in response.data["operations"]}
operations = {item["id"]: item for item in response.data["operations"]}
self.assertIn("histeq", operation_ids)
self.assertIn("box_filter", operation_ids)
self.assertIn("median_filter", operation_ids)
self.assertIn("noise_filter", operation_ids)
self.assertIn("rgb_to_gray", operation_ids)
self.assertEqual(operations["histeq"]["params"], {})
self.assertEqual(operations["noise_filter"]["label"], "Noise Filter")
self.assertEqual(operations["box_filter"]["label"], "Average / Box Filter")
self.assertEqual(operations["gaussian_filter"]["label"], "Gaussian Filter")
self.assertFalse(operations["negative"]["repeatable"])
self.assertFalse(operations["rgb_to_gray"]["repeatable"])
self.assertEqual(operations["rgb_to_gray"]["params"]["red_weight"]["default"], 0.299)
self.assertEqual(operations["rgb_to_gray"]["params"]["green_weight"]["default"], 0.587)
self.assertEqual(operations["rgb_to_gray"]["params"]["blue_weight"]["default"], 0.114)
self.assertNotIn("crop", operation_ids)
self.assertNotIn("identity", operation_ids)
self.assertNotIn("threshold", operation_ids)
self.assertNotIn("hist_match", operation_ids)
self.assertNotIn("convolution", operation_ids)
self.assertNotIn("bone_scan_workflow", operation_ids)
self.assertNotIn("gaussian_noise", operation_ids)
self.assertNotIn("salt_pepper_noise", operation_ids)
self.assertNotIn("speckle_noise", operation_ids)
self.assertNotIn("hsi_view", operation_ids)
self.assertNotIn("hsi_intensity_filter", operation_ids)
self.assertNotIn("frequency_filter", operation_ids)
self.assertNotIn("frequency_laplacian", operation_ids)
self.assertNotIn("pseudo_color_slices", operation_ids)
self.assertNotIn("gray_to_color_transform", operation_ids)
def test_upload_compacts_large_image(self):
with override_settings(IMAGE_WORKSPACE_MAX_DIMENSION=4):
upload = self.client.post("/api/images/", {"image": png_upload(size=(8, 4))}, format="multipart")
self.assertEqual(upload.status_code, 201)
self.assertEqual(upload.data["width"], 4)
self.assertEqual(upload.data["height"], 2)
def test_state_delete_removes_non_s0_and_keeps_children(self):
upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart")
s0_id = upload.data["states"][0]["state_id"]
first = self.client.post(
f"/api/states/{s0_id}/operations/",
{"operation": "negative", "params": {}},
format="json",
)
second = self.client.post(
f"/api/states/{first.data['state_id']}/operations/",
{"operation": "gamma", "params": {"gamma": 1, "c": 1}},
format="json",
)
image_path = Path(self.tmp.name) / first.data["image_path"]
self.assertTrue(image_path.exists())
deleted = self.client.delete(f"/api/states/{first.data['state_id']}/")
self.assertEqual(deleted.status_code, 204)
self.assertFalse(image_path.exists())
states = self.client.get(f"/api/sessions/{upload.data['session_id']}/states/")
child = next(item for item in states.data["states"] if item["state_id"] == second.data["state_id"])
self.assertIsNone(child["parent_state_id"])
def test_state_numbering_does_not_reuse_deleted_sequence(self):
upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart")
s0_id = upload.data["states"][0]["state_id"]
first = self.client.post(
f"/api/states/{s0_id}/operations/",
{"operation": "negative", "params": {}},
format="json",
)
second = self.client.post(
f"/api/states/{first.data['state_id']}/operations/",
{"operation": "gamma", "params": {"gamma": 1, "c": 1}},
format="json",
)
self.assertEqual(first.data["sequence"], 1)
self.assertEqual(second.data["sequence"], 2)
deleted = self.client.delete(f"/api/states/{first.data['state_id']}/")
self.assertEqual(deleted.status_code, 204)
third = self.client.post(
f"/api/states/{second.data['state_id']}/operations/",
{"operation": "negative", "params": {}},
format="json",
)
self.assertEqual(third.data["sequence"], 3)
self.assertTrue(third.data["label"].startswith("S3 "))
def test_state_delete_blocks_s0(self):
upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart")
s0_id = upload.data["states"][0]["state_id"]
response = self.client.delete(f"/api/states/{s0_id}/")
self.assertEqual(response.status_code, 400)
def test_single_image_operation_uses_repeat_count(self):
upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart")
s0_id = upload.data["states"][0]["state_id"]
response = self.client.post(
f"/api/states/{s0_id}/operations/",
{"operation": "box_filter", "params": {"_repeat": 2, "K": 3}},
format="json",
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.data["params"]["_repeat"], 2)
self.assertEqual(response.data["operation"], "box_filter")
def test_merged_noise_filter_creates_state(self):
upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart")
s0_id = upload.data["states"][0]["state_id"]
response = self.client.post(
f"/api/states/{s0_id}/operations/",
{"operation": "noise_filter", "params": {"kind": "salt_pepper", "amount": 0.1, "salt_ratio": 0.5}},
format="json",
)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.data["operation"], "noise_filter")
self.assertEqual(response.data["params"]["kind"], "salt_pepper")
def test_grayscale_operation_creates_state(self):
upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart")
s0_id = upload.data["states"][0]["state_id"]
gray = self.client.post(
f"/api/states/{s0_id}/operations/",
{"operation": "rgb_to_gray", "params": {"red_weight": 0.299, "green_weight": 0.587, "blue_weight": 0.114}},
format="json",
)
self.assertEqual(gray.status_code, 201)
self.assertEqual(gray.data["operation"], "rgb_to_gray")
@patch("processing.services.run_batch_job.delay")
def test_batch_returns_job_id(self, delay):
first = self.client.post("/api/images/", {"image": png_upload(name="a.png")}, format="multipart")

View File

@@ -9,6 +9,7 @@ from .views import (
ProcessView,
SessionStatesView,
StateCombineView,
StateDetailView,
StateHistogramView,
StateOperationView,
)
@@ -19,6 +20,7 @@ urlpatterns = [
path("images/", ImageUploadView.as_view(), name="image-upload"),
path("operations/", OperationsView.as_view(), name="operations"),
path("sessions/<uuid:session_id>/states/", SessionStatesView.as_view(), name="session-states"),
path("states/<uuid:state_id>/", StateDetailView.as_view(), name="state-detail"),
path("states/<uuid:state_id>/operations/", StateOperationView.as_view(), name="state-operation"),
path("states/<uuid:state_id>/histogram/", StateHistogramView.as_view(), name="state-histogram"),
path("states/combine/", StateCombineView.as_view(), name="state-combine"),

View File

@@ -11,6 +11,7 @@ from .services import (
image_session_create,
image_session_process,
image_state_apply_operation,
image_state_delete,
image_state_payload,
image_states_payload,
processing_job_payload,
@@ -105,6 +106,18 @@ class StateOperationView(APIView):
return error_response(str(exc), code)
class StateDetailView(APIView):
def delete(self, request, state_id):
state = image_state_get(state_id=state_id)
if state is None:
return error_response("Image state does not exist.", status.HTTP_404_NOT_FOUND)
try:
image_state_delete(state=state)
return Response(status=status.HTTP_204_NO_CONTENT)
except ProcessingError as exc:
return error_response(str(exc))
class StateCombineView(APIView):
class InputSerializer(serializers.Serializer):
operation = serializers.ChoiceField(choices=["add", "subtract", "dot_product", "average", "and", "or"])