diff --git a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/config.py b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/config.py index 70ea9dbef..593b0067a 100644 --- a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/config.py +++ b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/config.py @@ -159,6 +159,78 @@ WEED_DEFAULT_CONFIG = { "k_roi_shift_px_per_vnorm": 24.0, + # ============================================================ + # 7.1) Memória espacial por bico / predição até atuação + # ============================================================ + # Estratégia nova: + # Cada bico mantém uma "esteira" normalizada da imagem. + # cell 0 = topo da imagem + # cell N-1 = pé da imagem + # + # A câmera alimenta a esteira. + # A velocidade do rover desloca a memória para baixo na imagem. + # O bico atua quando a evidência chega na zona de atuação. + "usar_memoria_espacial_bicos": True, + + # ------------------------------------------------------------ + # Campo / operação + # ------------------------------------------------------------ + # Zona de atuação em percentual vertical da imagem, de cima para baixo. + # + # Exemplo: + # faixa_atuacao_bicos = 0.70 + # area_atuacao_bicos = 0.10 + # + # Resultado: + # zona de atuação = 70% até 80% do frame + # predição antes = 0% até 70% + # sobra depois = 80% até 100% + "faixa_atuacao_bicos": 0.70, + "area_atuacao_bicos": 0.10, + + # ------------------------------------------------------------ + # Técnico / baseline + # ------------------------------------------------------------ + # Quantidade de células verticais da esteira. + # 100 células = cada célula representa 1% da altura da imagem. + "memoria_num_cells": 100, + + # Fração vertical da imagem percorrida por segundo a 1 m/s. + # Exemplo: 0.90 significa que, a 1 m/s, o chão percorre ~90% + # da imagem por segundo. + "k_shift_frame_por_mps": None, + "comprimento_visao_chao_cm": 110.0, + + # Latência total estimada até o produto realmente atingir o alvo. + # Inclui worker + Redis/C# + CAN + solenoide + hidráulica. + "latencia_total_atuacao_ms": 120.0, + + # Atualização da confiança espacial. + "min_frac_erva_predicao": 0.004, + "percent_ref_evidencia": 0.020, + "score_up_gain": 0.25, + "score_down_decay": 0.06, + + # Histerese da memória no momento da atuação. + "score_ligar_bico": 0.55, + "score_desligar_bico": 0.25, + + # Agregação das células dentro da zona de atuação. + # Opções: "max", "mean", "max_mean". + "atuacao_score_mode": "max_mean", + + # Segurança contra saltos grandes de tempo entre frames. + "max_dt_shift_s": 0.50, + + # Com a memória espacial, o radar global não deve matar bico pendente. + # False evita perder erva já vista que ainda está chegando na atuação. + "radar_global_bloqueia_bicos": False, + + # Mantidos só para debug/telemetria aproximada, não como regulagem de campo. + "comprimento_visao_chao_cm": 110.0, + "largura_visao_chao_cm": 170.0, + + # ============================================================ # 8) Filtros opcionais da máscara # ============================================================ diff --git a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/weed_detector.py b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/weed_detector.py index 64d6f9962..a46614649 100644 --- a/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/weed_detector.py +++ b/AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/weed_worker/weed_detector.py @@ -9,18 +9,27 @@ import numpy as np class WeedDetector: """ - WeedDetector v1. + WeedDetector v2 - memória espacial por bico. Contrato oficial: predictions: np.ndarray uint8 HxW 0 = fundo / não pulverizar 1 = alvo pulverizável - Responsabilidade: - - Receber target_mask pronta do ONNX/TensorRT. - - Calcular presença global de alvo no frame. - - Calcular atuação dos bicos por faixa vertical. - - Aplicar EMA, histerese e debounce. + Estratégia: + - Recebe target_mask pronta do ONNX/TensorRT. + - Calcula radar global de alvo. + - Divide a imagem em colunas por bico. + - Divide o eixo vertical em células físicas no sentido do avanço. + - Mantém uma memória espacial de confiança por bico. + - Desloca a memória pela distância real percorrida pelo rover. + - Atua quando a evidência chega na zona física de atuação. + + Convenção espacial: + - cell index 0 = região mais próxima da linha dos bicos. + - cell index maior = região mais distante, vista antes pela câmera. + - Conforme o rover anda, a memória se desloca de índices maiores + para índices menores. Este arquivo NÃO: - gera imagem; @@ -44,11 +53,11 @@ class WeedDetector: if self.prediction_contract != self.CONTRATO_OFICIAL: raise RuntimeError( - f"WeedDetector v1 aceita apenas prediction_contract='{self.CONTRATO_OFICIAL}'. " + f"WeedDetector v2 aceita apenas prediction_contract='{self.CONTRATO_OFICIAL}'. " f"Recebido: {self.prediction_contract}" ) - resolucao = self.config.get("ia_resolution", [1024, 640]) + resolucao = self.config.get("ia_resolution", [640, 400]) self.resolucao: Tuple[int, int] = ( int(resolucao[0]), int(resolucao[1]), @@ -58,7 +67,9 @@ class WeedDetector: if self.qtd_bicos <= 0: self.qtd_bicos = 7 - self.inverter_ordem_bicos = bool(self.config.get("inverter_ordem_bicos", False)) + self.inverter_ordem_bicos = bool( + self.config.get("inverter_ordem_bicos", False) + ) self.detector_debug_perf = bool( self.config.get("detector_debug_perf", False) @@ -76,21 +87,27 @@ class WeedDetector: } # LUT booleana para interpretar target_mask. - # Isso evita comparação direta espalhada no código e protege - # caso venha algum valor extra inesperado. self._is_target = np.zeros(256, dtype=bool) self._is_target[self.TARGET_ID] = True - w, h = self.resolucao - self._inv_total = 1.0 / max(1, w * h) + # Geometria/memória espacial. + self._atualizar_geometria_memoria() + + self._score_grid = np.zeros( + (self.qtd_bicos, self.num_cells), + dtype=np.float32, + ) + + self._frac_grid_last = np.zeros_like(self._score_grid, dtype=np.float32) # Radar global. self._target_frac_global_ema = 0.0 + self._target_no_radar_frame = False + self._target_no_radar_memoria = False self._target_no_radar = False self._target_no_radar_percent = 0.0 # Estado temporal por bico. - self._ema_frac_bico = np.zeros(self.qtd_bicos, dtype=np.float32) self._on_cnt = np.zeros(self.qtd_bicos, dtype=np.int32) self._off_cnt = np.zeros(self.qtd_bicos, dtype=np.int32) self._estado_bico = np.zeros(self.qtd_bicos, dtype=bool) @@ -99,17 +116,82 @@ class WeedDetector: self._morf_cache_k: Optional[int] = None self._morf_kernel = None + # Tempo/movimento. + self._last_update_ts: Optional[float] = None + self._distancia_deslocada_total_cm = 0.0 + + def _atualizar_geometria_memoria(self): + cfg = self.config + + # Memória normalizada em percentual da altura da imagem. + # cell 0 = topo da imagem + # cell N-1 = pé da imagem + self.num_cells = int(cfg.get("memoria_num_cells", 100) or 100) + self.num_cells = max(10, self.num_cells) + + self.cell_frac = 1.0 / float(self.num_cells) + + self.cell_centers_frac = ( + np.arange(self.num_cells, dtype=np.float32) + 0.5 + ) * self.cell_frac + + # Mantém estes nomes só para não quebrar estatística/debug antigo. + self.comprimento_visao_chao_cm = float(cfg.get("comprimento_visao_chao_cm", 110.0) or 110.0) + self.largura_visao_chao_cm = float(cfg.get("largura_visao_chao_cm", 170.0) or 170.0) + self.cell_cm = self.comprimento_visao_chao_cm / float(self.num_cells) + self.largura_por_bico_cm = self.largura_visao_chao_cm / max(1, self.qtd_bicos) + + def _calcular_zona_atuacao_frac_from_config(self, cfg: dict): + """ + Converte os dois parâmetros de campo para zona de atuação normalizada. + + Convenção: + 0.0 = topo da imagem + 1.0 = pé da imagem + + Exemplo: + faixa_atuacao_bicos = 0.70 + area_atuacao_bicos = 0.10 + + Resultado: + zona de atuação = 70% até 80% da imagem + """ + + faixa = float(cfg.get("faixa_atuacao_bicos", 0.70) or 0.70) + area = float(cfg.get("area_atuacao_bicos", 0.10) or 0.10) + + faixa = max(0.0, min(1.0, faixa)) + area = max(0.001, min(1.0, area)) + + inicio = faixa + fim = faixa + area + + inicio = max(0.0, min(1.0, inicio)) + fim = max(0.0, min(1.0, fim)) + + if fim < inicio: + inicio, fim = fim, inicio + + return inicio, fim + def atualizar_config(self, config: Optional[dict] = None): """ Atualiza config do detector sem recriar o objeto. Útil para refletir mudanças vindas do Redis/C#. - Se qtd_bicos mudar, reinicializa o estado temporal. + Se qtd_bicos, inversão ou geometria mudarem, reinicializa o estado. """ if config is None: from weed_worker.config import load_seg_config config = load_seg_config() + old_qtd = self.qtd_bicos + old_inv = self.inverter_ordem_bicos + old_num_cells = getattr(self, "num_cells", None) + old_cell_cm = getattr(self, "cell_cm", None) + old_comp_cm = getattr(self, "comprimento_visao_chao_cm", None) + old_larg_cm = getattr(self, "largura_visao_chao_cm", None) + self.config = config contrato = str( @@ -118,7 +200,7 @@ class WeedDetector: if contrato != self.CONTRATO_OFICIAL: raise RuntimeError( - f"WeedDetector v1 aceita apenas prediction_contract='{self.CONTRATO_OFICIAL}'. " + f"WeedDetector v2 aceita apenas prediction_contract='{self.CONTRATO_OFICIAL}'. " f"Recebido: {contrato}" ) @@ -127,31 +209,35 @@ class WeedDetector: novo_qtd_bicos = int( self.config.get("qtd_bicos", self.qtd_bicos) or self.qtd_bicos ) - novo_inverter_ordem_bicos = bool( + if novo_qtd_bicos <= 0: + novo_qtd_bicos = 7 + + self.qtd_bicos = novo_qtd_bicos + self.inverter_ordem_bicos = bool( self.config.get( "inverter_ordem_bicos", self.inverter_ordem_bicos, ) ) - if novo_qtd_bicos <= 0: - novo_qtd_bicos = 7 + self._atualizar_geometria_memoria() mudou_mapeamento = ( - novo_qtd_bicos != self.qtd_bicos - or novo_inverter_ordem_bicos != self.inverter_ordem_bicos + old_qtd != self.qtd_bicos + or old_inv != self.inverter_ordem_bicos + or old_num_cells != self.num_cells + or old_cell_cm != self.cell_cm + or old_comp_cm != self.comprimento_visao_chao_cm + or old_larg_cm != self.largura_visao_chao_cm ) - self.qtd_bicos = novo_qtd_bicos - self.inverter_ordem_bicos = novo_inverter_ordem_bicos - - if mudou_mapeamento: - self._inicializar_estado() - self.detector_debug_perf = bool( self.config.get("detector_debug_perf", self.detector_debug_perf) ) + if mudou_mapeamento: + self._inicializar_estado() + # ============================================================ # API principal # ============================================================ @@ -180,12 +266,19 @@ class WeedDetector: predictions = predictions.astype(np.uint8, copy=False) t0 = time.perf_counter() + agora = time.time() cfg = self.config - vel_norm = float(cfg.get("velocidade_robo", 0.0) or 0.0) + vel_mps = float(cfg.get("velocidade_robo", 0.0) or 0.0) + if vel_mps < 0: + vel_mps = 0.0 + + # Mantém nome antigo vel_norm por compatibilidade conceitual, + # mas aqui tratamos como velocidade em m/s. + vel_norm = vel_mps t_rad0 = time.perf_counter() - target_no_radar, estat_target = self._decidir_target_no_radar( + target_no_radar_frame, estat_target = self._decidir_target_no_radar( predictions, cfg, vel_norm=vel_norm, @@ -193,14 +286,24 @@ class WeedDetector: t_rad1 = time.perf_counter() t_bic0 = time.perf_counter() - controle_bicos, estat_bicos = self._atuacao_por_mascara( - predictions, - cfg, - vel_norm=vel_norm, + controle_bicos, estat_bicos = self._atuacao_por_memoria_espacial( + predictions=predictions, + cfg=cfg, + vel_mps=vel_mps, + agora=agora, ) t_bic1 = time.perf_counter() - agora = time.time() + # Radar final: alvo visto no frame atual OU alvo ainda vivo + # na memória espacial. Isso é importante porque a erva pode + # já ter saído da imagem e ainda estar chegando no bico. + target_no_radar = bool( + target_no_radar_frame + or self._target_no_radar_memoria + or any(controle_bicos.values()) + ) + + self._target_no_radar = target_no_radar dados_visuais = { "timestamp": agora, @@ -208,15 +311,14 @@ class WeedDetector: "width": int(predictions.shape[1]), "controle": controle_bicos, - # Mantém o nome antigo para não quebrar payload/C# agora. - # Semântica atual: True se há alvo pulverizável no radar. + # Mantém nomes antigos para compatibilidade. "ervas_no_radar": bool(target_no_radar), "fracao_ervas_no_radar": float(estat_target.get("frac_global", 0.0)), - "percentual_ervas_no_radar": float(estat_target.get("frac_global", 0.0) * 100.0), + "percentual_ervas_no_radar": float( + estat_target.get("frac_global", 0.0) * 100.0 + ), "estatisticas": { - # Mantém a chave "erva" para compatibilidade do payload. - # Semântica atual: estatística de target/alvo pulverizável. "erva": estat_target, "bicos": estat_bicos, }, @@ -251,8 +353,7 @@ class WeedDetector: on_global = float(cfg.get("min_frac_erva_global_on", 0.0020)) off_global = float(cfg.get("min_frac_erva_global_off", 0.0015)) - # Ajuste por velocidade: - # valores maiores deixam o gate global mais sensível com velocidade. + # Ajuste legado por velocidade. k = float(cfg.get("erva_thresh_vel_gain", 0.0) or 0.0) if k: adj = 1.0 - k * self._clamp01(vel_norm) @@ -261,8 +362,11 @@ class WeedDetector: else: adj = 1.0 + h, w = predictions.shape[:2] + inv_total = 1.0 / max(1, h * w) + target_sum = int(self._is_target[predictions].sum()) - frac_global = target_sum * self._inv_total + frac_global = target_sum * inv_total alpha = float(cfg.get("erva_frac_ema", 0.30)) alpha = min(1.0, max(0.0, alpha)) @@ -272,15 +376,15 @@ class WeedDetector: + alpha * frac_global ) - prev = bool(self._target_no_radar) + prev = bool(self._target_no_radar_frame) thr = off_global if prev else on_global - target_no_radar = bool(ema_g >= thr) + target_no_radar_frame = bool(ema_g >= thr) - self._target_no_radar = target_no_radar + self._target_no_radar_frame = target_no_radar_frame self._target_no_radar_percent = float(frac_global) - return target_no_radar, { + return target_no_radar_frame, { "frac_global": float(frac_global), "ema_global": float(ema_g), "thr_on_global": float(on_global), @@ -289,10 +393,485 @@ class WeedDetector: } # ============================================================ - # Controle por bico + # Controle por memória espacial # ============================================================ - def _atuacao_por_mascara( + def _atuacao_por_memoria_espacial( + self, + predictions: np.ndarray, + cfg: dict, + vel_mps: float, + agora: float, + ): + qtd_bicos = self.qtd_bicos + + if qtd_bicos <= 0: + return {}, { + "motivo": "qtd_bicos_invalido", + "frac_erva_por_bico": np.zeros(0, dtype=np.float32), + "score_atuacao_por_bico": np.zeros(0, dtype=np.float32), + } + + usar_memoria = bool(cfg.get("usar_memoria_espacial_bicos", True)) + if not usar_memoria: + # Fallback raro: mantém compatibilidade caso queira desligar a estratégia nova. + return self._atuacao_por_roi_legada( + predictions=predictions, + cfg=cfg, + vel_norm=vel_mps, + ) + + # 1) Desloca a memória pela distância percorrida desde o último ciclo. + shift_info = self._deslocar_memoria_por_movimento( + vel_mps=vel_mps, + agora=agora, + cfg=cfg, + ) + + # 2) Calcula evidência espacial atual por bico/célula. + frac_grid, grid_info = self._calcular_frac_grid_por_bico_cell( + predictions=predictions, + cfg=cfg, + ) + + self._frac_grid_last = frac_grid + + # 3) Atualiza score espacial. + score_update_info = self._atualizar_score_memoria( + frac_grid=frac_grid, + cfg=cfg, + ) + + # 4) Decide bicos pela zona física de atuação, com compensação de latência. + atuacao_bicos, estat_atuacao = self._decidir_bicos_por_zona_atuacao( + cfg=cfg, + vel_mps=vel_mps, + ) + + # 5) Radar de memória: existe alvo relevante ainda vivo na esteira? + score_mem_max = float(self._score_grid.max()) if self._score_grid.size else 0.0 + score_mem_mean = float(self._score_grid.mean()) if self._score_grid.size else 0.0 + self._target_no_radar_memoria = bool( + score_mem_max >= float(cfg.get("score_desligar_bico", 0.25)) + ) + + estatisticas = { + "motivo": "memoria_espacial", + "ordem_bicos_invertida": bool(self.inverter_ordem_bicos), + + # Mantém nomes antigos em formato aproximado para compatibilidade. + # Aqui a fração por bico é a maior fração observada no frame atual + # dentro das células da visão. + "frac_erva_por_bico": np.max(frac_grid, axis=1).astype(np.float32), + "ema_frac_por_bico": np.max(self._score_grid, axis=1).astype(np.float32), + + # Novos dados úteis. + "score_atuacao_por_bico": estat_atuacao["score_atuacao_por_bico"], + "score_max_por_bico": np.max(self._score_grid, axis=1).astype(np.float32), + "score_mean_por_bico": np.mean(self._score_grid, axis=1).astype(np.float32), + "estado_bico": self._estado_bico.copy(), + "on_cnt": self._on_cnt.copy(), + "off_cnt": self._off_cnt.copy(), + + "memoria": { + "num_cells": int(self.num_cells), + "cell_cm": float(self.cell_cm), + "comprimento_visao_chao_cm": float(self.comprimento_visao_chao_cm), + "largura_visao_chao_cm": float(self.largura_visao_chao_cm), + "largura_por_bico_cm": float(self.largura_por_bico_cm), + "score_mem_max": score_mem_max, + "score_mem_mean": score_mem_mean, + "target_no_radar_memoria": bool(self._target_no_radar_memoria), + }, + + "movimento": shift_info, + "grid": grid_info, + "score_update": score_update_info, + "atuacao": estat_atuacao, + } + + return atuacao_bicos, estatisticas + + def _deslocar_memoria_por_movimento( + self, + vel_mps: float, + agora: float, + cfg: dict, + ): + if self._last_update_ts is None: + self._last_update_ts = agora + return { + "dt_s": 0.0, + "vel_mps": float(vel_mps), + "delta_frac": 0.0, + "shift_cells": 0.0, + "distancia_deslocada_total_cm": float(self._distancia_deslocada_total_cm), + } + + dt = float(agora - self._last_update_ts) + self._last_update_ts = agora + + max_dt = float(cfg.get("max_dt_shift_s", 0.50) or 0.50) + if max_dt > 0: + dt = min(dt, max_dt) + + if dt <= 0 or vel_mps <= 0: + return { + "dt_s": float(max(0.0, dt)), + "vel_mps": float(vel_mps), + "delta_frac": 0.0, + "shift_cells": 0.0, + "distancia_deslocada_total_cm": float(self._distancia_deslocada_total_cm), + } + + # Fração vertical da imagem percorrida por segundo a 1 m/s. + # Esse é parâmetro técnico, não ajuste de campo. + k_cfg = cfg.get("k_shift_frame_por_mps", None) + + if k_cfg is None: + comprimento_cm = float(cfg.get("comprimento_visao_chao_cm", 110.0) or 110.0) + k_shift = 100.0 / max(1.0, comprimento_cm) + else: + k_shift = float(k_cfg) + + k_shift = max(0.0, k_shift) + + delta_frac = vel_mps * dt * k_shift + shift_cells = delta_frac * float(self.num_cells) + + # Mantém apenas para debug antigo aproximado. + self._distancia_deslocada_total_cm += ( + delta_frac * float(self.comprimento_visao_chao_cm) + ) + + self._score_grid = self._shift_grid_towards_actuation( + self._score_grid, + shift_cells=shift_cells, + ) + + return { + "dt_s": float(dt), + "vel_mps": float(vel_mps), + "delta_frac": float(delta_frac), + "shift_cells": float(shift_cells), + "k_shift_frame_por_mps": float(k_shift), + "distancia_deslocada_total_cm": float(self._distancia_deslocada_total_cm), + } + + def _shift_grid_towards_actuation( + self, + grid: np.ndarray, + shift_cells: float, + ) -> np.ndarray: + """ + Desloca a memória para baixo na imagem. + + cell 0 = topo da imagem + cell N-1 = pé da imagem + + Se o rover avançou, o que estava na célula 20 vai para 21, 22... + Portanto: + new[i] = old[i - shift] + """ + if grid.size == 0: + return grid + + n = grid.shape[1] + + if shift_cells <= 0: + return grid + + if shift_cells >= n: + return np.zeros_like(grid, dtype=np.float32) + + idx = np.arange(n, dtype=np.float32) + src = idx - float(shift_cells) + + shifted = np.zeros_like(grid, dtype=np.float32) + + for b in range(grid.shape[0]): + shifted[b, :] = np.interp( + src, + idx, + grid[b, :], + left=0.0, + right=0.0, + ).astype(np.float32) + + return shifted + + def _calcular_frac_grid_por_bico_cell( + self, + predictions: np.ndarray, + cfg: dict, + ): + h, w = predictions.shape[:2] + + mask_target = self._is_target[predictions] + mask_target = self._aplicar_filtros_opcionais( + mask_target=mask_target, + cfg=cfg, + ) + + # Integral image para somar retângulos rápido. + ii = np.pad( + mask_target.astype(np.uint8).cumsum(axis=0).cumsum(axis=1), + ((1, 0), (1, 0)), + mode="constant", + constant_values=0, + ) + + x_edges = np.linspace(0, w, self.qtd_bicos + 1, dtype=np.int32) + y_edges = np.linspace(0, h, self.num_cells + 1, dtype=np.int32) + + frac_grid = np.zeros( + (self.qtd_bicos, self.num_cells), + dtype=np.float32, + ) + + valid_cell_count = 0 + + for c in range(self.num_cells): + y_top = int(y_edges[c]) + y_bot = int(y_edges[c + 1]) + + if y_bot <= y_top: + continue + + valid_cell_count += 1 + cell_h = y_bot - y_top + + for b in range(self.qtd_bicos): + x0 = int(x_edges[b]) + x1 = int(x_edges[b + 1]) + + if x1 <= x0: + continue + + area = float(cell_h * (x1 - x0)) + if area <= 0: + continue + + total = ( + ii[y_bot, x1] + - ii[y_top, x1] + - ii[y_bot, x0] + + ii[y_top, x0] + ) + + frac_grid[b, c] = float(total) / area + + # A segmentação é naturalmente da esquerda para a direita. + # Quando habilitado, inverte associação região imagem -> bico físico. + if self.inverter_ordem_bicos: + frac_grid = frac_grid[::-1, :].copy() + + return frac_grid, { + "valid_cell_count": int(valid_cell_count), + "height": int(h), + "width": int(w), + "x_edges": x_edges, + "y_edges": y_edges, + "zona_coord": "frac_top_to_bottom", + } + + def _atualizar_score_memoria( + self, + frac_grid: np.ndarray, + cfg: dict, + ): + min_frac_pred = float( + cfg.get( + "min_frac_erva_predicao", + max(0.001, float(cfg.get("min_frac_erva_por_bico_on", 0.02)) * 0.20), + ) + ) + + percent_ref = float( + cfg.get( + "percent_ref_evidencia", + max(0.001, float(cfg.get("min_frac_erva_por_bico_on", 0.02))), + ) + ) + percent_ref = max(1e-6, percent_ref) + + up_gain = float(cfg.get("score_up_gain", 0.25)) + down_decay = float(cfg.get("score_down_decay", 0.06)) + + up_gain = max(0.0, up_gain) + down_decay = max(0.0, down_decay) + + evidencia = np.clip(frac_grid / percent_ref, 0.0, 1.0).astype(np.float32) + cond_pos = frac_grid >= min_frac_pred + + # Decaimento geral, depois reforço positivo. + score = np.maximum(0.0, self._score_grid - down_decay) + + score = np.where( + cond_pos, + np.minimum(1.0, score + up_gain * evidencia), + score, + ).astype(np.float32) + + self._score_grid = score + + return { + "min_frac_predicao": float(min_frac_pred), + "percent_ref_evidencia": float(percent_ref), + "score_up_gain": float(up_gain), + "score_down_decay": float(down_decay), + "frac_grid_max": float(frac_grid.max()) if frac_grid.size else 0.0, + "evidencia_max": float(evidencia.max()) if evidencia.size else 0.0, + } + + def _decidir_bicos_por_zona_atuacao( + self, + cfg: dict, + vel_mps: float, + ): + zona_ini_frac, zona_fim_frac = self._calcular_zona_atuacao_frac_from_config(cfg) + + lat_ms = float(cfg.get("latencia_total_atuacao_ms", 120.0) or 0.0) + lat_s = max(0.0, lat_ms / 1000.0) + + k_cfg = cfg.get("k_shift_frame_por_mps", None) + + if k_cfg is None: + comprimento_cm = float(cfg.get("comprimento_visao_chao_cm", 110.0) or 110.0) + k_shift = 100.0 / max(1.0, comprimento_cm) + else: + k_shift = float(k_cfg) + + k_shift = max(0.0, k_shift) + + # Antecipação em fração da imagem: + # Se demora para atuar, avalia células um pouco acima da zona real, + # porque elas chegarão na zona quando o bico abrir de fato. + antecipacao_frac = max(0.0, vel_mps * lat_s * k_shift) + + zona_eval_ini_frac = zona_ini_frac - antecipacao_frac + zona_eval_fim_frac = zona_fim_frac - antecipacao_frac + + zona_eval_ini_frac = max(0.0, min(1.0, zona_eval_ini_frac)) + zona_eval_fim_frac = max(0.0, min(1.0, zona_eval_fim_frac)) + + if zona_eval_fim_frac < zona_eval_ini_frac: + zona_eval_ini_frac, zona_eval_fim_frac = zona_eval_fim_frac, zona_eval_ini_frac + + cell_ini = int(np.floor(zona_eval_ini_frac * self.num_cells)) + cell_fim = int(np.ceil(zona_eval_fim_frac * self.num_cells)) + + cell_ini = max(0, min(self.num_cells, cell_ini)) + cell_fim = max(0, min(self.num_cells, cell_fim)) + + if cell_fim <= cell_ini: + score_atuacao = np.zeros(self.qtd_bicos, dtype=np.float32) + zona_scores = np.zeros((self.qtd_bicos, 0), dtype=np.float32) + else: + zona_scores = self._score_grid[:, cell_ini:cell_fim] + score_atuacao = self._agregar_score_zona( + zona_scores=zona_scores, + cfg=cfg, + ) + + score_on = float(cfg.get("score_ligar_bico", 0.55)) + score_off = float(cfg.get("score_desligar_bico", 0.25)) + + on_frames = int(cfg.get("on_frames_required", 1)) + off_frames = int(cfg.get("off_frames_required", 2)) + + on_frames = max(1, on_frames) + off_frames = max(1, off_frames) + + cond_on = score_atuacao >= score_on + cond_off = score_atuacao <= score_off + + self._on_cnt = np.where(cond_on, self._on_cnt + 1, 0) + self._off_cnt = np.where(cond_off, self._off_cnt + 1, 0) + + estado = self._estado_bico.copy() + estado = np.where(self._on_cnt >= on_frames, True, estado) + estado = np.where(self._off_cnt >= off_frames, False, estado) + + radar_bloqueia = bool(cfg.get("radar_global_bloqueia_bicos", False)) + usar_gate_global = bool(cfg.get("usar_radar_global_gate", True)) + + if radar_bloqueia and usar_gate_global and not self._target_no_radar_frame: + estado[:] = False + self._on_cnt[:] = 0 + self._off_cnt[:] = 0 + + self._estado_bico = estado.astype(bool, copy=False) + + atuacao_bicos = { + i: bool(self._estado_bico[i]) + for i in range(self.qtd_bicos) + } + + self.ultimo_status_bicos = atuacao_bicos.copy() + + return atuacao_bicos, { + "score_atuacao_por_bico": score_atuacao.astype(np.float32), + "score_on": float(score_on), + "score_off": float(score_off), + + # Zona real configurada pelo operador. + "zona_atuacao_inicio_frac": float(zona_ini_frac), + "zona_atuacao_fim_frac": float(zona_fim_frac), + + # Zona avaliada com compensação de latência. + "latencia_total_atuacao_ms": float(lat_ms), + "antecipacao_frac": float(antecipacao_frac), + "zona_eval_inicio_frac": float(zona_eval_ini_frac), + "zona_eval_fim_frac": float(zona_eval_fim_frac), + + "cell_ini": int(cell_ini), + "cell_fim": int(cell_fim), + "on_frames_required": int(on_frames), + "off_frames_required": int(off_frames), + "radar_global_bloqueia_bicos": bool(radar_bloqueia), + + "zona_scores_max": ( + np.max(zona_scores, axis=1).astype(np.float32) + if zona_scores.size + else np.zeros(self.qtd_bicos, dtype=np.float32) + ), + "zona_scores_mean": ( + np.mean(zona_scores, axis=1).astype(np.float32) + if zona_scores.size + else np.zeros(self.qtd_bicos, dtype=np.float32) + ), + } + + def _agregar_score_zona( + self, + zona_scores: np.ndarray, + cfg: dict, + ) -> np.ndarray: + if zona_scores.size == 0: + return np.zeros(self.qtd_bicos, dtype=np.float32) + + mode = str(cfg.get("atuacao_score_mode", "max_mean")).lower() + + if mode == "mean": + return np.mean(zona_scores, axis=1).astype(np.float32) + + if mode == "max": + return np.max(zona_scores, axis=1).astype(np.float32) + + # max_mean: + # max captura touceira/local forte. + # mean ajuda tapete de mato falhado. + max_v = np.max(zona_scores, axis=1) + mean_v = np.mean(zona_scores, axis=1) + + return (0.70 * max_v + 0.30 * mean_v).astype(np.float32) + + # ============================================================ + # Fallback legado + # ============================================================ + + def _atuacao_por_roi_legada( self, predictions: np.ndarray, cfg: dict, @@ -311,7 +890,9 @@ class WeedDetector: zona_altura = float(cfg.get("area_atuacao_bicos", 0.1)) usar_gate_global = bool(cfg.get("usar_radar_global_gate", True)) - if usar_gate_global and not self._target_no_radar: + radar_bloqueia = bool(cfg.get("radar_global_bloqueia_bicos", True)) + + if usar_gate_global and radar_bloqueia and not self._target_no_radar_frame: return self._retornar_bicos_off( motivo="gate_global_off", y_top=0, @@ -358,9 +939,6 @@ class WeedDetector: larguras = np.diff(edges).astype(np.int32) area_por_bico = alturas * larguras - # A segmentação é calculada naturalmente da esquerda para a direita. - # Quando habilitado, inverte a associação entre região da imagem - # e numeração física dos bicos. if self.inverter_ordem_bicos: px_target_por_bico = px_target_por_bico[::-1].copy() larguras = larguras[::-1].copy() @@ -373,17 +951,20 @@ class WeedDetector: 0.0, ).astype(np.float32) - atuacao_bicos, estat_debounce = self._atualizar_estado_bicos( + # Usa o estado espacial como "EMA" compatível no fallback. + # Reinicializa o grid para evitar misturar fallback com memória antiga. + self._score_grid[:] = 0.0 + + atuacao_bicos, estat_debounce = self._atualizar_estado_bicos_legado( frac_target_por_bico=frac_target_por_bico, cfg=cfg, vel_norm=vel_norm, ) estatisticas = { - # Mantém nomes antigos por compatibilidade de payload. - # Semântica atual: fração de target/alvo pulverizável. + "motivo": "roi_legada", "frac_erva_por_bico": frac_target_por_bico, - "ema_frac_por_bico": self._ema_frac_bico.copy(), + "ema_frac_por_bico": np.max(self._score_grid, axis=1).astype(np.float32), "ordem_bicos_invertida": bool(self.inverter_ordem_bicos), "faixa": { "y_top": int(y_top), @@ -396,6 +977,66 @@ class WeedDetector: return atuacao_bicos, estatisticas + def _atualizar_estado_bicos_legado( + self, + frac_target_por_bico: np.ndarray, + cfg: dict, + vel_norm: float, + ): + thr_on = float(cfg.get("min_frac_erva_por_bico_on", 0.030)) + thr_off = float(cfg.get("min_frac_erva_por_bico_off", 0.015)) + + ema_alpha = float(cfg.get("ema_frac_bico", 0.35)) + ema_alpha = min(1.0, max(0.0, ema_alpha)) + + on_frames = int(cfg.get("on_frames_required", 2)) + off_frames = int(cfg.get("off_frames_required", 2)) + + vel_gain_local = float(cfg.get("erva_thresh_vel_gain_local", 0.6)) + scale = 1.0 + vel_gain_local * self._clamp01(vel_norm) + + thr_on_eff = thr_on * scale + thr_off_eff = thr_off * scale + + # Guarda EMA legada dentro da primeira célula do grid só para compat. + self._score_grid[:, 0] = ( + (1.0 - ema_alpha) * self._score_grid[:, 0] + + ema_alpha * frac_target_por_bico + ) + + cond_on = self._score_grid[:, 0] >= thr_on_eff + cond_off = self._score_grid[:, 0] < thr_off_eff + + self._on_cnt = np.where(cond_on, self._on_cnt + 1, 0) + self._off_cnt = np.where(cond_off, self._off_cnt + 1, 0) + + estado = self._estado_bico.copy() + estado = np.where(self._on_cnt >= on_frames, True, estado) + estado = np.where(self._off_cnt >= off_frames, False, estado) + + self._estado_bico = estado.astype(bool, copy=False) + + atuacao_bicos = { + i: bool(self._estado_bico[i]) + for i in range(self.qtd_bicos) + } + + self.ultimo_status_bicos = atuacao_bicos.copy() + + estat = { + "thr_on_eff": float(thr_on_eff), + "thr_off_eff": float(thr_off_eff), + "on_cnt": self._on_cnt.copy(), + "off_cnt": self._off_cnt.copy(), + "estado_bico": self._estado_bico.copy(), + } + + return atuacao_bicos, estat + + # ============================================================ + # Retorno OFF + # ============================================================ + def _retornar_bicos_off( self, motivo: str, @@ -410,7 +1051,8 @@ class WeedDetector: return {i: False for i in range(self.qtd_bicos)}, { "motivo": motivo, "frac_erva_por_bico": zeros_f, - "ema_frac_por_bico": self._ema_frac_bico.copy(), + "ema_frac_por_bico": np.max(self._score_grid, axis=1).astype(np.float32), + "score_atuacao_por_bico": zeros_f, "faixa": { "y_top": int(y_top), "y_bot": int(y_bot), @@ -419,6 +1061,18 @@ class WeedDetector: "shift_px": int(shift), } + def _reset_bicos_off(self): + self._estado_bico[:] = False + self._on_cnt[:] = 0 + self._off_cnt[:] = 0 + self.ultimo_status_bicos = { + i: False for i in range(self.qtd_bicos) + } + + # ============================================================ + # ROI legada + # ============================================================ + def _calcular_roi_vertical( self, h: int, @@ -427,7 +1081,7 @@ class WeedDetector: vel_norm: float, cfg: dict, ): - # Convenção atual: + # Convenção legada: # zona_inicio e zona_altura são frações medidas a partir da parte inferior da imagem. y_inicio = int((1.0 - zona_inicio) * h) y_fim = int((1.0 - (zona_inicio + zona_altura)) * h) @@ -443,6 +1097,10 @@ class WeedDetector: return int(y_top), int(y_bot), int(shift) + # ============================================================ + # Filtros opcionais + # ============================================================ + def _aplicar_filtros_opcionais( self, mask_target: np.ndarray, @@ -484,69 +1142,6 @@ class WeedDetector: return mask_target - def _atualizar_estado_bicos( - self, - frac_target_por_bico: np.ndarray, - cfg: dict, - vel_norm: float, - ): - thr_on = float(cfg.get("min_frac_erva_por_bico_on", 0.030)) - thr_off = float(cfg.get("min_frac_erva_por_bico_off", 0.015)) - - ema_alpha = float(cfg.get("ema_frac_bico", 0.35)) - ema_alpha = min(1.0, max(0.0, ema_alpha)) - - on_frames = int(cfg.get("on_frames_required", 2)) - off_frames = int(cfg.get("off_frames_required", 2)) - - vel_gain_local = float(cfg.get("erva_thresh_vel_gain_local", 0.6)) - scale = 1.0 + vel_gain_local * self._clamp01(vel_norm) - - thr_on_eff = thr_on * scale - thr_off_eff = thr_off * scale - - self._ema_frac_bico = ( - (1.0 - ema_alpha) * self._ema_frac_bico - + ema_alpha * frac_target_por_bico - ) - - cond_on = self._ema_frac_bico >= thr_on_eff - cond_off = self._ema_frac_bico < thr_off_eff - - self._on_cnt = np.where(cond_on, self._on_cnt + 1, 0) - self._off_cnt = np.where(cond_off, self._off_cnt + 1, 0) - - estado = self._estado_bico.copy() - estado = np.where(self._on_cnt >= on_frames, True, estado) - estado = np.where(self._off_cnt >= off_frames, False, estado) - - self._estado_bico = estado.astype(bool, copy=False) - - atuacao_bicos = { - i: bool(self._estado_bico[i]) - for i in range(self.qtd_bicos) - } - - self.ultimo_status_bicos = atuacao_bicos.copy() - - estat = { - "thr_on_eff": float(thr_on_eff), - "thr_off_eff": float(thr_off_eff), - "on_cnt": self._on_cnt.copy(), - "off_cnt": self._off_cnt.copy(), - "estado_bico": self._estado_bico.copy(), - } - - return atuacao_bicos, estat - - def _reset_bicos_off(self): - self._estado_bico[:] = False - self._on_cnt[:] = 0 - self._off_cnt[:] = 0 - self.ultimo_status_bicos = { - i: False for i in range(self.qtd_bicos) - } - # ============================================================ # Utils # ============================================================