Skip to content

Commit edc6da3

Browse files
authored
Available devices endpoint (#5048)
1 parent f55af09 commit edc6da3

File tree

5 files changed

+306
-0
lines changed

5 files changed

+306
-0
lines changed

application/backend/app/api/routers/system.py

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,11 +8,20 @@
88
from fastapi import APIRouter, Depends
99

1010
from app.api.dependencies import get_system_service
11+
from app.schemas.system import DeviceInfo
1112
from app.services import SystemService
1213

1314
router = APIRouter(prefix="/api")
1415

1516

17+
@router.get("/system/devices")
18+
async def get_devices(
19+
system_service: Annotated[SystemService, Depends(get_system_service)],
20+
) -> list[DeviceInfo]:
21+
"""Returns the list of available compute devices (CPU, Intel XPU, NVIDIA CUDA)."""
22+
return system_service.get_devices()
23+
24+
1625
@router.get("/system/metrics/memory")
1726
async def get_memory(
1827
system_service: Annotated[SystemService, Depends(get_system_service)],
Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
# Copyright (C) 2025 Intel Corporation
2+
# SPDX-License-Identifier: Apache-2.0
3+
4+
"""System schemas"""
5+
6+
from enum import StrEnum, auto
7+
8+
from pydantic import BaseModel, Field
9+
10+
11+
class DeviceType(StrEnum):
12+
"""Enumeration of device types"""
13+
14+
CPU = auto()
15+
XPU = auto()
16+
CUDA = auto()
17+
18+
19+
class DeviceInfo(BaseModel):
20+
"""Device information schema"""
21+
22+
type: DeviceType = Field(..., description="Device type (cpu, xpu, or cuda)")
23+
name: str = Field(..., description="Device name")
24+
memory: int | None = Field(None, description="Total memory available to the device, in bytes (null for CPU)")
25+
index: int | None = Field(None, description="Device index among those of the same type (null for CPU)")
26+
27+
model_config = {
28+
"json_schema_extra": {
29+
"examples": [
30+
{"type": "cpu", "name": "CPU"},
31+
{"type": "xpu", "name": "Intel Arc B580", "memory": 12884901888, "index": 0},
32+
{"type": "cuda", "name": "NVIDIA GeForce RTX 4090", "memory": 25769803776, "index": 0},
33+
]
34+
}
35+
}

application/backend/app/services/system_service.py

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,9 @@
22
# SPDX-License-Identifier: Apache-2.0
33

44
import psutil
5+
import torch
6+
7+
from app.schemas.system import DeviceInfo, DeviceType
58

69

710
class SystemService:
@@ -28,3 +31,42 @@ def get_cpu_usage(self) -> float:
2831
float: CPU usage in percentage
2932
"""
3033
return self.process.cpu_percent(interval=None)
34+
35+
@staticmethod
36+
def get_devices() -> list[DeviceInfo]:
37+
"""
38+
Get available compute devices (CPU, GPUs, ...)
39+
40+
Returns:
41+
list[DeviceInfo]: List of available devices
42+
"""
43+
# CPU is always available
44+
devices: list[DeviceInfo] = [DeviceInfo(type=DeviceType.CPU, name="CPU", memory=None, index=None)]
45+
46+
# Check for Intel XPU devices
47+
if torch.xpu.is_available():
48+
for device_idx in range(torch.xpu.device_count()):
49+
xpu_dp = torch.xpu.get_device_properties(device_idx)
50+
devices.append(
51+
DeviceInfo(
52+
type=DeviceType.XPU,
53+
name=xpu_dp.name,
54+
memory=xpu_dp.total_memory,
55+
index=device_idx,
56+
)
57+
)
58+
59+
# Check for NVIDIA CUDA devices
60+
if torch.cuda.is_available():
61+
for device_idx in range(torch.cuda.device_count()):
62+
cuda_dp = torch.cuda.get_device_properties(device_idx)
63+
devices.append(
64+
DeviceInfo(
65+
type=DeviceType.CUDA,
66+
name=cuda_dp.name,
67+
memory=cuda_dp.total_memory,
68+
index=device_idx,
69+
)
70+
)
71+
72+
return devices
Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
# Copyright (C) 2025 Intel Corporation
2+
# SPDX-License-Identifier: Apache-2.0
3+
4+
from unittest.mock import Mock
5+
6+
import pytest
7+
from fastapi import status
8+
from fastapi.testclient import TestClient
9+
10+
from app.api.dependencies import get_system_service
11+
from app.main import app
12+
from app.schemas.system import DeviceInfo, DeviceType
13+
from app.services import SystemService
14+
15+
16+
@pytest.fixture
17+
def fxt_client():
18+
return TestClient(app)
19+
20+
21+
@pytest.fixture
22+
def fxt_system_service() -> Mock:
23+
system_service = Mock(spec=SystemService)
24+
app.dependency_overrides[get_system_service] = lambda: system_service
25+
return system_service
26+
27+
28+
class TestSystemEndpoints:
29+
def test_get_devices_cpu_only(self, fxt_system_service: Mock, fxt_client: TestClient):
30+
"""Test GET /api/system/devices with CPU only"""
31+
fxt_system_service.get_devices.return_value = [
32+
DeviceInfo(type=DeviceType.CPU, name="CPU", memory=None, index=None),
33+
]
34+
35+
response = fxt_client.get("/api/system/devices")
36+
37+
assert response.status_code == status.HTTP_200_OK
38+
devices = response.json()
39+
assert len(devices) == 1
40+
assert devices[0]["type"] == "cpu"
41+
assert devices[0]["name"] == "CPU"
42+
assert devices[0]["memory"] is None
43+
assert devices[0]["index"] is None
44+
45+
def test_get_devices_with_xpu(self, fxt_system_service: Mock, fxt_client: TestClient):
46+
"""Test GET /api/system/devices with Intel XPU"""
47+
fxt_system_service.get_devices.return_value = [
48+
DeviceInfo(type=DeviceType.CPU, name="CPU", memory=None, index=None),
49+
DeviceInfo(type=DeviceType.XPU, name="Intel(R) Graphics [0x7d41]", memory=36022263808, index=0),
50+
]
51+
52+
response = fxt_client.get("/api/system/devices")
53+
54+
assert response.status_code == status.HTTP_200_OK
55+
devices = response.json()
56+
assert len(devices) == 2
57+
assert devices[0]["type"] == "cpu"
58+
assert devices[1]["type"] == "xpu"
59+
assert devices[1]["name"] == "Intel(R) Graphics [0x7d41]"
60+
assert devices[1]["memory"] == 36022263808
61+
assert devices[1]["index"] == 0
62+
63+
def test_get_devices_with_cuda(self, fxt_system_service: Mock, fxt_client: TestClient):
64+
"""Test GET /api/system/devices with NVIDIA CUDA"""
65+
fxt_system_service.get_devices.return_value = [
66+
DeviceInfo(type=DeviceType.CPU, name="CPU", memory=None, index=None),
67+
DeviceInfo(type=DeviceType.CUDA, name="NVIDIA GeForce RTX 4090", memory=25769803776, index=0),
68+
]
69+
70+
response = fxt_client.get("/api/system/devices")
71+
72+
assert response.status_code == status.HTTP_200_OK
73+
devices = response.json()
74+
assert len(devices) == 2
75+
assert devices[0]["type"] == "cpu"
76+
assert devices[1]["type"] == "cuda"
77+
assert devices[1]["name"] == "NVIDIA GeForce RTX 4090"
78+
assert devices[1]["memory"] == 25769803776
79+
assert devices[1]["index"] == 0
80+
81+
def test_get_devices_with_all_devices(self, fxt_system_service: Mock, fxt_client: TestClient):
82+
"""Test GET /api/system/devices with all device types"""
83+
fxt_system_service.get_devices.return_value = [
84+
DeviceInfo(type=DeviceType.CPU, name="CPU", memory=None, index=None),
85+
DeviceInfo(type=DeviceType.XPU, name="Intel(R) Graphics [0x7d41]", memory=36022263808, index=0),
86+
DeviceInfo(type=DeviceType.CUDA, name="NVIDIA GeForce RTX 4090", memory=25769803776, index=0),
87+
]
88+
89+
response = fxt_client.get("/api/system/devices")
90+
91+
assert response.status_code == status.HTTP_200_OK
92+
devices = response.json()
93+
assert len(devices) == 3
94+
assert devices[0]["type"] == "cpu"
95+
assert devices[1]["type"] == "xpu"
96+
assert devices[2]["type"] == "cuda"
97+
98+
def test_get_memory(self, fxt_system_service: Mock, fxt_client: TestClient):
99+
"""Test GET /api/system/metrics/memory"""
100+
fxt_system_service.get_memory_usage.return_value = (1024.5, 8192.0)
101+
102+
response = fxt_client.get("/api/system/metrics/memory")
103+
104+
assert response.status_code == status.HTTP_200_OK
105+
memory = response.json()
106+
assert memory["used"] == 1024
107+
assert memory["total"] == 8192
Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
# Copyright (C) 2025 Intel Corporation
2+
# SPDX-License-Identifier: Apache-2.0
3+
4+
from unittest.mock import MagicMock, patch
5+
6+
import pytest
7+
8+
from app.services.system_service import SystemService
9+
10+
11+
class TestSystemService:
12+
"""Test cases for SystemService"""
13+
14+
@pytest.fixture
15+
def fxt_system_service(self) -> SystemService:
16+
return SystemService()
17+
18+
def test_get_memory_usage(self, fxt_system_service: SystemService):
19+
"""Test getting memory usage"""
20+
used, total = fxt_system_service.get_memory_usage()
21+
22+
assert used > 0
23+
assert total > 0
24+
assert used <= total
25+
26+
def test_get_cpu_usage(self, fxt_system_service: SystemService):
27+
"""Test getting CPU usage"""
28+
cpu_usage = fxt_system_service.get_cpu_usage()
29+
30+
assert cpu_usage >= 0.0
31+
32+
def test_get_devices_cpu_only(self, fxt_system_service: SystemService):
33+
"""Test getting devices when only CPU is available"""
34+
with patch("app.services.system_service.torch") as mock_torch:
35+
# Simulate torch not being available
36+
mock_torch.xpu.is_available.return_value = False
37+
mock_torch.cuda.is_available.return_value = False
38+
39+
devices = fxt_system_service.get_devices()
40+
41+
assert len(devices) == 1
42+
assert devices[0].name == "CPU"
43+
assert devices[0].memory is None
44+
assert devices[0].index is None
45+
46+
def test_get_devices_with_xpu(self, fxt_system_service: SystemService):
47+
"""Test getting devices when Intel XPU is available"""
48+
with patch("app.services.system_service.torch") as mock_torch:
49+
# Mock XPU device
50+
mock_dp = MagicMock()
51+
mock_dp.name = "Intel(R) Graphics [0x7d41]"
52+
mock_dp.total_memory = 36022263808
53+
54+
mock_torch.xpu.is_available.return_value = True
55+
mock_torch.xpu.device_count.return_value = 1
56+
mock_torch.xpu.get_device_properties.return_value = mock_dp
57+
58+
# CUDA not available
59+
mock_torch.cuda.is_available.return_value = False
60+
61+
devices = fxt_system_service.get_devices()
62+
63+
assert len(devices) == 2
64+
assert devices[1].name == "Intel(R) Graphics [0x7d41]"
65+
assert devices[1].memory == 36022263808
66+
assert devices[1].index == 0
67+
68+
def test_get_devices_with_cuda(self, fxt_system_service: SystemService):
69+
"""Test getting devices when NVIDIA CUDA is available"""
70+
with patch("app.services.system_service.torch") as mock_torch:
71+
# XPU not available
72+
mock_torch.xpu.is_available.return_value = False
73+
74+
# Mock CUDA device
75+
mock_dp = MagicMock()
76+
mock_dp.name = "NVIDIA GeForce RTX 4090"
77+
mock_dp.total_memory = 25769803776
78+
79+
mock_torch.cuda.is_available.return_value = True
80+
mock_torch.cuda.device_count.return_value = 1
81+
mock_torch.cuda.get_device_properties.return_value = mock_dp
82+
83+
devices = fxt_system_service.get_devices()
84+
85+
assert len(devices) == 2
86+
assert devices[1].name == "NVIDIA GeForce RTX 4090"
87+
assert devices[1].memory == 25769803776
88+
assert devices[1].index == 0
89+
90+
def test_get_devices_with_multiple_devices(self, fxt_system_service: SystemService):
91+
"""Test getting devices when multiple GPUs are available"""
92+
with patch("app.services.system_service.torch") as mock_torch:
93+
# Mock XPU device
94+
mock_xpu_dp = MagicMock()
95+
mock_xpu_dp.name = "Intel(R) Graphics [0x7d41]"
96+
mock_xpu_dp.total_memory = 36022263808
97+
98+
mock_torch.xpu.is_available.return_value = True
99+
mock_torch.xpu.device_count.return_value = 1
100+
mock_torch.xpu.get_device_properties.return_value = mock_xpu_dp
101+
102+
# Mock CUDA device
103+
mock_cuda_dp = MagicMock()
104+
mock_cuda_dp.name = "NVIDIA GeForce RTX 4090"
105+
mock_cuda_dp.total_memory = 25769803776
106+
107+
mock_torch.cuda.is_available.return_value = True
108+
mock_torch.cuda.device_count.return_value = 1
109+
mock_torch.cuda.get_device_properties.return_value = mock_cuda_dp
110+
111+
devices = fxt_system_service.get_devices()
112+
113+
assert len(devices) == 3

0 commit comments

Comments
 (0)