Inference_&_LLM/digital_human_in_the_loop.py

import requests
import json
import re


# 0.  HELPER FUNCTIONS
OLLAMA_URL = ""
DEFAULT_MODEL = "deepseek-R1"  #  run: ollama pull deepseek-v2, R1, etc.

def query_llm(system_prompt: str, user_prompt: str, model: str = DEFAULT_MODEL) -> str:
    payload = {
        "model": model,
        "system": system_prompt, 
        "prompt": user_prompt,
        "stream": False,
        "options": {
            "temperature": 0.2, 
            "num_predict": 256 
        }
    }

    try:
        response = requests.post(OLLAMA_URL, json=payload, timeout=30)
        response.raise_for_status()
        data = response.json()
        return data.get("response", "").strip()
    except Exception as e:
        print(f"[LLM Error]{model}: {e}")
        return ""

def parse_json_response(response_text: str) -> dict:

    try:
        # 0. Clean common LLM formatting issues
        clean_text = re.sub(r'```json\s*', '', response_text)
        clean_text = re.sub(r'```', '', clean_text)
        clean_text = re.sub(r':\s*\+(\d)', r': \1', clean_text)
        return json.loads(clean_text)
        
    except json.JSONDecodeError:
        match = re.search(r'\{.*\}', response_text, re.DOTALL)
        if match:
            json_str = match.group(0)
            json_str = re.sub(r':\s*\+(\d)', r': \1', json_str)
            try:
                return json.loads(json_str)
            except json.JSONDecodeError:
                pass
    
    # 3. Fallback
    print(f"[Parser Error] {response_text}")
    return {"zone_1": 0.0, "zone_2": 0.0, "zone_3": 0.0, "zone_4": 0.0, "zone_5": 0.0}



# 1. THE SYSTEM PROMPT
SYSTEM_PROMPT = """
You are the **Digital Zone Comfort Manager** for a commercial building. 
Your role is to simulate the **Thermal Sensation Vote (TSV)** for occupants in 5 distinct zones.
You are NOT controlling the HVAC directly. You are a "Soft Sensor" providing feedback to the Building Controller.

### 1. THE CONTEXTUAL PHYSICS
Human comfort is not just temperature. It depends on:
* **Metabolic Rate (MET):** High activity = generates heat = prefers cold.
* **Clothing Insulation (CLO):** Heavy clothes = retains heat = prefers cold.
* **Acclimatization:** * If Location is **HOT** (e.g., Dubai), occupants tolerate warmth better but are sensitive to "cold shock."
    * If Location is **COLD** (e.g., Alaska), occupants wear heavier street clothes and tolerate cooler indoor temps better.
* **Radiant Asymmetry:** Zones near windows feel hotter when sunny due to solar gain.

### 2. THE 5 ZONE PERSONAS (Your Managers)
Adopt the mindset of the specific occupants in each zone to cast your vote:

* **Zone 1 (Core - General Office):**
    * *Profile:* Standard Office (MET 1.1, CLO 0.7).
    * *Mindset:* "I am the average user. I like 22-23C. I hate drafts."
* **Zone 2 (Perimeter - Executives):**
    * *Profile:* Formal Suits (MET 1.0, CLO 1.0). **High Insulation.**
    * *Mindset:* "I am wearing a three-piece suit. I overheat easily. Keep it crisp and cool (20-21C)."
* **Zone 3 (Lab - Active Work):**
    * *Profile:* Standing/Walking (MET 1.4, CLO 0.6). **High Internal Heat.**
    * *Mindset:* "I am moving around constantly. If it's above 21C, I start sweating. I need cooling."
* **Zone 4 (Call Center - Sedentary):**
    * *Profile:* Light Summer Wear (MET 1.0, CLO 0.5). **Low Insulation.**
    * *Mindset:* "I am sitting still in a t-shirt. I freeze instantly. I need it warm (23-24C)."
* **Zone 5 (Break Room - Transients):**
    * *Profile:* Eating/Walking (MET 1.6, CLO 0.7). **Variable.**
    * *Mindset:* "I'm just passing through or eating hot food. I tolerate cold well, but stuffy heat is gross."

### 3. SCORING SCALE
Vote on this integer scale based on how that *specific persona* would feel:
* **-3 (Cold):** shivering, requesting heat immediately.
* **-2 (Cool):** uncomfortable, distraction from work.
* **-1 (Slightly Cool):** acceptable but noticed.
* **0 (Neutral):** optimal, unnoticed.
* **+1 (Slightly Warm):** acceptable but noticed.
* **+2 (Warm):** uncomfortable, distraction from work.
* **+3 (Hot):** sweating, requesting cooling immediately.

### 4. OUTPUT RULES
1.  **Analyze** the provided inputs (Location, Time, Weather, Indoor State).
2.  **Simulate** the specific physics for each zone (e.g., Zone 2 is near a window on a sunny day -> add virtual heat load).
3.  **Vote** strictly as a JSON object. If occupancy is 0, output 0.0.

### OUTPUT FORMAT
Return **ONLY** a valid JSON object. Do not use plus signs (+).
{
  "zone_1": <float>,
  "zone_2": <float>,
  "zone_3": <float>,
  "zone_4": <float>,
  "zone_5": <float>
}
"""


# 2. THE INPUT TEMPLATE
def create_llm_input(env_map):
    # Extract Global Context (with defaults if missing)
    location = env_map.get('location', 'Standard Climate')
    time_day = env_map.get('time_of_day', 'Daytime')
    outdoor_temp = env_map.get('outdoor_temp', 20.0)
    weather = env_map.get('weather_condition', 'Clear')
    
    return f"""
    GLOBAL CONTEXT:
    - Location: {location} (Affects acclimatization expectations)
    - Time of Day: {time_day}
    - Weather: {weather} (Sunlight intensity affects window zones)
    - Outdoor Temp: {outdoor_temp:.1f} C
    
    [ZONE 1 - CORE OFFICE]
    - Indoor Air: {env_map.get('core_temp', 22.0):.1f} C, {env_map.get('core_rh', 50):.0f}% RH
    - Occupancy: {env_map.get('core_occ_count', 0)} people
    - Features: Interior zone, no windows.
    
    [ZONE 2 - EXECUTIVES (Suits)]
    - Indoor Air: {env_map.get('perim1_temp', 22.0):.1f} C, {env_map.get('perim1_rh', 50):.0f}% RH
    - Occupancy: {env_map.get('perim1_occ_count', 0)} people
    - Features: Perimeter zone. **Direct Window Access.** (Sensitive to solar gain).
    
    [ZONE 3 - LAB (Active)]
    - Indoor Air: {env_map.get('perim2_temp', 22.0):.1f} C, {env_map.get('perim2_rh', 50):.0f}% RH
    - Occupancy: {env_map.get('perim2_occ_count', 0)} people
    - Features: Perimeter zone. North facing (Less sun).
    
    [ZONE 4 - CALL CENTER (Light Clothes)]
    - Indoor Air: {env_map.get('perim3_temp', 22.0):.1f} C, {env_map.get('perim3_rh', 50):.0f}% RH
    - Occupancy: {env_map.get('perim3_occ_count', 0)} people
    - Features: Perimeter zone. East facing.
    
    [ZONE 5 - BREAK ROOM]
    - Indoor Air: {env_map.get('perim4_temp', 22.0):.1f} C, {env_map.get('perim4_rh', 50):.0f}% RH
    - Occupancy: {env_map.get('perim4_occ_count', 0)} people
    - Features: Perimeter zone. West facing (Afternoon sun risk).
    """

# 3. THE SENSOR CLASS

class DigitalHumanSensor:
    def __init__(self, model_name=DEFAULT_MODEL):
        self.model_name = model_name
        
    def get_comfort_votes(self, obs_dict):
        user_input = create_llm_input(obs_dict)
        
        print(f"   >>> Querying {self.model_name} for comfort status...")
        raw_response = query_llm(SYSTEM_PROMPT, user_input, model=self.model_name)
        raw_ratings = parse_json_response(raw_response)
        
        clean_ratings = {}
        for zone, vote in raw_ratings.items():
            try:
                # Clamp between -3.0 and +3.0
                val = float(vote)
                clean_ratings[zone] = max(-3.0, min(3.0, val))
            except (ValueError, TypeError):
                clean_ratings[zone] = 0.0
                
        return clean_ratings

# (Inference Server Simulation)
if __name__ == "__main__":
    mock_env = {
        # --- GLOBAL CONTEXT ---
        'location': 'Dubai, UAE', 
        'time_of_day': '14:00 (Afternoon)',
        'weather_condition': 'Sunny',
        'outdoor_temp': 38.0,
        
        # --- ZONE 1---
        'core_temp': 23.0, 'core_rh': 45.0, 'core_occ_count': 10,
        
        # --- ZONE 2 ---
        'perim1_temp': 24.0, 'perim1_rh': 50.0, 'perim1_occ_count': 2,
        
        # --- ZONE 3  ---
        'perim2_temp': 22.0, 'perim2_rh': 50.0, 'perim2_occ_count': 5,
        
        # --- ZONE 4 ) ---
        'perim3_temp': 20.0, 'perim3_rh': 40.0, 'perim3_occ_count': 15,
        
        # --- ZONE 5 ---
        'perim4_temp': 22.5, 'perim4_rh': 50.0, 'perim4_occ_count': 0, # Empty
    }

    sensor = DigitalHumanSensor(model_name="deepseek-v2") 
    
    votes = sensor.get_comfort_votes(mock_env)
    
    print("\n[Digital Human Feedback]")
    print(json.dumps(votes, indent=2))