network_speed_monitor_colab.py

import time
import threading
import ipywidgets as widgets
from IPython.display import display, HTML
import datetime

# Function to read current network usage
def get_network_usage():
    try:
        with open('/proc/net/dev', 'r') as f:
            lines = f.readlines()

        for line in lines:
            if 'eth0' in line:  # 'eth0' is the main Colab network interface
                data = line.split()
                download = int(data[1])  # Bytes received
                upload = int(data[9])  # Bytes sent
                return download, upload

        return 0, 0  # Default if no data found
    except Exception as e:
        return 0, 0  # Silently handle errors to avoid bleeding into other cells

# Create dashboard widgets
title = widgets.HTML(value="<h3 style='margin-bottom:10px'>📡 Network Monitor</h3>")
download_text = widgets.HTML(value="⬇️ Download: 0.000 MB/s")
upload_text = widgets.HTML(value="⬆️ Upload: 0.000 MB/s")
timestamp = widgets.HTML(value="Last update: -")
stop_button = widgets.Button(description="Stop", button_style="danger", 
                            layout=widgets.Layout(width='80px'))
status_indicator = widgets.HTML(value="<span style='color:green'>●</span> Active")

# Global control variable
monitor_active = {'value': True}

# Update function (to be called from the monitoring thread)
def update_widget_values(download_speed, upload_speed):
    download_text.value = f"⬇️ Download: {download_speed:.3f} MB/s"
    upload_text.value = f"⬆️ Upload: {upload_speed:.3f} MB/s"
    timestamp.value = f"Last update: {datetime.datetime.now().strftime('%H:%M:%S')}"

# The monitoring function that runs in background
def network_monitor_thread(stop_flag):
    prev_download, prev_upload = get_network_usage()
    
    while stop_flag['value']:
        try:
            time.sleep(1)
            curr_download, curr_upload = get_network_usage()
            
            # Calculate speeds
            download_speed = (curr_download - prev_download) / 1024 / 1024  # MB/s
            upload_speed = (curr_upload - prev_upload) / 1024 / 1024  # MB/s
            
            # Update previous values
            prev_download, prev_upload = curr_download, curr_upload
            
            # Update the widgets (thread-safe operation)
            update_widget_values(download_speed, upload_speed)
            
        except Exception:
            # Silent exception handling to avoid output bleeding
            pass

# Button click handler
def on_stop_button_clicked(b):
    monitor_active['value'] = False
    status_indicator.value = "<span style='color:red'>●</span> Stopped"
    b.description = "Stopped"
    b.disabled = True

# Start monitoring
def start_network_monitor():
    # Set up the monitoring thread
    stop_button.on_click(on_stop_button_clicked)
    
    # Create dashboard layout
    dashboard = widgets.VBox([
        title,
        widgets.HBox([download_text, upload_text]),
        widgets.HBox([timestamp, status_indicator, stop_button], 
                    layout=widgets.Layout(
                        justify_content='space-between',
                        margin='10px 0 0 0'
                    ))
    ], layout=widgets.Layout(
        border='1px solid #ddd',
        padding='10px',
        margin='10px 0',
        background_color='#f8f8f8',
        width='400px'
    ))
    
    # Display the dashboard
    display(dashboard)
    
    # Start the monitoring thread
    thread = threading.Thread(target=network_monitor_thread, args=(monitor_active,))
    thread.daemon = True  # Thread will exit when main program exits
    thread.start()
    
    # Return a message without polluting output
    return "Network monitor started. You can continue working in other cells."

# Call this function to start monitoring
start_network_monitor()