The Command Pattern is a behavioral design pattern that encapsulates a request as an object, thereby allowing you to parameterize clients with different requests, queue operations, and support undo functionality. This pattern decouples the object that invokes the operation from the object that performs it.
Core Concept
The fundamental idea behind the Command Pattern is to wrap a request or action in an object. This object contains all the information needed to perform the action, including the method to call, the method’s arguments, and a reference to the object that owns the method.
Key Components
The Command Pattern typically involves four main components:
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Client │───▶│ Invoker │───▶│ Command │───▶│ Receiver │
│ │ │ │ │ (Interface) │ │ │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
△
│
┌─────────────┐
│ Concrete │
│ Command │
└─────────────┘
Client: Creates and configures concrete command objects Invoker: Asks the command to carry out the request Command: Declares an interface for executing operations Receiver: Knows how to perform the operations associated with carrying out a request ConcreteCommand: Implements the Command interface and defines the binding between a Receiver and an action
When to Use the Command Pattern
The Command Pattern is particularly useful when you need to:
- Decouple the object that invokes the operation from the object that performs it
- Queue operations, schedule their execution, or execute them remotely
- Support undo/redo functionality
- Log operations for crash recovery
- Structure a system around high-level operations built on primitive operations
Benefits
Decoupling: The pattern separates the object that invokes the operation from the object that performs it, reducing dependencies between classes.
Flexibility: New commands can be added without changing existing code, following the Open/Closed Principle.
Macro Commands: You can assemble simple commands into complex ones by creating composite commands.
Undo/Redo Support: Commands can store state information to reverse their effect, enabling undo functionality.
Logging and Recovery: Operations can be logged and replayed for crash recovery or auditing purposes.
Potential Drawbacks
Increased Complexity: The pattern introduces additional classes and objects, which may increase the overall complexity of the codebase.
Memory Overhead: Each command object requires memory allocation, which could be a concern in memory-constrained environments.
Performance Impact: The additional layer of abstraction may introduce slight performance overhead, though this is typically negligible.
Implementation Strategy
When implementing the Command Pattern, consider the following approach:
- Define a common interface for all commands
- Create concrete command classes that implement this interface
- Each concrete command should hold a reference to a receiver object
- Implement the command’s execute method to call appropriate methods on the receiver
- Create an invoker class that can store and execute commands
- For undo functionality, add an undo method to the command interface
The Command Pattern promotes loose coupling and provides a clean way to handle complex operations while maintaining the flexibility to extend functionality without modifying existing code.This basic implementation demonstrates the fundamental concepts of the Command Pattern. The RemoteControl class acts as the invoker, while Light and Fan are receivers that know how to perform the actual operations. Each command encapsulates a specific request and stores the necessary state information to support undo functionality.
from abc import ABC, abstractmethod
from typing import List, Optional
# Command interface
class Command(ABC):
"""Abstract base class for all commands.
Defines the contract that all concrete commands must follow.
"""
@abstractmethod
def execute(self) -> None:
"""Execute the command."""
pass
@abstractmethod
def undo(self) -> None:
"""Undo the command (reverse its effect)."""
pass
# Receiver classes - these know how to perform the actual work
class Light:
"""A simple light that can be turned on/off and dimmed."""
def __init__(self, location: str) -> None:
self.location = location
self.is_on = False
self.brightness = 0 # 0-100
def turn_on(self) -> None:
"""Turn the light on."""
self.is_on = True
self.brightness = 100
print(f"{self.location} light is ON (brightness: {self.brightness}%)")
def turn_off(self) -> None:
"""Turn the light off."""
self.is_on = False
self.brightness = 0
print(f"{self.location} light is OFF")
def dim(self, brightness: int) -> None:
"""Set the light brightness (0-100)."""
if 0 <= brightness <= 100:
self.brightness = brightness
self.is_on = brightness > 0
status = "ON" if self.is_on else "OFF"
print(f"{self.location} light is {status} (brightness: {self.brightness}%)")
class Fan:
"""A simple fan that can be turned on/off and speed adjusted."""
def __init__(self, location: str) -> None:
self.location = location
self.is_on = False
self.speed = 0 # 0-3
def turn_on(self) -> None:
"""Turn the fan on at medium speed."""
self.is_on = True
self.speed = 2
print(f"{self.location} fan is ON (speed: {self.speed})")
def turn_off(self) -> None:
"""Turn the fan off."""
self.is_on = False
self.speed = 0
print(f"{self.location} fan is OFF")
def set_speed(self, speed: int) -> None:
"""Set fan speed (0-3, where 0 is off)."""
if 0 <= speed <= 3:
self.speed = speed
self.is_on = speed > 0
if self.is_on:
print(f"{self.location} fan speed set to {self.speed}")
else:
print(f"{self.location} fan is OFF")
# Concrete Command classes
class LightOnCommand(Command):
"""Command to turn a light on."""
def __init__(self, light: Light) -> None:
self._light = light
self._previous_state = False
self._previous_brightness = 0
def execute(self) -> None:
"""Execute the light on command."""
# Store previous state for undo
self._previous_state = self._light.is_on
self._previous_brightness = self._light.brightness
# Execute the command
self._light.turn_on()
def undo(self) -> None:
"""Undo the light on command."""
if self._previous_state:
self._light.dim(self._previous_brightness)
else:
self._light.turn_off()
class LightOffCommand(Command):
"""Command to turn a light off."""
def __init__(self, light: Light) -> None:
self._light = light
self._previous_state = False
self._previous_brightness = 0
def execute(self) -> None:
"""Execute the light off command."""
# Store previous state for undo
self._previous_state = self._light.is_on
self._previous_brightness = self._light.brightness
# Execute the command
self._light.turn_off()
def undo(self) -> None:
"""Undo the light off command."""
if self._previous_state:
self._light.dim(self._previous_brightness)
class FanOnCommand(Command):
"""Command to turn a fan on."""
def __init__(self, fan: Fan) -> None:
self._fan = fan
self._previous_state = False
self._previous_speed = 0
def execute(self) -> None:
"""Execute the fan on command."""
# Store previous state for undo
self._previous_state = self._fan.is_on
self._previous_speed = self._fan.speed
# Execute the command
self._fan.turn_on()
def undo(self) -> None:
"""Undo the fan on command."""
if self._previous_state:
self._fan.set_speed(self._previous_speed)
else:
self._fan.turn_off()
class FanOffCommand(Command):
"""Command to turn a fan off."""
def __init__(self, fan: Fan) -> None:
self._fan = fan
self._previous_state = False
self._previous_speed = 0
def execute(self) -> None:
"""Execute the fan off command."""
# Store previous state for undo
self._previous_state = self._fan.is_on
self._previous_speed = self._fan.speed
# Execute the command
self._fan.turn_off()
def undo(self) -> None:
"""Undo the fan off command."""
if self._previous_state:
self._fan.set_speed(self._previous_speed)
# No Operation Command (Null Object Pattern)
class NoCommand(Command):
"""A null command that does nothing - useful for initialization."""
def execute(self) -> None:
"""Do nothing."""
pass
def undo(self) -> None:
"""Do nothing."""
pass
# Invoker class
class RemoteControl:
"""A simple remote control that can store and execute commands.
This acts as the invoker in the Command Pattern.
"""
def __init__(self, slots: int = 7) -> None:
# Initialize all slots with NoCommand
self._on_commands: List[Command] = [NoCommand() for _ in range(slots)]
self._off_commands: List[Command] = [NoCommand() for _ in range(slots)]
self._last_command: Optional[Command] = None
def set_command(self, slot: int, on_command: Command, off_command: Command) -> None:
"""Set commands for a specific slot."""
if 0 <= slot < len(self._on_commands):
self._on_commands[slot] = on_command
self._off_commands[slot] = off_command
else:
raise ValueError(f"Slot {slot} is out of range")
def press_on_button(self, slot: int) -> None:
"""Press the ON button for a specific slot."""
if 0 <= slot < len(self._on_commands):
self._on_commands[slot].execute()
self._last_command = self._on_commands[slot]
else:
print(f"Invalid slot: {slot}")
def press_off_button(self, slot: int) -> None:
"""Press the OFF button for a specific slot."""
if 0 <= slot < len(self._off_commands):
self._off_commands[slot].execute()
self._last_command = self._off_commands[slot]
else:
print(f"Invalid slot: {slot}")
def press_undo_button(self) -> None:
"""Press the undo button to reverse the last command."""
if self._last_command:
self._last_command.undo()
else:
print("No command to undo")
def __str__(self) -> str:
"""String representation of the remote control."""
result = "\n------ Remote Control ------\n"
for i in range(len(self._on_commands)):
on_cmd = self._on_commands[i].__class__.__name__
off_cmd = self._off_commands[i].__class__.__name__
result += f"[slot {i}] {on_cmd:<15} {off_cmd}\n"
return result
# Example usage and demonstration
def main() -> None:
"""Demonstrate the Command Pattern with a remote control example."""
print("=== Command Pattern Demo ===\n")
# Create receiver objects
living_room_light = Light("Living Room")
kitchen_light = Light("Kitchen")
bedroom_fan = Fan("Bedroom")
# Create command objects
living_room_light_on = LightOnCommand(living_room_light)
living_room_light_off = LightOffCommand(living_room_light)
kitchen_light_on = LightOnCommand(kitchen_light)
kitchen_light_off = LightOffCommand(kitchen_light)
bedroom_fan_on = FanOnCommand(bedroom_fan)
bedroom_fan_off = FanOffCommand(bedroom_fan)
# Create the invoker (remote control)
remote = RemoteControl()
# Configure the remote control
remote.set_command(0, living_room_light_on, living_room_light_off)
remote.set_command(1, kitchen_light_on, kitchen_light_off)
remote.set_command(2, bedroom_fan_on, bedroom_fan_off)
# Display remote control configuration
print(remote)
# Test the commands
print("Testing commands:")
print("1. Turning on living room light...")
remote.press_on_button(0)
print("\n2. Turning on kitchen light...")
remote.press_on_button(1)
print("\n3. Turning on bedroom fan...")
remote.press_on_button(2)
print("\n4. Turning off kitchen light...")
remote.press_off_button(1)
print("\n5. Testing undo functionality...")
print("Undoing last command (kitchen light off):")
remote.press_undo_button()
print("\nUndoing again (bedroom fan on):")
remote.press_undo_button()
print("\nUndoing again (kitchen light on):")
remote.press_undo_button()
if __name__ == "__main__":
main()
The example shows how commands can be easily configured, executed, and undone without the invoker needing to know the specifics of how each operation is performed. This creates a flexible system where new commands can be added without modifying existing code.This macro commands implementation demonstrates how the Command Pattern can be extended to create complex operations by combining simpler commands. The MacroCommand class itself implements the Command interface, making it interchangeable with simple commands from the invoker’s perspective.
from abc import ABC, abstractmethod
from typing import List
# Base Command interface (same as before)
class Command(ABC):
"""Abstract base class for all commands."""
@abstractmethod
def execute(self) -> None:
"""Execute the command."""
pass
@abstractmethod
def undo(self) -> None:
"""Undo the command."""
pass
# Simple receiver classes for demonstration
class Light:
"""A light that can be controlled."""
def __init__(self, location: str) -> None:
self.location = location
self.is_on = False
def turn_on(self) -> None:
self.is_on = True
print(f"{self.location} light turned ON")
def turn_off(self) -> None:
self.is_on = False
print(f"{self.location} light turned OFF")
class Stereo:
"""A stereo system that can be controlled."""
def __init__(self, location: str) -> None:
self.location = location
self.is_on = False
self.volume = 0
self.cd_inserted = False
def turn_on(self) -> None:
self.is_on = True
print(f"{self.location} stereo turned ON")
def turn_off(self) -> None:
self.is_on = False
self.volume = 0
print(f"{self.location} stereo turned OFF")
def set_cd(self) -> None:
self.cd_inserted = True
print(f"{self.location} stereo: CD inserted")
def set_volume(self, volume: int) -> None:
if self.is_on:
self.volume = volume
print(f"{self.location} stereo volume set to {volume}")
class TV:
"""A television that can be controlled."""
def __init__(self, location: str) -> None:
self.location = location
self.is_on = False
self.channel = 1
def turn_on(self) -> None:
self.is_on = True
print(f"{self.location} TV turned ON")
def turn_off(self) -> None:
self.is_on = False
print(f"{self.location} TV turned OFF")
def set_input_channel(self, channel: int) -> None:
if self.is_on:
self.channel = channel
print(f"{self.location} TV set to channel {channel}")
# Simple command implementations
class LightOnCommand(Command):
"""Command to turn a light on."""
def __init__(self, light: Light) -> None:
self._light = light
self._previous_state = False
def execute(self) -> None:
self._previous_state = self._light.is_on
self._light.turn_on()
def undo(self) -> None:
if not self._previous_state:
self._light.turn_off()
class LightOffCommand(Command):
"""Command to turn a light off."""
def __init__(self, light: Light) -> None:
self._light = light
self._previous_state = False
def execute(self) -> None:
self._previous_state = self._light.is_on
self._light.turn_off()
def undo(self) -> None:
if self._previous_state:
self._light.turn_on()
class StereoOnCommand(Command):
"""Command to turn stereo on with CD and volume."""
def __init__(self, stereo: Stereo, volume: int = 11) -> None:
self._stereo = stereo
self._volume = volume
self._previous_state = False
self._previous_volume = 0
def execute(self) -> None:
self._previous_state = self._stereo.is_on
self._previous_volume = self._stereo.volume
self._stereo.turn_on()
self._stereo.set_cd()
self._stereo.set_volume(self._volume)
def undo(self) -> None:
if not self._previous_state:
self._stereo.turn_off()
else:
self._stereo.set_volume(self._previous_volume)
class StereoOffCommand(Command):
"""Command to turn stereo off."""
def __init__(self, stereo: Stereo) -> None:
self._stereo = stereo
self._previous_state = False
self._previous_volume = 0
def execute(self) -> None:
self._previous_state = self._stereo.is_on
self._previous_volume = self._stereo.volume
self._stereo.turn_off()
def undo(self) -> None:
if self._previous_state:
self._stereo.turn_on()
self._stereo.set_volume(self._previous_volume)
class TVOnCommand(Command):
"""Command to turn TV on."""
def __init__(self, tv: TV, channel: int = 3) -> None:
self._tv = tv
self._channel = channel
self._previous_state = False
self._previous_channel = 1
def execute(self) -> None:
self._previous_state = self._tv.is_on
self._previous_channel = self._tv.channel
self._tv.turn_on()
self._tv.set_input_channel(self._channel)
def undo(self) -> None:
if not self._previous_state:
self._tv.turn_off()
else:
self._tv.set_input_channel(self._previous_channel)
class TVOffCommand(Command):
"""Command to turn TV off."""
def __init__(self, tv: TV) -> None:
self._tv = tv
self._previous_state = False
self._previous_channel = 1
def execute(self) -> None:
self._previous_state = self._tv.is_on
self._previous_channel = self._tv.channel
self._tv.turn_off()
def undo(self) -> None:
if self._previous_state:
self._tv.turn_on()
self._tv.set_input_channel(self._previous_channel)
# Macro Command - This is the key concept
class MacroCommand(Command):
"""
A command that executes a sequence of other commands.
This allows creating complex operations by combining simpler commands.
MacroCommand follows the Composite Pattern as well.
"""
def __init__(self, commands: List[Command]) -> None:
"""Initialize with a list of commands to execute."""
self._commands = commands.copy() # Create a copy to avoid external modifications
def execute(self) -> None:
"""Execute all commands in sequence."""
print("--- Executing Macro Command ---")
for command in self._commands:
command.execute()
print("--- Macro Command Completed ---")
def undo(self) -> None:
"""Undo all commands in reverse order."""
print("--- Undoing Macro Command ---")
# Important: Undo in reverse order
for command in reversed(self._commands):
command.undo()
print("--- Macro Command Undo Completed ---")
def add_command(self, command: Command) -> None:
"""Add a command to the macro (optional convenience method)."""
self._commands.append(command)
def remove_command(self, command: Command) -> None:
"""Remove a command from the macro (optional convenience method)."""
if command in self._commands:
self._commands.remove(command)
# Enhanced Remote Control that supports macro commands
class RemoteControl:
"""Remote control that can handle both simple and macro commands."""
def __init__(self) -> None:
self._commands = {}
self._last_command = None
def set_command(self, slot: str, command: Command) -> None:
"""Set a command for a named slot."""
self._commands[slot] = command
def press_button(self, slot: str) -> None:
"""Press a button to execute a command."""
if slot in self._commands:
print(f"\nPressed {slot} button:")
self._commands[slot].execute()
self._last_command = self._commands[slot]
else:
print(f"No command set for slot: {slot}")
def press_undo(self) -> None:
"""Press undo button to reverse the last command."""
if self._last_command:
print("\nPressed UNDO button:")
self._last_command.undo()
else:
print("\nNo command to undo")
def get_available_commands(self) -> List[str]:
"""Get list of available command slots."""
return list(self._commands.keys())
def main() -> None:
"""Demonstrate Macro Commands with a home automation scenario."""
print("=== Macro Command Pattern Demo ===\n")
# Create receiver objects (devices)
living_room_light = Light("Living Room")
kitchen_light = Light("Kitchen")
stereo = Stereo("Living Room")
tv = TV("Living Room")
# Create individual commands
living_room_light_on = LightOnCommand(living_room_light)
living_room_light_off = LightOffCommand(living_room_light)
kitchen_light_on = LightOnCommand(kitchen_light)
kitchen_light_off = LightOffCommand(kitchen_light)
stereo_on = StereoOnCommand(stereo, volume=15)
stereo_off = StereoOffCommand(stereo)
tv_on = TVOnCommand(tv, channel=5)
tv_off = TVOffCommand(tv)
# Create macro commands for different scenarios
# Party mode: Turn on lights, stereo, and TV
party_on_commands = [
living_room_light_on,
kitchen_light_on,
stereo_on,
tv_on
]
party_on_macro = MacroCommand(party_on_commands)
# Party off: Turn everything off
party_off_commands = [
living_room_light_off,
kitchen_light_off,
stereo_off,
tv_off
]
party_off_macro = MacroCommand(party_off_commands)
# Movie mode: Dim lights (turn off kitchen), turn on TV and stereo
movie_mode_commands = [
kitchen_light_off, # Dim environment
tv_on,
stereo_on
]
movie_mode_macro = MacroCommand(movie_mode_commands)
# Create and configure remote control
remote = RemoteControl()
remote.set_command("party_on", party_on_macro)
remote.set_command("party_off", party_off_macro)
remote.set_command("movie_mode", movie_mode_macro)
remote.set_command("living_room_light", living_room_light_on)
# Display available commands
print(f"Available commands: {remote.get_available_commands()}\n")
# Demonstrate macro commands
print("🎉 Starting party mode...")
remote.press_button("party_on")
print("\n⏯️ Switching to movie mode...")
remote.press_button("movie_mode")
print("\n⏪ Undoing movie mode...")
remote.press_undo()
print("\n🔚 Ending party...")
remote.press_button("party_off")
print("\n⏪ Undoing party off (should restore party mode)...")
remote.press_undo()
print("\n🔚 Final cleanup - turning everything off...")
remote.press_button("party_off")
# Demonstrate creating dynamic macro commands
print("\n" + "="*50)
print("Creating a custom macro command dynamically:")
custom_macro = MacroCommand([])
custom_macro.add_command(living_room_light_on)
custom_macro.add_command(tv_on)
remote.set_command("custom", custom_macro)
print("Executing custom macro...")
remote.press_button("custom")
print("\nUndoing custom macro...")
remote.press_undo()
if __name__ == "__main__":
main()
Key features of this implementation:
Composite Structure: MacroCommand contains a list of other commands, following the Composite Pattern principles.
Sequential Execution: Commands are executed in the order they were added to the macro.
Reverse Undo: When undoing a macro command, the individual commands are undone in reverse order to properly restore the previous state.
Flexibility: Macro commands can be built dynamically, and new commands can be added or removed as needed.
This approach allows creating sophisticated automation scenarios like “party mode” or “movie mode” that involve multiple devices, while still maintaining the ability to undo the entire sequence of operations with a single command.
This advanced implementation demonstrates how the Command Pattern can be extended to support command queuing and logging in a real-world application. Here are the key features:
import time
import json
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from datetime import datetime
from queue import Queue
from threading import Thread, Lock
from dataclasses import dataclass, asdict
# Base Command interface with serialization support
class Command(ABC):
"""Abstract base class for all commands with serialization support."""
@abstractmethod
def execute(self) -> None:
"""Execute the command."""
pass
@abstractmethod
def undo(self) -> None:
"""Undo the command."""
pass
@abstractmethod
def serialize(self) -> Dict[str, Any]:
"""Serialize command to dictionary for logging/storage."""
pass
@classmethod
@abstractmethod
def deserialize(cls, data: Dict[str, Any]) -> 'Command':
"""Deserialize command from dictionary."""
pass
def get_description(self) -> str:
"""Get human-readable description of the command."""
return self.__class__.__name__
# Command execution result for logging
@dataclass
class CommandResult:
"""Represents the result of a command execution."""
command_type: str
description: str
timestamp: str
execution_time_ms: float
success: bool
error_message: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
return asdict(self)
# Enhanced receiver classes
class SmartDevice:
"""Base class for smart devices with state tracking."""
def __init__(self, device_id: str, name: str) -> None:
self.device_id = device_id
self.name = name
self._state_history: List[Dict[str, Any]] = []
self._lock = Lock()
def _save_state(self, state: Dict[str, Any]) -> None:
"""Save current state for history tracking."""
with self._lock:
state_entry = {
'timestamp': datetime.now().isoformat(),
'state': state.copy()
}
self._state_history.append(state_entry)
# Keep only last 10 states
if len(self._state_history) > 10:
self._state_history.pop(0)
def get_state_history(self) -> List[Dict[str, Any]]:
"""Get device state history."""
with self._lock:
return self._state_history.copy()
class SmartLight(SmartDevice):
"""A smart light with advanced features."""
def __init__(self, device_id: str, name: str) -> None:
super().__init__(device_id, name)
self.is_on = False
self.brightness = 0
self.color = "white"
self._save_state(self.get_current_state())
def get_current_state(self) -> Dict[str, Any]:
"""Get current device state."""
return {
'is_on': self.is_on,
'brightness': self.brightness,
'color': self.color
}
def turn_on(self, brightness: int = 100) -> None:
"""Turn the light on with specified brightness."""
old_state = self.get_current_state()
self.is_on = True
self.brightness = max(0, min(100, brightness))
print(f"🔆 {self.name} turned ON (brightness: {self.brightness}%)")
self._save_state(self.get_current_state())
def turn_off(self) -> None:
"""Turn the light off."""
old_state = self.get_current_state()
self.is_on = False
self.brightness = 0
print(f"🔅 {self.name} turned OFF")
self._save_state(self.get_current_state())
def set_color(self, color: str) -> None:
"""Set the light color."""
old_state = self.get_current_state()
self.color = color
if self.is_on:
print(f"🌈 {self.name} color changed to {color}")
self._save_state(self.get_current_state())
class SmartThermostat(SmartDevice):
"""A smart thermostat with temperature control."""
def __init__(self, device_id: str, name: str) -> None:
super().__init__(device_id, name)
self.target_temperature = 72
self.mode = "auto" # auto, heat, cool, off
self._save_state(self.get_current_state())
def get_current_state(self) -> Dict[str, Any]:
"""Get current device state."""
return {
'target_temperature': self.target_temperature,
'mode': self.mode
}
def set_temperature(self, temperature: int) -> None:
"""Set target temperature."""
old_state = self.get_current_state()
self.target_temperature = max(50, min(90, temperature))
print(f"🌡️ {self.name} temperature set to {self.target_temperature}°F")
self._save_state(self.get_current_state())
def set_mode(self, mode: str) -> None:
"""Set thermostat mode."""
old_state = self.get_current_state()
if mode in ["auto", "heat", "cool", "off"]:
self.mode = mode
print(f"🔄 {self.name} mode set to {mode}")
self._save_state(self.get_current_state())
# Concrete Command classes with serialization
class LightControlCommand(Command):
"""Command to control a smart light."""
def __init__(self, light: SmartLight, action: str, **kwargs) -> None:
self._light = light
self._action = action
self._params = kwargs
self._previous_state: Optional[Dict[str, Any]] = None
def execute(self) -> None:
"""Execute the light control command."""
# Store previous state
self._previous_state = self._light.get_current_state()
# Execute based on action
if self._action == "turn_on":
brightness = self._params.get('brightness', 100)
self._light.turn_on(brightness)
elif self._action == "turn_off":
self._light.turn_off()
elif self._action == "set_color":
color = self._params.get('color', 'white')
self._light.set_color(color)
def undo(self) -> None:
"""Undo the light control command."""
if self._previous_state:
# Restore previous state
if self._previous_state['is_on']:
self._light.turn_on(self._previous_state['brightness'])
self._light.set_color(self._previous_state['color'])
else:
self._light.turn_off()
def get_description(self) -> str:
"""Get description of the command."""
action_desc = self._action.replace('_', ' ').title()
params_desc = ', '.join([f"{k}={v}" for k, v in self._params.items()])
return f"{action_desc} {self._light.name}" + (f" ({params_desc})" if params_desc else "")
def serialize(self) -> Dict[str, Any]:
"""Serialize command to dictionary."""
return {
'command_type': 'LightControlCommand',
'device_id': self._light.device_id,
'device_name': self._light.name,
'action': self._action,
'params': self._params
}
@classmethod
def deserialize(cls, data: Dict[str, Any]) -> 'LightControlCommand':
"""Deserialize command from dictionary (simplified for demo)."""
# In real implementation, you'd need a device registry
# For demo purposes, this is a placeholder
raise NotImplementedError("Device registry needed for full deserialization")
class ThermostatControlCommand(Command):
"""Command to control a smart thermostat."""
def __init__(self, thermostat: SmartThermostat, action: str, **kwargs) -> None:
self._thermostat = thermostat
self._action = action
self._params = kwargs
self._previous_state: Optional[Dict[str, Any]] = None
def execute(self) -> None:
"""Execute the thermostat control command."""
self._previous_state = self._thermostat.get_current_state()
if self._action == "set_temperature":
temperature = self._params.get('temperature', 72)
self._thermostat.set_temperature(temperature)
elif self._action == "set_mode":
mode = self._params.get('mode', 'auto')
self._thermostat.set_mode(mode)
def undo(self) -> None:
"""Undo the thermostat control command."""
if self._previous_state:
self._thermostat.set_temperature(self._previous_state['target_temperature'])
self._thermostat.set_mode(self._previous_state['mode'])
def get_description(self) -> str:
"""Get description of the command."""
action_desc = self._action.replace('_', ' ').title()
params_desc = ', '.join([f"{k}={v}" for k, v in self._params.items()])
return f"{action_desc} {self._thermostat.name}" + (f" ({params_desc})" if params_desc else "")
def serialize(self) -> Dict[str, Any]:
"""Serialize command to dictionary."""
return {
'command_type': 'ThermostatControlCommand',
'device_id': self._thermostat.device_id,
'device_name': self._thermostat.name,
'action': self._action,
'params': self._params
}
@classmethod
def deserialize(cls, data: Dict[str, Any]) -> 'ThermostatControlCommand':
"""Deserialize command from dictionary (simplified for demo)."""
raise NotImplementedError("Device registry needed for full deserialization")
# Command Logger
class CommandLogger:
"""Logs command executions for audit trail and crash recovery."""
def __init__(self, log_file: str = "command_log.json") -> None:
self._log_file = log_file
self._log_entries: List[Dict[str, Any]] = []
self._lock = Lock()
def log_command(self, result: CommandResult) -> None:
"""Log a command execution result."""
with self._lock:
self._log_entries.append(result.to_dict())
# In a real system, you'd write to file here
print(f"📝 Logged: {result.description}")
def get_log_entries(self, limit: int = 10) -> List[Dict[str, Any]]:
"""Get recent log entries."""
with self._lock:
return self._log_entries[-limit:].copy()
def save_to_file(self) -> None:
"""Save log entries to file."""
with self._lock:
try:
with open(self._log_file, 'w') as f:
json.dump(self._log_entries, f, indent=2)
print(f"💾 Log saved to {self._log_file}")
except Exception as e:
print(f"❌ Error saving log: {e}")
def clear_log(self) -> None:
"""Clear all log entries."""
with self._lock:
self._log_entries.clear()
print("🗑️ Log cleared")
# Command Queue Processor
class CommandQueueProcessor:
"""Processes commands from a queue in a separate thread."""
def __init__(self, logger: CommandLogger) -> None:
self._queue: Queue[Command] = Queue()
self._logger = logger
self._running = False
self._worker_thread: Optional[Thread] = None
def start(self) -> None:
"""Start the queue processor."""
if not self._running:
self._running = True
self._worker_thread = Thread(target=self._process_queue, daemon=True)
self._worker_thread.start()
print("🚀 Command queue processor started")
def stop(self) -> None:
"""Stop the queue processor."""
self._running = False
if self._worker_thread:
self._worker_thread.join(timeout=1.0)
print("🛑 Command queue processor stopped")
def enqueue_command(self, command: Command) -> None:
"""Add a command to the execution queue."""
self._queue.put(command)
print(f"📥 Queued: {command.get_description()}")
def get_queue_size(self) -> int:
"""Get current queue size."""
return self._queue.qsize()
def _process_queue(self) -> None:
"""Process commands from the queue (runs in separate thread)."""
while self._running:
try:
# Wait for command with timeout to allow clean shutdown
command = self._queue.get(timeout=0.5)
# Execute command and measure time
start_time = time.time()
success = True
error_message = None
try:
print(f"⚡ Executing: {command.get_description()}")
command.execute()
except Exception as e:
success = False
error_message = str(e)
print(f"❌ Error executing command: {e}")
end_time = time.time()
execution_time = (end_time - start_time) * 1000 # Convert to milliseconds
# Log the result
result = CommandResult(
command_type=command.__class__.__name__,
description=command.get_description(),
timestamp=datetime.now().isoformat(),
execution_time_ms=execution_time,
success=success,
error_message=error_message
)
self._logger.log_command(result)
self._queue.task_done()
# Small delay to prevent overwhelming the system
time.sleep(0.1)
except Exception:
# Timeout or queue empty - continue
continue
# Smart Home Control System
class SmartHomeController:
"""Main controller that manages devices and command execution."""
def __init__(self) -> None:
self._devices: Dict[str, SmartDevice] = {}
self._logger = CommandLogger()
self._queue_processor = CommandQueueProcessor(self._logger)
self._command_history: List[Command] = []
self._max_history = 20
def start(self) -> None:
"""Start the smart home controller."""
self._queue_processor.start()
print("🏠 Smart Home Controller started")
def stop(self) -> None:
"""Stop the smart home controller."""
self._queue_processor.stop()
self._logger.save_to_file()
print("🏠 Smart Home Controller stopped")
def add_device(self, device: SmartDevice) -> None:
"""Add a device to the system."""
self._devices[device.device_id] = device
print(f"➕ Added device: {device.name} ({device.device_id})")
def get_device(self, device_id: str) -> Optional[SmartDevice]:
"""Get a device by ID."""
return self._devices.get(device_id)
def execute_command_sync(self, command: Command) -> None:
"""Execute a command synchronously."""
start_time = time.time()
success = True
error_message = None
try:
command.execute()
self._command_history.append(command)
# Keep history size manageable
if len(self._command_history) > self._max_history:
self._command_history.pop(0)
except Exception as e:
success = False
error_message = str(e)
print(f"❌ Error executing command: {e}")
end_time = time.time()
execution_time = (end_time - start_time) * 1000
# Log the result
result = CommandResult(
command_type=command.__class__.__name__,
description=command.get_description(),
timestamp=datetime.now().isoformat(),
execution_time_ms=execution_time,
success=success,
error_message=error_message
)
self._logger.log_command(result)
def execute_command_async(self, command: Command) -> None:
"""Execute a command asynchronously via queue."""
self._queue_processor.enqueue_command(command)
def undo_last_command(self) -> None:
"""Undo the last executed command."""
if self._command_history:
last_command = self._command_history.pop()
print(f"↩️ Undoing: {last_command.get_description()}")
try:
last_command.undo()
print("✅ Undo completed")
except Exception as e:
print(f"❌ Error during undo: {e}")
else:
print("❌ No commands to undo")
def get_system_status(self) -> Dict[str, Any]:
"""Get overall system status."""
return {
'devices': len(self._devices),
'queue_size': self._queue_processor.get_queue_size(),
'command_history_size': len(self._command_history),
'recent_logs': self._logger.get_log_entries(5)
}
def list_devices(self) -> None:
"""List all devices in the system."""
print("\n📱 Connected Devices:")
print("-" * 40)
for device_id, device in self._devices.items():
print(f" {device.name} ({device_id})")
if isinstance(device, SmartLight):
status = f"{'ON' if device.is_on else 'OFF'}"
if device.is_on:
status += f", {device.brightness}%, {device.color}"
print(f" Light: {status}")
elif isinstance(device, SmartThermostat):
print(f" Thermostat: {device.target_temperature}°F, {device.mode}")
def main() -> None:
"""Demonstrate command queuing and logging in a smart home system."""
print("=== Command Queue & Logging Demo ===\n")
# Create the smart home controller
controller = SmartHomeController()
controller.start()
# Add devices
living_room_light = SmartLight("light_001", "Living Room Light")
bedroom_light = SmartLight("light_002", "Bedroom Light")
main_thermostat = SmartThermostat("thermo_001", "Main Thermostat")
controller.add_device(living_room_light)
controller.add_device(bedroom_light)
controller.add_device(main_thermostat)
# List devices
controller.list_devices()
# Create and execute commands
print("\n🎬 Executing commands...")
# Synchronous commands
commands = [
LightControlCommand(living_room_light, "turn_on", brightness=75),
LightControlCommand(living_room_light, "set_color", color="blue"),
LightControlCommand(bedroom_light, "turn_on", brightness=50),
ThermostatControlCommand(main_thermostat, "set_temperature", temperature=68),
ThermostatControlCommand(main_thermostat, "set_mode", mode="cool"),
]
# Execute some commands synchronously
for i, cmd in enumerate(commands[:3]):
print(f"\nSync command {i+1}:")
controller.execute_command_sync(cmd)
# Execute some commands asynchronously
print(f"\nEnqueuing async commands...")
for cmd in commands[3:]:
controller.execute_command_async(cmd)
# Wait a bit for async commands to process
print("\n⏳ Waiting for async commands to complete...")
time.sleep(2)
# Show system status
print("\n📊 System Status:")
status = controller.get_system_status()
print(f" Devices: {status['devices']}")
print(f" Queue size: {status['queue_size']}")
print(f" Command history: {status['command_history_size']}")
print(f" Recent logs: {len(status['recent_logs'])}")
# Test undo functionality
print("\n↩️ Testing undo...")
controller.undo_last_command()
controller.undo_last_command()
# Show final system status
print("\n📋 Final System Status:")
controller.list_devices()
# Cleanup
print("\n🧹 Shutting down...")
controller.stop()
if __name__ == "__main__":
main()
Command Queuing: The CommandQueueProcessor allows commands to be executed asynchronously in a separate thread, which is useful for long-running operations or when you want to decouple command creation from execution.
Command Logging: The CommandLogger provides a complete audit trail of all command executions, including timing information, success/failure status, and error messages. This is crucial for debugging, monitoring, and crash recovery.
State Management: Smart devices maintain their state history, allowing for more sophisticated undo operations and system monitoring.
Serialization Support: Commands can be serialized to JSON format, enabling persistence, network transmission, or crash recovery scenarios.
Thread Safety: The implementation uses proper locking mechanisms to ensure thread-safe operations when multiple threads are accessing shared resources.
Error Handling: Robust error handling ensures that failures in one command don’t crash the entire system.