基于Python实现单实例后台倒计时计时器及命令行交互功能

Alright, let's build that custom CLI countdown timer you're looking for. I'll break down each key component step by step, so you can put it all together and tweak it as needed.

We’ll split this tool into two parts:

• A background daemon process that handles the countdown logic and plays alert sounds.
• A CLI control client that sends commands (start/stop/remaining) and communicates with the daemon.
• We’ll enforce single-instance behavior to avoid conflicting timers running at the same time.

1. Single-Instance Enforcement #

To prevent multiple daemon processes from running, we’ll use a PID file to track the active process. Before starting, we check if the file exists and if the process is still alive with psutil:

import os
import psutil

PID_FILE = "/tmp/countdown_timer.pid"

def is_timer_running():
    if os.path.exists(PID_FILE):
        with open(PID_FILE, "r") as f:
            pid = int(f.read().strip())
        return psutil.pid_exists(pid)
    return False

def save_pid():
    with open(PID_FILE, "w") as f:
        f.write(str(os.getpid()))

def cleanup_pid():
    if os.path.exists(PID_FILE):
        os.unlink(PID_FILE)

For more robustness, you could use file locking (via fcntl on Unix) instead of a PID file—this avoids race conditions where two processes check the PID file at the same time.

2. Inter-Process Communication (IPC) #

The CLI client needs to talk to the background daemon. Unix domain sockets (or named pipes on Windows) are lightweight and perfect for this. Here’s how the flow works:

• The daemon creates and listens on a socket.
• The CLI client connects to the socket, sends a command, and receives a response.

3. CLI Command Parsing #

Use Python’s built-in argparse to handle the three command types cleanly:

import argparse

def parse_cli_args():
    parser = argparse.ArgumentParser(description="Custom CLI Countdown Timer")
    subparsers = parser.add_subparsers(dest="command", required=True)
    
    # Start command with minute input
    start_cmd = subparsers.add_parser("start")
    start_cmd.add_argument("minutes", type=int, help="Number of minutes to count down from")
    
    # Stop command
    subparsers.add_parser("stop")
    
    # Remaining time query command
    subparsers.add_parser("remaining")
    
    return parser.parse_args()

4. Countdown Logic & Alert Sounds #

The daemon handles the countdown loop, and plays beeps when the timer is near ending or finishes. We’ll use simpleaudio for cross-platform sound playback (install it with pip install simpleaudio numpy):

import time
import numpy as np
import simpleaudio as sa

def play_short_beep():
    # Generate a sharp 440Hz beep for 0.2 seconds
    freq = 440
    duration = 0.2
    sample_rate = 44100
    t = np.linspace(0, duration, int(sample_rate * duration), False)
    note = np.sin(freq * t * 2 * np.pi)
    # Convert to 16-bit audio for playback
    audio = note * (2**15 - 1) / np.max(np.abs(note))
    audio = audio.astype(np.int16)
    play_obj = sa.play_buffer(audio, 1, 2, sample_rate)
    play_obj.wait_done()

def play_long_beep():
    # Longer beep to signal countdown completion
    freq = 440
    duration = 1.0
    sample_rate = 44100
    t = np.linspace(0, duration, int(sample_rate * duration), False)
    note = np.sin(freq * t * 2 * np.pi)
    audio = note * (2**15 - 1) / np.max(np.abs(note))
    audio = audio.astype(np.int16)
    play_obj = sa.play_buffer(audio, 1, 2, sample_rate)
    play_obj.wait_done()

def run_countdown(seconds_left):
    while seconds_left > 0:
        time.sleep(1)
        seconds_left -= 1
        # Beep every second for the last 10 seconds
        if seconds_left <= 10:
            play_short_beep()
    # Final alert when timer finishes
    play_long_beep()

Here’s how to tie all these parts together into a working script:

import os
import sys
import time
import argparse
import socket
import psutil
import numpy as np
import simpleaudio as sa

# Configuration
PID_FILE = "/tmp/countdown_timer.pid"
SOCKET_PATH = "/tmp/countdown_timer.sock"

def is_timer_running():
    if os.path.exists(PID_FILE):
        with open(PID_FILE, "r") as f:
            pid = int(f.read().strip())
        return psutil.pid_exists(pid)
    return False

def save_pid():
    with open(PID_FILE, "w") as f:
        f.write(str(os.getpid()))

def cleanup_pid():
    if os.path.exists(PID_FILE):
        os.unlink(PID_FILE)

def play_short_beep():
    freq = 440
    duration = 0.2
    sample_rate = 44100
    t = np.linspace(0, duration, int(sample_rate * duration), False)
    note = np.sin(freq * t * 2 * np.pi)
    audio = note * (2**15 - 1) / np.max(np.abs(note))
    audio = audio.astype(np.int16)
    play_obj = sa.play_buffer(audio, 1, 2, sample_rate)
    play_obj.wait_done()

def play_long_beep():
    freq = 440
    duration = 1.0
    sample_rate = 44100
    t = np.linspace(0, duration, int(sample_rate * duration), False)
    note = np.sin(freq * t * 2 * np.pi)
    audio = note * (2**15 - 1) / np.max(np.abs(note))
    audio = audio.astype(np.int16)
    play_obj = sa.play_buffer(audio, 1, 2, sample_rate)
    play_obj.wait_done()

def daemonize_process():
    # Basic daemonization for Unix systems
    if os.fork() > 0:
        sys.exit(0)
    os.setsid()
    if os.fork() > 0:
        sys.exit(0)
    # Redirect standard streams to /dev/null to detach from terminal
    sys.stdout.flush()
    sys.stderr.flush()
    with open('/dev/null', 'r') as f:
        os.dup2(f.fileno(), sys.stdin.fileno())
    with open('/dev/null', 'w') as f:
        os.dup2(f.fileno(), sys.stdout.fileno())
    with open('/dev/null', 'w') as f:
        os.dup2(f.fileno(), sys.stderr.fileno())

def run_daemon():
    save_pid()
    # Clean up old socket if it exists from a previous run
    if os.path.exists(SOCKET_PATH):
        os.unlink(SOCKET_PATH)
    
    # Set up socket listening for CLI commands
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.bind(SOCKET_PATH)
    sock.listen(1)
    
    active_seconds = 0
    is_running = False
    
    try:
        while True:
            # Accept incoming CLI connections
            conn, _ = sock.accept()
            command = conn.recv(1024).decode().strip()
            response = ""
            
            if command.startswith("start"):
                _, mins = command.split()
                active_seconds = int(mins) * 60
                is_running = True
                response = f"Countdown started: {mins} minutes"
            elif command == "stop":
                is_running = False
                active_seconds = 0
                response = "Countdown stopped"
            elif command == "remaining":
                if is_running:
                    mins = active_seconds // 60
                    secs = active_seconds % 60
                    response = f"Remaining: {mins}m {secs}s"
                else:
                    response = "No active countdown"
            
            conn.sendall(response.encode())
            conn.close()
            
            # Run countdown loop if active
            while is_running and active_seconds > 0:
                time.sleep(1)
                active_seconds -= 1
                if active_seconds <= 10:
                    play_short_beep()
                
                # Check for new commands without blocking the countdown
                sock.setblocking(False)
                try:
                    conn, _ = sock.accept()
                    cmd = conn.recv(1024).decode().strip()
                    if cmd == "stop":
                        is_running = False
                        active_seconds = 0
                        conn.sendall("Countdown stopped".encode())
                    conn.close()
                except BlockingIOError:
                    pass
                sock.setblocking(True)
            
            # Play final beep when countdown finishes
            if active_seconds == 0 and is_running:
                is_running = False
                play_long_beep()
    finally:
        cleanup_pid()
        if os.path.exists(SOCKET_PATH):
            os.unlink(SOCKET_PATH)

def send_cli_command(command):
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        sock.connect(SOCKET_PATH)
        sock.sendall(command.encode())
        response = sock.recv(1024).decode()
        print(response)
    except ConnectionRefusedError:
        print("Error: Timer daemon is not running")
    finally:
        sock.close()

def parse_cli_args():
    parser = argparse.ArgumentParser(description="Custom CLI Countdown Timer")
    subparsers = parser.add_subparsers(dest="command", required=True)
    
    start_cmd = subparsers.add_parser("start")
    start_cmd.add_argument("minutes", type=int, help="Number of minutes to count down from")
    
    subparsers.add_parser("stop")
    subparsers.add_parser("remaining")
    
    return parser.parse_args()

def main():
    args = parse_cli_args()
    
    if args.command == "start":
        if is_timer_running():
            print("Error: A countdown timer is already running")
            sys.exit(1)
        daemonize_process()
        run_daemon()
    elif args.command == "stop":
        send_cli_command("stop")
    elif args.command == "remaining":
        send_cli_command("remaining")

if __name__ == "__main__":
    main()

• Windows Compatibility: Replace Unix domain sockets with named pipes, and use win32api/win32con for daemonization instead of the Unix fork method.
• Sound Alternatives: If simpleaudio doesn’t work for you, use winsound (Windows-only) or playsound (cross-platform).
• File Locking: For better single-instance safety, replace the PID file with fcntl.flock (Unix) or msvcrt.locking (Windows) to prevent race conditions.

内容的提问来源于stack exchange，提问作者Jersey