Spaces:
Running
on
Zero
Running
on
Zero
File size: 3,703 Bytes
df6c67d |
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 |
import platform
import re
import socket
import time
import uuid
from inference.core.cache import cache
from inference.core.logger import logger
from inference.core.version import __version__
def get_model_metrics(
inference_server_id: str, model_id: str, min: float = -1, max: float = float("inf")
) -> dict:
"""
Gets the metrics for a given model between a specified time range.
Args:
device_id (str): The identifier of the device.
model_id (str): The identifier of the model.
start (float, optional): The starting timestamp of the time range. Defaults to -1.
stop (float, optional): The ending timestamp of the time range. Defaults to float("inf").
Returns:
dict: A dictionary containing the metrics of the model:
- num_inferences (int): The number of inferences made.
- avg_inference_time (float): The average inference time.
- num_errors (int): The number of errors occurred.
"""
now = time.time()
inferences_with_times = cache.zrangebyscore(
f"inference:{inference_server_id}:{model_id}", min=min, max=max, withscores=True
)
num_inferences = len(inferences_with_times)
inference_times = []
for inference, t in inferences_with_times:
response = inference["response"]
if isinstance(response, list):
times = [r["time"] for r in response if "time" in r]
inference_times.extend(times)
else:
if "time" in response:
inference_times.append(response["time"])
avg_inference_time = (
sum(inference_times) / len(inference_times) if len(inference_times) > 0 else 0
)
errors_with_times = cache.zrangebyscore(
f"error:{inference_server_id}:{model_id}", min=min, max=max, withscores=True
)
num_errors = len(errors_with_times)
return {
"num_inferences": num_inferences,
"avg_inference_time": avg_inference_time,
"num_errors": num_errors,
}
def get_system_info() -> dict:
"""Collects system information such as platform, architecture, hostname, IP address, MAC address, and processor details.
Returns:
dict: A dictionary containing detailed system information.
"""
info = {}
try:
info["platform"] = platform.system()
info["platform_release"] = platform.release()
info["platform_version"] = platform.version()
info["architecture"] = platform.machine()
info["hostname"] = socket.gethostname()
info["ip_address"] = socket.gethostbyname(socket.gethostname())
info["mac_address"] = ":".join(re.findall("..", "%012x" % uuid.getnode()))
info["processor"] = platform.processor()
return info
except Exception as e:
logger.exception(e)
finally:
return info
def get_inference_results_for_model(
inference_server_id: str, model_id: str, min: float = -1, max: float = float("inf")
):
inferences_with_times = cache.zrangebyscore(
f"inference:{inference_server_id}:{model_id}", min=min, max=max, withscores=True
)
inference_results = []
for result, score in inferences_with_times:
# Don't send large image files
if result.get("request", {}).get("image"):
del result["request"]["image"]
responses = result.get("response")
if responses:
if not isinstance(responses, list):
responses = [responses]
for resp in responses:
if resp.get("image"):
del resp["image"]
inference_results.append({"request_time": score, "inference": result})
return inference_results
|