Python

status_command.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import time
import json
import subprocess
import select
import os
import re

# --- Dependency-Free Stat Functions ---

# Global variables to store the previous CPU times for calculating usage percentage
prev_idle_time, prev_total_time = 0, 0

# Global variables for network speed calculation
prev_rx_bytes, prev_tx_bytes, last_speed_check_time = 0, 0, 0

def get_cpu_usage():
 """
 Calculates CPU usage and formats it as 'usage%/cores'.
 Returns a tuple of (usage_string, usage_percentage).
 """
 global prev_idle_time, prev_total_time
 try:
 # Get number of CPU cores
 num_cores = os.cpu_count()

 with open('/proc/stat') as f:
 line = f.readline()
 parts = line.split()
 idle_time = int(parts[4]) + int(parts[5])
 total_time = sum(map(int, parts[1:]))

 delta_idle = idle_time - prev_idle_time
 delta_total = total_time - prev_total_time

 cpu_percentage = 0.0
 if delta_total > 0:
 cpu_percentage = (1.0 - (delta_idle / delta_total)) * 100

 prev_idle_time, prev_total_time = idle_time, total_time
 
 # Format the output string to include core count
 formatted_string = f"{cpu_percentage:.1f}%/{num_cores}"
 return formatted_string, cpu_percentage
 except (IOError, IndexError):
 return "N/A", 0

def get_ram_usage():
 """
 Calculates RAM usage and formats it as 'used_GB/total_GB'.
 Returns a tuple of (usage_string, usage_percentage).
 """
 try:
 with open('/proc/meminfo') as f:
 lines = f.readlines()
 
 def find_mem(key):
 for line in lines:
 if line.startswith(key):
 return int(line.split()[1])
 return 0

 mem_total_kib = find_mem('MemTotal:')
 mem_available_kib = find_mem('MemAvailable:')
 
 if mem_total_kib > 0:
 mem_used_kib = mem_total_kib - mem_available_kib
 ram_percentage = (mem_used_kib / mem_total_kib) * 100
 
 # Convert used and total memory from KiB to GB
 mem_used_gb = mem_used_kib / (1024 * 1024)
 mem_total_gb = round(mem_total_kib / (1024 * 1024))
 
 # Format the output string as 'used_GB/total_GB' (e.g., "2.1/16")
 formatted_string = f"{mem_used_gb:.1f}/{mem_total_gb}"
 return formatted_string, ram_percentage
 
 return "N/A", 0
 except IOError:
 return "N/A", 0

def get_temp():
 """
 Gets CPU temperature from the first available thermal zone.
 Returns a tuple of (temp_string, temp_celsius).
 """
 try:
 for i in range(5):
 path = f"/sys/class/thermal/thermal_zone{i}/temp"
 if os.path.exists(path):
 with open(path) as f:
 temp_milli_c = int(f.read().strip())
 temp_c = temp_milli_c / 1000
 return f"{temp_c:.0f}°C", temp_c
 return "N/A", 0
 except (IOError, ValueError):
 return "N/A", 0

def get_conditional_color(value, warn_threshold, crit_threshold):
 """Returns a color based on the value against thresholds."""
 if value >= crit_threshold:
 return "#FF5555" # Red
 if value >= warn_threshold:
 return "#FFB86C" # Yellow/Orange
 return "#8BE9FD" # Default color (Cyan)

def get_system_stats():
 """Consolidates all system stats into a single Pango-formatted string."""
 cpu_str, cpu_val = get_cpu_usage()
 ram_str, ram_val = get_ram_usage()
 temp_str, temp_val = get_temp()

 # Get colors based on values
 cpu_color = get_conditional_color(cpu_val, 50, 85)
 ram_color = get_conditional_color(ram_val, 60, 85)
 temp_color = get_conditional_color(temp_val, 65, 80)

 # Build Pango markup string with the new memory icon
 cpu_icon = f"󰍛"
 ram_icon = f""
 temp_icon = f"🌡️"

 return f"{cpu_icon} {cpu_str} | {ram_icon} {ram_str} | {temp_icon} {temp_str}"
 
# --- New Modules with Dependencies ---

def get_active_interface():
 """Finds the default active network interface."""
 try:
 result = subprocess.run(
 ["ip", "route", "get", "1.1.1.1"], 
 capture_output=True, text=True, check=True, timeout=0.5
 )
 # Handle cases where output might be unexpected
 if "dev" in result.stdout:
 return result.stdout.split("dev")[1].split()[0]
 return None
 except (subprocess.CalledProcessError, FileNotFoundError, IndexError, subprocess.TimeoutExpired):
 return None

def format_speed(bytes_per_second):
 """Formats speed in a human-readable way (KB/s, MB/s)."""
 if bytes_per_second < 1024:
 return f"{bytes_per_second:.0f} B/s"
 elif bytes_per_second < 1024**2:
 return f"{bytes_per_second / 1024:.1f} K"
 else:
 return f"{bytes_per_second / (1024**2):.1f} M"

def format_usage(byte_total):
 """Formats total data usage in a human-readable way (MB, GB)."""
 if byte_total < 1024**2:
 return f"{byte_total / 1024:.1f} KB"
 if byte_total < 1024**3:
 return f"{byte_total / (1024**2):.1f} MB"
 else:
 return f"{byte_total / (1024**3):.2f} GB"

def get_network_speed():
 """
 Calculates and formats current download and upload speed.
 """
 global prev_rx_bytes, prev_tx_bytes, last_speed_check_time
 
 interface = get_active_interface()
 if not interface:
 return " -- K  -- K"

 try:
 with open('/proc/net/dev', 'r') as f:
 lines = f.readlines()
 
 for line in lines:
 if interface in line:
 data = line.split()
 # On some systems, the interface name is followed by a colon
 # e.g., 'wlan0:', so we adjust the index accordingly.
 offset = 1 if ":" in data[0] else 0
 rx_bytes = int(data[offset])
 tx_bytes = int(data[offset + 8])
 
 current_time = time.time()
 time_delta = current_time - last_speed_check_time

 # Calculate speed only if it's not the first run
 if last_speed_check_time > 0 and time_delta > 0:
 rx_speed = (rx_bytes - prev_rx_bytes) / time_delta
 tx_speed = (tx_bytes - prev_tx_bytes) / time_delta
 
 # Update previous values for the next calculation
 prev_rx_bytes, prev_tx_bytes = rx_bytes, tx_bytes
 last_speed_check_time = current_time
 
 return f" {format_speed(rx_speed)}  {format_speed(tx_speed)}"
 
 # For the first run, just set the initial values
 prev_rx_bytes, prev_tx_bytes = rx_bytes, tx_bytes
 last_speed_check_time = current_time
 return " -- K  -- K"

 except (IOError, IndexError, ValueError):
 pass # Silently fail and show default
 
 return " -- K  -- K"


def get_network_usage():
 """
 Tracks and displays total data usage for the day.
 Persists data in ~/.cache/status_command_usage.json
 """
 USAGE_FILE = os.path.expanduser("~/.cache/status_command_usage.json")
 interface = get_active_interface()
 if not interface:
 return " -- GB"

 try:
 # Get current total bytes from /proc/net/dev
 with open('/proc/net/dev', 'r') as f:
 lines = f.readlines()
 
 current_proc_total = 0
 for line in lines:
 if interface in line:
 data = line.split()
 offset = 1 if ":" in data[0] else 0
 current_proc_total = int(data[offset]) + int(data[offset + 8])
 break
 
 if current_proc_total == 0: return " -- GB"

 # Read persistent data
 today_str = time.strftime('%Y-%m-%d')
 daily_usage = 0
 
 try:
 with open(USAGE_FILE, 'r') as f:
 data = json.load(f)
 
 # Check if we are still on the same day
 if data.get('date') == today_str:
 session_usage = 0
 # If current /proc value is less, a reboot happened
 if current_proc_total < data.get('last_proc_total', 0):
 session_usage = current_proc_total
 else:
 session_usage = current_proc_total - data.get('last_proc_total', 0)
 
 daily_usage = data.get('total_usage', 0) + session_usage
 else: # New day, reset counter
 daily_usage = current_proc_total

 except (FileNotFoundError, json.JSONDecodeError):
 # If file doesn't exist or is invalid, start fresh
 daily_usage = current_proc_total

 # Write updated data back to the file
 os.makedirs(os.path.dirname(USAGE_FILE), exist_ok=True)
 with open(USAGE_FILE, 'w') as f:
 json.dump({
 'date': today_str,
 'total_usage': daily_usage,
 'last_proc_total': current_proc_total
 }, f)

 return f" {format_usage(daily_usage)}"

 except (IOError, IndexError, ValueError):
 return " -- GB"


def get_network_status():
 """
 Gets network status for Wi-Fi (SSID) or Ethernet (IP).
 Requires 'ip' and 'iw' commands.
 """
 interface = get_active_interface()
 if not interface:
 return "󰌙 Disconnected"
 
 try:
 # Check if it's Wi-Fi
 if interface.startswith("wl"):
 try:
 ssid_result = subprocess.run(["iwgetid", "-r"], capture_output=True, text=True, check=True)
 ssid = ssid_result.stdout.strip()
 return f"󰖩 {ssid}"
 except (subprocess.CalledProcessError, FileNotFoundError):
 return "󰖩 Wi-Fi"
 else: # Assumed Ethernet
 ip_result = subprocess.run(["ip", "-4", "addr", "show", interface], capture_output=True, text=True, check=True)
 ip_addr = ip_result.stdout.split("inet ")[1].split("/")[0]
 return f"󰈀 {ip_addr}"
 
 except (subprocess.CalledProcessError, FileNotFoundError, IndexError):
 return "󰌙 Disconnected"

def get_volume():
 """
 Gets current volume and mute status from PulseAudio/PipeWire.
 Requires 'pactl' command.
 Returns a tuple of (is_muted, display_text).
 """
 try:
 pactl_output = subprocess.check_output(["pactl", "get-sink-volume", "@DEFAULT_SINK@"], text=True)
 pactl_mute_output = subprocess.check_output(["pactl", "get-sink-mute", "@DEFAULT_SINK@"], text=True)
 
 is_muted = "yes" in pactl_mute_output.lower()
 
 if is_muted:
 return True, "󰝟 Muted"
 
 # Find the first percentage value
 match = re.search(r'(\d+)%', pactl_output)
 if match:
 volume = int(match.group(1))
 icon = "" # Speaker high
 if volume == 0:
 icon = "" # Speaker off
 elif volume < 50:
 icon = "" # Speaker low
 return False, f"{icon} {volume}%"
 
 return False, " N/A"
 
 except (subprocess.CalledProcessError, FileNotFoundError):
 return False, " N/A"


# --- Main Application Logic ---

input_buffer = ""

def handle_click_events(new_data):
 """
 Processes incoming click event data from the input buffer.
 This function no longer waits for input, it just processes it.
 """
 global input_buffer
 if new_data:
 input_buffer += new_data

 # Process all complete JSON objects in the buffer
 while '\n' in input_buffer:
 line, input_buffer = input_buffer.split('\n', 1)
 
 # Clean up potential leading commas or whitespace from swaybar
 clean_line = line.strip().lstrip(',')
 
 if not clean_line or clean_line == '[':
 continue
 
 try:
 event = json.loads(clean_line)
 name = event.get("name")
 
 # Define commands for each clickable block
 commands = {
 "pactl_mute": ["pactl", "set-sink-mute", "@DEFAULT_SINK@", "toggle"],
 "pactl_vol_down": ["pactl", "set-sink-volume", "@DEFAULT_SINK@", "-5%"],
 "pactl_vol_up": ["pactl", "set-sink-volume", "@DEFAULT_SINK@", "+5%"],
 "logout_button": ["swaynag", "-t", "warning", "-m", "Are you sure you want to log out?", "-b", "Yes, Logout", "swaymsg exit"],
 "shutdown_button": ["swaynag", "-t", "warning", "-m", "Are you sure you want to shut down?", "-b", "Yes, Shutdown", "systemctl poweroff"]
 }
 if name in commands:
 # Use Popen to run in the background without waiting
 subprocess.Popen(commands[name], stdin=subprocess.DEVNULL)
 except json.JSONDecodeError:
 continue

def main():
 """
 Main event loop to handle timed updates and click events without delay.
 """
 # Print the header for swaybar protocol version 1
 print(json.dumps({"version": 1, "click_events": True}))
 print('[')
 # Print an empty array first for faster startup
 print("[],")
 sys.stdout.flush()

 last_update_time = 0
 update_interval = 1.0 # seconds

 while True:
 current_time = time.time()
 
 # --- Timed Update: Refresh the status bar text every 'update_interval' ---
 if (current_time - last_update_time) >= update_interval:
 is_muted, volume_text = get_volume()
 
 # Build the list of blocks for the status bar
 status_blocks = [
 
 {"full_text": get_network_speed(), "name": "network_speed", "separator": True, "separator_block_width": 20, "markup": "pango"},
 {"full_text": get_network_usage(), "name": "network_usage", "separator": True, "separator_block_width": 20, "markup": "pango"},
 {"full_text": get_network_status(), "name": "network_status", "separator": True, "separator_block_width": 20, "markup": "pango"},
 {"full_text": get_system_stats(), "name": "system_stats", "separator": True, "separator_block_width": 20, "markup": "pango"},
 {"full_text": f" {time.strftime('%a %d %b | %H:%M:%S')}", "name": "datetime", "separator": True, "separator_block_width": 20, "markup": "pango"}
 ]

 # Conditionally add Volume Blocks
 if is_muted:
 status_blocks.append({
 "full_text": volume_text, "name": "pactl_mute", 
 "separator": True, "separator_block_width": 20
 })
 else:
 status_blocks.extend([
 {"full_text": "", "name": "pactl_vol_down", "separator": False, "separator_block_width": 8},
 {"full_text": volume_text, "name": "volume_text", "separator": False, "separator_block_width": 8},
 {"full_text": "", "name": "pactl_vol_up", "separator": True, "separator_block_width": 20}
 ])

 # Add final static right-aligned blocks
 status_blocks.extend([
 {"full_text": "󰗼 Logout", "name": "logout_button", "separator": True, "separator_block_width": 10},
 {"full_text": "󰐥 Shutdown", "name": "shutdown_button", "separator": False}
 ])
 
 print(json.dumps(status_blocks) + ",", flush=True)
 last_update_time = current_time

 # --- Event Handling: Wait for clicks or until the next update is due ---
 # Calculate how long select() should wait.
 # This is the key to making clicks responsive.
 timeout = max(0, update_interval - (time.time() - last_update_time))
 
 # select() waits for input on stdin OR until the timeout is reached.
 readable, _, _ = select.select([sys.stdin], [], [], timeout)
 
 if readable:
 # A click event has arrived. Read and process it immediately.
 data = os.read(sys.stdin.fileno(), 4096).decode("utf-8")
 handle_click_events(data)

if __name__ == "__main__":
 try:
 main()
 except KeyboardInterrupt:
 sys.exit(0)

Status command script for swaybar