Zendion/agrobot_base
Zendion
/
agrobot_base
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Workers atualizados

This commit is contained in:
Diego Freitas 2025-12-18 15:24:13 -03:00
parent b5ba765526
commit ff4e6116d6
4 changed files with 334 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

367
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/atuador.py
View File

@ -31,23 +31,28 @@ class ModuloAtuador(ModuloDiagnosticoBase):
SAUDE_MIN_ALERTA = 80 SAUDE_MIN_ALERTA = 80
sensor_massa_ativado = False sensor_massa_ativado = False
for tipo, sensores in dados_atu.get("sensores", {}).items(): sensores_ativos = dados_atu.get("sensores", {}).get("ativos", [])
for sensor_id, dados_sensor in sensores.items(): for key in sensores_ativos:
saude = 100 tipo, sensor_id = key.split(".")
motivos = [] sensores_tipo = dados_atu.get("sensores", {}).get(tipo, {})
condicoes_operacionais = [] dados_sensor = sensores_tipo.get(sensor_id)
aferir = dados_sensor.get("aferir", False) if not dados_sensor:
padroes = self.padroes_sensores.get(tipo, {}).get(sensor_id, {}) continue
conectado = dados_sensor.get("conectado", False) #for tipo, sensores in dados_atu.get("sensores", {}).items():
mandatorio = dados_sensor.get("mandatorio", False) # for sensor_id, dados_sensor in sensores.items():
saude = 100
if dados_sensor.get("label") == "MASRS" and conectado and mandatorio and aferir: sensor_massa_ativado = True motivos = []
condicoes_operacionais = []
if not conectado: aferir = dados_sensor.get("aferir", False)
saude = 0 padroes = self.padroes_sensores.get(tipo, {}).get(sensor_id, {})
motivos.append(f"desconectado") conectado = dados_sensor.get("conectado", False)
else: mandatorio = dados_sensor.get("mandatorio", False)
if dados_sensor.get("label") == "MASRS" and conectado and mandatorio and aferir: sensor_massa_ativado = True
if not conectado:
saude = 0
motivos.append(f"desconectado")
else:
for campo in dados_sensor.get("dados", []): for campo in dados_sensor.get("dados", []):
label = campo.get("label") label = campo.get("label")
if label is None: if label is None:
@ -73,135 +78,209 @@ class ModuloAtuador(ModuloDiagnosticoBase):
"severidade": grau, "severidade": grau,
"descricao": f"{label} fora do padrão" "descricao": f"{label} fora do padrão"
}) })
saude = max(0, saude) saude = max(0, saude)
chave = f"{S_Code(int(tipo)).name}_{sensor_id}" chave = f"{S_Code(int(tipo)).name}_{sensor_id}"
saude_mod = { saude_mod = {
"id": sensor_id, "id": sensor_id,
"label": dados_sensor.get("label"), "label": dados_sensor.get("label"),
"status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value, "status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value,
"saude": saude, "saude": saude,
"motivos": motivos, "motivos": motivos,
"condicoes_operacionais": condicoes_operacionais, "condicoes_operacionais": condicoes_operacionais,
"em_uso": aferir "em_uso": aferir
} }
saude_individual.append(saude_mod) saude_individual.append(saude_mod)
if mandatorio: if mandatorio:
total_ativos += 1 total_ativos += 1
saude_total += saude saude_total += saude
motivos_gerais.extend([f"{chave}: {m}" for m in motivos]) motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_sensor["saude"] = saude_mod dados_sensor["saude"] = saude_mod
bicos_ativos = dados_atu.get("bicos", {}).get("ativos", [])
for key in bicos_ativos:
tipo, bico_id = key.split(".")
bicos_tipo = dados_atu.get("bicos", {}).get(tipo, {})
dados_bico = bicos_tipo.get(bico_id)
if not dados_bico:
continue
for tipo, bicos in dados_atu.get("bicos", {}).items(): #for tipo, bicos in dados_atu.get("bicos", {}).items():
for bico_id, dados_bico in bicos.items(): # for bico_id, dados_bico in bicos.items():
saude = 100 saude = 100
motivos = [] motivos = []
condicoes_operacionais = [] condicoes_operacionais = []
comandar = dados_bico.get("comandar", False) comandar = dados_bico.get("comandar", False)
padroes = self.padroes_bicos.get(tipo, {}).get(bico_id, {}) padroes = self.padroes_bicos.get(tipo, {}).get(bico_id, {})
conectado = dados_bico.get("conectado", False) conectado = dados_bico.get("conectado", False)
mandatorio = dados_bico.get("mandatorio", False) mandatorio = dados_bico.get("mandatorio", False)
if not conectado: if not conectado:
saude = 0 saude = 0
motivos.append(f"desconectado") motivos.append(f"desconectado")
else: else:
for campo in dados_bico.get("dados", []): for campo in dados_bico.get("dados", []):
label = campo.get("label") label = campo.get("label")
if label is None: if label is None:
continue continue
freq = campo.get("freq", 0.0) freq = campo.get("freq", 0.0)
if freq == 0: if freq == 0:
saude = 0 saude = 0
motivos.append(f"{label} sem resposta (freq = 0 Hz)") motivos.append(f"{label} sem resposta (freq = 0 Hz)")
else: else:
penalidade = penalizar_frequencia(freq, FREQ_BASE, FREQ_MIN, (100 - SAUDE_MIN_ALERTA)) penalidade = penalizar_frequencia(freq, FREQ_BASE, FREQ_MIN, (100 - SAUDE_MIN_ALERTA))
saude -= penalidade saude -= penalidade
if (freq <= FREQ_MIN): if (freq <= FREQ_MIN):
motivos.append(f"{label} frequência baixa ({freq:.2f}/{FREQ_BASE} Hz, penalidade {penalidade}%)") motivos.append(f"{label} frequência baixa ({freq:.2f}/{FREQ_BASE} Hz, penalidade {penalidade}%)")
atual = campo.get("atual")
padrao = padroes.get(label) if atual else None
if padrao is not None:
# Avaliação de condição operacional (sem penalizar saúde)
if not padrao.esta_dentro_do_padrao(atual):
grau = padrao.calcular_quao_anormal(atual)
condicoes_operacionais.append({
"valor": atual,
"severidade": grau,
"descricao": f"{label} fora do padrão"
})
saude = max(0, saude)
chave = f"{S_Code(int(tipo)).name}_{bico_id}"
saude_mod = {
"id": bico_id,
"label": dados_bico.get("label"),
"status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value,
"saude": saude,
"motivos": motivos,
"condicoes_operacionais": condicoes_operacionais,
"em_uso": comandar
}
saude_individual.append(saude_mod)
if mandatorio:
total_ativos += 1
saude_total += saude
motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_bico["saude"] = saude_mod
bombas_ativos = dados_atu.get("bombas", {}).get("ativos", [])
for key in bombas_ativos:
tipo, bomba_id = key.split(".")
bombas_tipo = dados_atu.get("bombas", {}).get(tipo, {})
dados_bomba = bombas_tipo.get(bomba_id)
if not dados_bomba:
continue
#for tipo, bombas in dados_atu.get("bombas", {}).items():
# for bomba_id, dados_bomba in bombas.items():
saude = 100
motivos = []
condicoes_operacionais = []
comandar = dados_bomba.get("comandar", False)
padroes = self.padroes_bombas.get(tipo, {}).get(bomba_id, {})
conectado = dados_bomba.get("conectado", False)
mandatorio = dados_bomba.get("mandatorio", False)
if not conectado:
saude = 0
motivos.append(f"desconectado")
else:
for campo in dados_bomba.get("dados", []):
label = campo.get("label")
if label is None:
continue
freq = campo.get("freq", 0.0)
if freq == 0:
saude = 0
motivos.append(f"{label} sem resposta (freq = 0 Hz)")
else:
penalidade = penalizar_frequencia(freq, FREQ_BASE, FREQ_MIN, (100 - SAUDE_MIN_ALERTA))
saude -= penalidade
if (freq <= FREQ_MIN):
motivos.append(f"{label} frequência baixa ({freq:.2f}/{FREQ_BASE} Hz, penalidade {penalidade}%)")
atual = campo.get("atual")
padrao = padroes.get(label) if atual else None
if padrao is not None:
# Avaliação de condição operacional (sem penalizar saúde)
if not padrao.esta_dentro_do_padrao(atual):
grau = padrao.calcular_quao_anormal(atual)
condicoes_operacionais.append({
"valor": atual,
"severidade": grau,
"descricao": f"{label} fora do padrão"
})
saude = max(0, saude)
chave = f"{S_Code(int(tipo)).name}_{bomba_id}"
saude_mod = {
"id": bomba_id,
"label": dados_bomba.get("label"),
"status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value,
"saude": saude,
"motivos": motivos,
"condicoes_operacionais": condicoes_operacionais,
"em_uso": comandar
}
saude_individual.append(saude_mod)
if mandatorio:
total_ativos += 1
saude_total += saude
motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_bomba["saude"] = saude_mod
servos_ativos = dados_atu.get("servos", {}).get("ativos", [])
for key in servos_ativos:
tipo, servo_id = key.split(".")
servos_tipo = dados_atu.get("servos", {}).get(tipo, {})
dados_servo = servos_tipo.get(servo_id)
if not dados_servo:
continue
#for servo_id, dados_servo in dados_atu.get("servos", {}).items():
saude = 100
motivos = []
condicoes_operacionais = []
controlar = dados_servo.get("controlar", False)
conectado = dados_servo.get("conectado", False)
mandatorio = dados_servo.get("mandatorio", False)
if not conectado:
saude = 0
motivos.append(f"desconectado")
else:
for dado in dados_servo.get("dados", {}):
label = dado.get("label")
if label is None:
continue
freq = dado.get("freq", 0.0)
if freq == 0:
saude = 0
motivos.append(f"{label} sem resposta (freq = 0 Hz)")
else:
penalidade = penalizar_frequencia(freq, FREQ_BASE, FREQ_MIN, (100 - SAUDE_MIN_ALERTA))
saude -= penalidade
if freq <= FREQ_MIN:
motivos.append(f"{label} frequência baixa ({freq:.2f}/{FREQ_BASE} Hz, penalidade {penalidade}%)")
atual = campo.get("atual") if label == "ServoAnguloLeitura":
padrao = padroes.get(label) if atual else None angulo = dado.get("atual", 0.0)
if padrao is not None: angulo_sp = dados_servo.get("angulo_sp", 0.0)
# Avaliação de condição operacional (sem penalizar saúde) erro = abs(angulo - angulo_sp)
if not padrao.esta_dentro_do_padrao(atual): if erro > 0:
grau = padrao.calcular_quao_anormal(atual) condicoes_operacionais.append({
condicoes_operacionais.append({ "valor": angulo,
"valor": atual, "severidade": erro,
"severidade": grau, "descricao": f"{label} fora do padrão"
"descricao": f"{label} fora do padrão" })
}) saude = max(0, saude)
saude = max(0, saude) chave = f"servo_{servo_id}"
chave = f"{S_Code(int(tipo)).name}_{bico_id}" saude_mod = {
saude_mod = { "id": servo_id,
"id": bico_id, "label": dados_servo.get("label"),
"label": dados_bico.get("label"), "status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value,
"status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value, "saude": saude,
"saude": saude, "motivos": motivos,
"motivos": motivos, "condicoes_operacionais": condicoes_operacionais,
"condicoes_operacionais": condicoes_operacionais, "em_uso": controlar
"em_uso": comandar }
} saude_individual.append(saude_mod)
saude_individual.append(saude_mod) if mandatorio:
if mandatorio: total_ativos += 1
total_ativos += 1 saude_total += saude
saude_total += saude motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
motivos_gerais.extend([f"{chave}: {m}" for m in motivos]) dados_servo["saude"] = saude_mod
dados_bico["saude"] = saude_mod
for tipo, bombas in dados_atu.get("bombas", {}).items():
for bomba_id, dados_bomba in bombas.items():
saude = 100
motivos = []
condicoes_operacionais = []
comandar = dados_bomba.get("comandar", False)
padroes = self.padroes_bombas.get(tipo, {}).get(bomba_id, {})
conectado = dados_bomba.get("conectado", False)
mandatorio = dados_bomba.get("mandatorio", False)
if not conectado:
saude = 0
motivos.append(f"desconectado")
else:
for campo in dados_bomba.get("dados", []):
label = campo.get("label")
if label is None:
continue
freq = campo.get("freq", 0.0)
if freq == 0:
saude = 0
motivos.append(f"{label} sem resposta (freq = 0 Hz)")
else:
penalidade = penalizar_frequencia(freq, FREQ_BASE, FREQ_MIN, (100 - SAUDE_MIN_ALERTA))
saude -= penalidade
if (freq <= FREQ_MIN):
motivos.append(f"{label} frequência baixa ({freq:.2f}/{FREQ_BASE} Hz, penalidade {penalidade}%)")
atual = campo.get("atual")
padrao = padroes.get(label) if atual else None
if padrao is not None:
# Avaliação de condição operacional (sem penalizar saúde)
if not padrao.esta_dentro_do_padrao(atual):
grau = padrao.calcular_quao_anormal(atual)
condicoes_operacionais.append({
"valor": atual,
"severidade": grau,
"descricao": f"{label} fora do padrão"
})
saude = max(0, saude)
chave = f"{S_Code(int(tipo)).name}_{bomba_id}"
saude_mod = {
"id": bomba_id,
"label": dados_bomba.get("label"),
"status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value,
"saude": saude,
"motivos": motivos,
"condicoes_operacionais": condicoes_operacionais,
"em_uso": comandar
}
saude_individual.append(saude_mod)
if mandatorio:
total_ativos += 1
saude_total += saude
motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_bomba["saude"] = saude_mod
cond_op_mod = [] cond_op_mod = []
penalidade_percentual_herbicida = 0 penalidade_percentual_herbicida = 0

42
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/direcional.py
View File

@ -15,20 +15,24 @@ class ModuloDirecional(ModuloDiagnosticoBase):
total_drivers_em_uso = 0 total_drivers_em_uso = 0
motivos_dos_ativos = [] motivos_dos_ativos = []
saude_individual = [] saude_individual = []
drivers_respondendo = []
SAUDE_MIN_ALERTA = 80 SAUDE_MIN_ALERTA = 80
ENDERECOS_DRIVERS = {1, 2, 3, 4}
mod_conectado = False mod_conectado = False
dados_dir = ContextoGlobalRedis.get_modulo(self.t_code) dados_dir = ContextoGlobalRedis.get_modulo(self.t_code)
for endereco_str, modulo in dados_dir.items(): ativos = dados_dir.get("ativos", [])
try: for endereco in ativos:
endereco = int(endereco_str) endereco = str(endereco) # Redis costuma serializar chave como string
except ValueError: modulo = dados_dir.get(endereco)
continue if not modulo:
drivers_respondendo.append(endereco) continue # ativo listado, mas sem dados (race condition)
#for endereco_str, modulo in dados_dir.items():
# try:
# endereco = int(endereco_str)
# except ValueError:
# continue
mod_id = modulo.get("mod_id") mod_id = modulo.get("mod_id")
#print(f"atualizando saude do modulo {mod_id} com endereco {endereco}") #print(f"atualizando saude do modulo {mod_id} com endereco {endereco}")
FREQ_BASE = modulo.get("freq_base", 1) FREQ_BASE = modulo.get("freq_base", 1)
@ -96,26 +100,6 @@ class ModuloDirecional(ModuloDiagnosticoBase):
if conectado: if conectado:
mod_conectado = True mod_conectado = True
# Verifica se algum driver esperado está ausente
drivers_esperados = ENDERECOS_DRIVERS
drivers_ausentes = drivers_esperados - set(drivers_respondendo)
for endereco in drivers_ausentes:
saude_individual.append({
"id": endereco,
"label": "",
"status": StatusModulo.DESCONECTADO.value,
"saude": 0,
"motivos": ["Nenhuma resposta recebida"],
"condicoes_operacionais": [{
"label": "",
"valor": 0,
"severidade": 100,
"descricao": f"Sem comunicacao"
}]
})
motivos_dos_ativos.append(f"Driver {endereco}: ausente (sem nenhuma mensagem recebida)")
saude_individual.sort(key=lambda x: x["id"]) saude_individual.sort(key=lambda x: x["id"])
saude_final = (saude_total // total_drivers_em_uso) if total_drivers_em_uso > 0 else 0 saude_final = (saude_total // total_drivers_em_uso) if total_drivers_em_uso > 0 else 0
@ -123,7 +107,7 @@ class ModuloDirecional(ModuloDiagnosticoBase):
status = StatusModulo.OPERANTE status = StatusModulo.OPERANTE
if not mod_conectado: if not mod_conectado:
status = StatusModulo.DESCONECTADO status = StatusModulo.DESCONECTADO
elif saude_final <= 0 or len(drivers_ausentes) > 0: elif saude_final <= 0:
status = StatusModulo.FALHA status = StatusModulo.FALHA
elif saude_final < SAUDE_MIN_ALERTA or any(saude["saude"] < SAUDE_MIN_ALERTA for saude in saude_individual): elif saude_final < SAUDE_MIN_ALERTA or any(saude["saude"] < SAUDE_MIN_ALERTA for saude in saude_individual):
status = StatusModulo.ALERTA status = StatusModulo.ALERTA

41
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/movimentacao.py
View File

@ -26,21 +26,24 @@ class ModuloMovimentacao(ModuloDiagnosticoBase):
total_drivers_em_uso = 0 total_drivers_em_uso = 0
motivos_dos_ativos = [] motivos_dos_ativos = []
saude_individual = [] saude_individual = []
drivers_respondendo = []
SAUDE_MIN_ALERTA = 80 SAUDE_MIN_ALERTA = 80
ENDERECOS_DRIVERS = {5, 6, 7, 8}
LIMITE_ERRO_RPM = 30 LIMITE_ERRO_RPM = 30
mod_conectado = False mod_conectado = False
dados_mov = ContextoGlobalRedis.get_modulo(self.t_code) dados_mov = ContextoGlobalRedis.get_modulo(self.t_code)
for endereco_str, modulo in dados_mov.items(): ativos = dados_mov.get("ativos", [])
try: for endereco in ativos:
endereco = int(endereco_str) endereco = str(endereco) # Redis costuma serializar chave como string
except ValueError: modulo = dados_mov.get(endereco)
continue if not modulo:
drivers_respondendo.append(endereco) continue # ativo listado, mas sem dados (race condition)
#for endereco_str, modulo in dados_mov.items():
# try:
# endereco = int(endereco_str)
# except ValueError:
# continue
mod_id = modulo.get("mod_id") mod_id = modulo.get("mod_id")
FREQ_BASE = modulo.get("freq_base", 1) FREQ_BASE = modulo.get("freq_base", 1)
FREQ_MIN = FREQ_BASE * 0.5 FREQ_MIN = FREQ_BASE * 0.5
@ -184,26 +187,6 @@ class ModuloMovimentacao(ModuloDiagnosticoBase):
if conectado: if conectado:
mod_conectado = True mod_conectado = True
# Verificação de ausência de drivers esperados
drivers_esperados = ENDERECOS_DRIVERS
drivers_ausentes = drivers_esperados - set(drivers_respondendo)
for endereco in drivers_ausentes:
saude_individual.append({
"id": endereco,
"label": "",
"status": StatusModulo.DESCONECTADO.value,
"saude": 0,
"motivos": ["Nenhuma resposta recebida"],
"condicoes_operacionais": [{
"label": "",
"valor": 0,
"severidade": 100,
"descricao": f"Sem comunicacao"
}]
})
motivos_dos_ativos.append(f"Driver {endereco}: ausente (sem nenhuma mensagem recebida)")
saude_individual.sort(key=lambda x: x["id"]) saude_individual.sort(key=lambda x: x["id"])
saude_final = (saude_total // total_drivers_em_uso) if total_drivers_em_uso > 0 else 0 saude_final = (saude_total // total_drivers_em_uso) if total_drivers_em_uso > 0 else 0
@ -211,7 +194,7 @@ class ModuloMovimentacao(ModuloDiagnosticoBase):
status = StatusModulo.OPERANTE status = StatusModulo.OPERANTE
if not mod_conectado: if not mod_conectado:
status = StatusModulo.DESCONECTADO status = StatusModulo.DESCONECTADO
elif saude_final <= 0 or len(drivers_ausentes) > 0: elif saude_final <= 0:
status = StatusModulo.FALHA status = StatusModulo.FALHA
elif saude_final < SAUDE_MIN_ALERTA or any(saude["saude"] < SAUDE_MIN_ALERTA for saude in saude_individual): elif saude_final < SAUDE_MIN_ALERTA or any(saude["saude"] < SAUDE_MIN_ALERTA for saude in saude_individual):
status = StatusModulo.ALERTA status = StatusModulo.ALERTA

147
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/health_worker/modulos/sensoriamento.py
View File

@ -31,66 +31,77 @@ class ModuloSensoriamento(ModuloDiagnosticoBase):
sensor_bateria_ativado = False sensor_bateria_ativado = False
for tipo, sensores in dados_sen.get("sensores", {}).items(): sensores_ativos = dados_sen.get("sensores", {}).get("ativos", [])
for sensor_id, dados_sensor in sensores.items(): for key in sensores_ativos:
saude = 100 tipo, sensor_id = key.split(".")
motivos = [] sensores_tipo = dados_sen.get("sensores", {}).get(tipo, {})
condicoes_operacionais = [] dados_sensor = sensores_tipo.get(sensor_id)
aferir = dados_sensor.get("aferir", False) if not dados_sensor:
padroes = self.padroes_sensores.get(tipo, {}).get(sensor_id, {}) continue
conectado = dados_sensor.get("conectado", False) #for tipo, sensores in dados_sen.get("sensores", {}).items():
mandatorio = dados_sensor.get("mandatorio", False) # for sensor_id, dados_sensor in sensores.items():
saude = 100
motivos = []
condicoes_operacionais = []
aferir = dados_sensor.get("aferir", False)
padroes = self.padroes_sensores.get(tipo, {}).get(sensor_id, {})
conectado = dados_sensor.get("conectado", False)
mandatorio = dados_sensor.get("mandatorio", False)
if dados_sensor.get("label") == "AB36V" and conectado and mandatorio and aferir: sensor_bateria_ativado = True
if not conectado:
saude = 0
motivos.append(f"desconectado")
else:
for campo in dados_sensor.get("dados", []):
label = campo.get("label")
if label is None:
continue
freq = campo.get("freq", 0.0)
if freq == 0:
saude = 0
motivos.append(f"{label} sem resposta (freq = 0 Hz)")
else:
penalidade = penalizar_frequencia(freq, FREQ_BASE, FREQ_MIN, (100 - SAUDE_MIN_ALERTA))
saude -= penalidade
if (freq <= FREQ_MIN):
motivos.append(f"{label} frequência baixa ({freq:.2f}/{FREQ_BASE} Hz, penalidade {penalidade}%)")
atual = campo.get("atual")
padrao = padroes.get(label) if atual else None
if padrao is not None:
# Avaliação de condição operacional (sem penalizar saúde)
if not padrao.esta_dentro_do_padrao(atual):
grau = padrao.calcular_quao_anormal(atual)
condicoes_operacionais.append({
"valor": atual,
"severidade": grau,
"descricao": f"{label} fora do padrão"
})
saude = max(0, saude)
chave = f"{S_Code(int(tipo)).name}_{sensor_id}"
saude_mod = {
"id": sensor_id,
"label": dados_sensor.get("label"),
"status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value,
"saude": saude,
"motivos": motivos,
"condicoes_operacionais": condicoes_operacionais,
"em_uso": aferir
}
saude_individual.append(saude_mod)
if mandatorio:
total_ativos += 1
saude_total += saude
motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_sensor["saude"] = saude_mod
if dados_sensor.get("label") == "AB36V" and conectado and mandatorio and aferir: sensor_bateria_ativado = True servos_ativos = dados_sen.get("servos", {}).get("ativos", [])
for key in servos_ativos:
if not conectado: tipo, servo_id = key.split(".")
saude = 0 servos_tipo = dados_sen.get("servos", {}).get(tipo, {})
motivos.append(f"desconectado") dados_servo = servos_tipo.get(servo_id)
else: if not dados_servo:
for campo in dados_sensor.get("dados", []): continue
label = campo.get("label") #for servo_id, dados_servo in dados_sen.get("servos", {}).items():
if label is None:
continue
freq = campo.get("freq", 0.0)
if freq == 0:
saude = 0
motivos.append(f"{label} sem resposta (freq = 0 Hz)")
else:
penalidade = penalizar_frequencia(freq, FREQ_BASE, FREQ_MIN, (100 - SAUDE_MIN_ALERTA))
saude -= penalidade
if (freq <= FREQ_MIN):
motivos.append(f"{label} frequência baixa ({freq:.2f}/{FREQ_BASE} Hz, penalidade {penalidade}%)")
atual = campo.get("atual")
padrao = padroes.get(label) if atual else None
if padrao is not None:
# Avaliação de condição operacional (sem penalizar saúde)
if not padrao.esta_dentro_do_padrao(atual):
grau = padrao.calcular_quao_anormal(atual)
condicoes_operacionais.append({
"valor": atual,
"severidade": grau,
"descricao": f"{label} fora do padrão"
})
saude = max(0, saude)
chave = f"{S_Code(int(tipo)).name}_{sensor_id}"
saude_mod = {
"id": sensor_id,
"label": dados_sensor.get("label"),
"status": (StatusModulo.DESCONECTADO if not conectado else StatusModulo.OPERANTE if saude >= SAUDE_MIN_ALERTA else StatusModulo.ALERTA if saude > 0 else StatusModulo.FALHA).value,
"saude": saude,
"motivos": motivos,
"condicoes_operacionais": condicoes_operacionais,
"em_uso": aferir
}
saude_individual.append(saude_mod)
if mandatorio:
total_ativos += 1
saude_total += saude
motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_sensor["saude"] = saude_mod
for servo_id, dados_servo in dados_sen.get("servos", {}).items():
saude = 100 saude = 100
motivos = [] motivos = []
condicoes_operacionais = [] condicoes_operacionais = []
@ -143,7 +154,14 @@ class ModuloSensoriamento(ModuloDiagnosticoBase):
motivos_gerais.extend([f"{chave}: {m}" for m in motivos]) motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_servo["saude"] = saude_mod dados_servo["saude"] = saude_mod
for rele_id, dados_rele in dados_sen.get("reles", {}).items(): reles_ativos = dados_sen.get("reles", {}).get("ativos", [])
for key in reles_ativos:
tipo, rele_id = key.split(".")
reles_tipo = dados_sen.get("reles", {}).get(tipo, {})
dados_rele = reles_tipo.get(rele_id)
if not dados_rele:
continue
#for rele_id, dados_rele in dados_sen.get("reles", {}).items():
saude = 100 saude = 100
motivos = [] motivos = []
condicoes_operacionais = [] condicoes_operacionais = []
@ -186,7 +204,14 @@ class ModuloSensoriamento(ModuloDiagnosticoBase):
motivos_gerais.extend([f"{chave}: {m}" for m in motivos]) motivos_gerais.extend([f"{chave}: {m}" for m in motivos])
dados_rele["saude"] = saude_mod dados_rele["saude"] = saude_mod
for led_id, dados_led in dados_sen.get("sinaleiros", {}).items(): leds_ativos = dados_sen.get("sinaleiros", {}).get("ativos", [])
for key in leds_ativos:
tipo, led_id = key.split(".")
leds_tipo = dados_sen.get("sinaleiros", {}).get(tipo, {})
dados_led = leds_tipo.get(led_id)
if not dados_led:
continue
#for led_id, dados_led in dados_sen.get("sinaleiros", {}).items():
saude = 100 saude = 100
motivos = [] motivos = []
condicoes_operacionais = [] condicoes_operacionais = []