commit d36d96903543a8c472e1f329dd20abb2af447306
parent 19605dea2fe7f0085fc7545fced2ed2f838ba7a8
Author: AbhinavRai01 <abhinavrai004@gmail.com>
Date: Fri, 27 Mar 2026 19:40:31 +0530
agents done hopefully
Diffstat:
7 files changed, 240 insertions(+), 157 deletions(-)
diff --git a/agent/Marl/bandit.py b/agent/Marl/bandit.py
@@ -48,7 +48,7 @@ class ContextualBandit:
# 2. Expected Reward (Dot Product)
# This is the "Best Guess" for how good this action is.
- print(f"Action {a}: Theta shape: {theta.shape}, Context shape: {context_vector.shape}")
+ # print(f"Action {a}: Theta shape: {theta.shape}, Context shape: {context_vector.shape}")
pred = theta.dot(context_vector)
predicted_rewards[a] = pred
diff --git a/agent/main_agent.py b/agent/main_agent.py
@@ -16,12 +16,10 @@ from sub_agents.water_agent import WaterAgent
from sub_agents.Researcher import ResearcherAgent
from sub_agents.Supervisor import SupervisorAgent
-# Update this URL to your running simulator instance
SIMULATOR_ACTION_URL = os.getenv(
- "SIMULATOR_ACTION_URL", "http://localhost:3001/simulation/action"
+ "SIMULATOR_ACTION_URL", "http://localhost:8001/simulation/action"
)
-
def main():
print("š Initializing Demeter Orchestrator...")
@@ -31,9 +29,7 @@ def main():
researcher = ResearcherAgent()
atmos_agent = AtmosphericAgent()
water_agent = WaterAgent()
- # Pass researcher so Supervisor can share the RAG tools if needed
supervisor = SupervisorAgent(researcher_agent=researcher)
-
print("ā
Agents Online.")
except Exception as e:
print(f"ā Init Error: {e}")
@@ -44,83 +40,81 @@ def main():
print("ā±ļø STARTING NEW CYCLE")
print("=" * 50)
- # 1. Fetch
- fmu, sensor_snapshot, history, image_b64 = fetcher.fetch_and_process()
- if not fmu:
- print("ā ļø No FMU found. Waiting...")
+ crops_data = fetcher.fetch_and_process()
+ if not crops_data:
+ print("ā ļø No crops found. Waiting...")
time.sleep(10)
continue
- # 2. Judge
- time.sleep(2) # Small delay to ensure FMU is fully available before judging
- judge.review_previous_cycle(fmu, image_b64)
-
- # 3. š¢ GET BANDIT STRATEGY (The Brain)
- # The Supervisor consults the Bandit first to set the cycle's goal
- time.sleep(2) # Ensure judge's review is complete before strategy retrieval
- strat_name, strat_instr, action_idx = supervisor.get_strategic_goal(fmu)
- print(f"\nš° BANDIT STRATEGY: {strat_name}")
- print(f"š Instruction: {strat_instr}")
-
- # 4. Research
- crop = fmu.metadata.get("crop", "unknown")
- stage = fmu.metadata.get("stage", "unknown")
- query = f"optimal hydroponic conditions for {crop} in {stage} stage"
-
- time.sleep(2) # Small delay before research
- research_context = researcher.search(query)
-
- # 5. š¢ DELIBERATION (The Experts)
- # We pass the strategy instruction and history to the LangGraph agents
- print("\nš§ Agents Planning...")
-
- # Updated call signature to match the new 'reason' method
- time.sleep(2) # Ensure research context is ready before reasoning
- atmos_plan = atmos_agent.reason(
- sensors=sensor_snapshot,
- research=research_context,
- strategy=strat_instr, # Pass the instruction text (e.g. "LOWER pH...")
- history=history, # Pass history for context awareness
- image_b64=image_b64, # Pass the image data for visual diagnosis
- )
-
- print(f"\nš¬ļø Atmospheric Plan:\n{atmos_plan}")
-
- time.sleep(2) # Small delay between agent calls
-
- water_plan = water_agent.reason(
- sensors=sensor_snapshot,
- research=research_context,
- strategy=strat_instr,
- history=history,
- image_b64=image_b64,
- )
-
- print(f"\nš§ Water & Nutrient Plan:\n{water_plan}")
-
- # 6. Synthesis (The Supervisor)
- # Supervisor merges plans, checks conflicts, and ensures safety
- print("\nš® Supervisor Finalizing...")
- final_action = supervisor.synthesize_plan(
- atmos_plan,
- water_plan,
- fmu,
- history,
- strategy_info=(strat_name, strat_instr, action_idx),
- )
-
- print(f"šÆ FINAL COMMAND: {final_action}")
-
- # 7. Execute
+ batch_actions = []
+
+ for crop_data in crops_data:
+ fmu = crop_data["fmu"]
+ sensor_snapshot = crop_data["sensor_snapshot"]
+ history = crop_data["history"]
+ image_b64 = crop_data["image_b64"]
+ crop_id = crop_data["crop_id"]
+
+ print(f"\nš± --- PROCESSING CROP: {crop_id} ---")
+
+ time.sleep(1)
+ judge.review_previous_cycle(fmu, image_b64)
+
+ time.sleep(1)
+ strat_name, strat_instr, action_idx = supervisor.get_strategic_goal(fmu)
+ print(f"\nš° BANDIT STRATEGY: {strat_name}")
+
+ crop = fmu.metadata.get("crop", "unknown")
+ stage = fmu.metadata.get("stage", "unknown")
+ query = f"optimal hydroponic conditions for {crop} in {stage} stage"
+
+ time.sleep(1)
+ research_context = researcher.search(query)
+
+ print("\nš§ Agents Planning...")
+
+ time.sleep(1)
+ atmos_plan = atmos_agent.reason(
+ sensors=sensor_snapshot,
+ research=research_context,
+ strategy=strat_instr,
+ history=history,
+ image_b64=image_b64,
+ )
+
+ time.sleep(1)
+ water_plan = water_agent.reason(
+ sensors=sensor_snapshot,
+ research=research_context,
+ strategy=strat_instr,
+ history=history,
+ image_b64=image_b64,
+ )
+
+ print("\nš® Supervisor Finalizing...")
+ final_action = supervisor.synthesize_plan(
+ atmos_plan,
+ water_plan,
+ fmu,
+ history,
+ strategy_info=(strat_name, strat_instr, action_idx),
+ )
+
+ batch_actions.append({
+ "crop_id": crop_id,
+ "action": final_action
+ })
+
+ print(f"\nā
Final Action for {crop_id}: {final_action}")
try:
- requests.post(SIMULATOR_ACTION_URL, json=final_action)
- print("ā
Sent to Simulator.")
- except Exception as e:
- print(f"ā Connection Error: {e}")
+ requests.post(SIMULATOR_ACTION_URL, json=batch_actions)
- print("\nzzz Sleeping 15s...")
- time.sleep(15)
+ print(f"\nā
Batch sent to Simulator ({len(batch_actions)} actions).")
+ except Exception as e:
+ print(f"\nā Connection Error: {e}")
+ print("\nzzz Sleeping 2 minutes...")
+ time.sleep(120)
if __name__ == "__main__":
- main()
+ main()
+\ No newline at end of file
diff --git a/agent/sub_agents/Supervisor.py b/agent/sub_agents/Supervisor.py
@@ -253,7 +253,7 @@ class SupervisorAgent:
])
context_vector = np.concatenate([vis_vec, s_vec])
- print("Context Vector for Bandit:", context_vector.shape)
+ # print("Context Vector for Bandit:", context_vector.shape)
action_idx, _ = self.bandit.select_action(context_vector)
strategy_name = STRATEGIES[action_idx]
diff --git a/agent/sub_agents/atmospheric_agent.py b/agent/sub_agents/atmospheric_agent.py
@@ -32,7 +32,7 @@ Research: {research}
History: {history}
Critique from Simulation: {critique}
-TASK: Output a JSON dict with keys: 'air_temp' (C), 'humidity' (%), 'co2' (ppm), 'light_intensity' (umol).
+TASK: Output ONLY a valid JSON object with keys: 'air_temp', 'humidity', 'co2', 'light_intensity'. Do not include markdown formatting, code blocks, or any explanatory text outside the JSON. Return strictly the raw JSON.
"""
class AtmosphericAgent:
@@ -125,4 +125,6 @@ class AtmosphericAgent:
}
result = self.app.invoke(initial_state)
+
+ # print(f"\n[{self.name}] Final Result: {result}")
return result.get("final_action", {})
\ No newline at end of file
diff --git a/agent/sub_agents/fetching_agent.py b/agent/sub_agents/fetching_agent.py
@@ -2,6 +2,9 @@ import sys
import os
import requests
from pathlib import Path
+import base64
+from io import BytesIO
+from PIL import Image
current_file = Path(__file__).resolve()
project_root = current_file.parent.parent.parent
@@ -12,12 +15,11 @@ from Sentinel.agent import FMUBuilder
from Qdrant.Store import COLLECTION_NAME
from Qdrant.Client import client
-
class FetchingAgent:
def __init__(self, simulator_url=None):
if simulator_url is None:
simulator_url = os.environ.get(
- "SIMULATOR_STATE_URL", "http://localhost:3001/simulation/state"
+ "SIMULATOR_STATE_URL", "http://localhost:8001/simulation/state"
)
self.sim_url = simulator_url
self.builder = FMUBuilder()
@@ -29,78 +31,78 @@ class FetchingAgent:
response = requests.get(self.sim_url)
if response.status_code == 200:
- data = response.json()
-
- window_data = data.get("sensor_window", {})
- image_b64 = data.get("image", "")
- raw_meta = data.get("metadata", {})
-
- print(data)
-
- print("Rawmeta received: ", raw_meta)
-
- if not image_b64:
- from PIL import Image
- import base64
- from io import BytesIO
-
- img = Image.new("RGB", (512, 512), (50, 50, 50))
- buf = BytesIO()
- img.save(buf, format="PNG")
- image_b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
-
- wanted_keys = {
- "ph": "pH",
- "ec": "EC",
- "humidity": "humidity",
- "temp": "temp",
- "air_temp": "temp",
- }
- sensor_snapshot = {}
- for key, value_list in window_data.items():
- key_lower = key.lower()
- if key_lower in wanted_keys:
- out_name = wanted_keys[key_lower]
- val = (
- value_list[-1]
- if isinstance(value_list, list) and value_list
- else 0.0
- )
- sensor_snapshot[out_name] = val
-
- crop_id = raw_meta.get("crop_id", "UNKNOWN_CROP")
- next_seq = self._get_next_sequence(crop_id)
- print(f"[Fetcher] š¢ Sequence for {crop_id}: {next_seq}")
-
- filtered_metadata = {
- "crop": raw_meta.get("crop", "unknown"),
- "stage": raw_meta.get("stage", "unknown"),
- "crop_id": crop_id,
- "sequence_number": next_seq,
- "image_b64": image_b64,
- }
-
- fmu = self.builder.create_fmu(
- image_b64, sensor_snapshot, filtered_metadata
- )
-
- print(
- f"[Fetcher] š§ FMU Created (ID: {fmu.id}) - Handing off to Judge."
- )
-
- search_results = self.find_similar_instances(fmu)
-
- return fmu, sensor_snapshot, search_results, image_b64
+ data_list = response.json()
+
+ if not isinstance(data_list, list):
+ data_list = [data_list]
+
+ processed_crops = []
+
+ for data in data_list:
+ window_data = data.get("sensor_window", {})
+ image_b64 = data.get("image", "")
+ raw_meta = data.get("metadata", {})
+
+ if not image_b64:
+ img = Image.new("RGB", (512, 512), (50, 50, 50))
+ buf = BytesIO()
+ img.save(buf, format="PNG")
+ image_b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
+
+ wanted_keys = {
+ "ph": "pH",
+ "ec": "EC",
+ "humidity": "humidity",
+ "temp": "temp",
+ "air_temp": "temp",
+ }
+ sensor_snapshot = {}
+ for key, value_list in window_data.items():
+ key_lower = key.lower()
+ if key_lower in wanted_keys:
+ out_name = wanted_keys[key_lower]
+ val = (
+ value_list[-1]
+ if isinstance(value_list, list) and value_list
+ else 0.0
+ )
+ sensor_snapshot[out_name] = val
+
+ crop_id = raw_meta.get("crop_id", data.get("crop_id", "UNKNOWN_CROP"))
+ next_seq = self._get_next_sequence(crop_id)
+
+ filtered_metadata = {
+ "crop": raw_meta.get("crop", "unknown"),
+ "stage": raw_meta.get("stage", "unknown"),
+ "crop_id": crop_id,
+ "sequence_number": next_seq,
+ "image_b64": image_b64,
+ }
+
+ fmu = self.builder.create_fmu(
+ image_b64, sensor_snapshot, filtered_metadata
+ )
+
+ search_results = self.find_similar_instances(fmu)
+
+ processed_crops.append({
+ "crop_id": crop_id,
+ "fmu": fmu,
+ "sensor_snapshot": sensor_snapshot,
+ "history": search_results,
+ "image_b64": image_b64
+ })
+
+ return processed_crops
else:
print(f"[Fetcher] ā Error: Simulator returned {response.status_code}")
- return None, None, None, None
+ return []
except Exception as e:
print(f"[Fetcher] ā Critical Error: {e}")
- return None, None, None, None
+ return []
def _get_next_sequence(self, crop_id):
- """Queries Qdrant for count of existing points for this crop_id."""
try:
count_filter = models.Filter(
must=[
@@ -118,7 +120,6 @@ class FetchingAgent:
def find_similar_instances(self, current_fmu):
try:
- # Simple similarity search (excluding current crop to avoid bias if needed)
hits = client.search(
collection_name=COLLECTION_NAME,
query_vector=current_fmu.vector,
@@ -127,4 +128,4 @@ class FetchingAgent:
)
return [{"payload": hit.payload} for hit in hits]
except Exception:
- return []
+ return []
+\ No newline at end of file
diff --git a/agent/sub_agents/water_agent.py b/agent/sub_agents/water_agent.py
@@ -34,7 +34,7 @@ History: {history}
Critique from Simulation: {critique}
Visual Data: The latest camera image is available via the 'diagnose_plant' tool.
-TASK: Output a JSON dict with keys: 'ph', 'ec' (dS/m), 'water_temp' (C).
+TASK: Output ONLY a valid JSON object with keys: 'ph', 'ec', 'water_temp'. Do not include markdown formatting, code blocks, or any explanatory text outside the JSON. Return strictly the raw JSON.
If you suspect root rot or issues with nutrient uptake (e.g. yellowing leaves), call 'diagnose_plant()' (with no arguments) to verify.
"""
@@ -129,4 +129,5 @@ class WaterAgent:
}
result = self.app.invoke(initial_state)
+ # print(f"\n[{self.name}] Final Result: {result}")
return result.get("final_action", {})
\ No newline at end of file
diff --git a/simulator/main.py b/simulator/main.py
@@ -10,8 +10,8 @@ from pydantic import BaseModel
from typing import List
from PIL import Image
from dotenv import load_dotenv
-from pymongo import MongoClient
-from typing import Optional
+from pymongo import MongoClient, ReturnDocument
+from datetime import datetime
load_dotenv()
@@ -19,17 +19,51 @@ MONGO_URI = os.environ.get("MONGO_URI", "mongodb+srv://abhi:lovesv7@demeter.qfvt
mongo_client = MongoClient(MONGO_URI)
db = mongo_client["test"]
crops_collection = db["cropstates"]
+sim_state_collection = db["simulator_state"]
-MODEL_PATH = "/lettuce_brain_v1.zip"
+MODEL_PATH = "models/PPO/lettuce_brain_v1.zip"
HISTORY_LEN = 20
+CROP_LIFECYCLES = {
+ "lettuce": {
+ "stages": [
+ {"name": "seedling", "end_hour": 168},
+ {"name": "vegetative", "end_hour": 504},
+ {"name": "harvest", "end_hour": 999999}
+ ]
+ },
+ "tomato": {
+ "stages": [
+ {"name": "seedling", "end_hour": 336},
+ {"name": "vegetative", "end_hour": 1008},
+ {"name": "flowering", "end_hour": 1680},
+ {"name": "fruiting", "end_hour": 999999}
+ ]
+ },
+ "basil": {
+ "stages": [
+ {"name": "seedling", "end_hour": 168},
+ {"name": "vegetative", "end_hour": 672},
+ {"name": "harvest", "end_hour": 999999}
+ ]
+ },
+ "strawberry": {
+ "stages": [
+ {"name": "seedling", "end_hour": 336},
+ {"name": "vegetative", "end_hour": 1008},
+ {"name": "flowering", "end_hour": 1512},
+ {"name": "fruiting", "end_hour": 999999}
+ ]
+ }
+}
+
class FarmAction(BaseModel):
acid_dosage_ml: float = 0.0
base_dosage_ml: float = 0.0
nutrient_dosage_ml: float = 0.0
fan_speed_pct: float = 0.0
water_refill_l: float = 0.0
- debug_force_ph: Optional[float] = None
+ debug_force_ph: float | None = None
class BatchActionRequest(BaseModel):
crop_id: str
@@ -145,7 +179,6 @@ def sync_simulators_from_db():
if cid not in simulators:
sensors = crop.get("sensors", {})
-
ph_val = sensors.get("pH", [6.0])
ec_val = sensors.get("EC", [1.5])
temp_val = sensors.get("temp", [24.0])
@@ -164,10 +197,46 @@ def sync_simulators_from_db():
@app.get("/simulation/state")
async def get_all_states():
+ clock = sim_state_collection.find_one_and_update(
+ {"_id": "global_clock"},
+ {"$inc": {"tick_hours": 1}},
+ upsert=True,
+ return_document=ReturnDocument.AFTER
+ )
+ current_tick = clock["tick_hours"]
+
+ crops_collection.update_many({}, {"$inc": {"simulated_age_hours": 1}})
+
sync_simulators_from_db()
+ db_crops = list(crops_collection.find({}))
response = []
- for cid, sim in simulators.items():
+
+ for crop in db_crops:
+ cid = crop.get("crop_id")
+ if not cid or cid not in simulators:
+ continue
+
+ crop_type = crop.get("crop", "lettuce").lower()
+ age_hours = crop.get("simulated_age_hours", 0)
+ cycle_duration = crop.get("cycle_duration_hours", 1)
+
+ new_stage = crop.get("stage", "seedling")
+ if crop_type in CROP_LIFECYCLES:
+ for stage_info in CROP_LIFECYCLES[crop_type]["stages"]:
+ if age_hours <= stage_info["end_hour"]:
+ new_stage = stage_info["name"]
+ break
+
+ if new_stage != crop.get("stage"):
+ crops_collection.update_one({"crop_id": cid}, {"$set": {"stage": new_stage}})
+
+ print(f" current_tick: {current_tick} | crop_id: {cid} | age_hours: {age_hours} | stage: {new_stage} | cycle_duration: {cycle_duration}")
+
+ if current_tick % cycle_duration != 0:
+ continue
+
+ sim = simulators[cid]
pil_img = sim._generate_image()
buf = BytesIO()
pil_img.save(buf, format="PNG")
@@ -177,11 +246,16 @@ async def get_all_states():
"crop_id": cid,
"sensor_window": {k: list(v) for k, v in sim.history.items()},
"metadata": {
+ "crop": crop_type,
+ "stage": new_stage,
"health": round(float(sim.plant_health), 1),
"biomass_est": round(float(sim.state[6]), 2),
+ "age_hours": age_hours,
+ "global_tick": current_tick
},
"image": img_b64,
})
+
return response
@app.post("/simulation/action")
@@ -192,6 +266,7 @@ async def take_batch_actions(payload: List[BatchActionRequest]):
for req in payload:
cid = req.crop_id
if cid not in simulators:
+ results.append({"crop_id": cid, "status": "error", "message": "Crop not found"})
continue
sim = simulators[cid]
@@ -204,10 +279,16 @@ async def take_batch_actions(payload: List[BatchActionRequest]):
"sensors.humidity": {"$each": [float(sim.state[4])], "$slice": -5}
}
+ set_payload = {
+ "action_taken": req.action.dict(),
+ "last_updated": datetime.utcnow()
+ }
+
crops_collection.update_one(
{"crop_id": cid},
{
"$push": push_payload,
+ "$set": set_payload,
"$inc": {"sequence_number": 1}
}
)
@@ -217,7 +298,9 @@ async def take_batch_actions(payload: List[BatchActionRequest]):
"status": "success",
"new_state": {
"pH": float(sim.state[0]),
- "EC": float(sim.state[1])
+ "EC": float(sim.state[1]),
+ "temp": float(sim.state[3]),
+ "humidity": float(sim.state[4])
}
})