<s> import re <EOL> import curses <EOL> from itertools import count <EOL> from subprocess import Popen , PIPE <EOL> from sys import platform , exit <EOL> __version__ = "<STR_LIT>" <EOL> gunicorns = { } <EOL> selected_pid = None <EOL> screen_delay = <NUM_LIT> <EOL> ps_delay = <NUM_LIT:2> <EOL> tick = - <NUM_LIT:1> <EOL> title = "<STR_LIT>" <EOL> instructions = """<STR_LIT>""" <EOL> no_gunicorns = "<STR_LIT>" <EOL> screen_width = None <EOL> foreground_colour = curses . COLOR_BLACK <EOL> background_colour = curses . COLOR_GREEN <EOL> cmd_heading = "<STR_LIT>" <EOL> if re . search ( "<STR_LIT>" , platform ) : <EOL> cmd_heading = "<STR_LIT>" <EOL> if re . search ( "<STR_LIT>" , platform ) : <EOL> PS_ARGS = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> def ports_for_pids ( pids ) : <EOL> LSOF_ARGS = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> try : <EOL> lsof = Popen ( LSOF_ARGS , stdout = PIPE , <EOL> stderr = PIPE ) . communicate ( ) [ <NUM_LIT:0> ] . split ( "<STR_LIT:\n>" ) <EOL> except : <EOL> return <EOL> addr_pos = None <EOL> pid_pos = None <EOL> addr_heading = "<STR_LIT>" <EOL> pid_heading = "<STR_LIT>" <EOL> for row in lsof : <EOL> if addr_heading in row and pid_heading in row : <EOL> addr_pos = row . index ( addr_heading ) <EOL> pid_pos = row . index ( pid_heading ) - <NUM_LIT:2> <EOL> if addr_pos is not None : <EOL> pid = row [ pid_pos : ] . split ( "<STR_LIT:U+0020>" ) [ <NUM_LIT:0> ] <EOL> if pid in pids : <EOL> port = row [ addr_pos : ] . split ( "<STR_LIT::>" ) [ <NUM_LIT:1> ] . split ( "<STR_LIT:U+0020>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> yield ( pid , port ) <EOL> else : <EOL> PS_ARGS = [ "<STR_LIT>" , "<STR_LIT:x>" , "<STR_LIT>" ] <EOL> def ports_for_pids ( pids ) : <EOL> netstat = Popen ( [ "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE , <EOL> stderr = PIPE ) . communicate ( ) [ <NUM_LIT:0> ] . split ( "<STR_LIT:\n>" ) <EOL> addr_pos = None <EOL> pid_pos = None <EOL> addr_heading = "<STR_LIT>" <EOL> pid_heading = "<STR_LIT>" <EOL> for row in netstat : <EOL> if addr_heading in row and pid_heading in row : <EOL> addr_pos = row . index ( addr_heading ) <EOL> pid_pos = row . index ( pid_heading ) <EOL> if addr_pos is not None : <EOL> pid = row [ pid_pos : ] . split ( "<STR_LIT:/>" ) [ <NUM_LIT:0> ] <EOL> if pid in pids : <EOL> port = row [ addr_pos : ] . split ( "<STR_LIT:U+0020>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] . split ( "<STR_LIT::>" ) [ <NUM_LIT:1> ] <EOL> yield ( pid , port ) <EOL> def send_signal ( signal ) : <EOL> """<STR_LIT>""" <EOL> if selected_pid in gunicorns : <EOL> Popen ( [ "<STR_LIT>" , "<STR_LIT>" % signal , selected_pid ] ) <EOL> curses . flash ( ) <EOL> def move_selection ( reverse = False ) : <EOL> """<STR_LIT>""" <EOL> global selected_pid <EOL> if selected_pid not in gunicorns : <EOL> selected_pid = None <EOL> found = False <EOL> pids = sorted ( gunicorns . keys ( ) , reverse = reverse ) <EOL> for pid in pids + pids : <EOL> if selected_pid is None or found : <EOL> selected_pid = pid <EOL> return <EOL> found = pid == selected_pid <EOL> def update_gunicorns ( ) : <EOL> """<STR_LIT>""" <EOL> global tick <EOL> tick += <NUM_LIT:1> <EOL> if ( tick * screen_delay ) % ps_delay != <NUM_LIT:0> : <EOL> return <EOL> tick = <NUM_LIT:0> <EOL> for pid in gunicorns : <EOL> gunicorns [ pid ] . update ( { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT>" : <NUM_LIT:0> } ) <EOL> ps = Popen ( PS_ARGS , stdout = PIPE ) . communicate ( ) [ <NUM_LIT:0> ] . split ( "<STR_LIT:\n>" ) <EOL> headings = ps . pop ( <NUM_LIT:0> ) . split ( ) <EOL> name_col = headings . index ( cmd_heading ) <EOL> num_cols = len ( headings ) - <NUM_LIT:1> <EOL> for row in ps : <EOL> cols = row . split ( None , num_cols ) <EOL> if cols and "<STR_LIT>" in cols [ name_col ] : <EOL> if "<STR_LIT>" in cols [ name_col ] : <EOL> is_worker = True <EOL> else : <EOL> is_worker = False <EOL> if is_worker : <EOL> pid = cols [ headings . index ( "<STR_LIT>" ) ] <EOL> else : <EOL> pid = cols [ headings . index ( "<STR_LIT>" ) ] <EOL> if pid not in gunicorns : <EOL> gunicorns [ pid ] = { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT:port>" : None , "<STR_LIT:name>" : <EOL> cols [ name_col ] . strip ( ) . split ( "<STR_LIT:[>" , <NUM_LIT:1> ) [ <NUM_LIT:1> ] . split ( "<STR_LIT:]>" , <NUM_LIT:1> ) [ : - <NUM_LIT:1> ] } <EOL> gunicorns [ pid ] [ "<STR_LIT>" ] += int ( cols [ headings . index ( "<STR_LIT>" ) ] ) <EOL> if is_worker : <EOL> gunicorns [ pid ] [ "<STR_LIT>" ] += <NUM_LIT:1> <EOL> for pid in gunicorns . keys ( ) [ : ] : <EOL> if gunicorns [ pid ] [ "<STR_LIT>" ] == <NUM_LIT:0> : <EOL> del gunicorns [ pid ] <EOL> if not [ g for g in gunicorns . values ( ) if g [ "<STR_LIT:port>" ] is None ] : <EOL> return <EOL> for ( pid , port ) in ports_for_pids ( gunicorns . keys ( ) ) : <EOL> if pid in gunicorns : <EOL> gunicorns [ pid ] [ "<STR_LIT:port>" ] = port <EOL> def handle_keypress ( screen ) : <EOL> """<STR_LIT>""" <EOL> global selected_pid <EOL> try : <EOL> key = screen . getkey ( ) . upper ( ) <EOL> except : <EOL> return <EOL> if key in ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> move_selection ( ) <EOL> elif key in ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> move_selection ( reverse = True ) <EOL> elif key in ( "<STR_LIT:A>" , "<STR_LIT:+>" ) : <EOL> send_signal ( "<STR_LIT>" ) <EOL> if selected_pid in gunicorns : <EOL> gunicorns [ selected_pid ] [ "<STR_LIT>" ] = <NUM_LIT:0> <EOL> elif key in ( "<STR_LIT>" , "<STR_LIT:->" ) : <EOL> if selected_pid in gunicorns : <EOL> if gunicorns [ selected_pid ] [ "<STR_LIT>" ] != <NUM_LIT:1> : <EOL> send_signal ( "<STR_LIT>" ) <EOL> gunicorns [ selected_pid ] [ "<STR_LIT>" ] = <NUM_LIT:0> <EOL> elif key in ( "<STR_LIT:R>" , ) : <EOL> if selected_pid in gunicorns : <EOL> send_signal ( "<STR_LIT>" ) <EOL> del gunicorns [ selected_pid ] <EOL> selected_pid = None <EOL> elif key in ( "<STR_LIT:T>" , ) : <EOL> for pid in gunicorns . copy ( ) . iterkeys ( ) : <EOL> selected_pid = pid <EOL> send_signal ( "<STR_LIT>" ) <EOL> del gunicorns [ selected_pid ] <EOL> selected_pid = None <EOL> elif key in ( "<STR_LIT:M>" , "<STR_LIT:->" ) : <EOL> if selected_pid in gunicorns : <EOL> send_signal ( "<STR_LIT>" ) <EOL> del gunicorns [ selected_pid ] <EOL> selected_pid = None <EOL> elif key in ( "<STR_LIT>" , ) : <EOL> raise KeyboardInterrupt <EOL> def format_row ( pid = "<STR_LIT>" , port = "<STR_LIT>" , name = "<STR_LIT>" , mem = "<STR_LIT>" , workers = "<STR_LIT>" , prefix_char = "<STR_LIT:U+0020>" ) : <EOL> """<STR_LIT>""" <EOL> row = "<STR_LIT>" % ( prefix_char , pid , port , name , mem , workers ) <EOL> global screen_width <EOL> if screen_width is None : <EOL> screen_width = len ( row ) <EOL> return row <EOL> def display_output ( screen ) : <EOL> """<STR_LIT>""" <EOL> format_row ( ) <EOL> screen_height = len ( gunicorns ) + len ( instructions . split ( "<STR_LIT:\n>" ) ) + <NUM_LIT:9> <EOL> if not gunicorns : <EOL> screen_height += <NUM_LIT:2> <EOL> screen . erase ( ) <EOL> win = curses . newwin ( screen_height , screen_width + <NUM_LIT:6> , <NUM_LIT:1> , <NUM_LIT:3> ) <EOL> win . bkgd ( "<STR_LIT:U+0020>" , curses . color_pair ( <NUM_LIT:1> ) ) <EOL> win . border ( ) <EOL> x = <NUM_LIT:3> <EOL> blank_line = y = count ( <NUM_LIT:2> ) . next <EOL> win . addstr ( y ( ) , x , title . center ( screen_width ) , curses . A_NORMAL ) <EOL> blank_line ( ) <EOL> win . addstr ( y ( ) , x , format_row ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) , <EOL> curses . A_STANDOUT ) <EOL> if not gunicorns : <EOL> blank_line ( ) <EOL> win . addstr ( y ( ) , x , no_gunicorns . center ( screen_width ) , <EOL> curses . A_NORMAL ) <EOL> blank_line ( ) <EOL> else : <EOL> win . hline ( y ( ) , x , curses . ACS_HLINE , screen_width ) <EOL> for ( i , pid ) in enumerate ( sorted ( gunicorns . keys ( ) ) ) : <EOL> port = gunicorns [ pid ] [ "<STR_LIT:port>" ] <EOL> name = gunicorns [ pid ] [ "<STR_LIT:name>" ] <EOL> mem = "<STR_LIT>" % ( gunicorns [ pid ] [ "<STR_LIT>" ] / <NUM_LIT> ) <EOL> workers = gunicorns [ pid ] [ "<STR_LIT>" ] <EOL> if workers < <NUM_LIT:1> : <EOL> gunicorns [ pid ] [ "<STR_LIT>" ] -= <NUM_LIT:1> <EOL> chars = "<STR_LIT>" <EOL> workers *= - <NUM_LIT:1> <EOL> if workers == len ( chars ) : <EOL> gunicorns [ pid ] [ "<STR_LIT>" ] = workers = <NUM_LIT:0> <EOL> workers = chars [ workers ] <EOL> if pid == selected_pid : <EOL> attr = curses . A_STANDOUT <EOL> prefix_char = '<STR_LIT>' <EOL> else : <EOL> attr = curses . A_NORMAL <EOL> prefix_char = '<STR_LIT:U+0020>' <EOL> win . addstr ( y ( ) , x , format_row ( pid , port , name , mem , workers , <EOL> prefix_char ) , attr ) <EOL> win . hline ( y ( ) , x , curses . ACS_HLINE , screen_width ) <EOL> blank_line ( ) <EOL> for line in instructions . split ( "<STR_LIT:\n>" ) : <EOL> win . addstr ( y ( ) , x , line . center ( screen_width ) , curses . A_NORMAL ) <EOL> win . refresh ( ) <EOL> def main ( ) : <EOL> """<STR_LIT>""" <EOL> stdscr = curses . initscr ( ) <EOL> curses . start_color ( ) <EOL> curses . init_pair ( <NUM_LIT:1> , foreground_colour , background_colour ) <EOL> curses . noecho ( ) <EOL> stdscr . keypad ( True ) <EOL> stdscr . nodelay ( True ) <EOL> try : <EOL> curses . curs_set ( False ) <EOL> except : <EOL> pass <EOL> try : <EOL> while True : <EOL> try : <EOL> update_gunicorns ( ) <EOL> handle_keypress ( stdscr ) <EOL> display_output ( stdscr ) <EOL> curses . napms ( int ( screen_delay * <NUM_LIT:1000> ) ) <EOL> except KeyboardInterrupt : <EOL> break <EOL> finally : <EOL> curses . nocbreak ( ) <EOL> stdscr . keypad ( False ) <EOL> curses . echo ( ) <EOL> curses . endwin ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> try : <EOL> import setproctitle <EOL> except ImportError : <EOL> print <EOL> print "<STR_LIT>" <EOL> print <EOL> exit ( ) <EOL> main ( ) </s>
