# config_manager.py import json import os import re # Caminho para o ficheiro principal de onde vamos ler/escrever as constantes CONFIG_FILE_PATH = 'app.py' def _load_constants_from_file(): """Lê todas as CONSTANTES (ALL_CAPS) do ficheiro Python e retorna um dicionário.""" constants = {} BOOL_CONSTANTS = ["ALLOW_SLEEP_ON_CHANGE"] try: with open(CONFIG_FILE_PATH, 'r', encoding='utf-8') as f: content = f.read() except FileNotFoundError: print(f"ERRO: Ficheiro de configuração não encontrado em {CONFIG_FILE_PATH}") return constants matches = re.findall(r'^\s*([A-Z_]+)\s*=\s*(.*)', content, re.MULTILINE) for name, value_str in matches: value_str = value_str.strip() # Remove comentários inline value_str = re.sub(r'\s*#.*$', '', value_str).strip() try: if name in BOOL_CONSTANTS: constants[name] = value_str.lower() == "true" elif re.match(r'^-?\d+$', value_str): constants[name] = int(value_str) elif re.match(r'^-?\d+\.\d+$', value_str): constants[name] = float(value_str) elif re.match(r'^-?\d+e[+-]?\d+$', value_str, re.I): constants[name] = float(value_str) else: constants[name] = value_str.strip("'\"") except Exception: constants[name] = value_str.strip("'\"") return constants _runtime_globals = {} def set_runtime_globals(g: dict): """Chamado pelo app.py no endpoint /api/config para passar variáveis dinâmicas.""" _runtime_globals.update(g) def _get_runtime_var(key): return _runtime_globals.get(key) def map_backend_to_frontend(constants): """Mapeia as constantes do backend para a estrutura JSON do frontend.""" def safe_int(v, d=0): try: return int(v) except: return d def safe_float(v, d=0.0): try: return float(v) except: return d return { "preset": "custom", "hardware": { # Variáveis dinâmicas — lidas dos globals do app em runtime "cpu_count": safe_int(_get_runtime_var('logical_cpus') or constants.get('LOGICAL_CPUS', 0)), "omp_threads": safe_int(_get_runtime_var('omp_threads') or constants.get('OMP_THREADS', 0)), "dataloader_workers": safe_int(_get_runtime_var('dataloader_workers') or constants.get('DATALOADER_WORKERS', 0)), "total_ram_gb": round(safe_float(_get_runtime_var('total_ram_gb') or constants.get('TOTAL_RAM_GB', 0.0)), 1), }, "ramcfg": { "base_batch": safe_int(constants.get('BASE_BATCH_SIZE', 1)), "accum_min": safe_int(constants.get('INITIAL_ACCUMULATION_MIN_STEPS', 2)), "max_steps": safe_int(constants.get('INITIAL_ACCUMULATION_MAX_STEPS', constants.get('DYNAMIC_ACCUMULATION_MAX_STEPS', 4))), "target_utilization": safe_float(constants.get('TARGET_RAM_UTILIZATION', 0.95)), "estimated_batch_gb": safe_float(constants.get('ESTIMATED_BATCH_GB', 0.30)), "high_ram_limit_pct": safe_float(constants.get('DYNAMIC_ACCUMULATION_HIGH_RAM_LIMIT', 98.5)), "low_ram_limit_pct": safe_float(constants.get('DYNAMIC_ACCUMULATION_LOW_RAM_LIMIT', 40.0)), }, "mlcfg": { "base_max_len": safe_int(constants.get('BASE_MAX_LEN', 128)), "ml_target_utilization": safe_float(constants.get('TARGET_MAX_LEN_UTILIZATION', 0.75)), "increment": safe_int(constants.get('MAX_LEN_INCREMENT', 128)), "cap": safe_int(constants.get('MAX_LEN_CAP', 128)), "estimated_base_dataset_ram_gb": safe_float(constants.get('ESTIMATED_BASE_DATASET_RAM_GB', 1.6)), "cost_per_increment_gb": safe_float(constants.get('COST_PER_INCREMENT_GB', 0.30)), }, "training": { "base_batch": safe_int(constants.get('BASE_BATCH_SIZE', 1)), "base_eval_size": safe_int(constants.get('BASE_EVAL_SIZE', 5)), "base_lr": safe_float(constants.get('BASE_LEARNING_RATE', 1e-4)), "loggin_steps": safe_int(constants.get('LOGGIN_STEPS', constants.get('LOGGING_STEPS', 10))), "optim": constants.get('OPTIM', 'adamw_torch'), "scheduler": constants.get('LR_SCHEDULER_TYPE', 'constant_with_warmup'), } } def update_python_constants(new_data): """Atualiza as constantes no ficheiro Python original com os novos valores.""" # Mapeamento plano: chave JSON → nome da constante Python # Cobre ramcfg, mlcfg e training MAPPING = { # ramcfg "base_batch": "BASE_BATCH_SIZE", "max_steps": "DYNAMIC_ACCUMULATION_MAX_STEPS", "target_utilization": "TARGET_RAM_UTILIZATION", "estimated_batch_gb": "ESTIMATED_BATCH_GB", "high_ram_limit_pct": "DYNAMIC_ACCUMULATION_HIGH_RAM_LIMIT", "low_ram_limit_pct": "DYNAMIC_ACCUMULATION_LOW_RAM_LIMIT", # mlcfg "base_max_len": "BASE_MAX_LEN", "ml_target_utilization": "TARGET_MAX_LEN_UTILIZATION", "increment": "MAX_LEN_INCREMENT", "cap": "MAX_LEN_CAP", "estimated_base_dataset_ram_gb": "ESTIMATED_BASE_DATASET_RAM_GB", "cost_per_increment_gb": "COST_PER_INCREMENT_GB", # training "base_eval_size": "BASE_EVAL_SIZE", "base_lr": "BASE_LEARNING_RATE", "scheduler": "LR_SCHEDULER_TYPE", } # Achatar o JSON aninhado num dicionário plano flat = {} for section in ("ramcfg", "mlcfg", "training"): flat.update(new_data.get(section, {})) # base_batch pode vir de ramcfg ou training — usar o de training se presente if "base_batch" in new_data.get("training", {}): flat["base_batch"] = new_data["training"]["base_batch"] # Construir dicionário de actualizações { CONST_NAME: value } updates = {} for ui_key, const_name in MAPPING.items(): if ui_key in flat: updates[const_name] = flat[ui_key] if not updates: return False try: with open(CONFIG_FILE_PATH, 'r', encoding='utf-8') as f: lines = f.readlines() except FileNotFoundError: return False new_lines = [] for line in lines: match = re.match(r'^(\s*)([A-Z_]+)\s*=\s*(.*)', line) if match: indent, name, rest = match.groups() if name in updates: val = updates[name] if isinstance(val, str): formatted = f'"{val}"' elif isinstance(val, bool): formatted = str(val) elif isinstance(val, float): # Preserva notação científica para learning rate if abs(val) < 0.001 and val != 0: formatted = f'{val:.0e}' else: formatted = str(val) else: formatted = str(val) # Preserva comentário inline se existia comment_match = re.search(r'(#.*)$', rest) comment = f' {comment_match.group(1)}' if comment_match else '' new_lines.append(f"{indent}{name} = {formatted}{comment}\n") continue new_lines.append(line) try: with open(CONFIG_FILE_PATH, 'w', encoding='utf-8') as f: f.writelines(new_lines) return True except Exception as e: print(f"ERRO ao escrever no ficheiro: {e}") return False # Fim de config_manager.py