[Python] 네트워크 트래픽 모니터링 프로그램 만들기

JaehyoJJAng2024년 02월 10일

개요

네트워크 인터페이스별로 실시간 트래픽 모니터링을 수행하는 프로그램을 구현해봅시다.

필요 라이브러리 불러오기

import psutil  # 파이썬을 위한 실행중인 프로세스 및 시스템 리소스 및 정보 검색을 위한 크로스 플랫폼 라이브러리
import time  # 시간 라이브러리 
import os  # 운영체제에서 제공되는 여러 기능을 파이썬에서 수행시켜주는 라이브러리 
import pandas as pd # 판다스 라이브러리

psutil: 시스템 및 프로세스 정보를 조회하는 라이브러리입니다.
time: 시간과 관련된 기능을 제공하는 라이브러리입니다.
os: 운영체제와 관련된 기능을 수행하는 라이브러리입니다.
pandas: 데이터 조작 및 분석을 위한 라이브러리입니다.

바이트를 사람이 읽기 좋은 형식으로 변환하기

def get_size(bytes: int) -> str:
    """
    Returns size of bytes in a nice format
    """
    for unit in ['', 'K', 'M', 'G', 'T', 'P']:
        if bytes < 1024:        # IEC prefix names 기반 표기 기준 1 KiB = 1024 bytes
            return f"{bytes:.2f}{unit}B"
        bytes /= 1024

bytes 매개변수를 받아서
1024로 나누면서 적절한 단위(K, M, G 등)로 바꾸고
소수점 둘째 자리까지 표시된 값과 단위를 문자열로 반환합니다.

네트워크 인터페이스별 I/O 통계 추출

psutil 라이브러리를 사용하여 현재 네트워크 인터페이스별 I/O 통계를 가져옵니다.

io = psutil.net_io_counters(pernic=True)

pernic=True는 각 네트워크 인터페이스별로 통계를 가져오게 합니다.

다운로드 및 업로드 속도 계산하고 실시간 네트워크 사용량 출력

코드의 나머지 부분은 무한 루프로, 주기적으로 네트워크 트래픽을 모니터링합니다.

while True:
    time.sleep(UPDATE_DELAY)  # 'UPDATE_DELAY' time.sleep 
    io_2 = psutil.net_io_counters(pernic=True)  # 다시 가져오기
    
    data: list[dict] = []
    for iface, iface_io in io.items():
        upload_speed, download_speed = io_2[iface].bytes_sent - iface_io.bytes_sent, io_2[iface].bytes_recv - iface_io.bytes_recv
        data.append({
            "iface": iface,
            "Download": get_size(io_2[iface].bytes_recv),
            "Upload": get_size(io_2[iface].bytes_sent),
            "Upload Speed": f"{get_size(upload_speed / UPDATE_DELAY)}/s",
            "Download Speed": f"{get_size(download_speed / UPDATE_DELAY)}/s",
        })
        
    io = io_2  # 다음 반복을 위해 I/O 통계 업데이트 
    df = pd.DataFrame(data)  # 데이터를 판다스 데이터프레임으로 변환
    df.sort_values("Download", inplace=True, ascending=False)  # 다운로드 기준 내림차순 정렬 
    
    os.system("cls") if "nt" in os.name else os.system("clear")  # 화면 지우기 
    print(df.to_string())  # 최종 통계 출력

루프 시작: while True: 무한 루프를 사용하여 지속적으로 실행합니다.
지연 시간: time.sleep(UPDATE_DELAY)를 사용하여 1초(설정된 지연 시간) 동안 대기합니다.
I/O 통계 가져오기: io_2 = psutil.net_io_counters(pernic=True)를 사용하여 최신 네트워크 I/O 통계를 가져옵니다.
데이터 초기화: data라는 빈 리스트를 만듭니다.
네트워크 인터페이스별 데이터 계산:
- 각 네트워크 인터페이스(iface)에 대해 현재(io_2)와 이전(io)의 바이트 수를 비교하여 업로드 속도와 다운로드 속도를 계산합니다.
- data.append를 사용하여 각 인터페이스의 이름, 총 다운로드 및 업로드 바이트, 다운로드 속도 및 업로드 속도를 리스트에 추가합니다.
I/O 통계 업데이트: io = io_2를 사용하여 다음 반복을 위해 현재 통계를 저장합니다.
데이터프레임 생성 및 정렬: pd.DataFrame(data)를 사용하여 데이터를 판다스 데이터프레임으로 변환하고, 다운로드 바이트를 기준으로 내림차순으로 정렬합니다.
화면 지우기: os.system("cls") if "nt" in os.name else os.system("clear")를 사용하여 운영체제에 따라 콘솔 화면을 지웁니다.
결과 출력: print(df.to_string())를 사용하여 데이터프레임을 문자열 형식으로 변환하여 콘솔에 출력합니다.

python