agrobot_base/Python/gal5000/preview_4ch_autoexposure.py

548 lines
18 KiB
Python
Raw Normal View History

2026-01-28 18:05:51 +00:00
import math
import os
import time
import ctypes as C
from ctypes import wintypes as W
import numpy as np
import cv2
# =========================
# CONFIG
# =========================
SDK_DIR = os.path.join(os.path.dirname(__file__), "dlls")
DLL_NAME = "VT_SDK64.dll"
TIMEOUT_MS = 2000
WINDOW_NAME = "GAL5000 4CH Preview (E=AEC toggle, G=AGC toggle, Q=quit)"
# Camera scan/open
DEVICE_UDEF = 0
DEVICE_INDEX = 0
DATA_RAW = 0
# RAW geometry (como você já capturou)
RAW_W = 2592
RAW_H = 2056
# Bayer+NIR pattern (2x2):
# R G
# IR B
# => R = [0::2,0::2], G=[0::2,1::2], IR=[1::2,0::2], B=[1::2,1::2]
# =========================
# PARAM IDs (VT_Param.h)
# =========================
BUF_SIZE = 256
PARAM_ID_SENSOR_EXPOSURETIMERAW = 0x00003010
PARAM_ID_SENSOR_GAINANALOGRAW = 0x00003020
PARAM_ID_SENSOR_EXPOSUREAUTOENABLE = 0x00003011
PARAM_ID_SENSOR_GAINANALOGAUTOENABLE = 0x00003021
PARAM_ID_SENSOR_GAINANALOGAGCMAX = 0x00003022
PARAM_ID_COMMON_DISPLAYFPS = 0x00000203 # float R (média efetiva) :contentReference[oaicite:9]{index=9}
# PARAM_VALUETYPE
VALUE_INT = 0
VALUE_FLOAT = 1
VALUE_STR = 2
# =========================
# MANUAL EXPOSURE HOTKEYS
# =========================
EXP_MIN = 1
EXP_MAX = 20000 # ajuste depois conforme o sensor aceitar
EXP_STEP = 200 # passo “normal”
EXP_STEP_FAST = 1000 # passo “rápido”
TARGET_P95 = 160.0
SAT_LIMIT = 0.02
K = 0.35
MAX_STEP = 0.18
GAIN_A_MIN, GAIN_A_MAX = 0, 255 # ajuste conforme seu sensor
GAIN_STEP = 2
def clamp(v, lo, hi):
return lo if v < lo else hi if v > hi else v
def measure_raw_g_metrics(raw: np.ndarray):
"""
Mede brilho no canal G cru (8-bit) usando ROI (chão) e retorna:
- p90/p95 (brilho)
- sat (fração saturada)
"""
H, W = raw.shape[:2]
# canal G cru (mesmo que você já usa em soft_ae_step) :contentReference[oaicite:1]{index=1}
G = raw[0::2, 1::2] # tamanho ~ H/2 x W/2
h2, w2 = G.shape
# ROI: base da imagem, cortando laterais
y0, y1 = int(h2 * 0.55), int(h2 * 0.95)
x0, x1 = int(w2 * 0.15), int(w2 * 0.85)
roi = G[y0:y1, x0:x1]
p90 = float(np.percentile(roi, 90))
p95 = float(np.percentile(roi, 95))
sat = float(np.mean(roi >= 250))
return p90, p95, sat
def measure_brightness_and_sat(img_bgr):
h, w = img_bgr.shape[:2]
y0, y1 = int(h * 0.55), int(h * 0.95)
x0, x1 = int(w * 0.15), int(w * 0.85)
roi = img_bgr[y0:y1, x0:x1]
g = roi[:, :, 1].astype(np.uint8)
p95 = float(np.percentile(g, 95))
sat = float(np.mean(g >= 250))
return p95, sat
def auto_exposure_step(img_bgr, exp_raw, gain_a):
# mede
p95, sat = measure_brightness_and_sat(img_bgr)
# se está saturando, reduz exposição com prioridade
if sat > SAT_LIMIT:
# força erro “negativo”
err = math.log((TARGET_P95 + 1e-6) / (p95 + 1e-6)) # pode ser positivo/negativo
err = min(err, -0.15) # garante redução
else:
err = math.log((TARGET_P95 + 1e-6) / (p95 + 1e-6))
# limita o tamanho do passo por iteração (evita oscilar)
step = clamp(K * err, -MAX_STEP, +MAX_STEP)
# atualiza exposição (multiplicativo)
new_exp = int(round(exp_raw * math.exp(step)))
new_exp = clamp(new_exp, EXP_MIN, EXP_MAX)
# Ganho: só mexe se exposição já “bateu no teto/chão”
new_gain = gain_a
if new_exp >= EXP_MAX and p95 < (TARGET_P95 * 0.85):
new_gain = clamp(gain_a + GAIN_STEP, GAIN_A_MIN, GAIN_A_MAX)
elif new_exp <= EXP_MIN and (p95 > (TARGET_P95 * 1.15) or sat > SAT_LIMIT):
new_gain = clamp(gain_a - GAIN_STEP, GAIN_A_MIN, GAIN_A_MAX)
dbg = {"p95": p95, "sat": sat, "err": err, "step": step}
return new_exp, new_gain, dbg
class RobustAE:
def __init__(self,
exp_min=1, exp_max=20000,
target_p95=140.0,
deadband=6.0,
k=0.12,
max_step=0.10,
ema_alpha=0.20,
sat_limit=0.01):
self.exp_min = exp_min
self.exp_max = exp_max
self.target = target_p95
self.deadband = deadband
self.k = k
self.max_step = max_step
self.ema_alpha = ema_alpha
self.sat_limit = sat_limit
self.p95_ema = None
def step(self, raw, exp_raw):
p90, p95, sat = measure_raw_g_metrics(raw)
# EMA do p95 pra tirar tremedeira
if self.p95_ema is None:
self.p95_ema = p95
else:
self.p95_ema = (1 - self.ema_alpha) * self.p95_ema + self.ema_alpha * p95
e = self.target - self.p95_ema # erro em "nível de pixel"
# deadband: se tá perto do alvo, NÃO mexe
if abs(e) <= self.deadband and sat <= self.sat_limit:
return exp_raw, {"p90": p90, "p95": p95, "p95_ema": self.p95_ema, "sat": sat, "hold": True}
# Se saturou, força reduzir exposição
if sat > self.sat_limit:
# passo negativo garantido
step = -min(self.max_step, 0.12)
else:
# controle em log: step proporcional ao erro relativo
ratio = (self.target + 1e-6) / (self.p95_ema + 1e-6)
step = self.k * math.log(ratio)
step = max(-self.max_step, min(self.max_step, step))
new_exp = int(round(exp_raw * math.exp(step)))
new_exp = max(self.exp_min, min(self.exp_max, new_exp))
return new_exp, {"p90": p90, "p95": p95, "p95_ema": self.p95_ema, "sat": sat, "step": step, "hold": False}
# =========================
# STRUCTS (mínimo necessário)
# =========================
class VT_FRAMEINFO(C.Structure):
_fields_ = [
("lFrameID", W.DWORD),
("lBufSize", W.DWORD),
("lWidth", W.DWORD),
("lHeight", W.DWORD),
("lPixBits", C.c_ubyte),
("_pad0", C.c_ubyte * 3),
("pBufPtr", C.POINTER(C.c_ubyte)),
("lFrameStatus", W.DWORD),
("lPixType", W.DWORD),
("lTimeStamp", W.DWORD),
("_reserve", W.DWORD * 8),
]
class VT_DEVPARAM(C.Structure):
_fields_ = [
("bUseName", W.BOOL),
("lParamByID", W.DWORD),
("lParamByName", C.c_char * BUF_SIZE),
]
def devparam_by_id(pid: int) -> VT_DEVPARAM:
p = VT_DEVPARAM()
p.bUseName = False
p.lParamByID = pid
p.lParamByName = b"" # <- CORRETO: bytes (fica zerado / string vazia)
return p
# =========================
# DLL LOAD + prototypes
# =========================
os.add_dll_directory(SDK_DIR)
dll = C.WinDLL(os.path.join(SDK_DIR, DLL_NAME))
print("DLL carregada OK:", dll)
dll.VT_DeviceScan.argtypes = [C.POINTER(C.c_ubyte), C.c_int]
dll.VT_DeviceScan.restype = C.c_int
dll.VT_DeviceOpen.argtypes = [C.c_void_p, C.POINTER(W.HANDLE), C.c_int, C.c_int]
dll.VT_DeviceOpen.restype = C.c_int
dll.VT_SingleFrameCapture.argtypes = [W.HANDLE, C.POINTER(VT_FRAMEINFO), C.c_int, C.c_int, W.BOOL]
dll.VT_SingleFrameCapture.restype = C.c_int
dll.VT_DeviceClose.argtypes = [C.POINTER(W.HANDLE)]
dll.VT_DeviceClose.restype = C.c_int
# Param API
dll.VT_ParamGetValue.argtypes = [W.HANDLE, VT_DEVPARAM, C.c_void_p, C.c_int]
dll.VT_ParamGetValue.restype = C.c_int
dll.VT_ParamSetValue.argtypes = [W.HANDLE, VT_DEVPARAM, C.c_void_p, C.c_int]
dll.VT_ParamSetValue.restype = C.c_int
def ck(ret: int, name: str):
if ret != 0:
print(f"{name} falhou, ret={ret}")
raise RuntimeError(f"{name} falhou, ret={ret}")
def param_set_int(h: W.HANDLE, pid: int, value: int):
p = devparam_by_id(pid)
v = C.c_int(value)
ret = dll.VT_ParamSetValue(h, p, C.byref(v), VALUE_INT)
ck(ret, f"VT_ParamSetValue({hex(pid)})")
def param_get_int(h: W.HANDLE, pid: int) -> int:
p = devparam_by_id(pid)
v = C.c_int(0)
ret = dll.VT_ParamGetValue(h, p, C.byref(v), VALUE_INT)
ck(ret, f"VT_ParamGetValue({hex(pid)})")
return int(v.value)
def param_get_float(h: W.HANDLE, pid: int) -> float:
p = devparam_by_id(pid)
v = C.c_float(0)
ret = dll.VT_ParamGetValue(h, p, C.byref(v), VALUE_FLOAT)
ck(ret, f"VT_ParamGetValue({hex(pid)})")
return float(v.value)
def set_bool(h: W.HANDLE, pid: int, enabled: bool):
param_set_int(h, pid, 1 if enabled else 0) # boolean no SDK é int 0/1
def param_supported(h, pid, vtype):
p = devparam_by_id(pid)
minv = C.c_int()
maxv = C.c_int()
inc = C.c_int()
ret = dll.VT_ParamGetRange(h, p,
C.byref(minv),
C.byref(maxv),
C.byref(inc),
vtype)
return ret == 0
def capture_raw8(h: W.HANDLE) -> np.ndarray:
fi = VT_FRAMEINFO()
ret = dll.VT_SingleFrameCapture(h, C.byref(fi), DATA_RAW, TIMEOUT_MS, True)
ck(ret, "VT_SingleFrameCapture")
w, hh = int(fi.lWidth), int(fi.lHeight)
if w != RAW_W or hh != RAW_H:
# Se em algum momento você mudar resolução/ROI, aqui te avisa.
print(f"[WARN] Res mudou: {w}x{hh} (esperado {RAW_W}x{RAW_H})")
buf = C.string_at(fi.pBufPtr, fi.lBufSize)
arr = np.frombuffer(buf, dtype=np.uint8)
# garante reshape correto
needed = w * hh
if arr.size < needed:
arr = np.pad(arr, (0, needed - arr.size), mode="constant", constant_values=0)
arr = arr[:needed].reshape(hh, w)
return arr
def make_montage_4ch(raw: np.ndarray):
# pattern:
R = raw[0::2, 0::2]
G = raw[0::2, 1::2]
IR = raw[1::2, 0::2]
B = raw[1::2, 1::2]
# para visual: normaliza levemente (só pra ficar agradável)
# sem mexer nos dados crus do treino, isso é só display.
def norm8(x):
# estica por percentil p2-p98 pra ver melhor em campo
lo = np.percentile(x, 2)
hi = np.percentile(x, 98)
if hi <= lo + 1:
return x
y = (x.astype(np.float32) - lo) * (255.0 / (hi - lo))
return np.clip(y, 0, 255).astype(np.uint8)
Rn, Gn, IRn, Bn = map(norm8, [R, G, IR, B])
top = np.hstack([Rn, Gn])
bot = np.hstack([IRn, Bn])
mont = np.vstack([top, bot])
# labels (coloca texto no montage)
mont_bgr = cv2.cvtColor(mont, cv2.COLOR_GRAY2BGR)
h2, w2 = Rn.shape # cada plane é H/2 x W/2
# posições de texto
cv2.putText(mont_bgr, "R", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255,255,255), 2, cv2.LINE_AA)
cv2.putText(mont_bgr, "G", (w2 + 10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255,255,255), 2, cv2.LINE_AA)
cv2.putText(mont_bgr, "IR", (10, h2 + 30), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255,255,255), 2, cv2.LINE_AA)
cv2.putText(mont_bgr, "B", (w2 + 10, h2 + 30), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255,255,255), 2, cv2.LINE_AA)
return mont_bgr
def make_rgb_preview(raw: np.ndarray, upscale=2):
R = raw[0::2, 0::2]
G = raw[0::2, 1::2]
B = raw[1::2, 1::2]
# normalização leve só para display (p2-p98)
def norm8(x):
lo = np.percentile(x, 2)
hi = np.percentile(x, 98)
if hi <= lo + 1:
return x
y = (x.astype(np.float32) - lo) * (255.0 / (hi - lo))
return np.clip(y, 0, 255).astype(np.uint8)
Rn, Gn, Bn = map(norm8, [R, G, B])
rgb = np.dstack([Bn, Gn, Rn]) # OpenCV = BGR
if upscale and upscale != 1:
rgb = cv2.resize(rgb, (rgb.shape[1]*upscale, rgb.shape[0]*upscale), interpolation=cv2.INTER_NEAREST)
return rgb
def overlay_hud(img, aec_on, agc_on, exp_raw, gain_a, gain_d, fps):
lines = [
f"AEC: {'ON' if aec_on else 'OFF'} | AGC: {'ON' if agc_on else 'OFF'}",
f"ExposureRaw: {exp_raw}",
f"Gain A: {gain_a} | Gain D: {gain_d}",
f"FPS: {fps:.1f}",
"Keys: + - [ ] | A=AE | Q=quit",
]
y = 35
for s in lines:
draw_text(img, s, (12, y), scale=0.85)
y += 32
def draw_text(img, text, pos, scale=0.8):
x, y = pos
# sombra
cv2.putText(img, text, (x+2, y+2),
cv2.FONT_HERSHEY_SIMPLEX, scale,
(0, 0, 0), 3, cv2.LINE_AA)
# texto principal
cv2.putText(img, text, (x, y),
cv2.FONT_HERSHEY_SIMPLEX, scale,
(255, 255, 255), 2, cv2.LINE_AA)
def main():
# scan
n = C.c_ubyte(0)
ret = dll.VT_DeviceScan(C.byref(n), DEVICE_UDEF)
ck(ret, "VT_DeviceScan")
if n.value == 0:
raise RuntimeError("Nenhuma câmera encontrada.")
# open
idx = C.c_ubyte(0)
h = W.HANDLE()
ret = dll.VT_DeviceOpen(C.byref(idx), C.byref(h), DEVICE_INDEX, DEVICE_UDEF)
ck(ret, "VT_DeviceOpen")
print("DeviceOpen OK, handle=", h.value)
cv2.namedWindow(WINDOW_NAME, cv2.WINDOW_NORMAL)
cv2.namedWindow("RGB", cv2.WINDOW_NORMAL)
show_rgb = True
t0 = time.time()
frames = 0
fps = 0.0
ae = RobustAE(exp_min=EXP_MIN, exp_max=EXP_MAX, target_p95=140.0)
ae_every_n = 4
ae_i = 0
aec_on = False
agc_on = False
exp_raw = 1500 # valor inicial que você escolhe
gain_a = 0
gain_d = 0
has_exp_raw = param_supported(h, PARAM_ID_SENSOR_EXPOSURETIMERAW, VALUE_INT)
has_gain_a = param_supported(h, PARAM_ID_SENSOR_GAINANALOGRAW, VALUE_INT)
has_hw_aec = param_supported(h, PARAM_ID_SENSOR_EXPOSUREAUTOENABLE, VALUE_INT)
has_hw_agc = param_supported(h, PARAM_ID_SENSOR_GAINANALOGAUTOENABLE, VALUE_INT)
if (has_hw_aec):
aec_on = bool(param_get_int(h, PARAM_ID_SENSOR_EXPOSUREAUTOENABLE))
if (has_hw_agc):
agc_on = bool(param_get_int(h, PARAM_ID_SENSOR_GAINANALOGAUTOENABLE))
if (has_exp_raw):
exp_raw = param_get_int(h, PARAM_ID_SENSOR_EXPOSURETIMERAW)
if (has_gain_a):
gain_a = param_get_int(h, PARAM_ID_SENSOR_GAINANALOGRAW)
try:
def set_exposure_manual(new_exp: int):
new_exp = int(max(EXP_MIN, min(EXP_MAX, new_exp)))
param_set_int(h, PARAM_ID_SENSOR_EXPOSURETIMERAW, new_exp)
return new_exp
while True:
raw = capture_raw8(h)
if aec_on:
ae_i += 1
if ae_i % ae_every_n == 0:
new_exp, dbg = ae.step(raw, exp_raw)
if new_exp != exp_raw:
exp_raw = set_exposure_manual(new_exp)
# debug opcional:
# print(f"[AE] p95={dbg['p95']:.1f} ema={dbg['p95_ema']:.1f} sat={dbg['sat']*100:.2f}% exp={exp_raw} hold={dbg.get('hold')}")
montage = make_montage_4ch(raw)
# AUTO-EXPOSURE (sem GET)
if aec_on:
ae_i += 1
if ae_i % ae_every_n == 0:
try:
new_exp, new_gain_a, dbg = auto_exposure_step(montage, exp_raw, gain_a)
if new_exp != exp_raw:
exp_raw = set_exposure_manual(new_exp)
# Se você quiser mexer em ganho também:
if new_gain_a != gain_a:
param_set_int(h, PARAM_ID_SENSOR_GAINANALOGRAW, int(new_gain_a))
gain_a = int(new_gain_a)
# debug opcional
# print(f"[AE] p95={dbg['p95']:.1f} sat={dbg['sat']*100:.2f}% exp={exp_raw} gainA={gain_a}")
except Exception as e:
print("[AE] erro:", e)
frames += 1
dt = time.time() - t0
if dt >= 1.0:
fps = frames / dt
frames = 0
t0 = time.time()
overlay_hud(montage, aec_on, agc_on, exp_raw, gain_a, gain_d, fps)
cv2.imshow(WINDOW_NAME, montage)
if show_rgb:
rgb = make_rgb_preview(raw, upscale=2)
cv2.imshow("RGB", rgb)
k = cv2.waitKey(1) & 0xFF
if k in (ord('q'), ord('Q'), 27):
break
elif k in (ord('e'), ord('E')): # exemplo: E alterna AEC do hardware
aec_on = not aec_on
set_bool(h, PARAM_ID_SENSOR_EXPOSUREAUTOENABLE, aec_on)
print("AEC(hw) =", aec_on)
elif k in (ord('g'), ord('G')): # G alterna AGC do hardware
agc_on = not agc_on
set_bool(h, PARAM_ID_SENSOR_GAINANALOGAUTOENABLE, agc_on)
print("AGC(hw) =", agc_on)
elif k in (ord('v'), ord('V')):
show_rgb = not show_rgb
if not show_rgb:
cv2.destroyWindow("RGB")
else:
cv2.namedWindow("RGB", cv2.WINDOW_NORMAL)
elif aec_on == False:
if k in (ord('+'), ord('=')): # '=' costuma ser '+' sem shift em alguns teclados
try:
exp_raw = set_exposure_manual(exp_raw + EXP_STEP)
print(f"[MANUAL] ExposureRaw -> {exp_raw}")
except Exception as e:
print("[ERR] manual exp +:", e)
elif k in (ord('-'), ord('_')):
try:
exp_raw = set_exposure_manual(exp_raw - EXP_STEP)
print(f"[MANUAL] ExposureRaw -> {exp_raw}")
except Exception as e:
print("[ERR] manual exp -:", e)
elif k == ord(']'): # fast +
try:
exp_raw = set_exposure_manual(exp_raw + EXP_STEP_FAST)
print(f"[MANUAL] ExposureRaw (fast) -> {exp_raw}")
except Exception as e:
print("[ERR] manual exp fast +:", e)
elif k == ord('['): # fast -
try:
exp_raw = set_exposure_manual(exp_raw - EXP_STEP_FAST)
print(f"[MANUAL] ExposureRaw (fast) -> {exp_raw}")
except Exception as e:
print("[ERR] manual exp fast -:", e)
finally:
try:
ret = dll.VT_DeviceClose(C.byref(h))
if ret != 0:
print("VT_DeviceClose retornou:", ret)
except Exception as e:
print("Erro ao fechar:", e)
cv2.destroyAllWindows()
print("Fim.")
if __name__ == "__main__":
main()