"""
Pattern 101: Sensor Agent Pattern

Specialized agent that monitors and collects data from various sources (APIs, files, sensors).
Uses LangChain for processing sensor data and LangGraph for sensor orchestration.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional, Callable
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import time


class SensorType(Enum):
    """Types of sensors."""
    TEMPERATURE = "temperature"
    HUMIDITY = "humidity"
    PRESSURE = "pressure"
    MOTION = "motion"
    LIGHT = "light"
    API = "api"
    FILE = "file"
    DATABASE = "database"


@dataclass
class SensorReading:
    """Represents a sensor reading."""
    sensor_id: str
    sensor_type: SensorType
    value: Any
    unit: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)
    

@dataclass
class SensorAlert:
    """Alert triggered by sensor threshold violation."""
    sensor_id: str
    reading: SensorReading
    threshold_type: str  # "above", "below", "equal"
    threshold_value: float
    severity: str  # "low", "medium", "high", "critical"
    message: str
    timestamp: datetime = field(default_factory=datetime.now)


class SensorAgent:
    """Agent that monitors a specific sensor and generates alerts."""
    
    def __init__(
        self,
        sensor_id: str,
        sensor_type: SensorType,
        read_function: Callable[[], Any],
        unit: str = "",
        model_name: str = "gpt-4"
    ):
        self.sensor_id = sensor_id
        self.sensor_type = sensor_type
        self.read_function = read_function
        self.unit = unit
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
        self.readings: List[SensorReading] = []
        self.alerts: List[SensorAlert] = []
        self.thresholds: Dict[str, float] = {}
        
    def read_sensor(self) -> SensorReading:
        """Read from the sensor and create a reading."""
        value = self.read_function()
        reading = SensorReading(
            sensor_id=self.sensor_id,
            sensor_type=self.sensor_type,
            value=value,
            unit=self.unit
        )
        self.readings.append(reading)
        self._check_thresholds(reading)
        return reading
    
    def set_threshold(self, threshold_type: str, value: float):
        """Set a threshold for alerts."""
        self.thresholds[threshold_type] = value
    
    def _check_thresholds(self, reading: SensorReading):
        """Check if reading violates any thresholds."""
        value = float(reading.value)
        
        if "above" in self.thresholds and value > self.thresholds["above"]:
            self._create_alert(reading, "above", self.thresholds["above"])
        
        if "below" in self.thresholds and value  1 else "Threshold violation detected"
        
        alert = SensorAlert(
            sensor_id=self.sensor_id,
            reading=reading,
            threshold_type=threshold_type,
            threshold_value=threshold_value,
            severity=severity,
            message=message
        )
        self.alerts.append(alert)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics about sensor readings."""
        if not self.readings:
            return {"count": 0}
        
        values = [float(r.value) for r in self.readings]
        return {
            "sensor_id": self.sensor_id,
            "sensor_type": self.sensor_type.value,
            "count": len(self.readings),
            "min": min(values),
            "max": max(values),
            "avg": sum(values) / len(values),
            "latest": values[-1],
            "alerts": len(self.alerts)
        }


class SensorAgentPattern:
    """
    Coordinates multiple sensor agents for comprehensive monitoring.
    Implements sensor fusion and multi-sensor analysis.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.sensors: Dict[str, SensorAgent] = {}
        self.monitoring_active = False
        
    def register_sensor(self, sensor: SensorAgent):
        """Register a sensor agent."""
        self.sensors[sensor.sensor_id] = sensor
        
    def read_all_sensors(self) -> Dict[str, SensorReading]:
        """Read from all registered sensors."""
        readings = {}
        for sensor_id, sensor in self.sensors.items():
            readings[sensor_id] = sensor.read_sensor()
        return readings
    
    def get_all_alerts(self) -> List[SensorAlert]:
        """Get all alerts from all sensors."""
        alerts = []
        for sensor in self.sensors.values():
            alerts.extend(sensor.alerts)
        return sorted(alerts, key=lambda a: a.timestamp, reverse=True)
    
    def analyze_environment(self) -> str:
        """Analyze overall environment based on all sensors."""
        stats = {sid: s.get_statistics() for sid, s in self.sensors.items()}
        alerts = self.get_all_alerts()
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an environmental monitoring AI. Analyze sensor data and provide insights."),
            ("human", """Sensor Statistics:
{stats}

Recent Alerts:
{alerts}

Provide:
1. Overall environmental status
2. Any concerning patterns
3. Recommendations for action""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "stats": "\n".join([f"{k}: {v}" for k, v in stats.items()]),
            "alerts": "\n".join([f"- {a.severity.upper()}: {a.message}" for a in alerts[:5]])
        })
        
        return str(response.content)
    
    def monitor_continuous(self, duration_seconds: int, interval_seconds: int = 1):
        """Continuously monitor sensors for specified duration."""
        self.monitoring_active = True
        start_time = time.time()
        
        print(f"Starting continuous monitoring for {duration_seconds} seconds...")
        
        while self.monitoring_active and (time.time() - start_time)  (time.time() - interval_seconds):
                print(f"⚠️  ALERT: {new_alerts[0].message}")
            
            time.sleep(interval_seconds)
        
        print("Monitoring stopped.")
    
    def stop_monitoring(self):
        """Stop continuous monitoring."""
        self.monitoring_active = False


def main():
    """Example usage of Sensor Agent Pattern."""
    print("=" * 80)
    print("Sensor Agent Pattern - Environmental Monitoring Demo")
    print("=" * 80)
    
    # Create sensor pattern
    pattern = SensorAgentPattern()
    
    # Simulate sensor functions
    import random
    
    temp_base = 22.0
    humid_base = 60.0
    
    def read_temperature():
        return round(temp_base + random.uniform(-2, 5), 1)
    
    def read_humidity():
        return round(humid_base + random.uniform(-10, 15), 1)
    
    def read_light():
        return random.randint(100, 1000)
    
    # Create sensor agents
    temp_sensor = SensorAgent(
        sensor_id="temp-001",
        sensor_type=SensorType.TEMPERATURE,
        read_function=read_temperature,
        unit="°C"
    )
    temp_sensor.set_threshold("above", 25.0)
    temp_sensor.set_threshold("below", 18.0)
    
    humid_sensor = SensorAgent(
        sensor_id="humid-001",
        sensor_type=SensorType.HUMIDITY,
        read_function=read_humidity,
        unit="%"
    )
    humid_sensor.set_threshold("above", 70.0)
    humid_sensor.set_threshold("below", 40.0)
    
    light_sensor = SensorAgent(
        sensor_id="light-001",
        sensor_type=SensorType.LIGHT,
        read_function=read_light,
        unit="lux"
    )
    light_sensor.set_threshold("below", 200)
    
    # Register sensors
    pattern.register_sensor(temp_sensor)
    pattern.register_sensor(humid_sensor)
    pattern.register_sensor(light_sensor)
    
    # Read sensors multiple times
    print("\n📊 Reading sensors (10 samples)...")
    for i in range(10):
        readings = pattern.read_all_sensors()
        print(f"\nSample {i+1}:")
        for sensor_id, reading in readings.items():
            print(f"  {sensor_id}: {reading.value}{reading.unit}")
        time.sleep(0.1)
    
    # Show statistics
    print("\n" + "=" * 80)
    print("📈 Sensor Statistics")
    print("=" * 80)
    for sensor_id, sensor in pattern.sensors.items():
        stats = sensor.get_statistics()
        print(f"\n{sensor_id}:")
        for key, value in stats.items():
            print(f"  {key}: {value}")
    
    # Show alerts
    print("\n" + "=" * 80)
    print("⚠️  Alerts")
    print("=" * 80)
    alerts = pattern.get_all_alerts()
    if alerts:
        for alert in alerts[:10]:
            print(f"\n[{alert.severity.upper()}] {alert.sensor_id}")
            print(f"  Value: {alert.reading.value}{alert.reading.unit}")
            print(f"  Threshold: {alert.threshold_type} {alert.threshold_value}")
            print(f"  Message: {alert.message}")
    else:
        print("No alerts generated.")
    
    # Environment analysis
    print("\n" + "=" * 80)
    print("🌍 Environment Analysis")
    print("=" * 80)
    analysis = pattern.analyze_environment()
    print(analysis)


if __name__ == "__main__":
    main()

    

Agentic Architecture Pattern: Sensor Agent

Pattern 101: 101 Sensor Agent Pattern – Line-by-Line Explanation

Overview

Pattern 101: Sensor Agent Pattern

Agents that perceive and collect data from environment through sensors. Processes raw sensor data into meaningful information.

Code Structure Analysis

File Information

• Pattern Number: 101
• Pattern Name: 101 Sensor Agent Pattern
• File Path: patterns\perception_sensing\101_sensor_agent_pattern.py
• Total Lines: 213

Detailed Code Explanation

Module Docstring

"""
Pattern 101: Sensor Agent Pattern

Agents that perceive and collect data from environment through sensors.
Processes raw sensor data into meaningful information.
"""

Explanation: This module docstring describes the purpose and key features of the 101 Sensor Agent Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

SensorType (Line 17)

class SensorType(Enum):
    TEMPERATURE
    HUMIDITY
    PRESSURE
    LIGHT
    MOTION
    PROXIMITY
    TEXT
    IMAGE

Purpose: Types of sensors.

Values: – TEMPERATURE: Represents temperature – HUMIDITY: Represents humidity – PRESSURE: Represents pressure – LIGHT: Represents light – MOTION: Represents motion – PROXIMITY: Represents proximity – TEXT: Represents text – IMAGE: Represents image

Classes

SensorType (Line 17)

Description: Types of sensors.

SensorReading (Line 30)

Description: Single sensor reading.

SensorData (Line 41)

Description: Processed sensor data.

Sensor (Line 48)

Description: Individual sensor component.

Methods:

__init__ (Line 51)

def __init__(self, sensor_id, sensor_type, unit):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – sensor_id: Parameter for sensor_id – sensor_type: Parameter for sensor_type – unit: Parameter for unit

read (Line 57)

def read(self):
    """Simulate sensor reading."""

Purpose: Simulate sensor reading.

Parameters: – self: Instance reference

SensorAgent (Line 81)

Description: Agent that manages and processes sensors.

Methods:

__init__ (Line 84)

def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

add_sensor (Line 89)

def add_sensor(self, sensor):
    """Add a sensor to the agent."""

Purpose: Add a sensor to the agent.

Parameters: – self: Instance reference – sensor: Parameter for sensor

read_all_sensors (Line 94)

def read_all_sensors(self):
    """Read all sensors."""

Purpose: Read all sensors.

Parameters: – self: Instance reference

analyze_readings (Line 98)

def analyze_readings(self, readings):
    """Analyze sensor readings using LLM."""

Purpose: Analyze sensor readings using LLM.

Parameters: – self: Instance reference – readings: Parameter for readings

detect_anomalies (Line 114)

def detect_anomalies(self, readings):
    """Detect anomalies in sensor data."""

Purpose: Detect anomalies in sensor data.

Parameters: – self: Instance reference – readings: Parameter for readings

SensorAgentPattern (Line 129)

Description: Sensor Agent Pattern – Multi-sensor perception system.

Methods:

__init__ (Line 134)

def __init__(self, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

create_agent (Line 138)

def create_agent(self, name):
    """Create a sensor agent."""

Purpose: Create a sensor agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 144)

def run(self, agent_name):
    """Execute sensor reading and analysis."""

Purpose: Execute sensor reading and analysis.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = SensorType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities