| import numpy as np | |
| def track_anomalies(anomaly_counts): | |
| """Track anomaly counts and return statistics.""" | |
| if not anomaly_counts: | |
| return {"mean": 0, "std": 0} | |
| mean = np.mean(anomaly_counts) | |
| std = np.std(anomaly_counts) | |
| return {"mean": mean, "std": std} | |
| def predict_anomaly(anomaly_counts, threshold=2.5): | |
| """Predict potential issues based on anomaly spikes.""" | |
| if len(anomaly_counts) < 5: # Need enough data for prediction | |
| return False | |
| stats = track_anomalies(anomaly_counts) | |
| recent_counts = anomaly_counts[-5:] # Last 5 frames | |
| # Flag a potential issue if the recent anomaly count is significantly higher than the mean | |
| for count in recent_counts: | |
| if stats["std"] > 0 and (count - stats["mean"]) / stats["std"] > threshold: | |
| return True | |
| return False |