<s> """<STR_LIT>""" <EOL> import os <EOL> import time as time_module <EOL> import datetime <EOL> import hashlib <EOL> import operator <EOL> import subprocess <EOL> from . import exceptions <EOL> from . action import CmdAction , PythonAction <EOL> from . task import result_dep <EOL> result_dep <EOL> def create_folder ( dir_path ) : <EOL> """<STR_LIT>""" <EOL> os . makedirs ( dir_path , exist_ok = True ) <EOL> def title_with_actions ( task ) : <EOL> """<STR_LIT>""" <EOL> if task . actions : <EOL> title = "<STR_LIT>" . join ( [ str ( action ) for action in task . actions ] ) <EOL> else : <EOL> title = "<STR_LIT>" % "<STR_LIT:U+002CU+0020>" . join ( task . task_dep ) <EOL> return "<STR_LIT>" % ( task . name , title ) <EOL> def run_once ( task , values ) : <EOL> """<STR_LIT>""" <EOL> def save_executed ( ) : <EOL> return { '<STR_LIT>' : True } <EOL> task . value_savers . append ( save_executed ) <EOL> return values . get ( '<STR_LIT>' , False ) <EOL> class config_changed ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , config ) : <EOL> self . config = config <EOL> self . config_digest = None <EOL> def _calc_digest ( self ) : <EOL> if isinstance ( self . config , str ) : <EOL> return self . config <EOL> elif isinstance ( self . config , dict ) : <EOL> data = '<STR_LIT>' <EOL> for key in sorted ( self . config ) : <EOL> data += key + repr ( self . config [ key ] ) <EOL> byte_data = data . encode ( "<STR_LIT:utf-8>" ) <EOL> return hashlib . md5 ( byte_data ) . hexdigest ( ) <EOL> else : <EOL> raise Exception ( ( '<STR_LIT>' + <EOL> '<STR_LIT>' ) % ( type ( self . config ) , ) ) <EOL> def configure_task ( self , task ) : <EOL> task . value_savers . append ( lambda : { '<STR_LIT>' : self . config_digest } ) <EOL> def __call__ ( self , task , values ) : <EOL> """<STR_LIT>""" <EOL> self . config_digest = self . _calc_digest ( ) <EOL> last_success = values . get ( '<STR_LIT>' ) <EOL> if last_success is None : <EOL> return False <EOL> return ( last_success == self . config_digest ) <EOL> class timeout ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , timeout_limit ) : <EOL> if isinstance ( timeout_limit , datetime . timedelta ) : <EOL> self . limit_sec = ( ( timeout_limit . days * <NUM_LIT> * <NUM_LIT> ) + <EOL> timeout_limit . seconds ) <EOL> elif isinstance ( timeout_limit , int ) : <EOL> self . limit_sec = timeout_limit <EOL> else : <EOL> msg = "<STR_LIT>" <EOL> raise Exception ( msg % timeout_limit ) <EOL> def __call__ ( self , task , values ) : <EOL> def save_now ( ) : <EOL> return { '<STR_LIT>' : time_module . time ( ) } <EOL> task . value_savers . append ( save_now ) <EOL> last_success = values . get ( '<STR_LIT>' , None ) <EOL> if last_success is None : <EOL> return False <EOL> return ( time_module . time ( ) - last_success ) < self . limit_sec <EOL> class check_timestamp_unchanged ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , file_name , time = '<STR_LIT>' , cmp_op = operator . eq ) : <EOL> """<STR_LIT>""" <EOL> if time in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . _timeattr = '<STR_LIT>' <EOL> elif time in ( '<STR_LIT>' , '<STR_LIT:status>' ) : <EOL> self . _timeattr = '<STR_LIT>' <EOL> elif time in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . _timeattr = '<STR_LIT>' <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' % time ) <EOL> self . _file_name = file_name <EOL> self . _cmp_op = cmp_op <EOL> self . _key = '<STR_LIT:.>' . join ( [ self . _file_name , self . _timeattr ] ) <EOL> def _get_time ( self ) : <EOL> return getattr ( os . stat ( self . _file_name ) , self . _timeattr ) <EOL> def __call__ ( self , task , values ) : <EOL> """<STR_LIT>""" <EOL> def save_now ( ) : <EOL> return { self . _key : self . _get_time ( ) } <EOL> task . value_savers . append ( save_now ) <EOL> prev_time = values . get ( self . _key ) <EOL> if prev_time is None : <EOL> return False <EOL> current_time = self . _get_time ( ) <EOL> return self . _cmp_op ( prev_time , current_time ) <EOL> class LongRunning ( CmdAction ) : <EOL> """<STR_LIT>""" <EOL> def execute ( self , out = None , err = None ) : <EOL> action = self . expand_action ( ) <EOL> process = subprocess . Popen ( action , shell = self . shell , ** self . pkwargs ) <EOL> try : <EOL> process . wait ( ) <EOL> except KeyboardInterrupt : <EOL> pass <EOL> InteractiveAction = LongRunning <EOL> class Interactive ( CmdAction ) : <EOL> """<STR_LIT>""" <EOL> def execute ( self , out = None , err = None ) : <EOL> action = self . expand_action ( ) <EOL> process = subprocess . Popen ( action , shell = self . shell , ** self . pkwargs ) <EOL> process . wait ( ) <EOL> if process . returncode != <NUM_LIT:0> : <EOL> return exceptions . TaskFailed ( <EOL> "<STR_LIT>" % <EOL> ( action , process . returncode ) ) <EOL> class PythonInteractiveAction ( PythonAction ) : <EOL> """<STR_LIT>""" <EOL> def execute ( self , out = None , err = None ) : <EOL> kwargs = self . _prepare_kwargs ( ) <EOL> try : <EOL> returned_value = self . py_callable ( * self . args , ** kwargs ) <EOL> except Exception as exception : <EOL> return exceptions . TaskError ( "<STR_LIT>" , exception ) <EOL> if isinstance ( returned_value , str ) : <EOL> self . result = returned_value <EOL> elif isinstance ( returned_value , dict ) : <EOL> self . values = returned_value <EOL> self . result = returned_value <EOL> def set_trace ( ) : <EOL> """<STR_LIT>""" <EOL> import pdb <EOL> import sys <EOL> debugger = pdb . Pdb ( stdin = sys . __stdin__ , stdout = sys . __stdout__ ) <EOL> debugger . set_trace ( sys . _getframe ( ) . f_back ) <EOL> def register_doit_as_IPython_magic ( ) : <EOL> """<STR_LIT>""" <EOL> from IPython . core . magic import register_line_magic <EOL> from IPython . core . getipython import get_ipython <EOL> from doit . cmd_base import ModuleTaskLoader <EOL> from doit . doit_cmd import DoitMain <EOL> @ register_line_magic <EOL> def doit ( line ) : <EOL> """<STR_LIT>""" <EOL> ip = get_ipython ( ) <EOL> prof_dir = ip . profile_dir . location <EOL> opt_vals = { '<STR_LIT>' : os . path . join ( prof_dir , '<STR_LIT>' , '<STR_LIT>' ) } <EOL> commander = DoitMain ( ModuleTaskLoader ( ip . user_module ) , <EOL> extra_config = { '<STR_LIT>' : opt_vals } ) <EOL> commander . run ( line . split ( ) ) </s>
