"""Quantum Emotion Service - Background service for quantum-influenced emotion updates""" import os import sys import time import logging import requests import threading from collections import deque # Add parent directory to path for imports sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import MODEL_CONFIG class QuantumEmotionService: """Background service that collects quantum random numbers and applies them to emotions""" def __init__(self, emotional_agent, config=None): self.config = config or MODEL_CONFIG or {} self.emotional_agent = emotional_agent self.quantum_api_key = os.getenv("ANU_QUANTUM_API_KEY") self.running = False self.thread = None self.quantum_numbers = deque(maxlen=6) # Store last 6 numbers self.last_update_time = time.time() self.update_interval = 10.0 # Update emotions every 10 seconds self.call_interval = 1.0 # Call API once per second if not self.quantum_api_key: logging.info("[QuantumEmotionService] Quantum API key not found - service will not run") return quantum_config = self.config.get('quantum', {}) if self.config else {} self.api_endpoint = quantum_config.get('api_endpoint', 'https://api.quantumnumbers.anu.edu.au') logging.info("[QuantumEmotionService] Initialized - will call API once per second") def _fetch_quantum_number(self): """Fetch a single quantum random number from the API""" if not self.quantum_api_key: return None try: headers = {"x-api-key": self.quantum_api_key} params = {"length": 1, "type": "uint8"} response = requests.get(self.api_endpoint, headers=headers, params=params, timeout=5) if response.status_code == 200: result = response.json() if result.get('success') and 'data' in result and len(result['data']) > 0: # Normalize to 0-1 range normalized = result['data'][0] / 255.0 return normalized elif response.status_code == 429: # Rate limited - return None, will use pseudo-random logging.debug("[QuantumEmotionService] Rate limited, using pseudo-random") return None else: logging.debug(f"[QuantumEmotionService] API returned {response.status_code}") return None except Exception as e: logging.debug(f"[QuantumEmotionService] API call failed: {e}") return None return None def _apply_quantum_emotions(self): """Apply the collected 6 quantum numbers to influence emotion state""" if len(self.quantum_numbers) < 6: logging.debug(f"[QuantumEmotionService] Not enough numbers yet ({len(self.quantum_numbers)}/6)") return # Convert deque to list numbers = list(self.quantum_numbers) # Map 6 numbers to 5 emotions + overall variation # numbers[0] -> joy # numbers[1] -> sadness # numbers[2] -> anger # numbers[3] -> fear # numbers[4] -> curiosity # numbers[5] -> overall variation factor emotions = ["joy", "sadness", "anger", "fear", "curiosity"] current_state = self.emotional_agent.get_state() # Apply quantum influence (subtle changes, -0.05 to +0.05 range) for i, emotion in enumerate(emotions): if i < len(numbers) and numbers[i] is not None: # Convert 0-1 to -0.05 to +0.05 influence influence = (numbers[i] - 0.5) * 0.1 # Scale to ±0.05 # Apply with overall variation factor if len(numbers) > 5 and numbers[5] is not None: variation = (numbers[5] - 0.5) * 0.02 # Additional ±0.01 variation influence += variation # Update emotion new_value = current_state[emotion] + influence current_state[emotion] = max(0.05, min(1.0, new_value)) # Soft normalization to keep emotions balanced (only if they get too extreme) total = sum(current_state.values()) if total > 1.5 or total < 0.5: # Only normalize if emotions are way out of balance for emotion in emotions: current_state[emotion] = current_state[emotion] / total if total > 0 else 0.2 else: # Just ensure no emotion goes below minimum threshold for emotion in emotions: if current_state[emotion] < 0.05: current_state[emotion] = 0.05 # Update the emotional agent's state self.emotional_agent.emotional_state = current_state # Save to JSON after quantum update self.emotional_agent._save_to_json() logging.info(f"[QuantumEmotionService] Applied quantum influence: {current_state}") def _service_loop(self): """Main service loop - calls API once per second, updates emotions every 10 seconds""" logging.info("[QuantumEmotionService] Service started") while self.running: try: # Fetch quantum number (once per second) quantum_num = self._fetch_quantum_number() if quantum_num is not None: self.quantum_numbers.append(quantum_num) logging.debug(f"[QuantumEmotionService] Collected quantum number: {quantum_num:.4f}") else: # Use pseudo-random as fallback import random pseudo_random = random.random() self.quantum_numbers.append(pseudo_random) logging.debug(f"[QuantumEmotionService] Using pseudo-random: {pseudo_random:.4f}") # Check if it's time to update emotions (every 10 seconds) current_time = time.time() if current_time - self.last_update_time >= self.update_interval: if len(self.quantum_numbers) >= 6: self._apply_quantum_emotions() self.last_update_time = current_time logging.info(f"[QuantumEmotionService] Updated emotions with {len(self.quantum_numbers)} quantum numbers") # Wait 1 second before next call time.sleep(self.call_interval) except Exception as e: logging.error(f"[QuantumEmotionService] Error in service loop: {e}") time.sleep(self.call_interval) # Wait before retrying def start(self): """Start the background service""" if not self.quantum_api_key: logging.info("[QuantumEmotionService] Cannot start - no API key") return False if self.running: logging.warning("[QuantumEmotionService] Already running") return False self.running = True self.thread = threading.Thread(target=self._service_loop, daemon=True) self.thread.start() logging.info("[QuantumEmotionService] Background service started") return True def stop(self): """Stop the background service""" if not self.running: return self.running = False if self.thread: self.thread.join(timeout=2.0) logging.info("[QuantumEmotionService] Background service stopped") def is_running(self): """Check if service is running""" return self.running