from composabl import Agent, Sensor, Skill, Scenario, Perceptor
from composabl import MaintainGoal
# Create agent
agent = Agent(id="reactor-controller")
# Add sensors
agent.add_sensors([
Sensor("temp", "Temperature in Celsius", lambda obs: obs["temperature"]),
Sensor("pressure", "Pressure in bar", lambda obs: obs["pressure_reading"]),
Sensor("flow", "Flow rate L/min", lambda obs: obs["flow_rate"])
])
# Add perceptor
class RateCalculator(PerceptorImpl):
def __init__(self):
self.last_temp = None
async def compute(self, obs_spec, obs):
rate = 0
if self.last_temp is not None:
rate = obs["temperature"] - self.last_temp
self.last_temp = obs["temperature"]
return {"temp_rate": rate}
def filtered_sensor_space(self, obs):
return ["temperature"]
agent.add_perceptor(Perceptor("rate-calc", RateCalculator))
# Add skills
temp_skill = Skill("maintain-temp",
MaintainGoal("temp", "Keep temperature stable",
target=75.0, stop_distance=2.0))
agent.add_skill(temp_skill)
# Add scenarios
agent.add_scenario(Scenario({
"temperature": {"min": 70, "max": 80},
"pressure": {"min": 5, "max": 7}
}))
# Visualize
agent.draw()