Spaces:
Sleeping
Sleeping
| """ | |
| MycoController — autonomous AI agent that drives background game actions. | |
| The controller uses engine._llm() to decide what to do, then delegates | |
| execution to the same engine functions the player uses. | |
| """ | |
| from . import engine | |
| import json | |
| class MycoController: | |
| def __init__(self): | |
| self.position = [1, 1] | |
| self.history = [] | |
| # ------------------------------------------------------------------ | |
| # Decision layer — ask Gemma what to do next | |
| # ------------------------------------------------------------------ | |
| def get_agent_decision(self, current_mushroom, collection) -> dict: | |
| # Build a full context so Gemma has score/health/mystery data. | |
| ctx = engine._ctx(current_mushroom, list(collection or [])) | |
| mushroom_line = ( | |
| f"There is a {current_mushroom.get('rarity','?')} mushroom called " | |
| f"{current_mushroom.get('name','?')} in the clearing." | |
| if current_mushroom | |
| else "The clearing is empty." | |
| ) | |
| prompt = ( | |
| f"You are Myco, an autonomous forest agent. " | |
| f"Current position: {self.position}. {mushroom_line} " | |
| f"Collection count: {ctx.get('collection_count', 0)}. " | |
| f"Score: {ctx.get('score', 0)} spores. Health: {ctx.get('health', 3)}/3. " | |
| f"Mystery chapter: {ctx.get('mystery_title', 'The Wrong Memory')}. " | |
| "Decide the single best next action from: move, search, study, collect, wait. " | |
| "If the clearing is empty, prefer 'search'. " | |
| "If a mushroom is present and unstudied, prefer 'study'. " | |
| "If studied and safe, prefer 'collect'. " | |
| "If move, pick an adjacent grid coordinate within [0,2] x [0,2]. " | |
| 'Respond ONLY in valid JSON: {"action":"...","target":[x,y],"thought":"..."}' | |
| ) | |
| response = engine._llm(prompt, ctx) | |
| if not response: | |
| print("[Myco Controller] LLM returned no data — defaulting to wait.") | |
| return {"action": "wait", "target": None, "thought": "Engine returned no data."} | |
| try: | |
| parsed = json.loads(response) | |
| action = parsed.get("action", "wait") | |
| if action not in {"move", "search", "study", "collect", "wait"}: | |
| raise ValueError(f"Unknown action: {action!r}") | |
| return parsed | |
| except Exception as exc: | |
| print(f"[Myco Controller] Parse error: {exc} — raw: {response!r}") | |
| return {"action": "wait", "target": None, "thought": "Failed to parse response."} | |
| # ------------------------------------------------------------------ | |
| # Execution layer — carry out the decision using engine functions | |
| # ------------------------------------------------------------------ | |
| def run_tick(self, current_mushroom, collection) -> dict: | |
| """ | |
| Run one autonomous tick. Returns a result dict with: | |
| action_taken, thought, and optionally data (new current/collection). | |
| This is the ONLY entry point the UI should call — it keeps controller | |
| state (self.history, self.position) in sync. | |
| """ | |
| decision = self.get_agent_decision(current_mushroom, collection) | |
| action = decision.get("action", "wait") | |
| result = { | |
| "action_taken": action, | |
| "thought": decision.get("thought", ""), | |
| "data": {}, | |
| } | |
| if action == "move": | |
| target = decision.get("target") | |
| if isinstance(target, list) and len(target) == 2: | |
| x = max(0, min(2, int(target[0]))) | |
| y = max(0, min(2, int(target[1]))) | |
| self.position = [x, y] | |
| result["data"]["position"] = self.position | |
| else: | |
| result["thought"] = "Move had no valid target — staying put." | |
| elif action == "search": | |
| mushroom, current, history = engine.discover_mushroom(list(collection or [])) | |
| self.history = history | |
| result["data"] = { | |
| "mushroom": mushroom, | |
| "current": current, | |
| "history": history, | |
| } | |
| elif action == "study": | |
| if current_mushroom: | |
| reply, history = engine.study_current(current_mushroom, self.history) | |
| self.history = history | |
| result["data"] = { | |
| "study_reply": reply, | |
| "history": history, | |
| } | |
| else: | |
| result["thought"] = "Nothing to study — clearing is empty." | |
| elif action == "collect": | |
| if current_mushroom: | |
| coll, history = engine.collect_current( | |
| current_mushroom, list(collection or []), self.history | |
| ) | |
| self.history = history | |
| result["data"] = { | |
| "collection": coll, | |
| "history": history, | |
| } | |
| else: | |
| result["thought"] = "Nothing to collect — clearing is empty." | |
| # action == "wait" needs no execution | |
| return result | |