Можно использовать стандартную библиотеку `socket` в Python для проверки того, доступно ли устройство в сети по его IP-адресу и порту. Ниже пример кода, который выполняет подключение к устройству через TCP:
```python
import socket
def is_device_up(ip, port, timeout=3):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
try:
result = sock.connect_ex((ip, port))
if result == 0:
return True
else:
return False
except socket.error as e:
return False
finally:
sock.close()
# Пример использования
ip_address = '192.168.1.10'
port = 80
if is_device_up(ip_address, port):
print(f"Устройство {ip_address} на порту {port} доступно")
else:
print(f"Устройство {ip_address} на порту {port} не доступно")
```
Функция `is_device_up` возвращает `True`, если устройство успешно ответило на запрос подключения, и `False`, если подключение не удалось. Она использует метод `connect_ex`, который возвращает 0 при успешном подключении и код ошибки в противном случае. Вы также можете задать время ожидания `timeout`.
Если вам нужно проверить большое количество устройств, вы можете использовать библиотеку `concurrent.futures` для параллельной проверки:
```python
from concurrent.futures import ThreadPoolExecutor
# Список IP адресов и портов
devices = [('192.168.1.10', 80), ('192.168.1.11', 80), ...]
# Функция для одновременной проверки доступности устройств
def check_devices(devices, timeout=3, max_threads=20):
with ThreadPoolExecutor(max_threads) as executor:
futures = {executor.submit(is_device_up, ip, port, timeout): (ip, port) for ip, port in devices}
results = {}
for future in concurrent.futures.as_completed(futures):
ip, port = futures[future]
try:
result = future.result()
results[(ip, port)] = result
except Exception as e:
print(f"Ошибка при проверке устройства {ip} на порту {port}: {e}")
return results
# Запуск проверки
devices_status = check_devices(devices)
for (ip, port), is_up in devices_status.items():
print(f"Устройство {ip} на порту {port} {'доступно' if is_up else 'не доступно'}")
```
В этом примере используется `ThreadPoolExecutor` для создания пула потоков, которые параллельно осуществляют проверку устройств. Вы можете настроить количество потоков в пуле с помощью аргумента `max_threads`.
Обратите внимание, что выставление большого количества параллельных потоков может привести к нагрузке как на сетевую инфраструктуру, так и на целевые устройства, поэтому стоит заранее убедиться, что у вас есть разрешение на такую проверку и ваша сеть справится с подобной нагрузкой.