diff --git a/camera/imx230_pivariety_tuner/.gitignore b/camera/imx230_pivariety_tuner/.gitignore new file mode 100644 index 0000000..478a6ca --- /dev/null +++ b/camera/imx230_pivariety_tuner/.gitignore @@ -0,0 +1,19 @@ +venv/ +.venv/ +__pycache__/ +*.py[cod] + +settings.json +snapshots/ + +*.jpg +*.jpeg +*.png +*.raw +*.mkv +*.mp4 + +*.ko +*.dtbo +*.so +*.bin diff --git a/camera/imx230_pivariety_tuner/README.md b/camera/imx230_pivariety_tuner/README.md new file mode 100644 index 0000000..84305ae --- /dev/null +++ b/camera/imx230_pivariety_tuner/README.md @@ -0,0 +1,92 @@ +# Arducam Pivariety IMX230 Tuner + +Browser-based live preview and camera tuning example for the Arducam +Pivariety IMX230 camera on T3 Gemstone O1. + +## Hardware + +- T3 Gemstone O1 +- Arducam B0324/Pivariety IMX230, UC-788 Rev.B +- CSI0/J11 +- 2-lane MIPI CSI-2 + +## Required software support + +This example requires: + +- Arducam Pivariety Linux V4L2 driver +- T3 Gemstone O1 IMX230 Device Tree overlay +- IMX230 support in `edgeai-tiovx-modules` +- IMX230 support in `edgeai-gst-plugins` +- Matching IMX230 DCC files + +The default DCC directory is: + +```text +/opt/imaging/imx230/linear +``` + +The following files are expected: + +```text +dcc_viss.bin +dcc_2a.bin +``` + +DCC binaries are not included in this example. + +## Python environment + +Create a virtual environment with access to the system GObject, +OpenCV and NumPy packages: + +```bash +python3 -m venv --system-site-packages venv +venv/bin/pip install -r requirements.txt +``` + +System packages such as Python GObject bindings, OpenCV, NumPy, +GStreamer, `media-ctl` and `v4l2-ctl` must already be installed. + +## Run + +```bash +./run.sh +``` + +Then open: + +```text +http://BOARD_IP:8000 +``` + +## Custom build directories + +Custom GStreamer and TIOVX builds can be selected without editing the +source code: + +```bash +IMX230_GST_PLUGIN_DIR=/path/to/gst/plugins \ +IMX230_TIOVX_LIB_DIR=/path/to/tiovx/lib \ +IMX230_DCC_DIR=/path/to/imx230/dcc \ +./run.sh +``` + +## Optional environment variables + +- `IMX230_DCC_DIR` +- `IMX230_GST_PLUGIN_DIR` +- `IMX230_TIOVX_LIB_DIR` +- `IMX230_MEDIA_DEVICE` +- `IMX230_VIDEO_DEVICE` +- `IMX230_SENSOR_DEVICE` +- `IMX230_SETTINGS_FILE` +- `IMX230_SNAPSHOT_DIR` +- `IMX230_HOST` +- `IMX230_PORT` +- `IMX230_PYTHON` +- `IMX230_GST_REGISTRY` +- `T3_EDGEAI_ENV` + +Runtime settings, snapshots, virtual environments, DCC binaries and +compiled libraries are intentionally excluded from the repository. diff --git a/camera/imx230_pivariety_tuner/app.py b/camera/imx230_pivariety_tuner/app.py new file mode 100644 index 0000000..1fdad47 --- /dev/null +++ b/camera/imx230_pivariety_tuner/app.py @@ -0,0 +1,1957 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import atexit +from contextlib import asynccontextmanager +import json +import os +import pwd +import subprocess +import threading +import time +from datetime import datetime +from pathlib import Path +from typing import Any, AsyncIterator, Iterator + +import cv2 +import gi +import numpy as np +import uvicorn +from fastapi import Body, FastAPI, HTTPException +from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse + +gi.require_version("Gst", "1.0") +from gi.repository import Gst + + +APP_DIR = Path(__file__).resolve().parent + +PROFILE_FILE = Path( + os.environ.get("IMX230_SETTINGS_FILE", APP_DIR / "settings.json") +) +SNAPSHOT_DIR = Path( + os.environ.get("IMX230_SNAPSHOT_DIR", APP_DIR / "snapshots") +) + +MEDIA_DEVICE = os.environ.get("IMX230_MEDIA_DEVICE", "/dev/media0") +VIDEO_DEVICE = os.environ.get("IMX230_VIDEO_DEVICE", "/dev/video2") +SENSOR_DEVICE = os.environ.get("IMX230_SENSOR_DEVICE", "/dev/v4l-subdev2") + +DCC_DIR = Path( + os.environ.get("IMX230_DCC_DIR", "/opt/imaging/imx230/linear") +) +DCC_VISS = DCC_DIR / "dcc_viss.bin" +DCC_2A = DCC_DIR / "dcc_2a.bin" + +PREVIEW_WIDTH = 640 +PREVIEW_HEIGHT = 360 +STREAM_FPS = 8.0 + +# Canlı görüntüde ölçüm için kullanılan merkez alan. +ROI_WIDTH_RATIO = 0.34 +ROI_HEIGHT_RATIO = 0.40 + +# Yazılımsal otomatik exposure denetleyicisi. +# Sensör gain'i bu sürümde otomatik değiştirilmez; 100'de tutulması önerilir. +DEFAULT_AE_MODE = "auto" +AE_MODES = {"auto", "manual", "locked"} +AE_INTERVAL_SECONDS = 0.70 +AE_MIN_EXPOSURE = 500 +AE_MAX_EXPOSURE = 20000 + +# Tek-sefer kontrast autofocus ayarları. +# Merkez ROI üzerinde kaba + ince tarama yapılır. +AF_COARSE_STEP = 100 +AF_FINE_RADIUS = 100 +AF_FINE_STEP = 10 +AF_SETTLE_SECONDS = 0.16 +AF_SAMPLE_COUNT = 3 +AF_SAMPLE_INTERVAL = 0.05 +AF_LOW_TEXTURE_STD = 6.0 + +DEFAULT_SETTINGS: dict[str, Any] = { + "exposure": 13810, + "gain": 100, + "focus": 121, + "red_gain": 1.00, + "green_gain": 0.98, + "blue_gain": 0.98, + "gamma": 1.50, + "contrast": 1.06, + "saturation": 1.25, +} + +NEUTRAL_SETTINGS: dict[str, Any] = { + "exposure": 13810, + "gain": 100, + "focus": 121, + "red_gain": 1.00, + "green_gain": 1.00, + "blue_gain": 1.00, + "gamma": 1.00, + "contrast": 1.00, + "saturation": 1.00, +} + +LIMITS: dict[str, tuple[float, float]] = { + "exposure": (20, 65478), + "gain": (100, 800), + "focus": (0, 1000), + "red_gain": (0.50, 1.50), + "green_gain": (0.50, 1.50), + "blue_gain": (0.50, 1.50), + "gamma": (0.50, 2.50), + "contrast": (0.50, 1.80), + "saturation": (0.00, 2.00), +} + + +def run_command(command: list[str]) -> subprocess.CompletedProcess[str]: + print("+", " ".join(command), flush=True) + result = subprocess.run( + command, + check=True, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + if result.stdout: + print(result.stdout, end="") + return result + + +def clamp(name: str, value: float | int) -> float | int: + lower, upper = LIMITS[name] + bounded = max(lower, min(upper, float(value))) + + if name in {"exposure", "gain", "focus"}: + return int(round(bounded)) + + return round(bounded, 2) + + +class CameraController: + def __init__(self) -> None: + self.settings_lock = threading.RLock() + self.frame_lock = threading.Lock() + self.sensor_lock = threading.Lock() + self.measurement_lock = threading.Lock() + self.autofocus_lock = threading.Lock() + self.autofocus_state_lock = threading.Lock() + + self.settings, self.ae_mode = self._load_profile() + self.latest_frame: np.ndarray | None = None + self.pipeline: Gst.Pipeline | None = None + self.running = False + self.last_error = "" + self.bus_thread: threading.Thread | None = None + self.ae_thread: threading.Thread | None = None + self.autofocus_thread: threading.Thread | None = None + self.autofocus_cancel = threading.Event() + + self.autofocus_state: dict[str, Any] = { + "running": False, + "phase": "idle", + "progress": 0, + "current_focus": int(self.settings["focus"]), + "best_focus": int(self.settings["focus"]), + "best_score": 0.0, + "texture_std": 0.0, + "message": "Otomatik odak hazır.", + } + + self.ae_metrics: dict[str, float] = { + "raw_clipped_percent": 0.0, + "display_p50": 0.0, + "display_p75": 0.0, + "display_p90": 0.0, + } + self.ae_last_action = "Başlangıç değeri bekleniyor." + + self.last_measurement: dict[str, Any] | None = None + self.undo_settings: dict[str, Any] | None = None + + def _load_profile(self) -> tuple[dict[str, Any], str]: + settings = dict(DEFAULT_SETTINGS) + ae_mode = DEFAULT_AE_MODE + + if PROFILE_FILE.exists(): + try: + loaded = json.loads(PROFILE_FILE.read_text(encoding="utf-8")) + if isinstance(loaded, dict): + for name in settings: + if name in loaded: + settings[name] = clamp(name, loaded[name]) + + loaded_mode = str(loaded.get("ae_mode", DEFAULT_AE_MODE)) + if loaded_mode in AE_MODES: + ae_mode = loaded_mode + except (OSError, ValueError, TypeError) as exc: + print(f"Profil okunamadı; başlangıç değerleri kullanılacak: {exc}") + + return settings, ae_mode + + def get_settings(self) -> dict[str, Any]: + with self.settings_lock: + result: dict[str, Any] = dict(self.settings) + result["ae_mode"] = self.ae_mode + return result + + @staticmethod + def roi_bounds(width: int, height: int) -> tuple[int, int, int, int]: + roi_width = max(40, int(width * ROI_WIDTH_RATIO)) + roi_height = max(40, int(height * ROI_HEIGHT_RATIO)) + + x1 = (width - roi_width) // 2 + y1 = (height - roi_height) // 2 + x2 = x1 + roi_width + y2 = y1 + roi_height + + return x1, y1, x2, y2 + + def configure_media(self) -> None: + required_paths = [ + MEDIA_DEVICE, + VIDEO_DEVICE, + SENSOR_DEVICE, + DCC_VISS, + DCC_2A, + ] + + missing = [path for path in required_paths if not Path(path).exists()] + if missing: + raise RuntimeError("Eksik dosya veya cihaz: " + ", ".join(missing)) + + entities = [ + '"arducam-pivariety 5-000c":0', + '"cdns_csi2rx.30101000.csi-bridge":0', + '"cdns_csi2rx.30101000.csi-bridge":1', + '"30102000.ticsi2rx":0', + '"30102000.ticsi2rx":1', + ] + + for entity in entities: + run_command([ + "media-ctl", + "-d", + MEDIA_DEVICE, + "--set-v4l2", + f"{entity} [fmt:SBGGR10_1X10/1920x1080 field:none]", + ]) + + run_command([ + "v4l2-ctl", + "-d", + VIDEO_DEVICE, + "--set-fmt-video=width=1920,height=1080,pixelformat=BG10", + ]) + + current = self.get_settings() + self.set_sensor_controls( + exposure=int(current["exposure"]), + gain=int(current["gain"]), + focus=int(current["focus"]), + ) + + def set_sensor_controls( + self, + *, + exposure: int | None = None, + gain: int | None = None, + focus: int | None = None, + ) -> None: + controls: list[str] = [] + + if exposure is not None: + controls.append(f"exposure={int(clamp('exposure', exposure))}") + if gain is not None: + controls.append(f"analogue_gain={int(clamp('gain', gain))}") + if focus is not None: + controls.append(f"focus_absolute={int(clamp('focus', focus))}") + + if not controls: + return + + with self.sensor_lock: + run_command([ + "v4l2-ctl", + "-d", + SENSOR_DEVICE, + "--set-ctrl", + ",".join(controls), + ]) + + def start(self) -> None: + if self.running: + return + + Gst.init(None) + self.configure_media() + + pipeline_text = f""" + v4l2src + device={VIDEO_DEVICE} + io-mode=dmabuf-import + ! video/x-bayer,format=bggr10,width=1920,height=1080 + ! tiovxisp + sensor-name=SENSOR_SONY_IMX230_PIVARIETY + dcc-isp-file={DCC_VISS} + sink_0::dcc-2a-file={DCC_2A} + sink_0::device={SENSOR_DEVICE} + sink_0::ae-mode=2 + sink_0::awb-mode=2 + format-msb=9 + ! video/x-raw,format=NV12,width=1920,height=1080 + ! queue max-size-buffers=2 leaky=downstream + ! videoscale + ! video/x-raw,width={PREVIEW_WIDTH},height={PREVIEW_HEIGHT} + ! videoconvert + ! video/x-raw,format=BGR + ! appsink + name=preview_sink + emit-signals=true + max-buffers=1 + drop=true + sync=false + """ + + parsed = Gst.parse_launch(pipeline_text) + if not isinstance(parsed, Gst.Pipeline): + raise RuntimeError("GStreamer pipeline oluşturulamadı.") + + self.pipeline = parsed + sink = self.pipeline.get_by_name("preview_sink") + + if sink is None: + raise RuntimeError("GStreamer appsink bulunamadı.") + + sink.connect("new-sample", self._on_sample) + + state_result = self.pipeline.set_state(Gst.State.PLAYING) + if state_result == Gst.StateChangeReturn.FAILURE: + self.pipeline.set_state(Gst.State.NULL) + raise RuntimeError("GStreamer pipeline başlatılamadı.") + + self.running = True + self.last_error = "" + + self.bus_thread = threading.Thread( + target=self._monitor_bus, + name="gstreamer-bus", + daemon=True, + ) + self.bus_thread.start() + + self.ae_thread = threading.Thread( + target=self._auto_exposure_loop, + name="imx230-auto-exposure", + daemon=True, + ) + self.ae_thread.start() + + print("DOĞRU: IMX230 canlı pipeline başlatıldı.", flush=True) + + def stop(self) -> None: + self.autofocus_cancel.set() + self.running = False + + if self.pipeline is not None: + self.pipeline.set_state(Gst.State.NULL) + self.pipeline = None + + def _compute_ae_metrics(self) -> dict[str, float] | None: + raw = self.raw_frame() + display = self.adjusted_frame(draw_roi=False) + + if raw is None or display is None: + return None + + raw_maximum = np.max(raw, axis=2) + raw_clipped_percent = float( + np.mean(raw_maximum >= 250) * 100.0 + ) + + display_luma = cv2.cvtColor(display, cv2.COLOR_BGR2GRAY) + + return { + "raw_clipped_percent": raw_clipped_percent, + "display_p50": float(np.percentile(display_luma, 50)), + "display_p75": float(np.percentile(display_luma, 75)), + "display_p90": float(np.percentile(display_luma, 90)), + } + + def _auto_exposure_loop(self) -> None: + while self.running: + time.sleep(AE_INTERVAL_SECONDS) + + with self.settings_lock: + mode = self.ae_mode + current_exposure = int(self.settings["exposure"]) + + if mode != "auto": + continue + + metrics = self._compute_ae_metrics() + if metrics is None: + continue + + with self.settings_lock: + self.ae_metrics = dict(metrics) + + clipped = metrics["raw_clipped_percent"] + p75 = metrics["display_p75"] + p90 = metrics["display_p90"] + + factor = 1.0 + reason = "Pozlama dengeli; değer korunuyor." + + # Parlak alana dönüldüğünde hızlı tepki verir. + if clipped >= 10.0: + factor = 0.60 + reason = "Şiddetli kırpılma; exposure hızlı azaltıldı." + elif clipped >= 5.0: + factor = 0.72 + reason = "Yüksek kırpılma; exposure azaltıldı." + elif clipped >= 2.0: + factor = 0.84 + reason = "Parlak alan fazla; exposure azaltıldı." + elif clipped >= 1.0: + factor = 0.92 + reason = "Hafif kırpılma; exposure az miktarda azaltıldı." + + # Karanlık sahnede exposure yavaşça yükselir. + elif p75 < 55.0: + factor = 1.20 + reason = "Sahne çok karanlık; exposure artırıldı." + elif p75 < 75.0: + factor = 1.12 + reason = "Sahne karanlık; exposure artırıldı." + elif p75 < 90.0: + factor = 1.06 + reason = "Orta tonlar düşük; exposure hafif artırıldı." + + # Parlak fakat henüz tamamen kırpılmamış sahnede yumuşak azaltma. + elif p90 > 230.0 and clipped > 0.40: + factor = 0.95 + reason = "Üst tonlar sınıra yakın; exposure hafif azaltıldı." + + target_exposure = int(round(current_exposure * factor)) + target_exposure = max( + AE_MIN_EXPOSURE, + min(AE_MAX_EXPOSURE, target_exposure), + ) + + # Küçük değişiklikleri uygulamayarak titreşimi azaltır. + if abs(target_exposure - current_exposure) < 80: + with self.settings_lock: + self.ae_last_action = reason + continue + + try: + self.update_settings({"exposure": target_exposure}) + with self.settings_lock: + self.ae_last_action = ( + f"{reason} {current_exposure} → {target_exposure}" + ) + except subprocess.CalledProcessError as exc: + with self.settings_lock: + self.ae_last_action = f"AE kontrol hatası: {exc}" + + def _monitor_bus(self) -> None: + if self.pipeline is None: + return + + bus = self.pipeline.get_bus() + watched = Gst.MessageType.ERROR | Gst.MessageType.EOS + + while self.running: + message = bus.timed_pop_filtered(500 * Gst.MSECOND, watched) + if message is None: + continue + + if message.type == Gst.MessageType.ERROR: + error, debug = message.parse_error() + self.last_error = str(error) + print(f"GStreamer hatası: {error}", flush=True) + if debug: + print(debug, flush=True) + self.running = False + break + + if message.type == Gst.MessageType.EOS: + self.last_error = "Kamera akışı sona erdi." + self.running = False + break + + def _on_sample(self, sink: Any) -> Gst.FlowReturn: + sample = sink.emit("pull-sample") + if sample is None: + return Gst.FlowReturn.ERROR + + caps = sample.get_caps() + structure = caps.get_structure(0) + width = int(structure.get_value("width")) + height = int(structure.get_value("height")) + + buffer = sample.get_buffer() + success, map_info = buffer.map(Gst.MapFlags.READ) + if not success: + return Gst.FlowReturn.ERROR + + try: + expected = width * height * 3 + pixels = np.frombuffer(map_info.data, dtype=np.uint8) + + if pixels.size < expected: + self.last_error = ( + f"Eksik görüntü buffer'ı: beklenen {expected}, gelen {pixels.size}" + ) + return Gst.FlowReturn.ERROR + + frame = pixels[:expected].reshape((height, width, 3)).copy() + + with self.frame_lock: + self.latest_frame = frame + + return Gst.FlowReturn.OK + finally: + buffer.unmap(map_info) + + def raw_frame(self) -> np.ndarray | None: + with self.frame_lock: + if self.latest_frame is None: + return None + return self.latest_frame.copy() + + def adjusted_frame(self, *, draw_roi: bool = False) -> np.ndarray | None: + frame = self.raw_frame() + if frame is None: + return None + + settings = self.get_settings() + result = frame.astype(np.float32) + + # OpenCV kanal sırası BGR'dir. + result[:, :, 0] *= float(settings["blue_gain"]) + result[:, :, 1] *= float(settings["green_gain"]) + result[:, :, 2] *= float(settings["red_gain"]) + + output = np.clip(result, 0, 255).astype(np.uint8) + + # Kullanıcı gamma değeri: + # 1.00 = değişiklik yok + # 1.00'dan büyük = orta tonları açar + # 1.00'dan küçük = orta tonları koyulaştırır + gamma = float(settings["gamma"]) + if abs(gamma - 1.0) > 0.001: + gamma_lut = np.clip( + ((np.arange(256, dtype=np.float32) / 255.0) ** (1.0 / gamma)) + * 255.0, + 0, + 255, + ).astype(np.uint8) + output = cv2.LUT(output, gamma_lut) + + # Kontrastı orta gri (127.5) çevresinde uygula. + contrast = float(settings["contrast"]) + if abs(contrast - 1.0) > 0.001: + contrasted = ( + (output.astype(np.float32) - 127.5) * contrast + + 127.5 + ) + output = np.clip(contrasted, 0, 255).astype(np.uint8) + + # Saturation: gri görüntü ile renkli görüntü arasında doğrusal karışım. + # Bu yöntem hue değerini değiştirmeden doygunluğu artırıp azaltır. + saturation = float(settings["saturation"]) + if abs(saturation - 1.0) > 0.001: + gray = cv2.cvtColor(output, cv2.COLOR_BGR2GRAY).astype(np.float32) + gray3 = np.repeat(gray[:, :, None], 3, axis=2) + saturated = ( + gray3 + + saturation + * (output.astype(np.float32) - gray3) + ) + output = np.clip(saturated, 0, 255).astype(np.uint8) + + if draw_roi: + height, width = output.shape[:2] + x1, y1, x2, y2 = self.roi_bounds(width, height) + + cv2.rectangle(output, (x1, y1), (x2, y2), (0, 220, 255), 2) + cv2.putText( + output, + "Notr kart olcum alani", + (x1, max(22, y1 - 8)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.58, + (0, 220, 255), + 2, + cv2.LINE_AA, + ) + + return output + + def _set_autofocus_state(self, **changes: Any) -> None: + with self.autofocus_state_lock: + self.autofocus_state.update(changes) + + def get_autofocus_status(self) -> dict[str, Any]: + with self.autofocus_state_lock: + return dict(self.autofocus_state) + + def _set_focus_position(self, position: int) -> int: + bounded = int(clamp("focus", position)) + self.set_sensor_controls(focus=bounded) + + with self.settings_lock: + self.settings["focus"] = bounded + + self._set_autofocus_state(current_focus=bounded) + return bounded + + def _focus_score_once(self) -> tuple[float, float]: + frame = self.raw_frame() + if frame is None: + raise RuntimeError("Autofocus için kamera karesi alınamadı.") + + height, width = frame.shape[:2] + x1, y1, x2, y2 = self.roi_bounds(width, height) + roi = frame[y1:y2, x1:x2] + + if roi.size == 0: + raise RuntimeError("Autofocus ROI alanı boş.") + + gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY) + texture_std = float(np.std(gray)) + + # Hafif blur, sensör gürültüsünün yapay netlik puanı üretmesini azaltır. + gray = cv2.GaussianBlur(gray, (3, 3), 0) + + grad_x = cv2.Sobel(gray, cv2.CV_32F, 1, 0, ksize=3) + grad_y = cv2.Sobel(gray, cv2.CV_32F, 0, 1, ksize=3) + magnitude = grad_x * grad_x + grad_y * grad_y + + # En güçlü kenarların üst çeyreğini kullan. Düz arka plan, puanı + # gereksiz yere aşağı çekmez; gürültü ise blur ile bastırılmıştır. + threshold = float(np.percentile(magnitude, 75)) + strong_edges = magnitude[magnitude >= threshold] + + if strong_edges.size == 0: + return 0.0, texture_std + + score = float(np.mean(strong_edges)) + return score, texture_std + + def _sample_focus_score(self) -> tuple[float, float]: + scores: list[float] = [] + textures: list[float] = [] + + for _ in range(AF_SAMPLE_COUNT): + if self.autofocus_cancel.is_set(): + raise RuntimeError("Autofocus iptal edildi.") + + time.sleep(AF_SAMPLE_INTERVAL) + score, texture = self._focus_score_once() + scores.append(score) + textures.append(texture) + + return float(np.median(scores)), float(np.median(textures)) + + def start_autofocus(self) -> dict[str, Any]: + if not self.running: + raise RuntimeError("Kamera çalışmıyor.") + + if not self.autofocus_lock.acquire(blocking=False): + raise RuntimeError("Autofocus zaten çalışıyor.") + + self.autofocus_cancel.clear() + self._set_autofocus_state( + running=True, + phase="starting", + progress=0, + current_focus=int(self.get_settings()["focus"]), + best_focus=int(self.get_settings()["focus"]), + best_score=0.0, + texture_std=0.0, + message="Exposure kilitleniyor; autofocus başlatılıyor.", + ) + + self.autofocus_thread = threading.Thread( + target=self._autofocus_worker, + name="imx230-single-autofocus", + daemon=True, + ) + self.autofocus_thread.start() + + return self.get_autofocus_status() + + def _autofocus_worker(self) -> None: + previous_settings = self.get_settings() + previous_mode = str(previous_settings["ae_mode"]) + previous_focus = int(previous_settings["focus"]) + + try: + # AF sırasında exposure ve gain değişmemeli. + self.update_settings({"ae_mode": "locked"}) + time.sleep(0.35) + + focus_min = int(LIMITS["focus"][0]) + focus_max = int(LIMITS["focus"][1]) + + coarse_positions = list( + range(focus_min, focus_max + 1, AF_COARSE_STEP) + ) + if coarse_positions[-1] != focus_max: + coarse_positions.append(focus_max) + + best_focus = previous_focus + best_score = -1.0 + best_texture = 0.0 + + for index, position in enumerate(coarse_positions, start=1): + if self.autofocus_cancel.is_set(): + raise RuntimeError("Autofocus iptal edildi.") + + applied = self._set_focus_position(position) + time.sleep(AF_SETTLE_SECONDS) + score, texture = self._sample_focus_score() + + if score > best_score: + best_score = score + best_focus = applied + best_texture = texture + + self._set_autofocus_state( + phase="coarse", + progress=int(index / len(coarse_positions) * 55), + current_focus=applied, + best_focus=best_focus, + best_score=round(best_score, 2), + texture_std=round(best_texture, 2), + message=( + f"Kaba tarama: focus={applied}, " + f"puan={score:.1f}" + ), + ) + + fine_start = max(focus_min, best_focus - AF_FINE_RADIUS) + fine_end = min(focus_max, best_focus + AF_FINE_RADIUS) + fine_positions = list( + range(fine_start, fine_end + 1, AF_FINE_STEP) + ) + + if fine_positions[-1] != fine_end: + fine_positions.append(fine_end) + + for index, position in enumerate(fine_positions, start=1): + if self.autofocus_cancel.is_set(): + raise RuntimeError("Autofocus iptal edildi.") + + applied = self._set_focus_position(position) + time.sleep(AF_SETTLE_SECONDS) + score, texture = self._sample_focus_score() + + if score > best_score: + best_score = score + best_focus = applied + best_texture = texture + + self._set_autofocus_state( + phase="fine", + progress=55 + int(index / len(fine_positions) * 44), + current_focus=applied, + best_focus=best_focus, + best_score=round(best_score, 2), + texture_std=round(best_texture, 2), + message=( + f"İnce tarama: focus={applied}, " + f"puan={score:.1f}" + ), + ) + + self._set_focus_position(best_focus) + time.sleep(AF_SETTLE_SECONDS) + + if best_texture < AF_LOW_TEXTURE_STD: + message = ( + f"AF tamamlandı: focus={best_focus}. " + "ROI düşük detaylı; sonucu görsel olarak kontrol et." + ) + else: + message = ( + f"AF tamamlandı: en iyi focus={best_focus}, " + f"netlik puanı={best_score:.1f}." + ) + + self._set_autofocus_state( + running=False, + phase="completed", + progress=100, + current_focus=best_focus, + best_focus=best_focus, + best_score=round(best_score, 2), + texture_std=round(best_texture, 2), + message=message, + ) + + except Exception as exc: + try: + if self.running: + self._set_focus_position(previous_focus) + except Exception: + pass + + self._set_autofocus_state( + running=False, + phase="error", + progress=0, + current_focus=previous_focus, + best_focus=previous_focus, + message=f"Autofocus başarısız: {exc}", + ) + + finally: + try: + if self.running and previous_mode in AE_MODES: + self.update_settings({"ae_mode": previous_mode}) + except Exception as exc: + self._set_autofocus_state( + message=( + self.get_autofocus_status().get("message", "") + + f" Exposure modu geri yüklenemedi: {exc}" + ) + ) + + self.autofocus_lock.release() + + def update_settings(self, changes: dict[str, Any]) -> dict[str, Any]: + allowed = set(DEFAULT_SETTINGS) | {"ae_mode"} + unknown = set(changes) - allowed + + if unknown: + raise ValueError("Bilinmeyen ayarlar: " + ", ".join(sorted(unknown))) + + sensor_changes: dict[str, int] = {} + + with self.settings_lock: + if "ae_mode" in changes: + requested_mode = str(changes["ae_mode"]) + if requested_mode not in AE_MODES: + raise ValueError( + "AE modu auto, manual veya locked olmalıdır." + ) + self.ae_mode = requested_mode + self.ae_last_action = { + "auto": "Otomatik exposure etkin.", + "manual": "Manuel exposure etkin.", + "locked": "Mevcut exposure kilitlendi.", + }[requested_mode] + + for name, raw_value in changes.items(): + if name == "ae_mode": + continue + + value = clamp(name, raw_value) + self.settings[name] = value + + if name in {"exposure", "gain", "focus"}: + sensor_changes[name] = int(value) + + if sensor_changes: + self.set_sensor_controls( + exposure=sensor_changes.get("exposure"), + gain=sensor_changes.get("gain"), + focus=sensor_changes.get("focus"), + ) + + return self.get_settings() + + def save_profile(self) -> None: + PROFILE_FILE.write_text( + json.dumps(self.get_settings(), indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + + def reset(self, neutral: bool = False) -> dict[str, Any]: + target: dict[str, Any] = dict( + NEUTRAL_SETTINGS if neutral else DEFAULT_SETTINGS + ) + target["ae_mode"] = "manual" if neutral else DEFAULT_AE_MODE + return self.update_settings(target) + + def stats(self) -> dict[str, Any]: + frame = self.adjusted_frame() + if frame is None: + return { + "ready": False, + "running": self.running, + "error": self.last_error, + } + + b_mean, g_mean, r_mean = np.mean(frame, axis=(0, 1)) + maximum = np.max(frame, axis=2) + minimum = np.min(frame, axis=2) + + clipped = float(np.mean(maximum >= 250) * 100.0) + dark = float(np.mean(maximum <= 5) * 100.0) + + return { + "ready": True, + "running": self.running, + "error": self.last_error, + "mean_r": round(float(r_mean), 1), + "mean_g": round(float(g_mean), 1), + "mean_b": round(float(b_mean), 1), + "clipped_percent": round(clipped, 2), + "dark_percent": round(dark, 2), + "ae_mode": self.ae_mode, + "ae_raw_clipped_percent": round( + float(self.ae_metrics["raw_clipped_percent"]), + 2, + ), + "ae_p50": round(float(self.ae_metrics["display_p50"]), 1), + "ae_p75": round(float(self.ae_metrics["display_p75"]), 1), + "ae_p90": round(float(self.ae_metrics["display_p90"]), 1), + "ae_last_action": self.ae_last_action, + "current_focus": int(self.get_settings()["focus"]), + "autofocus": self.get_autofocus_status(), + } + + def measure_neutral(self, target_kind: str) -> dict[str, Any]: + frame = self.raw_frame() + if frame is None: + raise RuntimeError("Henüz kamera karesi alınmadı.") + + if target_kind not in {"gray", "white"}: + raise ValueError("Hedef türü gray veya white olmalıdır.") + + height, width = frame.shape[:2] + x1, y1, x2, y2 = self.roi_bounds(width, height) + roi = frame[y1:y2, x1:x2] + + blue = roi[:, :, 0].astype(np.float32) + green = roi[:, :, 1].astype(np.float32) + red = roi[:, :, 2].astype(np.float32) + + maximum = np.maximum(np.maximum(red, green), blue) + minimum = np.minimum(np.minimum(red, green), blue) + + clipped_mask = maximum >= 250 + dark_mask = maximum <= 10 + valid_mask = (~clipped_mask) & (~dark_mask) + + valid_count = int(np.count_nonzero(valid_mask)) + total_count = int(valid_mask.size) + + if valid_count < max(500, int(total_count * 0.25)): + raise RuntimeError( + "Ölçüm alanında yeterli geçerli piksel yok. " + "Kart çok karanlık veya patlamış olabilir." + ) + + r_median = float(np.median(red[valid_mask])) + g_median = float(np.median(green[valid_mask])) + b_median = float(np.median(blue[valid_mask])) + + if min(r_median, g_median, b_median) < 1.0: + raise RuntimeError("Kanal ölçümü güvenilir değil.") + + # En düşük kanalı 1.00 kabul ederek yalnızca yüksek kanalları azaltır. + # Böylece yazılımsal WB ek parlaklık kırpılması üretmez. + neutral_level = min(r_median, g_median, b_median) + + red_gain = float(clamp("red_gain", neutral_level / r_median)) + green_gain = float(clamp("green_gain", neutral_level / g_median)) + blue_gain = float(clamp("blue_gain", neutral_level / b_median)) + + luminance = ( + 0.2126 * red + + 0.7152 * green + + 0.0722 * blue + ) + luma_median = float(np.median(luminance[valid_mask])) + + clipped_percent = float(np.mean(clipped_mask) * 100.0) + dark_percent = float(np.mean(dark_mask) * 100.0) + valid_percent = float(valid_count / total_count * 100.0) + + target_luma = 128.0 if target_kind == "gray" else 190.0 + current = self.get_settings() + current_exposure = int(current["exposure"]) + + if clipped_percent > 1.0: + exposure_factor = min(0.80, target_luma / max(luma_median, 1.0)) + else: + exposure_factor = target_luma / max(luma_median, 1.0) + + # Tek ölçümde aşırı sıçrama yapılmasını engeller. + exposure_factor = max(0.60, min(1.60, exposure_factor)) + recommended_exposure = int( + clamp("exposure", round(current_exposure * exposure_factor)) + ) + + channel_values = { + "kırmızı": r_median, + "yeşil": g_median, + "mavi": b_median, + } + highest_name = max(channel_values, key=channel_values.get) + lowest_name = min(channel_values, key=channel_values.get) + channel_ratio = max(channel_values.values()) / min(channel_values.values()) + + if channel_ratio <= 1.05: + color_diagnosis = "Nötr kanallar birbirine yakın." + else: + color_diagnosis = ( + f"{highest_name.capitalize()} kanal yüksek, " + f"{lowest_name} kanal düşük görünüyor." + ) + + lower_ok = target_luma * 0.82 + upper_ok = target_luma * 1.12 + + if clipped_percent > 1.0: + exposure_diagnosis = ( + f"Ölçüm alanının %{clipped_percent:.2f} kadarı patlamış. " + "Exposure azaltılmalı." + ) + elif luma_median < lower_ok: + exposure_diagnosis = ( + f"Kart karanlık (medyan {luma_median:.1f}). " + "Exposure artırılabilir." + ) + elif luma_median > upper_ok: + exposure_diagnosis = ( + f"Kart fazla parlak (medyan {luma_median:.1f}). " + "Exposure azaltılabilir." + ) + else: + exposure_diagnosis = ( + f"Kart parlaklığı uygun aralıkta (medyan {luma_median:.1f})." + ) + + warning = "" + if channel_ratio > 1.80: + warning = ( + "Kanallar arasındaki fark çok büyük. Ölçüm alanında yalnızca " + "gri/beyaz kart olduğundan ve tek tip ışık kullanıldığından emin ol." + ) + + measurement = { + "target_kind": target_kind, + "roi": {"x1": x1, "y1": y1, "x2": x2, "y2": y2}, + "median_r": round(r_median, 2), + "median_g": round(g_median, 2), + "median_b": round(b_median, 2), + "median_luma": round(luma_median, 2), + "clipped_percent": round(clipped_percent, 2), + "dark_percent": round(dark_percent, 2), + "valid_percent": round(valid_percent, 2), + "color_diagnosis": color_diagnosis, + "exposure_diagnosis": exposure_diagnosis, + "warning": warning, + "recommended": { + "red_gain": red_gain, + "green_gain": green_gain, + "blue_gain": blue_gain, + "exposure": recommended_exposure, + }, + } + + with self.measurement_lock: + self.last_measurement = measurement + + return measurement + + def apply_recommendation( + self, + *, + apply_wb: bool, + apply_exposure: bool, + ) -> dict[str, Any]: + with self.measurement_lock: + measurement = self.last_measurement + + if measurement is None: + raise RuntimeError("Önce nötr kart ölçümü yapmalısın.") + + recommended = measurement["recommended"] + changes: dict[str, Any] = {} + + if apply_wb: + changes.update({ + "red_gain": recommended["red_gain"], + "green_gain": recommended["green_gain"], + "blue_gain": recommended["blue_gain"], + }) + + if apply_exposure: + changes["exposure"] = recommended["exposure"] + + if not changes: + raise ValueError("Uygulanacak öneri seçilmedi.") + + self.undo_settings = self.get_settings() + return self.update_settings(changes) + + def undo_last(self) -> dict[str, Any]: + if self.undo_settings is None: + raise RuntimeError("Geri alınacak bir öneri yok.") + + target = self.undo_settings + self.undo_settings = None + return self.update_settings(target) + + def snapshot(self) -> Path: + frame = self.adjusted_frame(draw_roi=False) + if frame is None: + raise RuntimeError("Henüz kamera karesi alınmadı.") + + SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") + output = SNAPSHOT_DIR / f"imx230-{timestamp}.jpg" + + if not cv2.imwrite(str(output), frame): + raise RuntimeError("JPEG dosyası yazılamadı.") + + try: + account = pwd.getpwnam("gemstone") + os.chown(output, account.pw_uid, account.pw_gid) + except (KeyError, PermissionError): + pass + + return output + + def mjpeg_stream(self) -> Iterator[bytes]: + frame_interval = 1.0 / STREAM_FPS + + while True: + started = time.monotonic() + frame = self.adjusted_frame(draw_roi=True) + + if frame is None: + frame = np.zeros( + (PREVIEW_HEIGHT, PREVIEW_WIDTH, 3), + dtype=np.uint8, + ) + cv2.putText( + frame, + "Kamera bekleniyor...", + (145, 185), + cv2.FONT_HERSHEY_SIMPLEX, + 0.8, + (255, 255, 255), + 2, + cv2.LINE_AA, + ) + + ok, encoded = cv2.imencode( + ".jpg", + frame, + [int(cv2.IMWRITE_JPEG_QUALITY), 82], + ) + + if ok: + yield ( + b"--frame\r\n" + b"Content-Type: image/jpeg\r\n\r\n" + + encoded.tobytes() + + b"\r\n" + ) + + elapsed = time.monotonic() - started + time.sleep(max(0.0, frame_interval - elapsed)) + + +camera = CameraController() + + +@asynccontextmanager +async def lifespan(_: FastAPI) -> AsyncIterator[None]: + camera.start() + try: + yield + finally: + camera.stop() + + +app = FastAPI( + title="IMX230 Pivariety Tuner", + lifespan=lifespan, +) + +atexit.register(camera.stop) + + +@app.get("/", response_class=HTMLResponse) +def index() -> str: + return HTML_PAGE + + +@app.get("/video") +def video() -> StreamingResponse: + return StreamingResponse( + camera.mjpeg_stream(), + media_type="multipart/x-mixed-replace; boundary=frame", + ) + + +@app.get("/api/settings") +def get_settings() -> dict[str, Any]: + return camera.get_settings() + + +@app.post("/api/settings") +def set_settings(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: + try: + return camera.update_settings(payload) + except (ValueError, TypeError, subprocess.CalledProcessError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/save") +def save_profile() -> dict[str, str]: + camera.save_profile() + return {"status": "saved", "path": str(PROFILE_FILE)} + + +@app.post("/api/reset") +def reset_profile(payload: dict[str, Any] = Body(default={})) -> dict[str, Any]: + neutral = bool(payload.get("neutral", False)) + return camera.reset(neutral=neutral) + + +@app.post("/api/snapshot") +def snapshot() -> dict[str, str]: + try: + output = camera.snapshot() + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + + return {"status": "saved", "path": str(output)} + + +@app.post("/api/autofocus") +def autofocus() -> JSONResponse: + try: + result = camera.start_autofocus() + except RuntimeError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + return JSONResponse(result) + + +@app.get("/api/stats") +def stats() -> JSONResponse: + return JSONResponse(camera.stats()) + + +@app.post("/api/measure-neutral") +def measure_neutral(payload: dict[str, Any] = Body(default={})) -> JSONResponse: + try: + result = camera.measure_neutral( + str(payload.get("target_kind", "gray")) + ) + except (RuntimeError, ValueError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + return JSONResponse(result) + + +@app.post("/api/apply-recommendation") +def apply_recommendation( + payload: dict[str, Any] = Body(default={}), +) -> dict[str, Any]: + try: + return camera.apply_recommendation( + apply_wb=bool(payload.get("apply_wb", False)), + apply_exposure=bool(payload.get("apply_exposure", False)), + ) + except (RuntimeError, ValueError, subprocess.CalledProcessError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/undo") +def undo() -> dict[str, Any]: + try: + return camera.undo_last() + except (RuntimeError, subprocess.CalledProcessError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +HTML_PAGE = r""" + + + + + +IMX230 Renk Ayarı V8 + + + +
+

IMX230 Canlı Renk Ayarı V8

+ Bağlanıyor… +
+ +
+
+ Canlı IMX230 görüntüsü + +
+
–Genel R
+
–Genel G
+
–Genel B
+
–Patlayan %
+
–Karanlık %
+
+ +

+ Sarı çerçevenin içini yalnızca mat gri kart veya mat beyaz A4 ile doldur. + Kartta parlama olmamalı; aynı anda iki farklı ışık kaynağı kullanma. + Ölçüm, kullanıcı RGB ve gamma düzeltmesinden önceki ISP görüntüsünden yapılır. + Bu sürümde otomatik sistem yalnızca exposure değerini değiştirir; + analogue gain 100'de sabit kalır. + Gamma orta tonları, kontrast açık-koyu ayrımını, + renk doygunluğu ise renklerin canlılığını değiştirir. +

+
+ + +
+ + + + +""" + + +if __name__ == "__main__": + uvicorn.run( + app, + host=os.environ.get("IMX230_HOST", "0.0.0.0"), + port=int(os.environ.get("IMX230_PORT", "8000")), + log_level="info", + access_log=False, + ) diff --git a/camera/imx230_pivariety_tuner/requirements.txt b/camera/imx230_pivariety_tuner/requirements.txt new file mode 100644 index 0000000..97dc7cd --- /dev/null +++ b/camera/imx230_pivariety_tuner/requirements.txt @@ -0,0 +1,2 @@ +fastapi +uvicorn diff --git a/camera/imx230_pivariety_tuner/run.sh b/camera/imx230_pivariety_tuner/run.sh new file mode 100755 index 0000000..c67def2 --- /dev/null +++ b/camera/imx230_pivariety_tuner/run.sh @@ -0,0 +1,50 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +SCRIPT_DIR="$( + cd -- "$(dirname -- "${BASH_SOURCE[0]}")" + pwd +)" + +if [[ ${EUID} -ne 0 ]]; then + exec sudo \ + --preserve-env=T3_EDGEAI_ENV,IMX230_GST_PLUGIN_DIR,IMX230_TIOVX_LIB_DIR,IMX230_DCC_DIR,IMX230_PYTHON,IMX230_MEDIA_DEVICE,IMX230_VIDEO_DEVICE,IMX230_SENSOR_DEVICE,IMX230_SETTINGS_FILE,IMX230_SNAPSHOT_DIR,IMX230_HOST,IMX230_PORT,IMX230_GST_REGISTRY \ + -- "$0" "$@" +fi + +EDGEAI_ENV="${T3_EDGEAI_ENV:-/opt/t3-edgeai-env}" + +if [[ ! -r "$EDGEAI_ENV" ]]; then + echo "Error: T3 Edge AI environment file not found: $EDGEAI_ENV" >&2 + exit 1 +fi + +# shellcheck disable=SC1090 +source "$EDGEAI_ENV" +unset LD_PRELOAD + +if [[ -n ${IMX230_GST_PLUGIN_DIR:-} ]]; then + export GST_PLUGIN_PATH="${IMX230_GST_PLUGIN_DIR}${GST_PLUGIN_PATH:+:${GST_PLUGIN_PATH}}" +fi + +if [[ -n ${IMX230_TIOVX_LIB_DIR:-} ]]; then + export LD_LIBRARY_PATH="${IMX230_TIOVX_LIB_DIR}${LD_LIBRARY_PATH:+:${LD_LIBRARY_PATH}}" +fi + +export IMX230_DCC_DIR="${IMX230_DCC_DIR:-/opt/imaging/imx230/linear}" +export GST_REGISTRY="${IMX230_GST_REGISTRY:-/tmp/gst-registry-imx230-tuner.bin}" + +rm -f "$GST_REGISTRY" + +PYTHON="${IMX230_PYTHON:-${SCRIPT_DIR}/venv/bin/python}" + +if [[ ! -x "$PYTHON" ]]; then + PYTHON="$(command -v python3 || true)" +fi + +if [[ -z "$PYTHON" || ! -x "$PYTHON" ]]; then + echo "Error: no usable Python interpreter was found." >&2 + exit 1 +fi + +exec "$PYTHON" "$SCRIPT_DIR/app.py" diff --git a/camera/imx230_pivariety_tuner/settings.example.json b/camera/imx230_pivariety_tuner/settings.example.json new file mode 100644 index 0000000..5e5f679 --- /dev/null +++ b/camera/imx230_pivariety_tuner/settings.example.json @@ -0,0 +1,12 @@ +{ + "exposure": 13810, + "gain": 100, + "focus": 121, + "red_gain": 1.0, + "green_gain": 0.98, + "blue_gain": 0.98, + "gamma": 1.5, + "contrast": 1.06, + "saturation": 1.25, + "ae_mode": "auto" +}