utils.py

import os
import re
import csv
import json
import time
import yaml
import autogen
import pandas as pd
import xml.etree.ElementTree as ET

from tqdm import tqdm
from TOON_parser import *
from typing import List, Tuple
from transformers import AutoTokenizer
from toon_format import encode, decode as toon_decode, ToonDecodeError

CARBON_INTENSITY = 0.475

MODELS = {
    "GPT-oss-20B": {
        "name": "gpt-oss-20b",
        "folder": "GPT-oss-20B",
        "ollama_model": "gpt-oss:20b"
    },
    "GPT-oss-120B": {
        "name": "gpt-oss-120b",
        "folder": "GPT-oss-120B",
        "ollama_model": "gpt-oss:120b"
    },
    "Gemma3-4B": {
        "name": "gemma-3-4b",
        "folder": "Gemma3-4B",
        "ollama_model": "gemma:4b",
        "no_system_prompt": True
    },
    "Gemma3-12B": {
        "name": "gemma-3-12b",
        "folder": "Gemma3-12B",
        "ollama_model": "gemma:12b",
        "no_system_prompt": True
    },
    "Gemma3-27B": {
        "name": "gemma-3-27b",
        "folder": "Gemma3-27B",
        "ollama_model": "gemma:27b",
        "no_system_prompt": True
    },
    "Mistral-7B": {
        "name": "mistral-7b",
        "folder": "Mistral-7B",
        "ollama_model": "mistral:7b"
    },
    "Llama-3.1-405B": {
        "name": "llama-3.1-405b",
        "folder": "Llama-3.1-405B",
        "ollama_model": "llama3.1:405b"
    },
    "Llama-3.3-70B": {
        "name": "llama-3.3-70b",
        "folder": "Llama-3.3-70B",
        "ollama_model": "llama3.3:70b"
    },
    "Qwen3-4B": {
        "name": "qwen3-4b",
        "folder": "Qwen3-4B",
        "ollama_model": "qwen:4b"
    },
}

MODEL_PARAMS = {
    "gpt-oss:20b": 20e9,
    "gpt-oss:120b": 120e9,
    "gemma:4b": 4e9,
    "gemma:12b": 12e9,
    "gemma:27b": 27e9,
    "mistral:7b": 7e9,
    "llama3.1:405b": 405e9,
    "llama3.3:70b": 70e9,
    "qwen:4b": 4e9,
}


def calculate_api_emissions(model_name: str, input_tokens: int, output_tokens: int, 
                           duration: float, carbon_intensity: float = 0.475) -> dict:
    model_key = model_name.lower().replace("_", "-")
    
    model_params = None
    for key, params in MODEL_PARAMS.items():
        if key in model_key:
            model_params = params
            break
    
    if model_params is None:
        raise ValueError(f"Modello '{model_name}' non trovato. Modelli supportati: {list(MODEL_PARAMS.keys())}")
    
    base_power = 30
    power_per_billion_params = 0.9
    estimated_power_watts = base_power + (model_params / 1e9) * power_per_billion_params
    
    duration_hours = duration / 3600
    energy_kwh = (estimated_power_watts * duration_hours) / 1000
    
    # Emissioni in kg CO2
    emissions_kg = energy_kwh * carbon_intensity
    
    total_tokens = input_tokens + output_tokens
    
    return {
        "emissions": emissions_kg,
        "energy_consumed": energy_kwh,
        "power_watts": estimated_power_watts,
        "duration": duration,
        "total_tokens": total_tokens,
        "model_params": model_params,
        "carbon_intensity": carbon_intensity
    }

D_START = 0.00
D_END = 1.00
D_STEP = 0.01

D_VALUES = [round(i * D_STEP, 2) for i in range(int(D_START / D_STEP), int(D_END / D_STEP) + 1)]
BASE_COLUMNS = ["ID"]
GCS_ENV_COLUMNS = [f"GCS_env({d:.2f})" for d in D_VALUES]

def extract_keys_json(data, parent_key=""):
    keys = []

    if isinstance(data, dict):
        for k, v in data.items():
            full_key = f"{parent_key}.{k}" if parent_key else k
            keys.append(full_key)
            keys.extend(extract_keys_json(v, full_key))

    elif isinstance(data, list):
        for i, item in enumerate(data):
            full_key = f"{parent_key}[{i}]"
            keys.append(full_key)
            keys.extend(extract_keys_json(item, full_key))

    return keys

def extract_keys_xml(element, parent_key=""):
    keys = []

    current_key = f"{parent_key}.{element.tag}" if parent_key else element.tag
    keys.append(current_key)

    for attr in element.attrib:
        attr_key = f"{current_key}.@{attr}"
        keys.append(attr_key)

    if element.text and element.text.strip():
        text_key = f"{current_key}.#text"
        keys.append(text_key)

    tag_count = {}
    for child in element:
        tag = child.tag
        tag_count[tag] = tag_count.get(tag, 0) + 1

    tag_seen = {}
    for child in element:
        tag = child.tag
        tag_seen[tag] = tag_seen.get(tag, 0)

        if tag_count[tag] > 1:
            child_key = f"{current_key}.{tag}[{tag_seen[tag]}]"
            keys.append(child_key)
            keys.extend(extract_keys_xml(child, child_key))
        else:
            keys.extend(extract_keys_xml(child, current_key))

        tag_seen[tag] += 1

    return keys

def extract_keys_yaml(data, parent_key=""):
    keys = []

    if isinstance(data, dict):
        for k, v in data.items():
            full_key = f"{parent_key}.{k}" if parent_key else k
            keys.append(full_key)
            keys.extend(extract_keys_yaml(v, full_key))

    elif isinstance(data, list):
        for i, item in enumerate(data):
            full_key = f"{parent_key}[{i}]"
            keys.append(full_key)
            keys.extend(extract_keys_yaml(item, full_key))

    return keys

def clean_llm_response(response: str) -> str:
    cleaned = response.strip()

    pattern = r'^```(?:json|xml|yaml|toon)?\s*\n?'
    cleaned = re.sub(pattern, '', cleaned, flags=re.IGNORECASE | re.MULTILINE)
    
    # Rimuove i ``` finali
    if cleaned.endswith('```'):
        cleaned = cleaned[:-3].rstrip()
    
    return cleaned

def is_valid_toon(toon_string: str) -> bool:
    try:
        toon_decode(toon_string)
        return True
    except ToonDecodeError:
        return False

def load_json(path):
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)

def from_string_to_json(text):
    try:
        json_data = json.loads(text)
        return json_data
    except json.JSONDecodeError:
        match = re.search(r'{(.*?)}', text, re.DOTALL)
        if match:
            json_text = match.group(1)
            json_text = json_text.strip()
            try:
                json_data = json.loads("{" + json_text + "}")
                return json_data
            except json.JSONDecodeError as e:
                print(f"JSON string parsing error: {e}")

    return {}

def read_from_file(filename, folder):
    file_path = os.path.join(folder, filename)

    if not os.path.exists(file_path):
        return "File not found."

    with open(file_path, 'r', encoding='utf-8') as file:
        content = file.read()
    return content

system_prompt_JSON = read_from_file("system_prompt_JSON.txt", "prompts")
system_prompt_TOON = read_from_file("system_prompt_TOON.txt", "prompts")
system_prompt_XML = read_from_file("system_prompt_XML.txt", "prompts")
system_prompt_YAML = read_from_file("system_prompt_YAML.txt", "prompts")

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=0,
    code_execution_config=False,
    default_auto_reply="default_auto_reply"
)