scripts/generate_synthetic.py

#!/usr/bin/env python3
"""
scripts/generate_synthetic.py
Gerador de dataset sintetico para fine-tuning do Agente CBMGO (CERCON)
Gera exemplos no formato SFT (Supervised Fine-Tuning) para Llama-3 / Mistral
"""

import json
import random
import argparse
from pathlib import Path
from datetime import datetime

OCUPACOES = [
    {"grupo": "A-1", "desc": "Residencial unifamiliar", "ex": "casa, sobrado, chale"},
    {"grupo": "A-2", "desc": "Residencial multifamiliar", "ex": "apartamento, flat, kitnet"},
    {"grupo": "B-1", "desc": "Servico de hospedagem", "ex": "hotel, motel, pousada"},
    {"grupo": "B-2", "desc": "Hospedagem longa permanencia", "ex": "apart-hotel, residencial senior"},
    {"grupo": "C-1", "desc": "Comercio pequeno porte", "ex": "loja, boutique, sapataria"},
    {"grupo": "C-2", "desc": "Comercio medio e grande porte", "ex": "supermercado, shopping"},
    {"grupo": "D-1", "desc": "Reuniao publica - esporte e lazer", "ex": "ginasio, estadio, piscina"},
    {"grupo": "D-2", "desc": "Reuniao publica - cultura e religiao", "ex": "teatro, cinema, igreja"},
    {"grupo": "D-3", "desc": "Reuniao publica - ensino", "ex": "escola, universidade, creche"},
    {"grupo": "E-1", "desc": "Saude e assistencia social", "ex": "hospital, clinica, UBS"},
    {"grupo": "F-1", "desc": "Servico profissional e negocios", "ex": "escritorio, banco, consultorio"},
    {"grupo": "F-2", "desc": "Servico de transporte", "ex": "aeroporto, rodoviaria, correios"},
    {"grupo": "G-1", "desc": "Servicos automotivos", "ex": "garagem, estacionamento, posto"},
    {"grupo": "H-1", "desc": "Industrial baixo risco", "ex": "industria alimenticia, textil"},
    {"grupo": "H-2", "desc": "Industrial medio risco", "ex": "metalurgica, moveleira, grafica"},
    {"grupo": "H-3", "desc": "Industrial alto risco", "ex": "quimica, petroquimica, fogos"},
    {"grupo": "I-1", "desc": "Deposito baixo risco", "ex": "arquivo, deposito de papel"},
    {"grupo": "I-2", "desc": "Deposito medio risco", "ex": "deposito de madeira, borracha"},
    {"grupo": "I-3", "desc": "Deposito alto risco", "ex": "deposito de liquidos inflamaveis"},
    {"grupo": "J-1", "desc": "Explosivos e inflamaveis especiais", "ex": "refinaria, base de gas"},
]

CIDADES_GOIAS = [
    "Goiania", "Aparecida de Goiania", "Anapolis", "Rio Verde", "Luziania",
    "Aguas Lindas de Goias", "Valparaiso de Goias", "Trindade", "Formosa", "Novo Gama",
    "Itumbiara", "Senador Canedo", "Catalao", "Jatai", "Planaltina",
    "Caldas Novas", "Goianesia", "Mineiros", "Inhumas", "Goiatuba",
]

RESPONSAVEIS = [
    ("Joao Silva", "CREA-GO 12345-D"),
    ("Maria Santos", "CREA-GO 67890-D"),
    ("Carlos Oliveira", "CAU A123456-0"),
    ("Ana Ferreira", "CREA-GO 11111-D"),
    ("Pedro Alves", "CAU A654321-0"),
    ("Lucia Martins", "CREA-GO 99999-D"),
]


PERGUNTAS_CLASSIFICACAO = [
    "Qual e o grupo de ocupacao para {desc}?",
    "Uma {desc} se enquadra em qual grupo da NT-01/2025?",
    "Como classificar uma {desc} segundo a NT-01/2025 do CBMGO?",
    "Qual a classificacao de ocupacao para {desc} no CBMGO?",
    "Preciso classificar uma {desc}. Qual grupo de ocupacao devo usar?",
]

PERGUNTAS_CALCULO = [
    "Quantos extintores sao necessarios para uma {desc} com {area}m2?",
    "Como calcular as medidas de seguranca para uma {desc} de {area}m2?",
    "Qual o numero minimo de hidrantes para uma {desc} com {area}m2?",
    "Uma {desc} de {area}m2 e {pavimentos} pavimentos precisa de SPDA?",
    "Quais medidas de seguranca sao obrigatorias para uma {desc} de {area}m2?",
]

PERGUNTAS_MEMORIAL = [
    "Gere um memorial descritivo de PPCI para uma {desc} em {cidade}.",
    "Preciso do memorial de prevencao contra incendio para {desc} com {area}m2.",
    "Elabore o memorial descritivo conforme NT-01/2025 para {desc}.",
    "Crie um memorial de PPCI completo para {desc} de {area}m2 em {cidade}.",
]

PERGUNTAS_NORMAS = [
    "Quais sao as exigencias da NT-01/2025 para {desc}?",
    "O que a NT-01/2025 determina sobre {desc}?",
    "Quais normas se aplicam para {desc} segundo o CBMGO?",
    "Quais sao os requisitos minimos de prevencao de incendio para {desc}?",
]


def gerar_resposta_classificacao(ocupacao):
    grupo = ocupacao["grupo"]
    desc = ocupacao["desc"]
    ex = ocupacao["ex"]
    letra = grupo.split("-")[0]
    descricoes = {
        "A": "residencial", "B": "hospedagem", "C": "comercial",
        "D": "reuniao de publico", "E": "saude", "F": "servicos",
        "G": "automotivo", "H": "industrial", "I": "deposito", "J": "especial",
    }
    return f"""Com base na NT-01/2025 do CBMGO, a classificacao correta e:

**Grupo: {grupo} - {desc}**

Divisao {letra} ({descricoes.get(letra, "especial")}): edificacoes como {ex}.

**Implicacoes para o PPCI:**
- Grupo {grupo} determina medidas de seguranca obrigatorias
- Consulte Tabela 1 da NT-01/2025 para requisitos especificos
- Area total e pavimentos influenciam exigencias adicionais

]

PERGUNTAS_CALCULO = [
    "Quantos extintores sao necessarios para uma {desc} com {area}m2?",
    "Como calcular as medidas de seguranca para uma {desc} de {area}m2?",
    "Qual o numero minimo de hidrantes para uma {desc} com {area}m2?",
    "Uma {desc} de {area}m2 e {pavimentos} pavimentos precisa de SPDA?",
    "Quais medidas de seguranca sao obrigatorias para uma {desc} de {area}m2?",
]

PERGUNTAS_MEMORIAL = [
    "Gere um memorial descritivo de PPCI para uma {desc} em {cidade}.",
    "Preciso do memorial de prevencao contra incendio para {desc} com {area}m2.",
    "Elabore o memorial descritivo conforme NT-01/2025 para {desc}.",
    "Crie um memorial de PPCI completo para {desc} de {area}m2 em {cidade}.",
]

PERGUNTAS_NORMAS = [
    "Quais sao as exigencias da NT-01/2025 para {desc}?",
    "O que a NT-01/2025 determina sobre {desc}?",
    "Quais normas se aplicam para {desc} segundo o CBMGO?",
    "Quais sao os requisitos minimos de prevencao de incendio para {desc}?",
]


def gerar_resposta_classificacao(ocupacao):
    grupo = ocupacao["grupo"]
    desc = ocupacao["desc"]
    ex = ocupacao["ex"]
    letra = grupo.split("-")[0]
    descricoes = {
        "A": "residencial", "B": "hospedagem", "C": "comercial",
        "D": "reuniao de publico", "E": "saude", "F": "servicos",
        "G": "automotivo", "H": "industrial", "I": "deposito", "J": "especial",
    }
    return f"""Com base na NT-01/2025 do CBMGO, a classificacao correta e:

**Grupo: {grupo} - {desc}**

Divisao {letra} ({descricoes.get(letra, "especial")}): edificacoes como {ex}.

**Implicacoes para o PPCI:**
- Grupo {grupo} determina medidas de seguranca obrigatorias
- Consulte Tabela 1 da NT-01/2025 para requisitos especificos
- Area total e pavimentos influenciam exigencias adicionais

Para duvidas, consulte o CBMGO ou profissional habilitado (CREA/CAU)."""


def gerar_resposta_calculo(ocupacao, area, pavimentos):
    grupo = ocupacao["grupo"]
    risco = "alto" if grupo[0] in ["H","I","J"] else ("medio" if grupo[0] in ["C","D","G"] else "baixo")
    area_ext = 150 if risco == "alto" else (250 if risco == "medio" else 500)
    num_ext = max(2, int(area / area_ext) + 1)
    precisa_hidrante = area > 750 or pavimentos >= 3
    altura = pavimentos * 3.0
    precisa_spda = altura > 15 or area > 1000
    precisa_ilum = area > 200 or pavimentos >= 2
    
    return f"""Para edificacao {grupo} com {area:.0f}m2 e {pavimentos} pavimento(s):

**EXTINTORES:**
- Risco: {risco.upper()} | Area/extintor: {area_ext}m2
- Quantidade minima: {num_ext} extintores ABC 6kg

**HIDRANTES:**
- {"OBRIGATORIO" if precisa_hidrante else "Verificar com CBMGO"}

**SPDA:**
- {"OBRIGATORIO (altura {:.1f}m)".format(altura) if precisa_spda else "Avaliar necessidade"}

**ILUMINACAO DE EMERGENCIA:**
- {"OBRIGATORIA - autonomia minima 1h (NBR 10898)" if precisa_ilum else "Recomendada"}

*Nota: Estes calculos sao estimativas. PPCI deve ser elaborado por profissional habilitado.*"""


def gerar_memorial(ocupacao, area, pavimentos, cidade, responsavel):
    grupo = ocupacao["grupo"]
    desc = ocupacao["desc"]
    nome_resp, registro = responsavel
    data_atual = datetime.now().strftime("%d/%m/%Y")
    altura = pavimentos * 3.0
    risco = "alto" if grupo[0] in ["H","I","J"] else ("medio" if grupo[0] in ["C","D","G"] else "baixo")
    area_ext = 150 if risco == "alto" else (250 if risco == "medio" else 500)
    num_ext = max(2, int(area / area_ext) + 1)
    carga = "500 MJ/m2" if risco == "baixo" else ("1.000 MJ/m2" if risco == "medio" else "2.000 MJ/m2")
    
    return f"""MEMORIAL DESCRITIVO DE PREVENCAO E PROTECAO CONTRA INCENDIO E PANICO
PLANO DE PREVENCAO CONTRA INCENDIO - PPCI

1. IDENTIFICACAO DO PROJETO
Proprietario: [A PREENCHER]
Endereco: [Endereco completo], {cidade} - GO
Uso/Ocupacao: {desc} (Grupo {grupo} - NT-01/2025)
Area Total: {area:.2f} m2 | Pavimentos: {pavimentos} | Altura: {altura:.2f} m
Data: {data_atual}
Responsavel Tecnico: {nome_resp} | {registro}

2. OBJETIVO
Este memorial apresenta medidas de seguranca conforme:
- NT-01/2025 do CBMGO | ABNT NBR 9077:2001 | Lei Estadual 15.802/2006

3. CARACTERISTICAS DA EDIFICACAO
- Grupo: {grupo} - {desc}
- Risco de Incendio: {risco.upper()}
- Carga de Incendio Especifica: {carga}
- Estrutura: Alvenaria convencional / Concreto armado

4. MEDIDAS DE SEGURANCA
4.1 Extintores: {num_ext} extintores ABC 6kg (1 a cada {area_ext}m2)
4.2 Saidas de Emergencia: largura minima 1,20m, abertura no sentido do escape
4.3 Iluminacao de Emergencia: autonomia 1h, 30 lux nas saidas (NBR 10898)
4.4 Sinalizacao: conforme ABNT NBR 13434

5. RESPONSABILIDADE TECNICA
{nome_resp} | {registro}
ART/RRT No: [A PREENCHER]
{cidade}, {data_atual}"""


def gerar_exemplo_classificacao():
    ocupacao = random.choice(OCUPACOES)
    tmpl = random.choice(PERGUNTAS_CLASSIFICACAO)
    pergunta = tmpl.format(desc=ocupacao["desc"], grupo=ocupacao["grupo"])
    return {
        "instruction": pergunta, "input": "",
        "output": gerar_resposta_classificacao(ocupacao),
        "category": "classificacao", "grupo": ocupacao["grupo"],
        "source": "synthetic_nt01_2025"
    }


def gerar_exemplo_calculo():
    ocupacao = random.choice(OCUPACOES)
    area = random.choice([100,150,200,300,500,750,1000,1500,2000,3000,5000])
    pavimentos = random.choice([1,2,3,4,5,8,10,15])
    tmpl = random.choice(PERGUNTAS_CALCULO)
    pergunta = tmpl.format(desc=ocupacao["desc"], area=area, pavimentos=pavimentos, grupo=ocupacao["grupo"])
    return {
        "instruction": pergunta,
        "input": f"Area: {area}m2, Pavimentos: {pavimentos}, Grupo: {ocupacao['grupo']}",
        "output": gerar_resposta_calculo(ocupacao, area, pavimentos),
        "category": "calculo", "grupo": ocupacao["grupo"],
        "area": area, "pavimentos": pavimentos, "source": "synthetic_nt01_2025"
    }


def gerar_exemplo_memorial():
    ocupacao = random.choice(OCUPACOES)
    area = random.choice([150,200,300,500,750,1000,1500,2000])
    pavimentos = random.choice([1,2,3,4,5])
    cidade = random.choice(CIDADES_GOIAS)
    responsavel = random.choice(RESPONSAVEIS)
    tmpl = random.choice(PERGUNTAS_MEMORIAL)
    pergunta = tmpl.format(desc=ocupacao["desc"], area=area, cidade=cidade, grupo=ocupacao["grupo"])
    return {
        "instruction": pergunta,
        "input": f"Edificacao: {ocupacao['desc']}, Area: {area}m2, Pavimentos: {pavimentos}, Cidade: {cidade}",
        "output": gerar_memorial(ocupacao, area, pavimentos, cidade, responsavel),
        "category": "memorial", "grupo": ocupacao["grupo"],
        "area": area, "cidade": cidade, "source": "synthetic_nt01_2025"
    }


def gerar_exemplo_normas():
    ocupacao = random.choice(OCUPACOES)
    tmpl = random.choice(PERGUNTAS_NORMAS)
    pergunta = tmpl.format(desc=ocupacao["desc"], grupo=ocupacao["grupo"])
    grupo = ocupacao["grupo"]
    letra = grupo.split("-")[0]
    normas_por_grupo = {
        "A": ["NBR 9077","NBR 10897"], "B": ["NBR 9077","NBR 13714","NBR 12693"],
        "C": ["NBR 9077","NBR 12693"], "D": ["NBR 9077","NBR 13434","NR-23"],
        "E": ["RDC ANVISA","NBR 9077","NBR 12693"], "H": ["NBR 14276","NR-23","NBR 12693"],
    }
    normas = normas_por_grupo.get(letra, ["NBR 9077","NBR 12693","NT-01/2025"])
    normas_str = chr(10).join(f"- {n}" for n in normas)
    resposta = f"""Para {ocupacao["desc"]} (grupo {grupo}), NT-01/2025 determina:

**Normas Aplicaveis:**
{normas_str}
- NT-01/2025 CBMGO | Lei Estadual 15.802/2006

**Exigencias Principais:**
- Extintores com manutencao anual (NBR 12693)
- Sinalizacao de emergencia (NBR 13434)
- Rotas de saida desobstruidas (NBR 9077)
- PPCI aprovado pelo CBMGO antes da ocupacao

**Periodicidade de Vistorias:**
- Inicial: Antes da ocupacao
- Renovacao: 3 anos (baixo risco) / 1 ano (alto risco)"""
    return {
        "instruction": pergunta, "input": "",
        "output": resposta, "category": "normas",
        "grupo": grupo, "source": "synthetic_nt01_2025"
    }


def formatar_chatml(exemplo):
    system_msg = "Voce e o CERCON, assistente do CBMGO especializado em NT-01/2025 e prevencao de incendio em Goias."
    user_content = exemplo["instruction"]
    if exemplo.get("input"):
        user_content += f"\n\nDados fornecidos:\n{exemplo['input']}"
    return {
        "messages": [
            {"role": "system", "content": system_msg},
            {"role": "user", "content": user_content},
            {"role": "assistant", "content": exemplo["output"]}
        ]
    }


def formatar_alpaca(exemplo):
    return {
        "instruction": exemplo["instruction"],
        "input": exemplo.get("input", ""),
        "output": exemplo["output"]
    }


def formatar_llama3(exemplo):
    system_msg = "Voce e o CERCON, assistente do CBMGO especializado em NT-01/2025."
    user_content = exemplo["instruction"]
    if exemplo.get("input"):
        user_content += f"\n\n{exemplo['input']}"
    text = f"<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n\n{system_msg}<|eot_id|><|start_header_id|>user<|end_header_id|>\n\n{user_content}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n{exemplo['output']}<|eot_id|>"
    return {"text": text}


def gerar_dataset(n_total=2000, output_dir="data", formato="chatml", seed=42):
    random.seed(seed)
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    
    n_class = int(n_total * 0.30)
    n_calc = int(n_total * 0.25)
    n_mem = int(n_total * 0.30)
    n_norm = n_total - n_class - n_calc - n_mem
    
    print(f"Gerando dataset sintetico NT-01/2025 CBMGO")
    print(f"Total: {n_total} | Classificacao: {n_class} | Calculo: {n_calc} | Memorial: {n_mem} | Normas: {n_norm}")
    
    exemplos = []
    print("Gerando exemplos de classificacao...")
    for _ in range(n_class): exemplos.append(gerar_exemplo_classificacao())
    print("Gerando exemplos de calculo...")
    for _ in range(n_calc): exemplos.append(gerar_exemplo_calculo())
    print("Gerando exemplos de memorial...")
    for _ in range(n_mem): exemplos.append(gerar_exemplo_memorial())
    print("Gerando exemplos de normas...")
    for _ in range(n_norm): exemplos.append(gerar_exemplo_normas())
    
    random.shuffle(exemplos)
    
    formatadores = {"alpaca": formatar_alpaca, "chatml": formatar_chatml, "llama3": formatar_llama3}
    fmt = formatadores.get(formato, formatar_chatml)
    formatados = [fmt(e) for e in exemplos]
    
    split = int(len(formatados) * 0.9)
    treino, val = formatados[:split], formatados[split:]
    
    out = Path(output_dir)
    
    with open(out / f"train_{formato}.jsonl", "w", encoding="utf-8") as f:
        for ex in treino: f.write(json.dumps(ex, ensure_ascii=False) + "\n")
    print(f"Treino: {out}/train_{formato}.jsonl ({len(treino)} exemplos)")
    
    with open(out / f"validation_{formato}.jsonl", "w", encoding="utf-8") as f:
        for ex in val: f.write(json.dumps(ex, ensure_ascii=False) + "\n")
    print(f"Validacao: {out}/validation_{formato}.jsonl ({len(val)} exemplos)")
    
    meta = {
        "total": n_total, "treino": len(treino), "validacao": len(val),
        "formato": formato, "seed": seed,
        "distribuicao": {"classificacao": n_class, "calculo": n_calc, "memorial": n_mem, "normas": n_norm},
        "base_normativa": "NT-01/2025 CBMGO",
        "gerado_em": datetime.now().isoformat(),
        "versao": "1.0.0",
    }
    with open(out / "dataset_metadata.json", "w", encoding="utf-8") as f:
        json.dump(meta, f, ensure_ascii=False, indent=2)
    print(f"Metadados: {out}/dataset_metadata.json")
    print(f"\nDataset gerado com sucesso! Total: {n_total} exemplos")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Gerador de dataset sintetico NT-01/2025 CBMGO")
    parser.add_argument("--n", type=int, default=2000, help="Numero de exemplos (default: 2000)")
    parser.add_argument("--output", type=str, default="data", help="Diretorio de saida (default: data)")
    parser.add_argument("--formato", type=str, default="chatml",
                        choices=["alpaca","chatml","llama3"], help="Formato SFT (default: chatml)")
    parser.add_argument("--seed", type=int, default=42, help="Semente aleatoria (default: 42)")
    args = parser.parse_args()
    gerar_dataset(n_total=args.n, output_dir=args.output, formato=args.formato, seed=args.seed)