杰瑞科技汇

python select stdin

Of course! Using Python's select module to monitor stdin is a powerful technique for writing non-blocking or event-driven command-line applications. It allows your program to check if there's data ready to be read from standard input without getting stuck waiting, which is crucial for handling other tasks simultaneously (like network sockets or timers).

python select stdin-图1
(图片来源网络,侵删)

Here’s a comprehensive guide, from the basic concept to practical examples.

The Core Concept: select.select()

The select module uses a system call of the same name to monitor multiple file descriptors (FDs). A file descriptor is just an integer that the operating system uses to identify an open I/O resource. For your terminal, stdin is file descriptor 0.

The select.select() function has this signature: select.select(rlist, wlist, xlist[, timeout])

  • rlist: A list of file descriptors to monitor for read readiness. We will put sys.stdin here.
  • wlist: A list of file descriptors to monitor for write readiness. We'll ignore this.
  • xlist: A list of file descriptors to monitor for "exceptional" conditions. We'll ignore this.
  • timeout: A floating-point number specifying the maximum time to wait.
    • If timeout is a number, select will block for at most that many seconds.
    • If timeout is None, select will block indefinitely until a file descriptor is ready.
    • If timeout is 0, select will not block and will return immediately, checking the current status of the FDs.

Return Value: select() returns three new lists:

python select stdin-图2
(图片来源网络,侵删)
  1. A list of FDs from rlist that are ready for reading.
  2. A list of FDs from wlist that are ready for writing.
  3. A list of FDs from xlist with exceptional conditions.

For our purpose, we only care about the first returned list.


Setup: Importing and Getting stdin's File Descriptor

First, you need to import the necessary modules and get the file descriptor for stdin.

import sys
import select
# sys.stdin is a file object. We need its underlying file descriptor (an integer).
# On most systems, stdin is file descriptor 0.
stdin_fd = sys.stdin.fileno()
print("Ready for input. Press Ctrl+D (on Unix) or Ctrl+Z then Enter (on Windows) to exit.")

Example 1: Non-blocking Read

This is the most common use case. You want to check if the user has typed something, but if not, you don't want your program to halt. You can then do other work.

import sys
import select
def non_blocking_stdin_example():
    """
    Reads from stdin in a non-blocking way.
    """
    print("Type something and press Enter. I'll print it when it's ready.")
    print("I'll also print a message every second if there's no input.")
    # sys.stdin is a file object. We need its underlying file descriptor (an integer).
    stdin_fd = sys.stdin.fileno()
    while True:
        # Check if stdin has data to be read.
        # The select() call will return immediately (timeout=0).
        # The first list in the returned tuple contains the FDs ready for reading.
        ready_to_read, _, _ = select.select([stdin_fd], [], [], 0.0)
        if ready_to_read:
            # If the list is not empty, it means stdin is ready.
            data = sys.stdin.readline()
            if data: # readline() returns an empty string on EOF (Ctrl+D / Ctrl+Z)
                print(f"You typed: {data.strip()}")
            else:
                print("EOF detected. Exiting.")
                break
        else:
            # If the list is empty, no data is available right now.
            # We can do other things here.
            print(".", end="", flush=True)
            # We add a small sleep to prevent the loop from consuming 100% CPU.
            # Note: In a real app, you'd do work here instead of sleeping.
            import time
            time.sleep(1)
if __name__ == "__main__":
    non_blocking_stdin_example()

How to run it:

python select stdin-图3
(图片来源网络,侵删)
  1. Save the code as a file (e.g., non_blocking_stdin.py).
  2. Run it from your terminal: python non_blocking_stdin.py.
  3. You'll see a printed every second.
  4. Type something and press Enter. Your input will be printed immediately, and the will continue.

Example 2: Timeout-based Read

Sometimes you want to wait for input for a specific period, but then give up and continue with other logic.

import sys
import select
def timeout_stdin_example():
    """
    Waits for up to 5 seconds for input from stdin.
    """
    print("Waiting for input for 5 seconds...")
    print("Type something and press Enter within that time.")
    stdin_fd = sys.stdin.fileno()
    # Wait for up to 5 seconds for stdin to become ready.
    # The first list in the returned tuple contains the FDs ready for reading.
    ready_to_read, _, _ = select.select([stdin_fd], [], [], 5.0)
    if ready_to_read:
        # Data was received within the 5-second timeout.
        data = sys.stdin.readline()
        print(f"Success! You typed: {data.strip()}")
    else:
        # The 5-second timeout elapsed with no input.
        print("Timeout! No input received within 5 seconds.")
if __name__ == "__main__":
    timeout_stdin_example()

How to run it:

  1. Save the code as timeout_stdin.py.
  2. Run it: python timeout_stdin.py.
  3. You have 5 seconds to type and press Enter. If you do, it prints your input. If you don't, it prints "Timeout!".

Advanced Example: Monitoring stdin and a Timer

This example demonstrates the real power of select: handling multiple "events" in a single loop. We'll create a fake timer using time.time() and check for both input and the timer's expiration.

import sys
import select
import time
def multi_event_example():
    """
    Monitors both stdin and a timer using select.
    """
    print("Monitoring stdin and a 10-second timer.")
    print("Type something to see it printed immediately.")
    print("The program will exit after 10 seconds or if you send EOF.")
    stdin_fd = sys.stdin.fileno()
    start_time = time.time()
    timeout_duration = 10  # seconds
    while True:
        # Calculate the remaining time for the timeout
        elapsed_time = time.time() - start_time
        remaining_time = max(0, timeout_duration - elapsed_time)
        if remaining_time == 0:
            print("\nTimer expired. Exiting.")
            break
        # We select on stdin, but with a dynamic timeout.
        # select will return when either stdin is ready OR the timeout is reached.
        ready_to_read, _, _ = select.select([stdin_fd], [], [], remaining_time)
        if ready_to_read:
            # Handle input from the user
            data = sys.stdin.readline()
            if data:
                print(f"Input received: {data.strip()}")
            else:
                print("EOF detected by user. Exiting.")
                break
        else:
            # This block is executed if the select() call timed out
            # (i.e., remaining_time passed without any input).
            # This is our "timer" event.
            print(f"\n{remaining_time:.1f} seconds remaining...", end="", flush=True)
            # We don't need a sleep here, as select() handles the timing.
if __name__ == "__main__":
    multi_event_example()

How to run it:

  1. Save as multi_event.py.
  2. Run it: python multi_event.py.
  3. The program will print the remaining time every second. If you type something, it will print your input and continue counting down. It will exit after 10 seconds or if you send EOF.

Important Considerations and Caveats

  1. Platform Dependency: select works on Unix, Linux, and macOS. On Windows, it has limited functionality and generally only works with sockets, not with regular files like stdin, stdout, or pipes. For cross-platform non-blocking I/O, the msvcrt module (Windows-only) or the asyncio module (recommended for modern Python) are better choices.

  2. sys.stdin Buffering: select tells you that the operating system has data for your process. However, Python's sys.stdin object has its own internal buffer. When select says stdin is ready, calling sys.stdin.readline() will read from that buffer until a newline is found. This is usually the desired behavior.

  3. The Modern Alternative: asyncio: For any new, non-trivial application, the asyncio library is the recommended, modern way to handle concurrent I/O. It's more powerful, more readable, and cross-platform.

    Here's how you could write the first non-blocking example using asyncio:

    import asyncio
    import sys
    async def read_stdin():
        """Coroutine to read from stdin line by line."""
        loop = asyncio.get_event_loop()
        while True:
            # Use loop.run_in_executor to avoid blocking the event loop
            line = await loop.run_in_executor(None, sys.stdin.readline)
            if not line: # EOF
                print("EOF detected. Exiting.")
                break
            print(f"You typed: {line.strip()}")
    async def main():
        print("Asyncio example: Type something. Press Ctrl+D to exit.")
        # Create a task for the stdin reader
        await asyncio.create_task(read_stdin())
    if __name__ == "__main__":
        try:
            asyncio.run(main())
        except KeyboardInterrupt:
            print("\nProgram interrupted.")

    The asyncio version is often cleaner because it avoids manual polling and uses a cooperative multitasking model.

分享:
扫描分享到社交APP
上一篇
下一篇