DarkyCodez/MetaxScaler

1import os2import sys3import json4import requests5from typing import Any6from openai import OpenAI7 8# --- CONFIGURATION ---9API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")10MODEL_NAME = os.getenv("MODEL_NAME", "Qwen/Qwen2.5-72B-Instruct")11HF_TOKEN = os.getenv("HF_TOKEN")12ENV_URL = "https://darkycodez-metaxscaler.hf.space"13 14TASKS = ["task_thirsty_crop", "task_nutrient_balance", "task_heatwave_crisis"]15 16# Constraints for the Validator17MIN_SCORE = 0.0118MAX_SCORE = 0.9919SUCCESS_THRESHOLD = 0.520 21client = OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN)22 23SYSTEM_PROMPT = """You are an agricultural AI managing a greenhouse.24Observations include soil moisture, nitrogen levels, and crop health.25Choose one action per turn:260: Do nothing271: Irrigate282: Fertilize293: Harvest30Reply ONLY with a single integer (0, 1, 2, or 3)."""31 32# --- UTILS ---33def _strict_score(value: float) -> float:34    """Clamps score strictly between (0, 1) and rounds to 3 decimals."""35    score = max(MIN_SCORE, min(MAX_SCORE, value))36    return round(score, 3)37 38def _post_json(path: str, payload: dict[str, Any]) -> dict[str, Any] | None:39    """Handles POST requests with timeout and basic error catching."""40    try:41        response = requests.post(f"{ENV_URL.rstrip('/')}{path}", json=payload, timeout=15)42        response.raise_for_status()43        return response.json()44    except Exception as e:45        print(f"[DEBUG] Request failed: {str(e)}", file=sys.stderr)46        return None47 48# --- CORE LOGIC ---49def run_simulation(task_id: str):50    print(f"[START] task={task_id} env=precision-ag-env model={MODEL_NAME}")51    52    # 1. Reset Environment53    reset_data = _post_json("/reset", {"task_id": task_id})54    if not reset_data:55        print(f"[END] success=false steps=0 score={MIN_SCORE} rewards= error=reset_failed")56        return MIN_SCORE57 58    obs = reset_data.get("observation", {})59    rewards = []60    turn = 061    done = False62 63    # 2. Step Loop64    while not done and turn < 15:65        # Get Action from LLM66        try:67            completion = client.chat.completions.create(68                model=MODEL_NAME,69                messages=[70                    {"role": "system", "content": SYSTEM_PROMPT},71                    {"role": "user", "content": f"Observation: {obs}\nAction (0-3):"}72                ],73                temperature=0.1,74                max_tokens=575            )76            content = completion.choices[0].message.content.strip()77            action_int = int(''.join(filter(str.isdigit, content)))78        except Exception:79            action_int = 0 # Default to 'Do nothing' on LLM failure80 81        # Execute Step82        step_data = _post_json("/step", {"action": action_int})83        if not step_data:84            error_text = "step_failed"85            reward = 0.086            done = True87        else:88            obs = step_data.get("observation", {})89            reward = float(step_data.get("reward", 0.0))90            done = bool(step_data.get("done", False))91            error_text = "null"92            rewards.append(reward)93 94        # Mandatory Step Logging95        step_log = {96            "turn": turn,97            "action": action_int,98            "reward": reward,99            "done": done,100            "error": error_text101        }102        print(f"[STEP] {json.dumps(step_log)}", flush=True)103        104        turn += 1105        if done: break106 107    # 3. Final Scoring & Clamping108    raw_score = sum(rewards) / len(rewards) if rewards else MIN_SCORE109    final_score = _strict_score(raw_score)110    success = final_score >= SUCCESS_THRESHOLD111    112    rewards_str = ",".join(f"{r:.2f}" for r in rewards)113    print(f"[END] success={str(success).lower()} steps={len(rewards)} score={final_score:.3f} rewards={rewards_str}")114    return final_score115 116def main():117    for task in TASKS:118        try:119            run_simulation(task)120        except Exception as e:121            # Last resort catch-all to ensure [END] is always printed122            print(f"[END] success=false steps=0 score={MIN_SCORE} error=unhandled_exception:{type(e).__name__}")123 124if __name__ == "__main__":125    main()