Hugging Face's logo

Spaces:
mrwabnalas40
/
Test
Runtime error

App Files Community
Test
File size: 3,144 Bytes 
eebf941
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
 
import subprocess
import GPUtil
import psutil
import logging
from peer_discovery import PORT

logging.getLogger().setLevel(logging.CRITICAL)  # ุตุงู…ุช

class DeviceManager:
    def __init__(self):
        self.devices = {
            "GPU": self._detect_gpus(),
            "DSP": self._detect_dsps(),
            "NIC": self._detect_nics(),
            "STORAGE": self._detect_storage(),
            "CAPTURE": self._detect_capture(),
            "ACCELERATOR": self._detect_accelerators()
        }

    # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ ุงูƒุชุดุงู ุงู„ูƒุฑูˆุช โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
    def _detect_gpus(self):
        try:
            return GPUtil.getGPUs()
        except:
            return []

    def _detect_dsps(self):
        try:
            output = subprocess.check_output(["aplay", "-l"], stderr=subprocess.DEVNULL).decode()
            return ["DSP_Audio"] if "card" in output.lower() else []
        except:
            return []

    def _detect_nics(self):
        try:
            return list(psutil.net_if_addrs().keys())
        except:
            return []

    def _detect_storage(self):
        try:
            output = subprocess.check_output(["lsblk", "-o", "NAME"], stderr=subprocess.DEVNULL).decode()
            return output.split() if output else []
        except:
            return []

    def _detect_capture(self):
        try:
            output = subprocess.check_output(["v4l2-ctl", "--list-devices"], stderr=subprocess.DEVNULL).decode()
            return output.split(":")[0::2] if output else []
        except:
            return []

    def _detect_accelerators(self):
        # ุงูุชุฑุงุถ: ููŠ ุงู„ู…ุณุชู‚ุจู„ ูŠู…ูƒู† ุฅุถุงูุฉ ุงูƒุชุดุงู ุญู‚ูŠู‚ูŠ ู„ู€ FPGA/TPU
        return []

    # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ ูุญุต ุงู„ุญู…ู„ โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
    def get_device_load(self, device_type, index=0):
        try:
            if device_type == "GPU" and self.devices["GPU"]:
                return self.devices["GPU"][index].load * 100
            elif device_type == "DSP" and self.devices["DSP"]:
                return 10  # ุงูุชุฑุงุถูŠ: ู…ุง ููŠ API ู„ู„ุญู…ู„
            elif device_type == "NIC" and self.devices["NIC"]:
                return psutil.net_io_counters().bytes_sent / (1024 * 1024)  # ู…ุซุงู„ ุจุณูŠุท
            elif device_type == "STORAGE" and self.devices["STORAGE"]:
                return psutil.disk_usage('/').percent
            elif device_type == "CAPTURE" and self.devices["CAPTURE"]:
                return 20  # ุงูุชุฑุงุถูŠ: ุญู…ู„ ู…ู†ุฎูุถ
            elif device_type == "ACCELERATOR" and self.devices["ACCELERATOR"]:
                return 15  # ุงูุชุฑุงุถูŠ
            return 0
        except:
            return 0

    # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ ู…ู†ุทู‚ 30% / 70% โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
    def can_receive(self, device_type, index=0):
        return self.get_device_load(device_type, index) <= 30

    def should_offload(self, device_type, index=0):
        return self.get_device_load(device_type, index) >= 70