Spaces:
Configuration error
Configuration error
File size: 9,492 Bytes
77bcbf1 | 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 | """
CASCADE Logging Integration
Plug-and-play logging for existing CASCADE components.
Retrofits existing systems with world-class logging without major surgery.
"""
import functools
import time
from typing import Any, Callable, Dict, Optional
from .log_manager import get_log_manager, LogLevel, ImpactLevel
def log_component(component_name: str, system: str = "CASCADE"):
"""Decorator to add logging to any class or function"""
def decorator(target):
if isinstance(target, type):
# Decorating a class
return _log_class(target, component_name, system)
else:
# Decorating a function
return _log_function(target, component_name, system)
return decorator
def _log_class(cls, component_name: str, system: str):
"""Add logging to all methods of a class"""
manager = get_log_manager()
manager.register_component(component_name, system)
for attr_name in dir(cls):
if not attr_name.startswith('_'):
attr = getattr(cls, attr_name)
if callable(attr):
setattr(cls, attr_name, _log_method(attr, component_name))
return cls
def _log_function(func, component_name: str, system: str):
"""Add logging to a function"""
manager = get_log_manager()
manager.register_component(component_name, system)
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
# Log start
get_log_manager().log_operation(
component_name, f"{func.__name__}_start",
level=LogLevel.DEBUG,
impact=ImpactLevel.TRACE,
details={
"context": f"Starting {func.__name__}",
"consequence": f"Will execute {func.__name__}",
"metrics": {"args": len(args), "kwargs": len(kwargs)}
}
)
try:
result = func(*args, **kwargs)
# Log success
duration = time.time() - start_time
get_log_manager().log_operation(
component_name, f"{func.__name__}_complete",
level=LogLevel.INFO,
impact=ImpactLevel.LOW,
details={
"context": f"Completed {func.__name__}",
"consequence": f"Result ready",
"metrics": {"duration_seconds": duration}
}
)
return result
except Exception as e:
# Log error
get_log_manager().log_operation(
component_name, f"{func.__name__}_error",
level=LogLevel.ERROR,
impact=ImpactLevel.HIGH,
details={
"context": f"Failed in {func.__name__}",
"consequence": "Operation failed",
"metrics": {"error": str(e)}
}
)
raise
return wrapper
def _log_method(method, component_name: str):
"""Add logging to a method"""
@functools.wraps(method)
def wrapper(self, *args, **kwargs):
start_time = time.time()
try:
result = method(self, *args, **kwargs)
# Log successful method call
get_log_manager().log_operation(
component_name, f"{method.__name__}",
level=LogLevel.DEBUG,
impact=ImpactLevel.TRACE,
details={
"metrics": {"duration": time.time() - start_time}
}
)
return result
except Exception as e:
# Log method error
get_log_manager().log_operation(
component_name, f"{method.__name__}_error",
level=LogLevel.ERROR,
impact=ImpactLevel.HIGH,
details={
"context": f"Method {method.__name__} failed",
"metrics": {"error": str(e)}
}
)
raise
return wrapper
def log_kleene_iterations(operation_name: str):
"""Decorator specifically for Kleene fixed point iterations"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
get_log_manager().log_operation(
"KleeneEngine", f"{operation_name}_start",
level=LogLevel.INFO,
impact=ImpactLevel.MEDIUM,
details={
"context": f"Starting fixed point iteration for {operation_name}",
"consequence": "Will iterate until convergence"
}
)
start_time = time.time()
result = func(*args, **kwargs)
# Extract iteration info from result if available
iterations = getattr(result, 'iterations', 0)
converged = getattr(result, 'converged', True)
get_log_manager().log_operation(
"KleeneEngine", f"{operation_name}_complete",
level=LogLevel.INFO,
impact=ImpactLevel.LOW if converged else ImpactLevel.HIGH,
details={
"context": f"Fixed point iteration {'converged' if converged else 'diverged'}",
"consequence": f"Processed {iterations} iterations",
"metrics": {
"iterations": iterations,
"converged": converged,
"duration": time.time() - start_time
},
"fixed_point": converged
}
)
return result
return wrapper
return decorator
def log_model_observation(model_id: str):
"""Decorator for model observation functions"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
get_log_manager().log_operation(
"ModelObserver", f"observe_{model_id}",
level=LogLevel.INFO,
impact=ImpactLevel.MEDIUM,
details={
"context": f"Starting observation of model {model_id}",
"consequence": "Will generate cryptographic proof"
}
)
result = func(*args, **kwargs)
# Extract observation details
layers = getattr(result, 'layer_count', 0)
merkle = getattr(result, 'merkle_root', 'unknown')
get_log_manager().log_operation(
"ModelObserver", f"observed_{model_id}",
level=LogLevel.INFO,
impact=ImpactLevel.LOW,
details={
"context": f"Model observation complete",
"consequence": "Cryptographic proof generated",
"metrics": {
"model": model_id,
"layers": layers,
"merkle": merkle[:16] + "..."
},
"fixed_point": True
}
)
return result
return wrapper
return decorator
def log_data_processing(dataset_name: str):
"""Decorator for data processing functions"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
get_log_manager().log_operation(
"DataProcessor", f"process_{dataset_name}",
level=LogLevel.INFO,
impact=ImpactLevel.MEDIUM,
details={
"context": f"Processing dataset {dataset_name}",
"consequence": "Will extract and analyze data"
}
)
result = func(*args, **kwargs)
# Extract processing stats
records = getattr(result, 'record_count', 0)
operations = getattr(result, 'operations', [])
get_log_manager().log_operation(
"DataProcessor", f"processed_{dataset_name}",
level=LogLevel.INFO,
impact=ImpactLevel.LOW,
details={
"context": f"Dataset processing complete",
"consequence": f"Processed {records} records",
"metrics": {
"dataset": dataset_name,
"records": records,
"operations": len(operations)
}
}
)
return result
return wrapper
return decorator
# Quick integration function
def integrate_cascade_logging():
"""One-call integration for entire CASCADE system"""
from ..system.observer import SystemObserver
from ..core.provenance import ProvenanceTracker
from data_unity import run_kleene_iteration
# Register main components
manager = get_log_manager()
manager.register_component("SystemObserver", "System Observatory")
manager.register_component("ProvenanceTracker", "Model Observatory")
manager.register_component("DataUnity", "Data Unity")
manager.register_component("KleeneEngine", "NEXUS")
print("✅ CASCADE logging integrated across all components")
return manager
|