ping из python скрипта
Вопрос из конца: Возможно ли написать это на python более лаконично, не теряя в отказоустойчивости, избегая страшных bash регулярок?
Для начала хочу сказать что на python я пишу буквально около недели (до этого только Си/С++).
Мне необходимо вызвать команду ping
из моего python скрипта, причём важен не только ответ о доступности по адресу, но и данные времени и работы самого ping
. Важно отметить что скрипт будет запускаться на машине, где нет возможности доставить пакеты из pip.
Для начала я поступил довольно прямым способом - написал функцию-парсер для вывода subprocess.run
и выгядит она так
SYSTEM_PING_ITERATIONS: int = 1
SYSTEM_CMD_PING : str = "ping -4 -c{0} {1} | tail -n2"
def run_system_cmd(cmd: str) -> Tuple[List[str], int]:
ret = subprocess.run(cmd,
shell=True,
executable='/bin/bash',
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT
)
output = ret.stdout.decode('utf-8', errors='ignore')
out_split: List[str] = output.split('\n')
return (out_split[:-1] if out_split[-1] == '' else out_split, ret.returncode)
def process_ping(interations: int, address: Union[str, List[int]]) -> Dict[str, Union[int, float]]:
address = address if isinstance(address, str) else '.'.join(map(str, address))
from_cmd = run_system_cmd(str.format(SYSTEM_CMD_PING, interations, address))
if from_cmd[1] != 0:
return {}
output = from_cmd[0]
if not output or len(output) != 2:
return {}
suffixes: Dict[str, float] = {
"ns": 1000000.0,
"us": 1000.0,
"ms": 1.0,
"s" : 0.001
}
result: Dict[str, Union[int, float]] = {
"transmitted": 0 ,
"received" : 0 ,
"loss" : 0.0,
"total" : 0.0,
"min" : 0.0,
"max" : 0.0,
"avg" : 0.0
}
data1 = output[0].split(' ')
if len(data1) < 10:
return {}
result['transmitted'] = int(data1[0])
result['received'] = int(data1[3])
result['loss'] = float(re.sub('\\D', '', data1[5]))
result['total'] = float(re.sub('\\D', '', data1[9])) * suffixes[re.sub(r'[^A-Za-z]', '', data1[9])]
# do not parse further, if destination unreachable
if result['loss'] == 100.0:
return result
data2 = output[1].split(' ')[-2:]
splitted = data2[0].split('/')
result['min'] = float(splitted[0]) * suffixes[data2[1]]
result['max'] = float(splitted[2]) * suffixes[data2[1]]
result['avg'] = float(splitted[1]) * suffixes[data2[1]]
return result
Довольно много кода, для такой простой задачи...
Следующее, что мне пришло в голову в отсутствие достаточных знаний по python - изменить команду bash:
ping -c{0} {1} | tail -n2 | awk -F'[ |,]' '/packets transmitted/ {transmitted=$1; received=$5; loss=$8; time=$13} /rtt min\/avg\/max/ {split($4, rtt, "/"); min=rtt[1]; avg=rtt[2]; max=rtt[3]} END {print transmitted, received, loss, time, min, avg, max, $5}'
И вывод у неё будет такой для доступного и недоступного адреса соотвественно:
1 1 0% 0ms 36.477 36.477 36.477 ms
1 0 100% 0ms
Это парсить и проверять всяко легче, нежели изначальный вариант.
Так вот теперь вопрос: Возможно ли написать это на python более лаконично, не теряя в отказоустойчивости, избегая страшных bash регулярок?
Ответы (1 шт):
Ваше решение в одну строку выглядит вполне читаемо, ИМХО
Тру-Пайтон-стайл многословный, длинный, ни разу не однострочный. Поэтому стоит ли городить страницы текста для замены скрипта в одну строку?
#!/usr/bin/env python3
"ping"
import argparse
from subprocess import CompletedProcess, run, PIPE
from itertools import dropwhile
from typing import Iterable, Tuple
def run_ping(address: str, count: int = 4, interval: float = 0.1, args:Iterable = []) -> CompletedProcess:
'''Запускает ping и ждёт завершения процесса'''
return run(['ping', '-i', str(interval), '-c', str(count), '-q', address, *args],
stdout=PIPE,
stderr=PIPE,
check=False,
text=True,
encoding='ascii',
env={'LANG': 'C'},
)
def get_ping_stats_text(ping_proc: CompletedProcess) -> Tuple[str | None, str | None]:
'''Возвращает две последние строки из вывода ping.
Если строк меньше двух, возвращает (None, None).'''
lines = dropwhile(
lambda line: not line.startswith('---'),
ping_proc.stdout.splitlines())
# lines состоит из строки '--- ping statistics ---' и двух строк с данными
lines = list(lines)
if len(lines) < 3:
return None, None
return lines[1:]
TRANSMITTED = 'packets transmitted'
RECEIVED = 'received'
LOST = 'packet loss'
TIME = 'time'
ERRORS = 'errors'
def parse_packet_data(line: str) -> dict:
'''Возвращает набор пар с ключами 'packets transmitted', 'received', 'packet loss', 'time'.'''
return dict(
map(
# в выводе ping сначала идёт значение, затем ключ
# кроме случая с 'time', там наоборот
lambda p: (p[1], p[0]) if p[0] != 'time' else (p[0], p[1]),
map(
# разделить пару на первое слово и остальной текст
lambda _str: _str.strip().split(' ', 1),
# Разбить строку на пары, разделённые запятыми
line.split(',')
)
)
)
def parse_ping_rtt(line: str) -> Tuple[list, str]:
'''Возвращает список времён RTT в мс.'''
if ' = ' not in line:
return None, None
times_str, unit = line.split(' = ')[1].split()
times = times_str.split('/')
return times, unit
def get_ping_stats(ping_proc: CompletedProcess) -> dict|None:
'''Возвращает словарь с данными о пинге.'''
first, second = get_ping_stats_text(ping_proc)
if first is None:
return None
stats = parse_packet_data(first)
times, unit = parse_ping_rtt(second)
return {
'packets': stats,
'times': times ,
'unit': unit,
}
def flatten_ping_stats(stats: dict) -> list:
'''Возвращает список значений из словаря stats.'''
result = [
stats['packets'][TRANSMITTED],
stats['packets'][RECEIVED],
stats['packets'][LOST],
stats['packets'][TIME],
]
if stats['times']:
result.extend(stats['times'])
result.append(stats['unit'])
return result
def main():
'''Вызывает ping и выдаёт сокращённые результаты.'''
arg_parser = argparse.ArgumentParser(
description='Ping a host and print the results')
arg_parser.add_argument(
'address',
help='The address to ping')
arg_parser.add_argument(
'-c', '--count',
type=int,
default=10,
help='The number of packets to send')
arg_parser.add_argument(
'-i', '--interval',
type=float,
default=0.1,
help='The interval between packets')
arg_parser.add_argument(
'extra_args',
nargs=argparse.REMAINDER,
help='Extra arguments to pass to ping')
args = arg_parser.parse_args()
address = args.address
ping_proc = run_ping(address, args.count, args.interval, args.extra_args)
stats = get_ping_stats(ping_proc)
if stats is None:
print('No data')
return
print(' '.join(flatten_ping_stats(stats)))
if __name__ == '__main__':
main()