gistfile1.txt
· 690 B · Text
Bruto
Пример вывода при успешном
```
Ping monitor for example.com
Status: pinging... (connection stable)
Current series: packets_lost=0 | duration=0.0 sec | 2023-05-15 14:30:45
Press Ctrl+C to exit...
```
Пример вывода при потере пакетов
```
Ping monitor for example.com
Status: CONNECTION LOST!
Current series: packets_lost=3 | duration=3.0 sec | started at 2023-05-15 14:31:02
Outage history (sorted by packets lost):
------------------------------------------------------------
Start: 2023-05-15 14:25:10 | Packets lost: 5 | Duration: 5.0 sec
Start: 2023-05-15 14:20:30 | Packets lost: 2 | Duration: 2.0 sec
Press Ctrl+C to exit...
```
1 | Пример вывода при успешном |
2 | ``` |
3 | Ping monitor for example.com |
4 | |
5 | Status: pinging... (connection stable) |
6 | Current series: packets_lost=0 | duration=0.0 sec | 2023-05-15 14:30:45 |
7 | |
8 | Press Ctrl+C to exit... |
9 | ``` |
10 | |
11 | Пример вывода при потере пакетов |
12 | ``` |
13 | Ping monitor for example.com |
14 | |
15 | Status: CONNECTION LOST! |
16 | Current series: packets_lost=3 | duration=3.0 sec | started at 2023-05-15 14:31:02 |
17 | |
18 | Outage history (sorted by packets lost): |
19 | ------------------------------------------------------------ |
20 | Start: 2023-05-15 14:25:10 | Packets lost: 5 | Duration: 5.0 sec |
21 | Start: 2023-05-15 14:20:30 | Packets lost: 2 | Duration: 2.0 sec |
22 | |
23 | Press Ctrl+C to exit... |
24 | ``` |
ping_monitor.py
· 4.3 KiB · Python
Bruto
#!/usr/bin/env python3
import sys
import subprocess
import time
from datetime import datetime
from collections import deque
class PingMonitor:
def __init__(self, host):
self.host = host
self.current_outage = None
self.outages = deque(maxlen=100) # Ограничиваем количество хранимых серий
self.last_success_time = datetime.now()
def ping_host(self):
try:
# -c 1: один пакет, -W 1: таймаут 1 секунда (для Linux)
# Для Windows нужно использовать "-n 1 -w 1000"
result = subprocess.run(['ping', '-c', '1', '-W', '1', self.host],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True)
return result.returncode == 0
except Exception as e:
print(f"Error pinging host: {e}")
return False
def monitor(self):
while True:
success = self.ping_host()
now = datetime.now()
if not success and self.current_outage is None:
# Начало новой серии разрывов
self.current_outage = {
'start': now,
'packets_lost': 1,
'end': None
}
elif not success and self.current_outage is not None:
# Продолжение текущей серии разрывов
self.current_outage['packets_lost'] += 1
elif success and self.current_outage is not None:
# Серия разрывов закончилась
self.current_outage['end'] = now
self.outages.append(self.current_outage)
self.current_outage = None
self.last_success_time = now
elif success:
# Успешный пинг, обновляем время последнего успеха
self.last_success_time = now
self.print_status()
time.sleep(1) # Пауза между пингами
def print_status(self):
# Очищаем экран (работает в терминалах, поддерживающих ANSI коды)
print("\033c", end="")
print(f"Ping monitor for {self.host}\n")
# Всегда показываем текущий статус
if self.current_outage is None:
print("Status: pinging... (connection stable)")
current_duration = (datetime.now() - self.last_success_time).total_seconds()
print(f"Current series: packets_lost=0 | duration={current_duration:.1f} sec | {datetime.now()}")
else:
current_duration = (datetime.now() - self.current_outage['start']).total_seconds()
print(f"Status: CONNECTION LOST!")
print(f"Current series: packets_lost={self.current_outage['packets_lost']} | "
f"duration={current_duration:.1f} sec | started at {self.current_outage['start']}")
# Выводим историю разрывов, если они есть
if self.outages:
print("\nOutage history (sorted by packets lost):")
print("-" * 60)
# Сортируем серии по количеству потерянных пакетов (по убыванию)
sorted_outages = sorted(self.outages,
key=lambda x: x['packets_lost'],
reverse=True)
for outage in sorted_outages:
duration = (outage['end'] - outage['start']).total_seconds()
print(f"Start: {outage['start']} | "
f"Packets lost: {outage['packets_lost']} | "
f"Duration: {duration:.1f} sec")
print("\nPress Ctrl+C to exit...")
def main():
if len(sys.argv) != 2:
print("Usage: ./ping_monitor.py <host>")
sys.exit(1)
host = sys.argv[1]
monitor = PingMonitor(host)
try:
monitor.monitor()
except KeyboardInterrupt:
print("\nMonitoring stopped.")
if __name__ == "__main__":
main()
1 | #!/usr/bin/env python3 |
2 | import sys |
3 | import subprocess |
4 | import time |
5 | from datetime import datetime |
6 | from collections import deque |
7 | |
8 | class PingMonitor: |
9 | def __init__(self, host): |
10 | self.host = host |
11 | self.current_outage = None |
12 | self.outages = deque(maxlen=100) # Ограничиваем количество хранимых серий |
13 | self.last_success_time = datetime.now() |
14 | |
15 | def ping_host(self): |
16 | try: |
17 | # -c 1: один пакет, -W 1: таймаут 1 секунда (для Linux) |
18 | # Для Windows нужно использовать "-n 1 -w 1000" |
19 | result = subprocess.run(['ping', '-c', '1', '-W', '1', self.host], |
20 | stdout=subprocess.PIPE, |
21 | stderr=subprocess.PIPE, |
22 | text=True) |
23 | return result.returncode == 0 |
24 | except Exception as e: |
25 | print(f"Error pinging host: {e}") |
26 | return False |
27 | |
28 | def monitor(self): |
29 | while True: |
30 | success = self.ping_host() |
31 | now = datetime.now() |
32 | |
33 | if not success and self.current_outage is None: |
34 | # Начало новой серии разрывов |
35 | self.current_outage = { |
36 | 'start': now, |
37 | 'packets_lost': 1, |
38 | 'end': None |
39 | } |
40 | elif not success and self.current_outage is not None: |
41 | # Продолжение текущей серии разрывов |
42 | self.current_outage['packets_lost'] += 1 |
43 | elif success and self.current_outage is not None: |
44 | # Серия разрывов закончилась |
45 | self.current_outage['end'] = now |
46 | self.outages.append(self.current_outage) |
47 | self.current_outage = None |
48 | self.last_success_time = now |
49 | elif success: |
50 | # Успешный пинг, обновляем время последнего успеха |
51 | self.last_success_time = now |
52 | |
53 | self.print_status() |
54 | time.sleep(1) # Пауза между пингами |
55 | |
56 | def print_status(self): |
57 | # Очищаем экран (работает в терминалах, поддерживающих ANSI коды) |
58 | print("\033c", end="") |
59 | print(f"Ping monitor for {self.host}\n") |
60 | |
61 | # Всегда показываем текущий статус |
62 | if self.current_outage is None: |
63 | print("Status: pinging... (connection stable)") |
64 | current_duration = (datetime.now() - self.last_success_time).total_seconds() |
65 | print(f"Current series: packets_lost=0 | duration={current_duration:.1f} sec | {datetime.now()}") |
66 | else: |
67 | current_duration = (datetime.now() - self.current_outage['start']).total_seconds() |
68 | print(f"Status: CONNECTION LOST!") |
69 | print(f"Current series: packets_lost={self.current_outage['packets_lost']} | " |
70 | f"duration={current_duration:.1f} sec | started at {self.current_outage['start']}") |
71 | |
72 | # Выводим историю разрывов, если они есть |
73 | if self.outages: |
74 | print("\nOutage history (sorted by packets lost):") |
75 | print("-" * 60) |
76 | |
77 | # Сортируем серии по количеству потерянных пакетов (по убыванию) |
78 | sorted_outages = sorted(self.outages, |
79 | key=lambda x: x['packets_lost'], |
80 | reverse=True) |
81 | |
82 | for outage in sorted_outages: |
83 | duration = (outage['end'] - outage['start']).total_seconds() |
84 | print(f"Start: {outage['start']} | " |
85 | f"Packets lost: {outage['packets_lost']} | " |
86 | f"Duration: {duration:.1f} sec") |
87 | |
88 | print("\nPress Ctrl+C to exit...") |
89 | |
90 | def main(): |
91 | if len(sys.argv) != 2: |
92 | print("Usage: ./ping_monitor.py <host>") |
93 | sys.exit(1) |
94 | |
95 | host = sys.argv[1] |
96 | monitor = PingMonitor(host) |
97 | |
98 | try: |
99 | monitor.monitor() |
100 | except KeyboardInterrupt: |
101 | print("\nMonitoring stopped.") |
102 | |
103 | if __name__ == "__main__": |
104 | main() |