CODEWITHSUNDEEP
python
worenga
worenga

Reputation: 5856

check if input is available on stdin with timeout

I have a python script that publishes messages that are read from stdin onto a message queue in the network. However, if there wasn't any message on the pipe for a specified amount of time I have to send a heartbeat.

So I have to distinguish between 3 cases:

  1. There is input on the pipe that can be processed
  2. There was no input for some specified amount of time
  3. The piping process has been closed and we can gracefully terminate.

Currently, my code looks as follow:

import sys
for i, line in enumerate(sys.stdin):
    connection.publish(line)
connection.close()

However I need to interrupt the for look if there was any timeout. I.e. sys.stdin has not delivered any data for some time.

Upvotes: 6

Views: 4607

Answers (3)

petiepooo
petiepooo

Reputation: 226

A python3 version of the OP's answer without using exception handling:

import sys
import select

while True:
  if select.select([sys.stdin,],[],[],2.0)[0]:
    line = sys.stdin.readline()
    if line:
      print('Got:', line, end='')
    else:
      print('EOF!')
      break
  else:
    print('No data for 2 secs')

Upvotes: -1

ChrisB
ChrisB

Reputation: 3714

If anybody is looking for a portable solution, this is what I came up with after lots of trial and error. It requires the pywin32 pypi package to work on windows

import os, sys

if sys.platform == "win32":
    import win32api, win32event, win32file, pywintypes
else:
    import select

def stdin_has_content(timeout: float) -> bool:
    assert timeout >= 0
    if sys.platform == "win32":
        try:
            # without this flush, WaitForSingleObject might return early
            win32file.FlushFileBuffers(win32api.STD_INPUT_HANDLE)
        except pywintypes.error:         
            pass # this happens if stdin is already closed, but that's ok
        return win32event.WaitForSingleObject(
            win32api.STD_INPUT_HANDLE, int(timeout * 1000)
        ) == win32event.WAIT_OBJECT_0       
    else:
        rlist, _, _ = select.select([sys.stdin], [], [], timeout)
        return bool(rlist)

Upvotes: 3

worenga
worenga

Reputation: 5856

This is what I finally came up with

import sys
import select


while True:
    try:
        if select.select([sys.stdin,],[],[],2.0)[0]:
            line = sys.stdin.next()
            print "Got:", line
        else:
            print "No data for 2 secs"

    except StopIteration:
        print 'EOF!'
        break

Upvotes: 8

Related Questions