testing async subroutine

This commit is contained in:
2024-12-05 20:18:14 -05:00
parent 24d2502bc4
commit 4622005cf9

View File

@ -1,5 +1,5 @@
#!/usr/bin/env python3
import os, sys, json, requests, psutil, socket, subprocess, logging, asyncio
import os, sys, json, requests, psutil, socket, subprocess, logging
from typing import Dict, Any, List
# Create a logger
@ -56,7 +56,6 @@ class SystemHealthMonitor:
'memory_health': self._check_memory_usage(),
'cpu_health': self._check_cpu_usage(),
'network_health': self._check_network_status()
# 'temperature_health': self._check_system_temperatures() # Optional temperature check
}
return health_report
@ -263,7 +262,7 @@ class SystemHealthMonitor:
'used_space': self._convert_bytes(usage.used),
'free_space': self._convert_bytes(usage.free),
'usage_percent': usage.percent,
'disk_usage_status': disk_usage_status
'usage_status': disk_usage_status
})
# Update overall status based on usage
if disk_usage_status == 'CRITICAL_HIGH_USAGE':
@ -280,13 +279,13 @@ class SystemHealthMonitor:
)
output = result.stdout + result.stderr
drive_smart_status = 'HEALTHY' if 'PASSED' in output else 'UNHEALTHY'
drive_report['drive_smart_status'] = drive_smart_status
drive_report['smart_status'] = drive_smart_status
# Update overall status if SMART status is unhealthy
if drive_smart_status == 'UNHEALTHY' and overall_status != 'CRITICAL_HIGH_USAGE':
overall_status = 'UNHEALTHY'
except Exception as e:
print(f"Error checking SMART status for {partition.device}: {str(e)}")
drive_report['drive_smart_status'] = 'ERROR'
drive_report['smart_status'] = 'ERROR'
except Exception as e:
logger.error(f"Could not check drive: {str(e)}")
drive_report['error'] = str(e)
@ -296,6 +295,7 @@ class SystemHealthMonitor:
logger.error(f"Error checking drives health: {str(e)}")
return drives_health
@staticmethod
def _convert_bytes(bytes_value: int, suffix: str = 'B') -> str:
"""
Convert bytes to a human-readable format.
@ -337,6 +337,7 @@ class SystemHealthMonitor:
'status': 'OK' if cpu_usage_percent < 90 else 'WARNING'
}
return cpu_health
def _check_network_status(self) -> Dict[str, Any]:
"""
Check the status of network interfaces and report any issues.