-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy patherror_handler.py
More file actions
executable file
·293 lines (243 loc) · 10.5 KB
/
error_handler.py
File metadata and controls
executable file
·293 lines (243 loc) · 10.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
"""
Error Handling Module for SmartHome System
Provides custom exceptions, error recovery strategies, and graceful degradation.
"""
import logging
import traceback
import sys
from typing import Optional, Callable, Any, Dict
from functools import wraps
from enum import Enum
class ErrorSeverity(Enum):
"""Error severity levels."""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class SmartHomeError(Exception):
"""Base exception for SmartHome system."""
def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.MEDIUM,
component: str = "unknown", recoverable: bool = True):
self.message = message
self.severity = severity
self.component = component
self.recoverable = recoverable
super().__init__(self.message)
class CameraError(SmartHomeError):
"""Camera-related errors."""
def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.HIGH):
super().__init__(message, severity, "camera", True)
class VoiceError(SmartHomeError):
"""Voice/TTS-related errors."""
def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.MEDIUM):
super().__init__(message, severity, "voice", True)
class FacialRecognitionError(SmartHomeError):
"""Facial recognition errors."""
def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.MEDIUM):
super().__init__(message, severity, "facial_recognition", True)
class SmartLightError(SmartHomeError):
"""Smart light control errors."""
def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.MEDIUM):
super().__init__(message, severity, "smart_lights", True)
class WeatherError(SmartHomeError):
"""Weather API errors."""
def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.LOW):
super().__init__(message, severity, "weather", True)
class ConfigurationError(SmartHomeError):
"""Configuration-related errors."""
def __init__(self, message: str, severity: ErrorSeverity = ErrorSeverity.HIGH):
super().__init__(message, severity, "configuration", False)
class ErrorHandler:
"""Centralized error handling for the SmartHome system."""
def __init__(self):
self.logger = logging.getLogger(__name__)
self.error_count: Dict[str, int] = {}
self.recovery_strategies: Dict[str, Callable] = {}
self._setup_recovery_strategies()
def _setup_recovery_strategies(self):
"""Setup default recovery strategies for different error types."""
self.recovery_strategies = {
'camera': self._recover_camera,
'voice': self._recover_voice,
'facial_recognition': self._recover_facial_recognition,
'smart_lights': self._recover_smart_lights,
'weather': self._recover_weather,
'configuration': self._recover_configuration
}
def handle_error(self, error: SmartHomeError, context: Optional[Dict] = None) -> bool:
"""
Handle a SmartHome error with appropriate logging and recovery.
Args:
error: The SmartHome error to handle
context: Additional context information
Returns:
True if error was handled successfully, False otherwise
"""
try:
# Log the error
self._log_error(error, context)
# Update error count
self._update_error_count(error.component)
# Attempt recovery if error is recoverable
if error.recoverable:
return self._attempt_recovery(error)
else:
self.logger.critical(f"Non-recoverable error in {error.component}: {error.message}")
return False
except Exception as e:
self.logger.error(f"Error in error handler: {e}")
return False
def _log_error(self, error: SmartHomeError, context: Optional[Dict] = None):
"""Log error with appropriate level based on severity."""
log_message = f"[{error.component.upper()}] {error.message}"
if context:
log_message += f" | Context: {context}"
if error.severity == ErrorSeverity.LOW:
self.logger.warning(log_message)
elif error.severity == ErrorSeverity.MEDIUM:
self.logger.error(log_message)
elif error.severity == ErrorSeverity.HIGH:
self.logger.error(log_message)
self.logger.error(f"Stack trace: {traceback.format_exc()}")
elif error.severity == ErrorSeverity.CRITICAL:
self.logger.critical(log_message)
self.logger.critical(f"Stack trace: {traceback.format_exc()}")
def _update_error_count(self, component: str):
"""Update error count for a component."""
self.error_count[component] = self.error_count.get(component, 0) + 1
def _attempt_recovery(self, error: SmartHomeError) -> bool:
"""Attempt to recover from an error."""
try:
recovery_func = self.recovery_strategies.get(error.component)
if recovery_func:
self.logger.info(f"Attempting recovery for {error.component}")
return recovery_func(error)
else:
self.logger.warning(f"No recovery strategy for {error.component}")
return False
except Exception as e:
self.logger.error(f"Recovery failed for {error.component}: {e}")
return False
def _recover_camera(self, error: SmartHomeError) -> bool:
"""Recovery strategy for camera errors."""
try:
# Try to reinitialize camera with different source
self.logger.info("Attempting camera recovery...")
# This would be implemented in the camera module
return True
except Exception as e:
self.logger.error(f"Camera recovery failed: {e}")
return False
def _recover_voice(self, error: SmartHomeError) -> bool:
"""Recovery strategy for voice errors."""
try:
self.logger.info("Attempting voice system recovery...")
# Try to reinitialize TTS engine
return True
except Exception as e:
self.logger.error(f"Voice recovery failed: {e}")
return False
def _recover_facial_recognition(self, error: SmartHomeError) -> bool:
"""Recovery strategy for facial recognition errors."""
try:
self.logger.info("Attempting facial recognition recovery...")
# Try to reload encodings or restart detection
return True
except Exception as e:
self.logger.error(f"Facial recognition recovery failed: {e}")
return False
def _recover_smart_lights(self, error: SmartHomeError) -> bool:
"""Recovery strategy for smart light errors."""
try:
self.logger.info("Attempting smart light recovery...")
# Try to reconnect to lights
return True
except Exception as e:
self.logger.error(f"Smart light recovery failed: {e}")
return False
def _recover_weather(self, error: SmartHomeError) -> bool:
"""Recovery strategy for weather errors."""
try:
self.logger.info("Attempting weather service recovery...")
# Switch to mock data or retry API call
return True
except Exception as e:
self.logger.error(f"Weather recovery failed: {e}")
return False
def _recover_configuration(self, error: SmartHomeError) -> bool:
"""Recovery strategy for configuration errors."""
try:
self.logger.info("Attempting configuration recovery...")
# Reload configuration or use defaults
return True
except Exception as e:
self.logger.error(f"Configuration recovery failed: {e}")
return False
def get_error_stats(self) -> Dict[str, Any]:
"""Get error statistics."""
return {
'total_errors': sum(self.error_count.values()),
'error_count_by_component': self.error_count.copy(),
'most_error_prone_component': max(self.error_count.items(), key=lambda x: x[1])[0] if self.error_count else None
}
def reset_error_count(self, component: Optional[str] = None):
"""Reset error count for a component or all components."""
if component:
self.error_count[component] = 0
else:
self.error_count.clear()
def handle_errors(func: Callable) -> Callable:
"""
Decorator to handle errors in functions.
Args:
func: Function to wrap with error handling
Returns:
Wrapped function with error handling
"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except SmartHomeError as e:
error_handler = ErrorHandler()
error_handler.handle_error(e, {'function': func.__name__})
return None
except Exception as e:
# Convert generic exceptions to SmartHomeError
smart_home_error = SmartHomeError(
f"Unexpected error in {func.__name__}: {str(e)}",
ErrorSeverity.HIGH,
"system"
)
error_handler = ErrorHandler()
error_handler.handle_error(smart_home_error, {'function': func.__name__})
return None
return wrapper
def safe_execute(func: Callable, *args, default_return: Any = None, **kwargs) -> Any:
"""
Safely execute a function with error handling.
Args:
func: Function to execute
*args: Function arguments
default_return: Value to return if function fails
**kwargs: Function keyword arguments
Returns:
Function result or default_return if execution fails
"""
try:
return func(*args, **kwargs)
except SmartHomeError as e:
error_handler = ErrorHandler()
error_handler.handle_error(e, {'function': func.__name__})
return default_return
except Exception as e:
smart_home_error = SmartHomeError(
f"Unexpected error in {func.__name__}: {str(e)}",
ErrorSeverity.HIGH,
"system"
)
error_handler = ErrorHandler()
error_handler.handle_error(smart_home_error, {'function': func.__name__})
return default_return
# Global error handler instance
error_handler = ErrorHandler()