scripts/classify_equipments.py

#!/usr/bin/env python3
"""Classificacao v4: media por dia ativo (ignora dias parados).

Filosofia: classificar pelo que a escavadeira faz QUANDO esta operando.
- avg_dia_ativo = total_viagens / dias_com_alguma_viagem
- max_dia = melhor dia do periodo
- Producao = avg_dia_ativo >= 80 E max_dia >= 120
- Marginal = avg_dia_ativo >= 40 E max_dia >= 60
- Infra = resto

Assim escavadeiras em corretiva longa nao sao penalizadas pela
quantidade de dias parados - o que importa e a performance quando
estao realmente produzindo.
"""
import duckdb, json
from datetime import datetime
from collections import defaultdict

DB = "/Users/gabrielsapucaia/Code/cco-apn/f2m_local.duckdb"
OUT = "/Users/gabrielsapucaia/Code/cco-apn/dashboard_apn/equipment_categories.json"

JANELA_DIAS = 21  # janela maior pra capturar dias ativos mesmo com muita manutencao
MIN_DIAS_ATIVOS = 2  # pelo menos 2 dias produziu algo no periodo
PROD_AVG_ATIVO = 80
PROD_MAX_DIA = 120
MARG_AVG_ATIVO = 40
MARG_MAX_DIA = 60

c = duckdb.connect(DB, read_only=True)

tables = [r[0] for r in c.execute("SHOW TABLES").fetchall()]
pivots = sorted([t for t in tables if t.startswith("pivot_transporte_") and "20" not in t])[-3:]
parts = (["current_transporte"] if "current_transporte" in tables else []) + pivots
union = " UNION ALL ".join(f'SELECT * FROM "{t}"' for t in parts)

rows = c.execute(f"""
    SELECT TRIM(carga) as eq,
           CAST(production_date AS VARCHAR)[:10] as dia,
           COUNT(*) n
    FROM ({union})
    WHERE carga IS NOT NULL
      AND production_date >= CAST(NOW() - INTERVAL {JANELA_DIAS} DAY AS VARCHAR)[:10]
    GROUP BY 1, 2
""").fetchall()

by_eq = defaultdict(lambda: {"dias": 0, "total": 0, "max_dia": 0, "dias_50": 0, "dias_100": 0})
for eq, dia, n in rows:
    if not eq or eq == "-":
        continue
    by_eq[eq]["dias"] += 1
    by_eq[eq]["total"] += n
    if n > by_eq[eq]["max_dia"]:
        by_eq[eq]["max_dia"] = n
    if n >= 50:
        by_eq[eq]["dias_50"] += 1
    if n >= 100:
        by_eq[eq]["dias_100"] += 1

all_tipos = dict(c.execute("""
    SELECT DISTINCT equipamento, tipo_equipamento FROM current_registro_estados
    WHERE equipamento IS NOT NULL
""").fetchall())

candidatos = set(by_eq.keys())
for eq, tipo in all_tipos.items():
    if tipo in ("Escavadeira", "Pá Carregadeira"):
        candidatos.add(eq)

c.close()

producao = set()
producao_marginal = set()
infra = set()

for eq in candidatos:
    s = by_eq.get(eq, {"dias": 0, "total": 0, "max_dia": 0, "dias_50": 0, "dias_100": 0})
    dias = s["dias"]
    total = s["total"]
    avg_ativo = total / dias if dias else 0
    max_dia = s["max_dia"]

    if dias >= MIN_DIAS_ATIVOS and avg_ativo >= PROD_AVG_ATIVO and max_dia >= PROD_MAX_DIA:
        producao.add(eq)
    elif dias >= MIN_DIAS_ATIVOS and avg_ativo >= MARG_AVG_ATIVO and max_dia >= MARG_MAX_DIA:
        producao_marginal.add(eq)
    else:
        infra.add(eq)

# Ordena detalhes por avg_ativo desc
detalhes = []
for eq in candidatos:
    s = by_eq.get(eq, {"dias": 0, "total": 0, "max_dia": 0, "dias_50": 0, "dias_100": 0})
    dias = s["dias"]
    total = s["total"]
    avg_ativo = round(total / dias, 1) if dias else 0
    detalhes.append({
        "equip": eq,
        "tipo": all_tipos.get(eq, "?"),
        "dias_ativos": dias,
        "total": total,
        "avg_dia_ativo": avg_ativo,
        "max_dia": s["max_dia"],
        "dias_com_50plus": s["dias_50"],
        "dias_com_100plus": s["dias_100"],
        "categoria": "producao" if eq in producao else ("producao_marginal" if eq in producao_marginal else "infra"),
    })
detalhes.sort(key=lambda x: (-x["avg_dia_ativo"], -x["total"]))

result = {
    "_description": (
        f"Classificacao v4 (janela {JANELA_DIAS}d, media por dia ativo): "
        f"producao = avg_dia_ativo >= {PROD_AVG_ATIVO} E max_dia >= {PROD_MAX_DIA}, "
        f"marginal >= {MARG_AVG_ATIVO}/{MARG_MAX_DIA}. "
        "Ignora dias parados - mede performance quando operando."
    ),
    "_thresholds": {
        "janela_dias": JANELA_DIAS,
        "min_dias_ativos": MIN_DIAS_ATIVOS,
        "producao_avg_dia_ativo": PROD_AVG_ATIVO,
        "producao_max_dia": PROD_MAX_DIA,
        "marginal_avg_dia_ativo": MARG_AVG_ATIVO,
        "marginal_max_dia": MARG_MAX_DIA,
    },
    "_generated_at": datetime.now().isoformat(),
    "producao": {
        "escavadeiras": sorted(e for e in producao if all_tipos.get(e) == "Escavadeira"),
        "carregadeiras": sorted(e for e in producao if all_tipos.get(e) == "Pá Carregadeira"),
        "outros_producao": sorted(e for e in producao if all_tipos.get(e) not in ("Escavadeira", "Pá Carregadeira")),
    },
    "producao_marginal": {
        "escavadeiras": sorted(e for e in producao_marginal if all_tipos.get(e) == "Escavadeira"),
        "carregadeiras": sorted(e for e in producao_marginal if all_tipos.get(e) == "Pá Carregadeira"),
    },
    "infra": {
        "escavadeiras": sorted(e for e in infra if all_tipos.get(e) == "Escavadeira"),
        "carregadeiras": sorted(e for e in infra if all_tipos.get(e) == "Pá Carregadeira"),
    },
    "_detalhes": detalhes[:25],
}

with open(OUT, "w") as f:
    json.dump(result, f, indent=2, ensure_ascii=False)

print(f"=== Classificacao v4 (media por dia ATIVO, janela {JANELA_DIAS}d) ===\n")
print(f"PRODUCAO ({len(result['producao']['escavadeiras'])}): {result['producao']['escavadeiras']}")
print(f"MARGINAL ({len(result['producao_marginal']['escavadeiras'])}): {result['producao_marginal']['escavadeiras']}")
print(f"INFRA ({len(result['infra']['escavadeiras'])}): {result['infra']['escavadeiras']}")
print()
print("DETALHE (so escavadeiras):")
print(f"  {'equip':<10} {'tipo':<16} {'dias_at':<8} {'total':<7} {'avg_at':<8} {'max':<5} {'50+':<4} {'100+':<5} categoria")
for d in detalhes:
    if d["tipo"] == "Escavadeira" or d["total"] > 50:
        print(f"  {d['equip']:<10} {d['tipo'][:14]:<16} {d['dias_ativos']:<8} {d['total']:<7} {d['avg_dia_ativo']:<8} {d['max_dia']:<5} {d['dias_com_50plus']:<4} {d['dias_com_100plus']:<5} {d['categoria']}")