Merlin 1.2.1: Fixed Ruff checks and applied single responsability principal to lights setup.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "Merlin"
-version = "1.2.0"
+version = "1.2.1"
 description = "Merlin - Your AI Assistant with multi-agent architecture"
 readme = "README.md"
 requires-python = ">=3.14"

src/config/app_config.py

@@ -1,7 +1,8 @@
 from pathlib import Path
-import yaml
 from types import SimpleNamespace
 
+import yaml
+
 
 class Config:
     """Main Config Class for application-level configuration."""
@@ -183,13 +184,22 @@ class Weather:
 
     @classmethod  # Load from YAML
     def load_yaml(cls, file_path="src/config/config.yaml"):
+        """Load configuration from YAML file.
+
+        Args:
+            file_path (str): Path to the YAML configuration file.
+
+        Returns:
+            AppConfig: Configuration instance with loaded settings.
+
+        """
         yaml_file = Path(file_path)
         if not yaml_file.exists():
             default_dict = {"DEBUG": True}
             with open(yaml_file, "w") as f:
                 yaml.dump(default_dict, f)
 
-        with open(yaml_file, "r") as f:
+        with open(yaml_file) as f:
             config_data = yaml.safe_load(f)
 
         # Populate lights and rooms

src/config/user_config_template.py

@@ -15,12 +15,12 @@ class Model:
         # Uncomment and modify as needed
 
         # Base model overrides (affects all agents unless specifically overridden)
-        PROVIDER = 'lm_studio'
-        MODEL_NAME = 'openai/gpt-oss-20b'
-        HOST_ADDRESS = '127.0.0.1'
-        HOST_PORT = '1234'
-        HOST_API_KEY = 'your-personal-key'
-        HOST_API_PATH = 'v1'
+        PROVIDER = "lm_studio"
+        MODEL_NAME = "openai/gpt-oss-20b"
+        HOST_ADDRESS = "127.0.0.1"
+        HOST_PORT = "1234"
+        HOST_API_KEY = "your-personal-key"
+        HOST_API_PATH = "v1"
 
         # Orchestrator personal config
         ORCHESTRATOR = {

src/experts/memory.py

@@ -1,30 +1,36 @@
 import os
-import yaml
-from typing import Any, Optional, List, Dict
+from typing import Any
+
 import dspy
+import yaml
 
 
 class MemoryExpertSignature(dspy.Signature):
-    """You are the long term memory expert responsible for managing memories that persist across sessions
-    Important, your keys should always be lower case and snake_case to ensure consistency.
+    """You are the long term memory expert responsible for managing memories.
+
+    Important, your keys should always be lower case and snake_case to
+    ensure consistency.
     Your role:
     - memory management, adding, removing and updating memories
     - returning memory content to the requestor
-    - Confirm whether tasks were completed successfully or failed"""
+    - Confirm whether tasks were completed successfully or failed
+    """
 
     command: str = dspy.InputField(
         desc="A natural language command describing what you should do with the memory"
     )
     answer: str = dspy.OutputField(
-        desc="A confirmation message indicating success or failure of the requested task"
+        desc="A confirmation message indicating success or failure of the "
+        "requested task"
     )
 
 
 class MemoryAgent(dspy.Module):
-    """A Lighting Agent that has access to Light based tools"""
+    """A Memory Agent that has access to memory-based tools."""
 
     def __init__(self):
-        self.memory_file_path = 'src/config/memory.yaml'
+        """Initialize the Memory Agent with configuration and tools."""
+        self.memory_file_path = "src/config/memory.yaml"
         self.tools = [
             self.add_or_update_memory,
             self.load_memory,
@@ -37,55 +43,55 @@ def __init__(self):
         )
 
     def add_or_update_memory(self, key: str, value: Any) -> bool:
-        """
-        Save or update a memory to long term storage
+        """Save or update a memory to long term storage.
 
         Args:
             key (str): The key to store/update
             value (any): The value to store
 
         Returns:
             bool: True if successful, False if failed
+
         """
         memory_content = self._load_memory_file()
         memory_content[key] = value
         succsess = self._save_memory_file(memory_content)
         return succsess
 
-    def load_memory(self, key: str) -> Optional[Any]:
-        """
-        Load a memory value from long term storage by key
+    def load_memory(self, key: str) -> Any | None:
+        """Load a memory value from long term storage by key.
 
         Args:
             key (str): The key to retrieve
 
         Returns:
             Any: The stored value if found, None if key doesn't exist
+
         """
         memory_content = self._load_memory_file()
 
         return memory_content.get(key)
 
-    def list_memory_keys(self) -> Optional[List[str]]:
-        """
-        List all keys from long term storage
+    def list_memory_keys(self) -> list[str] | None:
+        """List all keys from long term storage.
 
         Returns:
             List[str]: List of all keys if successful, None if failed
+
         """
         memory_content = self._load_memory_file()
 
         return list(memory_content.keys())
 
     def remove_memory(self, key: str) -> bool:
-        """
-        Remove a key-value pair from long term storage
+        """Remove a key-value pair from long term storage.
 
         Args:
             key (str): The key to remove
 
         Returns:
             bool: True if successful, False if failed
+
         """
         memory_content = self._load_memory_file()
         # Remove the key if it exists
@@ -94,16 +100,16 @@ def remove_memory(self, key: str) -> bool:
         succsess = self._save_memory_file(memory_content)
         return succsess
 
-    def load_all_memories(self) -> Dict[str, Any]:
-        "shows all things stored in long term memory"
+    def load_all_memories(self) -> dict[str, Any]:
+        """Return all things stored in long term memory."""
         memory_content = self._load_memory_file()
         return memory_content
 
-    def _load_memory_file(self) -> Dict[str, Any]:
-        memory_file_path = 'src/config/memory.yaml'
+    def _load_memory_file(self) -> dict[str, Any]:
+        memory_file_path = "src/config/memory.yaml"
         # Load existing memory content
         if os.path.exists(memory_file_path):
-            with open(memory_file_path, "r") as file:
+            with open(memory_file_path) as file:
                 memory_content = yaml.safe_load(file) or {}
         else:
             memory_content = {}

src/experts/orchestrator.py

@@ -4,8 +4,8 @@
 
 from .game import GameAgent
 from .lights import LightingAgent
-from .weather import WeatherAgent
 from .memory import MemoryAgent
+from .weather import WeatherAgent
 
 
 class OrchestratorSignature(dspy.Signature):
@@ -67,9 +67,13 @@ def consult_lighting_expert(self, command: str) -> str:
             return result.answer
 
     def consult_memory_expert(self, command: str) -> str:
-        """Use this expert when you want to save or retrieve any information that should be stored for future use.
-        Use this expert whenever you encounter information that should persist beyond the current interaction
-        Use this expert freely and proactively - it's designed to help you maintain a persistent knowledge base.
+        """Use this expert when you want to save or retrieve information.
+
+        Use this expert when you want to save or retrieve any information
+        that should be stored for future use. Use this expert whenever you
+        encounter information that should persist beyond the current
+        interaction. Use this expert freely and proactively - it's designed
+        to help you maintain a persistent knowledge base.
         """
         with dspy.context(lm=ModelFactory.create_memory_model()):
             result = MemoryAgent().memory_agent(command=command)

src/main.py

@@ -160,7 +160,7 @@ def setup_logging():
 
     # Create a file handler with rotation every 5MB
     file_handler = RotatingFileHandler(
-        "merlin.log", maxBytes=5*1024*1024, backupCount=3
+        "merlin.log", maxBytes=5 * 1024 * 1024, backupCount=3
     )
     file_handler.setLevel(logging.DEBUG)
 

src/setup/lights/hue_bridge_set_up.py

@@ -10,16 +10,14 @@
 
 
 def is_bridge_is_set_up() -> bool:
-    """Check if the bridge is already set up.
+    """Check if Hue bridge is properly configured.
 
     Returns False if not set up
     Returns True if set up
     """
     load_dotenv()
     bridge_user = os.getenv("HUE_CLIENTKEY")
-    if not bridge_user:
-        return False
-    return True
+    return bool(bridge_user)
 
 
 def get_hue_bridge_ip() -> str:
@@ -35,7 +33,9 @@ def get_hue_bridge_ip() -> str:
 def wait_for_hue_bridge_link_button(
     bridge_ip, max_attempts=30, interval=2, save_to_env=True
 ):
-    """Wait for user to press the Hue bridge link button and return the success response.
+    """Wait for user to press the Hue bridge link button.
+
+    Return the success response.
 
     Args:
         bridge_ip (str): IP address of the Hue bridge
@@ -82,7 +82,8 @@ def wait_for_hue_bridge_link_button(
                 # Check for error response (link button not pressed)
                 elif "error" in first_item and first_item["error"]["type"] == 101:
                     print(
-                        f"Link button not pressed yet. Attempt {attempt + 1}/{max_attempts}"
+                        f"Link button not pressed yet. "
+                        f"Attempt {attempt + 1}/{max_attempts}"
                     )
                     time.sleep(interval)
                     continue
@@ -114,6 +115,7 @@ def save_credentials_to_env(credentials, home_ip_address):
 
     Args:
         credentials (dict): Dictionary containing 'username' and 'clientkey'
+        home_ip_address (str): IP address of the Hue bridge
 
     """
     # Create or update .env file