Рефакторинг. docstring
This commit is contained in:
parent
060f4de175
commit
37f9e3abfb
14 changed files with 1234 additions and 243 deletions
|
|
@ -23,16 +23,35 @@ import subprocess
|
|||
|
||||
|
||||
class NetworkService:
|
||||
"""Сервис для взаимодействия с NetworkManager через D-Bus"""
|
||||
"""Сервис для взаимодействия с NetworkManager через D-Bus.
|
||||
|
||||
Предоставляет асинхронные методы для получения информации о сетевых интерфейсах,
|
||||
их конфигурациях IP, Wi-Fi и состояниях.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""Инициализирует сервис NetworkManager.
|
||||
|
||||
Устанавливает системный D-Bus и создаёт объекты для взаимодействия с
|
||||
NetworkManager и его настройками.
|
||||
"""
|
||||
self.system_bus = sd_bus_open_system()
|
||||
self.nm = NetworkManager(self.system_bus)
|
||||
self.nm_settings = NetworkManagerSettings(self.system_bus)
|
||||
|
||||
async def get_ip_method(self, device: NetworkDeviceGeneric,
|
||||
ip_type: str) -> str:
|
||||
"""Получает метод IP для устройства"""
|
||||
async def get_ip_method(self, device: NetworkDeviceGeneric, ip_type: str) -> str:
|
||||
"""Получает метод IP (например, "auto", "manual") для указанного устройства.
|
||||
|
||||
Args:
|
||||
device (NetworkDeviceGeneric): Объект устройства NetworkManager.
|
||||
ip_type (str): Тип IP (например, "ipv4" или "ipv6").
|
||||
|
||||
Returns:
|
||||
str: Метод IP или "unknown" в случае ошибки.
|
||||
|
||||
Raises:
|
||||
Exception: Если произошла ошибка при получении настроек соединения.
|
||||
"""
|
||||
try:
|
||||
config = await device.get_applied_connection()
|
||||
settings = config[0].get(ip_type, {})
|
||||
|
|
@ -41,64 +60,80 @@ class NetworkService:
|
|||
return "unknown"
|
||||
|
||||
async def get_connection_name(self, device: NetworkDeviceGeneric) -> str:
|
||||
"""Извлекает имя соединения (GENERAL.CONNECTION) для устройства"""
|
||||
"""Извлекает имя соединения (GENERAL.CONNECTION) для устройства.
|
||||
|
||||
Args:
|
||||
device (NetworkDeviceGeneric): Объект устройства NetworkManager.
|
||||
|
||||
Returns:
|
||||
str: Имя соединения или "unknown" в случае ошибки.
|
||||
|
||||
Raises:
|
||||
Exception: Если произошла ошибка при получении настроек соединения.
|
||||
"""
|
||||
try:
|
||||
applied_connection = await device.get_applied_connection()
|
||||
connection_settings = applied_connection[0]
|
||||
connection_info = connection_settings.get("connection", {})
|
||||
return str(connection_info.get("id", ("s", "unknown"))[1])
|
||||
except Exception as e:
|
||||
# print(f"Ошибка при получении соединения для устройства {await device.interface}: {e}")
|
||||
# срёт ошибки дл отключенных интерфейсов
|
||||
# добавить потом условия для отключенных интерфейсов
|
||||
return "unknown"
|
||||
|
||||
async def get_ip_config(self, device: NetworkDeviceGeneric,
|
||||
ip_version: str) -> List[IndividualIPConfig]:
|
||||
"""Получает конфигурацию IP для устройства"""
|
||||
async def get_ip_config(self, device: NetworkDeviceGeneric, ip_version: str) -> List[IndividualIPConfig]:
|
||||
"""Получает конфигурацию IP для устройства.
|
||||
|
||||
Args:
|
||||
device (NetworkDeviceGeneric): Объект устройства NetworkManager.
|
||||
ip_version (str): Версия IP ("4" или "6").
|
||||
|
||||
Returns:
|
||||
List[IndividualIPConfig]: Список конфигураций IP, включая адреса и маршруты.
|
||||
|
||||
Raises:
|
||||
Exception: Если произошла ошибка при получении данных IP.
|
||||
"""
|
||||
configs = []
|
||||
device_ip = await getattr(device, f"ip{ip_version}_config")
|
||||
if device_ip != "/":
|
||||
ip_config = globals()[f"IPv{ip_version}Config"](device_ip,
|
||||
self.system_bus)
|
||||
ip_config = globals()[f"IPv{ip_version}Config"](device_ip, self.system_bus)
|
||||
address_data = await ip_config.address_data
|
||||
for address in address_data:
|
||||
config = IndividualIPConfig(
|
||||
address=str(address["address"][1]),
|
||||
routes=[
|
||||
globals()[f"IPv{ip_version}Route"](
|
||||
destination=route.get(
|
||||
"dest",
|
||||
("s",
|
||||
"0.0.0.0" if ip_version == "4" else "::"))[1],
|
||||
destination=route.get("dest", ("s", "0.0.0.0" if ip_version == "4" else "::"))[1],
|
||||
prefix=route.get("prefix", ("u", 0))[1],
|
||||
next_hop=route.get(
|
||||
"next-hop",
|
||||
("s",
|
||||
"0.0.0.0" if ip_version == "4" else "::"),
|
||||
)[1],
|
||||
next_hop=route.get("next-hop", ("s", "0.0.0.0" if ip_version == "4" else "::"))[1],
|
||||
metric=route.get("metric", ("u", 0))[1],
|
||||
) for route in await ip_config.route_data
|
||||
],
|
||||
method=await self.get_ip_method(device,
|
||||
f"ipv{ip_version}"),
|
||||
method=await self.get_ip_method(device, f"ipv{ip_version}"),
|
||||
gateway=str(await ip_config.gateway),
|
||||
dns=[
|
||||
str(ns["address"][1]) if ip_version == "4" else str(ns)
|
||||
for ns in await getattr(
|
||||
ip_config,
|
||||
"nameserver_data" if ip_version ==
|
||||
"4" else "nameservers",
|
||||
"nameserver_data" if ip_version == "4" else "nameservers",
|
||||
)
|
||||
],
|
||||
)
|
||||
configs.append(config)
|
||||
return configs
|
||||
|
||||
async def get_wifi_info(
|
||||
self, wifi_device: NetworkDeviceWireless
|
||||
) -> Dict[str, Optional[Union[str, int]]]:
|
||||
"""Получает информацию о Wi-Fi устройстве"""
|
||||
async def get_wifi_info(self, wifi_device: NetworkDeviceWireless) -> Dict[str, Optional[Union[str, int]]]:
|
||||
"""Получает информацию о Wi-Fi устройстве.
|
||||
|
||||
Args:
|
||||
wifi_device (NetworkDeviceWireless): Объект Wi-Fi устройства NetworkManager.
|
||||
|
||||
Returns:
|
||||
Dict[str, Optional[Union[str, int]]]: Словарь с данными Wi-Fi (SSID, BSSID,
|
||||
bitrate, strength, frequency, mode), где значения могут быть None.
|
||||
|
||||
Raises:
|
||||
Exception: Если произошла ошибка при получении данных точки доступа.
|
||||
"""
|
||||
active_ap = await wifi_device.active_access_point
|
||||
wifi_info = {
|
||||
"ssid": None,
|
||||
|
|
@ -123,7 +158,18 @@ class NetworkService:
|
|||
self,
|
||||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||||
) -> List[Union[NetworkInterface, WiFiDevice]]:
|
||||
"""Получает список сетевых интерфейсов"""
|
||||
"""Получает список сетевых интерфейсов.
|
||||
|
||||
Args:
|
||||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||||
координат (x, y) для интерфейсов. По умолчанию None (используются (100, 100)).
|
||||
|
||||
Returns:
|
||||
List[Union[NetworkInterface, WiFiDevice]]: Список объектов интерфейсов,
|
||||
включая Wi-Fi устройства.
|
||||
|
||||
Исключает интерфейсы типа LOOPBACK и неактивные состояния (state == "10").
|
||||
"""
|
||||
devices_paths = await self.nm.get_devices()
|
||||
network_interfaces = []
|
||||
initial_positions = initial_positions or {}
|
||||
|
|
@ -141,43 +187,37 @@ class NetworkService:
|
|||
continue
|
||||
|
||||
common_data = {
|
||||
"name":
|
||||
interface_name,
|
||||
"type":
|
||||
device_type,
|
||||
"device_state":
|
||||
device_state,
|
||||
"active_connection":
|
||||
await self.get_connection_name(generic_device),
|
||||
"mac_address":
|
||||
str(await generic_device.hw_address),
|
||||
"device_mtu":
|
||||
str(await generic_device.mtu),
|
||||
"ipv4":
|
||||
IPConfig(configurations=await self.get_ip_config(
|
||||
generic_device, "4")),
|
||||
"ipv6":
|
||||
IPConfig(configurations=await self.get_ip_config(
|
||||
generic_device, "6")),
|
||||
"x":
|
||||
initial_positions.get(interface_name, (100, 100))[0],
|
||||
"y":
|
||||
initial_positions.get(interface_name, (100, 100))[1],
|
||||
"name": interface_name,
|
||||
"type": device_type,
|
||||
"device_state": device_state,
|
||||
"active_connection": await self.get_connection_name(generic_device),
|
||||
"mac_address": str(await generic_device.hw_address),
|
||||
"device_mtu": str(await generic_device.mtu),
|
||||
"ipv4": IPConfig(configurations=await self.get_ip_config(generic_device, "4")),
|
||||
"ipv6": IPConfig(configurations=await self.get_ip_config(generic_device, "6")),
|
||||
"x": initial_positions.get(interface_name, (100, 100))[0],
|
||||
"y": initial_positions.get(interface_name, (100, 100))[1],
|
||||
}
|
||||
|
||||
if device_type == "WIFI":
|
||||
wifi_device = NetworkDeviceWireless(device_path,
|
||||
self.system_bus)
|
||||
wifi_device = NetworkDeviceWireless(device_path, self.system_bus)
|
||||
wifi_info = await self.get_wifi_info(wifi_device)
|
||||
network_interfaces.append(
|
||||
WiFiDevice(**common_data, **wifi_info))
|
||||
network_interfaces.append(WiFiDevice(**common_data, **wifi_info))
|
||||
else:
|
||||
network_interfaces.append(NetworkInterface(**common_data))
|
||||
|
||||
return network_interfaces
|
||||
|
||||
async def get_hostname(self) -> Dict[str, str]:
|
||||
"""Возвращает имя хоста из команды hostname в формате JSON"""
|
||||
"""Возвращает имя хоста из команды hostname в формате JSON.
|
||||
|
||||
Returns:
|
||||
Dict[str, str]: Словарь с ключом "hostname" и значением имени хоста.
|
||||
|
||||
Raises:
|
||||
RuntimeError: Если произошла ошибка при выполнении команды hostname
|
||||
или обработке вывода.
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(["hostname"],
|
||||
capture_output=True,
|
||||
|
|
@ -195,8 +235,15 @@ class NetworkService:
|
|||
state: str,
|
||||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||||
) -> List[NetworkInterface]:
|
||||
"""
|
||||
Получает список интерфейсов по состоянию
|
||||
"""Получает список интерфейсов по указанному состоянию.
|
||||
|
||||
Args:
|
||||
state (str): Состояние интерфейса (например, "100" для активных).
|
||||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||||
координат (x, y) для интерфейсов. По умолчанию None.
|
||||
|
||||
Returns:
|
||||
List[NetworkInterface]: Список интерфейсов с указанным состоянием.
|
||||
"""
|
||||
all_interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return [
|
||||
|
|
@ -211,10 +258,21 @@ class NetworkService:
|
|||
container_x: int = 0,
|
||||
container_y: int = 0,
|
||||
) -> NetworkInterfacesContainer:
|
||||
"""Получает объект, содержащий список интерфейсов и координаты контейнера"""
|
||||
"""Получает объект, содержащий список интерфейсов и координаты контейнера.
|
||||
|
||||
Args:
|
||||
state (Optional[str], optional): Состояние интерфейсов (например, "100").
|
||||
По умолчанию None (все интерфейсы).
|
||||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||||
координат (x, y) для интерфейсов. По умолчанию None.
|
||||
container_x (int, optional): Координата X контейнера. По умолчанию 0.
|
||||
container_y (int, optional): Координата Y контейнера. По умолчанию 0.
|
||||
|
||||
Returns:
|
||||
NetworkInterfacesContainer: Объект с интерфейсами и координатами контейнера.
|
||||
"""
|
||||
if state:
|
||||
interfaces = await self.get_interfaces_by_state(
|
||||
state, initial_positions)
|
||||
interfaces = await self.get_interfaces_by_state(state, initial_positions)
|
||||
else:
|
||||
interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return NetworkInterfacesContainer(interfaces=interfaces,
|
||||
|
|
@ -225,7 +283,15 @@ class NetworkService:
|
|||
self,
|
||||
initial_positions: Optional[Dict[str, Tuple[int, int]]] = None,
|
||||
) -> NetworkInterfacesContainer:
|
||||
"""Получает контейнер с сетевыми интерфейсами и координатами контейнера"""
|
||||
"""Получает контейнер с сетевыми интерфейсами и координатами контейнера.
|
||||
|
||||
Args:
|
||||
initial_positions (Optional[Dict[str, Tuple[int, int]]], optional):
|
||||
Словарь начальных координат (x, y) для интерфейсов. По умолчанию None.
|
||||
|
||||
Returns:
|
||||
NetworkInterfacesContainer: Объект с интерфейсами и координатами (0, 0).
|
||||
"""
|
||||
interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return NetworkInterfacesContainer(interfaces=interfaces,
|
||||
container_x=0,
|
||||
|
|
@ -235,14 +301,30 @@ class NetworkService:
|
|||
self,
|
||||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||||
) -> List[NetworkInterface]:
|
||||
"""Получает список активных сетевых интерфейсов"""
|
||||
"""Получает список активных сетевых интерфейсов.
|
||||
|
||||
Args:
|
||||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||||
координат (x, y) для интерфейсов. По умолчанию None.
|
||||
|
||||
Returns:
|
||||
List[NetworkInterface]: Список активных интерфейсов (state == "100").
|
||||
"""
|
||||
return await self.get_interfaces_by_state("100", initial_positions)
|
||||
|
||||
async def get_deactivated_interfaces(
|
||||
self,
|
||||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||||
) -> List[NetworkInterface]:
|
||||
"""Получает список неактивных сетевых интерфейсов"""
|
||||
"""Получает список неактивных сетевых интерфейсов.
|
||||
|
||||
Args:
|
||||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||||
координат (x, y) для интерфейсов. По умолчанию None.
|
||||
|
||||
Returns:
|
||||
List[NetworkInterface]: Список интерфейсов с state != "100".
|
||||
"""
|
||||
all_interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return [
|
||||
interface for interface in all_interfaces
|
||||
|
|
@ -255,9 +337,18 @@ class NetworkService:
|
|||
container_x: int = 0,
|
||||
container_y: int = 0,
|
||||
) -> NetworkInterfacesContainer:
|
||||
"""Получает объект, содержащий список активных интерфейсов и координаты контейнера"""
|
||||
return await self.get_interfaces_container("100", initial_positions,
|
||||
container_x, container_y)
|
||||
"""Получает объект, содержащий список активных интерфейсов и координаты контейнера.
|
||||
|
||||
Args:
|
||||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||||
координат (x, y) для интерфейсов. По умолчанию None.
|
||||
container_x (int, optional): Координата X контейнера. По умолчанию 0.
|
||||
container_y (int, optional): Координата Y контейнера. По умолчанию 0.
|
||||
|
||||
Returns:
|
||||
NetworkInterfacesContainer: Объект с активными интерфейсами и координатами.
|
||||
"""
|
||||
return await self.get_interfaces_container("100", initial_positions, container_x, container_y)
|
||||
|
||||
async def get_deactivated_interfaces_container(
|
||||
self,
|
||||
|
|
@ -265,8 +356,16 @@ class NetworkService:
|
|||
container_x: int = 0,
|
||||
container_y: int = 0,
|
||||
) -> NetworkInterfacesContainer:
|
||||
"""Получает объект, содержащий список неактивных интерфейсов и координаты контейнера"""
|
||||
"""Получает объект, содержащий список неактивных интерфейсов и координаты контейнера.
|
||||
|
||||
Args:
|
||||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||||
координат (x, y) для интерфейсов. По умолчанию None.
|
||||
container_x (int, optional): Координата X контейнера. По умолчанию 0.
|
||||
container_y (int, optional): Координата Y контейнера. По умолчанию 0.
|
||||
|
||||
Returns:
|
||||
NetworkInterfacesContainer: Объект с неактивными интерфейсами и координатами.
|
||||
"""
|
||||
interfaces = await self.get_deactivated_interfaces(initial_positions)
|
||||
return NetworkInterfacesContainer(interfaces=interfaces,
|
||||
container_x=container_x,
|
||||
container_y=container_y)
|
||||
return NetworkInterfacesContainer(interfaces=interfaces, container_x=container_x, container_y=container_y)
|
||||
Loading…
Add table
Add a link
Reference in a new issue