ajuste no imu

This commit is contained in:
Diego Freitas 2026-06-18 07:29:04 -03:00
parent 1297a898d5
commit afc91db881
2 changed files with 296 additions and 39 deletions

View File

@ -30,7 +30,7 @@ def main():
T_Code.Sen: ModuloSensoriamento(),
T_Code.Atu: ModuloAtuador(),
T_Code.Lra: ModuloLoRa(),
T_Code.Imu: CameraIMU(),
T_Code.Imu: CameraIMU(modulos_imu=[T_Code.Snr]),
T_Code.Npc: ModuloPC(),
T_Code.Lvx: ModuloLivox(),
T_Code.Ipb: ModuloIPBribge(),

View File

@ -51,6 +51,7 @@ class CameraIMU(ModuloDiagnosticoBase):
pitch_rate_critical_dps=75.0,
filtro_alpha=0.35,
historico_len=100,
modulos_imu=None,
):
self.freq = float(freq)
self.publish_hz = float(publish_hz)
@ -80,6 +81,17 @@ class CameraIMU(ModuloDiagnosticoBase):
self.filtro_alpha = float(filtro_alpha)
# Quais módulos podem participar da IMU final.
# None mantém o comportamento histórico: frontal + traseira.
# Para usar somente a frontal:
# CameraIMU(modulos_imu=[T_Code.Snr])
self.modulos_imu = self._preparar_modulos_imu(modulos_imu)
self._modulos_imu_configurados = {
self._normalizar_identificador(modulo)
for modulo in self.modulos_imu
}
self.redundancia_configurada = len(self._modulos_imu_configurados) >= 2
self.ativo = True
self.last_data = None
self.ultima_saude = None
@ -110,7 +122,11 @@ class CameraIMU(ModuloDiagnosticoBase):
)
self.thread.start()
self.mostrar_log("Task iniciada")
modulos_log = ", ".join(self._nome_codigo(m) for m in self.modulos_imu)
self.mostrar_log(
f"Task iniciada | módulos IMU considerados=[{modulos_log}] "
f"| redundância_exigida={self.redundancia_configurada}"
)
# ============================================================
# Ciclo de vida
@ -163,26 +179,124 @@ class CameraIMU(ModuloDiagnosticoBase):
def _ema(self, anterior, atual, alpha):
return (1.0 - alpha) * float(anterior) + alpha * float(atual)
def _modelo_peso_base(self, cam_data, mx_id):
def _normalizar_identificador(self, valor):
if valor is None:
return ""
# Para Enum/IntEnum, o nome é mais estável que o value.
nome = getattr(valor, "name", None)
if nome:
valor = nome
else:
valor_enum = getattr(valor, "value", None)
if valor_enum is not None:
valor = valor_enum
texto = str(valor).strip().lower()
if "." in texto:
texto = texto.rsplit(".", 1)[-1]
return "".join(ch for ch in texto if ch.isalnum())
def _aliases_codigo(self, codigo):
aliases = set()
for valor in (codigo, getattr(codigo, "name", None), getattr(codigo, "value", None)):
normalizado = self._normalizar_identificador(valor)
if normalizado:
aliases.add(normalizado)
return aliases
def _nome_codigo(self, codigo):
nome = getattr(codigo, "name", None)
if nome:
return str(nome)
valor = getattr(codigo, "value", codigo)
return str(valor)
def _preparar_modulos_imu(self, modulos_imu):
if modulos_imu is None:
modulos_imu = [T_Code.Snr, T_Code.Cam]
unicos = []
vistos = set()
for modulo in list(modulos_imu):
chave = self._normalizar_identificador(modulo)
if not chave or chave in vistos:
continue
vistos.add(chave)
unicos.append(modulo)
if not unicos:
raise ValueError(
"modulos_imu deve conter pelo menos um módulo, por exemplo [T_Code.Snr]"
)
return tuple(unicos)
def _identificar_modulo_camera(self, cam_data):
if not isinstance(cam_data, dict):
cam_data = {}
candidatos = set()
for campo in ("dispositivo", "t_code", "codigo", "tipo_modulo", "modulo"):
valor = cam_data.get(campo)
candidatos.update(self._aliases_codigo(valor))
conhecidos = (T_Code.Snr, T_Code.Cam)
for codigo in conhecidos:
if candidatos.intersection(self._aliases_codigo(codigo)):
return self._normalizar_identificador(codigo)
# Fallback por modelo para instalações antigas que não publicam dispositivo.
modelo = str(cam_data.get("modelo", "") or "").lower()
if "oak-d" in modelo or "oak d" in modelo:
return self._normalizar_identificador(T_Code.Snr)
if (
"ffc" in modelo
or "multiespectral" in modelo
or "multispectral" in modelo
):
return self._normalizar_identificador(T_Code.Cam)
return ""
def _modulo_imu_habilitado(self, codigo_normalizado):
return codigo_normalizado in self._modulos_imu_configurados
def _modelo_peso_base(self, cam_data, mx_id, codigo_normalizado=""):
modelo = str(cam_data.get("modelo", "") or "").lower()
dispositivo = str(cam_data.get("dispositivo", "") or "").lower()
# OAK-D Lite presa no chassi/cabine deve ser a principal.
if "oak-d" in modelo or "snr" in dispositivo:
codigo_snr = self._normalizar_identificador(T_Code.Snr)
codigo_cam = self._normalizar_identificador(T_Code.Cam)
# OAK-D Lite presa ao chassi/cabine é a referência principal.
if codigo_normalizado == codigo_snr or "oak-d" in modelo or "snr" in dispositivo:
return 1.0
# Multiespectral na haste entra como redundância.
if "ffc" in modelo or "multiespectral" in modelo or "multispectral" in modelo:
return 0.0
# A traseira pode participar quando estiver habilitada, mas com menor influência.
# Para removê-la completamente, retire T_Code.Cam de modulos_imu.
if (
codigo_normalizado == codigo_cam
or "ffc" in modelo
or "multiespectral" in modelo
or "multispectral" in modelo
):
return 0.35
# Fallback seguro.
# Fallback seguro para uma fonte habilitada ainda não catalogada.
return 0.50
# ============================================================
# Leitura das IMUs das câmeras
# ============================================================
def _buscar_camera_e_imu_individual(self, mx_id):
def _buscar_camera_e_imu_individual(self, mx_id, cam_detalhada=None):
"""
Busca os dados detalhados da câmera e a IMU individual dela.
@ -200,7 +314,8 @@ class CameraIMU(ModuloDiagnosticoBase):
dentro da chave detalhada da câmera.
"""
cam_detalhada = ContextoGlobalRedis.get_camera(mx_id) or {}
if cam_detalhada is None:
cam_detalhada = ContextoGlobalRedis.get_camera(mx_id) or {}
imu_ctx = None
try:
@ -238,7 +353,14 @@ class CameraIMU(ModuloDiagnosticoBase):
return cam_detalhada, imu_dados
def _extrair_imu_da_camera(self, mx_id, cam_data, imu, agora):
def _extrair_imu_da_camera(
self,
mx_id,
cam_data,
imu,
agora,
codigo_normalizado,
):
"""
Extrai uma fonte de IMU individual.
@ -295,7 +417,11 @@ class CameraIMU(ModuloDiagnosticoBase):
except Exception:
imu_saude = None
peso_base = self._modelo_peso_base(cam_data, mx_id)
peso_base = self._modelo_peso_base(
cam_data,
mx_id,
codigo_normalizado=codigo_normalizado,
)
motivos = []
confiavel = True
@ -345,6 +471,7 @@ class CameraIMU(ModuloDiagnosticoBase):
"mx_id": str(mx_id),
"modelo": cam_data.get("modelo"),
"dispositivo": cam_data.get("dispositivo"),
"t_code": codigo_normalizado,
"valido": valido,
"calibrado": calibrado,
@ -375,33 +502,50 @@ class CameraIMU(ModuloDiagnosticoBase):
def _ler_fontes_imu(self):
"""
as IMUs individuais das câmeras.
apenas as IMUs configuradas em ``modulos_imu``.
get_cameras():
usado apenas como inventário/lista de câmeras conhecidas.
get_camera(mx_id) + CamImuKey(mx_id):
usados para buscar os dados reais de cada câmera.
Câmeras fora da seleção são classificadas como desabilitadas e não
participam da fusão, da divergência, da saúde ou da exigência de
redundância.
"""
agora = time.time()
cameras_mapeadas = ContextoGlobalRedis.get_cameras() or {}
fontes = []
ignoradas = []
desabilitadas = []
modulos_detectados = set()
if not isinstance(cameras_mapeadas, dict):
return [], []
cameras_mapeadas = {}
for mx_id in list(cameras_mapeadas.keys()):
cam_detalhada, imu_dados = self._buscar_camera_e_imu_individual(mx_id)
cam_detalhada = ContextoGlobalRedis.get_camera(mx_id) or {}
codigo_normalizado = self._identificar_modulo_camera(cam_detalhada)
if not self._modulo_imu_habilitado(codigo_normalizado):
desabilitadas.append({
"mx_id": str(mx_id),
"modelo": cam_detalhada.get("modelo") if isinstance(cam_detalhada, dict) else None,
"dispositivo": cam_detalhada.get("dispositivo") if isinstance(cam_detalhada, dict) else None,
"t_code": codigo_normalizado or None,
})
continue
modulos_detectados.add(codigo_normalizado)
cam_detalhada, imu_dados = self._buscar_camera_e_imu_individual(
mx_id,
cam_detalhada=cam_detalhada,
)
fonte = self._extrair_imu_da_camera(
mx_id=mx_id,
cam_data=cam_detalhada,
imu=imu_dados,
agora=agora,
codigo_normalizado=codigo_normalizado,
)
if fonte is None:
@ -409,6 +553,7 @@ class CameraIMU(ModuloDiagnosticoBase):
"mx_id": str(mx_id),
"modelo": cam_detalhada.get("modelo") if isinstance(cam_detalhada, dict) else None,
"dispositivo": cam_detalhada.get("dispositivo") if isinstance(cam_detalhada, dict) else None,
"t_code": codigo_normalizado,
"confiavel": False,
"idade_s": 999.0,
"confidence": 0.0,
@ -427,7 +572,29 @@ class CameraIMU(ModuloDiagnosticoBase):
else:
ignoradas.append(fonte)
return fontes, ignoradas
# Também representa módulos configurados que nem apareceram no inventário.
for codigo in sorted(self._modulos_imu_configurados):
if codigo in modulos_detectados:
continue
ignoradas.append({
"mx_id": None,
"modelo": None,
"dispositivo": None,
"t_code": codigo,
"confiavel": False,
"idade_s": 999.0,
"confidence": 0.0,
"motivos": ["modulo_imu_configurado_nao_detectado"],
"roll": 0.0,
"pitch": 0.0,
"yaw": 0.0,
"roll_rate_dps": 0.0,
"pitch_rate_dps": 0.0,
"yaw_rate_dps": 0.0,
})
return fontes, ignoradas, desabilitadas
# ============================================================
# Fusão
@ -575,7 +742,15 @@ class CameraIMU(ModuloDiagnosticoBase):
return nivel, motivos
def _avaliar_risco(self, roll, pitch, roll_rate, pitch_rate, sensores_agree, n_fontes):
def _avaliar_risco(
self,
roll,
pitch,
roll_rate,
pitch_rate,
sensores_agree,
redundancia_disponivel,
):
abs_roll = abs(float(roll))
abs_pitch = abs(float(pitch))
abs_roll_rate = abs(float(roll_rate))
@ -623,7 +798,7 @@ class CameraIMU(ModuloDiagnosticoBase):
nivel = max(nivel, 2)
motivos.append("imus_divergentes")
if n_fontes <= 1:
if self.redundancia_configurada and not redundancia_disponivel:
motivos.append("sem_redundancia")
risk_score = 0.0
@ -686,7 +861,7 @@ class CameraIMU(ModuloDiagnosticoBase):
**data
)
def _montar_saida_sem_imu(self, fontes_ignoradas, agora):
def _montar_saida_sem_imu(self, fontes_ignoradas, fontes_desabilitadas, agora):
return {
"valido": False,
"modo": "sem_imu",
@ -704,15 +879,23 @@ class CameraIMU(ModuloDiagnosticoBase):
"risk_level": "emergency",
"risk_level_num": 4,
"risk_score": 100.0,
"motivos_risco": ["sem_imu_valida"],
"acao_sugerida": "emergency_stop",
"velocidade_factor": 0.0,
"bloquear_avanco": True,
"stop": True,
"emergency_stop": True,
"confidence": 0.0,
"sensors_count": 0,
"modulos_imu_configurados": [
self._nome_codigo(modulo) for modulo in self.modulos_imu
],
"redundancia_configurada": self.redundancia_configurada,
"redundancia_disponivel": False,
"sensors_expected_count": len(self._modulos_imu_configurados),
"sensors_count": len(fontes_ignoradas),
"sensors_valid_count": 0,
"sensors_ignored_count": len(fontes_ignoradas),
"sensors_disabled_count": len(fontes_desabilitadas),
"sensors_agree": False,
"roll_disagreement_deg": 0.0,
"pitch_disagreement_deg": 0.0,
@ -720,6 +903,7 @@ class CameraIMU(ModuloDiagnosticoBase):
"fonte_principal": None,
"fontes": [],
"fontes_ignoradas": fontes_ignoradas,
"fontes_desabilitadas": fontes_desabilitadas,
"timestamp": agora * 1000.0,
"last_packet_ts": agora,
"last_publish_ts": agora,
@ -728,7 +912,17 @@ class CameraIMU(ModuloDiagnosticoBase):
"frequencia": self.frequencia,
}
def _montar_saida(self, fontes, ignoradas, fusao, divergencia, risco, agora):
def _montar_saida(
self,
fontes,
ignoradas,
desabilitadas,
fusao,
divergencia,
risco,
redundancia_disponivel,
agora,
):
roll = float(fusao["roll"])
pitch = float(fusao["pitch"])
yaw = float(fusao["yaw"])
@ -769,8 +963,8 @@ class CameraIMU(ModuloDiagnosticoBase):
estabilidade_score = 100.0 - self.norm_to_100(oscilacao_total, 0.5, 8.0)
estabilidade_score = self.clamp(estabilidade_score, 0.0, 100.0)
# Penaliza estabilidade se a redundância caiu.
if len(fontes) <= 1:
# Penaliza apenas se a redundância foi solicitada e realmente caiu.
if self.redundancia_configurada and not redundancia_disponivel:
estabilidade_score = max(0.0, estabilidade_score - 10.0)
# Penaliza se sensores divergem.
@ -779,9 +973,14 @@ class CameraIMU(ModuloDiagnosticoBase):
fonte_principal = fusao.get("fonte_principal")
if self.redundancia_configurada:
modo = "redundante" if redundancia_disponivel else "redundancia_degradada"
else:
modo = "single_source_configurada"
return {
"valido": True,
"modo": "redundante" if len(fontes) >= 2 else "single_source",
"modo": modo,
# Dado rápido.
"roll": round(roll, 3),
@ -822,9 +1021,16 @@ class CameraIMU(ModuloDiagnosticoBase):
"fonte_principal": fonte_principal,
"single_source": bool(fusao.get("single_source", False)),
"modulos_imu_configurados": [
self._nome_codigo(modulo) for modulo in self.modulos_imu
],
"redundancia_configurada": self.redundancia_configurada,
"redundancia_disponivel": redundancia_disponivel,
"sensors_expected_count": len(self._modulos_imu_configurados),
"sensors_count": len(fontes) + len(ignoradas),
"sensors_valid_count": len(fontes),
"sensors_ignored_count": len(ignoradas),
"sensors_disabled_count": len(desabilitadas),
"sensors_agree": divergencia["sensors_agree"],
"roll_disagreement_deg": round(divergencia["roll_disagreement_deg"], 3),
@ -836,6 +1042,7 @@ class CameraIMU(ModuloDiagnosticoBase):
{
"mx_id": f["mx_id"],
"modelo": f.get("modelo"),
"t_code": f.get("t_code"),
"roll": round(f["roll"], 3),
"pitch": round(f["pitch"], 3),
"yaw": round(f["yaw"], 3),
@ -849,13 +1056,23 @@ class CameraIMU(ModuloDiagnosticoBase):
],
"fontes_ignoradas": [
{
"mx_id": f["mx_id"],
"mx_id": f.get("mx_id"),
"modelo": f.get("modelo"),
"t_code": f.get("t_code"),
"idade_s": round(f["idade_s"], 3),
"motivos": f.get("motivos", []),
}
for f in ignoradas
],
"fontes_desabilitadas": [
{
"mx_id": f.get("mx_id"),
"modelo": f.get("modelo"),
"dispositivo": f.get("dispositivo"),
"t_code": f.get("t_code"),
}
for f in desabilitadas
],
# Índices úteis para painel/manager.
"estabilidade_idx": round(estabilidade_score, 1),
@ -885,10 +1102,24 @@ class CameraIMU(ModuloDiagnosticoBase):
try:
agora = time.time()
fontes, ignoradas = self._ler_fontes_imu()
fontes, ignoradas, desabilitadas = self._ler_fontes_imu()
modulos_validos = {
f.get("t_code")
for f in fontes
if f.get("t_code")
}
redundancia_disponivel = bool(
self.redundancia_configurada
and self._modulos_imu_configurados.issubset(modulos_validos)
)
if len(fontes) <= 0:
data = self._montar_saida_sem_imu(ignoradas, agora)
data = self._montar_saida_sem_imu(
ignoradas,
desabilitadas,
agora,
)
else:
fusao = self._fundir_fontes(fontes)
divergencia = self._calcular_divergencia(fontes)
@ -899,15 +1130,17 @@ class CameraIMU(ModuloDiagnosticoBase):
roll_rate=fusao["roll_rate_dps"],
pitch_rate=fusao["pitch_rate_dps"],
sensores_agree=divergencia["sensors_agree"],
n_fontes=len(fontes),
redundancia_disponivel=redundancia_disponivel,
)
data = self._montar_saida(
fontes=fontes,
ignoradas=ignoradas,
desabilitadas=desabilitadas,
fusao=fusao,
divergencia=divergencia,
risco=risco,
redundancia_disponivel=redundancia_disponivel,
agora=agora,
)
@ -955,6 +1188,13 @@ class CameraIMU(ModuloDiagnosticoBase):
valido = bool(data.get("valido", False))
sensors_valid_count = int(data.get("sensors_valid_count", 0) or 0)
sensors_count = int(data.get("sensors_count", 0) or 0)
sensors_expected_count = int(
data.get("sensors_expected_count", len(self._modulos_imu_configurados)) or 0
)
redundancia_configurada = bool(
data.get("redundancia_configurada", self.redundancia_configurada)
)
redundancia_disponivel = bool(data.get("redundancia_disponivel", False))
sensors_agree = bool(data.get("sensors_agree", False))
risk_level_num = int(data.get("risk_level_num", 4) or 0)
@ -984,7 +1224,9 @@ class CameraIMU(ModuloDiagnosticoBase):
sensors_ignored_count = int(data.get("sensors_ignored_count", 0) or 0)
if sensors_count <= 0:
motivos.append("Nenhuma câmera com IMU detectada")
motivos.append(
"Nenhum dos módulos IMU configurados foi detectado"
)
elif sensors_valid_count <= 0 and sensors_ignored_count > 0:
motivos.append("Câmeras detectadas, mas nenhuma IMU válida")
else:
@ -995,11 +1237,14 @@ class CameraIMU(ModuloDiagnosticoBase):
saude = 0
motivos.append("Sem sensores IMU válidos")
elif sensors_valid_count == 1:
elif redundancia_configurada and not redundancia_disponivel:
saude -= 15
motivos.append("Operando sem redundância de IMU")
motivos.append(
"Redundância de IMU configurada indisponível "
f"({sensors_valid_count}/{sensors_expected_count})"
)
if sensors_count >= 2 and not sensors_agree:
if sensors_valid_count >= 2 and not sensors_agree:
saude -= 30
motivos.append("IMUs divergentes")
@ -1061,6 +1306,11 @@ class CameraIMU(ModuloDiagnosticoBase):
"status": status.value,
"saude": saude,
"motivos": motivos,
"modulos_imu_configurados": [
self._nome_codigo(modulo) for modulo in self.modulos_imu
],
"redundancia_configurada": redundancia_configurada,
"redundancia_disponivel": redundancia_disponivel,
"saude_individual": [],
}
@ -1117,9 +1367,16 @@ class CameraIMU(ModuloDiagnosticoBase):
"stop": True,
"emergency_stop": True,
"confidence": 0.0,
"modulos_imu_configurados": [
self._nome_codigo(modulo) for modulo in self.modulos_imu
],
"redundancia_configurada": self.redundancia_configurada,
"redundancia_disponivel": False,
"sensors_expected_count": len(self._modulos_imu_configurados),
"sensors_count": 0,
"sensors_valid_count": 0,
"sensors_ignored_count": 0,
"sensors_disabled_count": 0,
"sensors_agree": False,
"estabilidade_idx": 0.0,
"timestamp": 0.0,