<s> from contextlib import contextmanager <EOL> from atom . api import DefaultValue , Typed <EOL> from atom . datastructures . api import sortedmap <EOL> from . dynamic_scope import DynamicScope , Nonlocals <EOL> from . exceptions import DeclarativeError <EOL> from . funchelper import call_func <EOL> from . standard_inverter import StandardInverter <EOL> from . standard_tracer import StandardTracer <EOL> class OperatorBase ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = '<STR_LIT>' <EOL> def __init__ ( self , binding ) : <EOL> """<STR_LIT>""" <EOL> self . binding = binding <EOL> def get_locals ( self , owner ) : <EOL> """<STR_LIT>""" <EOL> scope_member = self . binding . parent . scope_member <EOL> if scope_member : <EOL> return scope_member . get_slot ( owner ) or { } <EOL> return { } <EOL> def release ( self , owner ) : <EOL> """<STR_LIT>""" <EOL> scope_member = self . binding . parent . scope_member <EOL> if scope_member : <EOL> scope_member . set_slot ( owner , None ) <EOL> class OpSimple ( OperatorBase ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( ) <EOL> def eval ( self , owner ) : <EOL> """<STR_LIT>""" <EOL> overrides = { '<STR_LIT>' : Nonlocals ( owner , None ) , '<STR_LIT>' : owner } <EOL> f_locals = self . get_locals ( owner ) <EOL> func = self . binding . func <EOL> scope = DynamicScope ( <EOL> owner , f_locals , overrides , func . func_globals , None <EOL> ) <EOL> return call_func ( func , ( ) , { } , scope ) <EOL> class DeprecatedNotificationEvent ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> _warning_registry = { } <EOL> def __init__ ( self , change , binding ) : <EOL> self . _change = change <EOL> self . _binding = binding <EOL> def _raise_warning ( self ) : <EOL> msg = "<STR_LIT>" <EOL> msg += "<STR_LIT>" <EOL> binding = self . _binding <EOL> filename = binding . root ( ) . filename <EOL> lineno = binding . lineno <EOL> reg = self . _warning_registry <EOL> import warnings <EOL> warnings . warn_explicit ( msg , FutureWarning , filename , lineno , '<STR_LIT>' , reg ) <EOL> @ property <EOL> def obj ( self ) : <EOL> self . _raise_warning ( ) <EOL> return self . _change [ '<STR_LIT:object>' ] <EOL> @ property <EOL> def name ( self ) : <EOL> self . _raise_warning ( ) <EOL> return self . _change [ '<STR_LIT:name>' ] <EOL> @ property <EOL> def old ( self ) : <EOL> self . _raise_warning ( ) <EOL> return self . _change . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def new ( self ) : <EOL> self . _raise_warning ( ) <EOL> return self . _change . get ( '<STR_LIT:value>' ) <EOL> class OpNotify ( OperatorBase ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( ) <EOL> def notify ( self , change ) : <EOL> """<STR_LIT>""" <EOL> owner = change [ '<STR_LIT:object>' ] <EOL> nonlocals = Nonlocals ( owner , None ) <EOL> overrides = { '<STR_LIT>' : change , '<STR_LIT>' : nonlocals , '<STR_LIT>' : owner } <EOL> overrides [ '<STR_LIT>' ] = DeprecatedNotificationEvent ( change , self . binding ) <EOL> f_locals = self . get_locals ( owner ) <EOL> func = self . binding . func <EOL> scope = DynamicScope ( <EOL> owner , f_locals , overrides , func . func_globals , None <EOL> ) <EOL> call_func ( func , ( ) , { } , scope ) <EOL> class OpUpdate ( OperatorBase ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( ) <EOL> def notify ( self , change ) : <EOL> """<STR_LIT>""" <EOL> owner = change [ '<STR_LIT:object>' ] <EOL> nonlocals = Nonlocals ( owner , None ) <EOL> overrides = { '<STR_LIT>' : nonlocals , '<STR_LIT>' : owner } <EOL> inverter = StandardInverter ( nonlocals ) <EOL> f_locals = self . get_locals ( owner ) <EOL> func = self . binding . func <EOL> scope = DynamicScope ( <EOL> owner , f_locals , overrides , func . func_globals , None <EOL> ) <EOL> call_func ( func , ( inverter , change . get ( '<STR_LIT:value>' ) ) , { } , scope ) <EOL> class SubscriptionObserver ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT:name>' ) <EOL> def __init__ ( self , owner , name ) : <EOL> """<STR_LIT>""" <EOL> self . owner = owner <EOL> self . name = name <EOL> def __nonzero__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . owner is not None <EOL> def __call__ ( self , change ) : <EOL> """<STR_LIT>""" <EOL> owner = self . owner <EOL> if owner is not None : <EOL> name = self . name <EOL> setattr ( owner , name , owner . _run_eval_operator ( name ) ) <EOL> class OpSubscribe ( OperatorBase ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( ) <EOL> def get_storage ( self , owner ) : <EOL> """<STR_LIT>""" <EOL> member = owner . get_member ( '<STR_LIT>' ) <EOL> if member is None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> return member . do_getattr ( owner ) <EOL> def release ( self , owner ) : <EOL> """<STR_LIT>""" <EOL> super ( OpSubscribe , self ) . release ( owner ) <EOL> storage = self . get_storage ( owner ) <EOL> observer = storage . pop ( self , None ) <EOL> if observer is not None : <EOL> observer . owner = None <EOL> def eval ( self , owner ) : <EOL> """<STR_LIT>""" <EOL> tracer = StandardTracer ( ) <EOL> overrides = { '<STR_LIT>' : Nonlocals ( owner , tracer ) , '<STR_LIT>' : owner } <EOL> f_locals = self . get_locals ( owner ) <EOL> func = self . binding . func <EOL> scope = DynamicScope ( <EOL> owner , f_locals , overrides , func . func_globals , tracer <EOL> ) <EOL> result = call_func ( func , ( tracer , ) , { } , scope ) <EOL> storage = self . get_storage ( owner ) <EOL> old_observer = storage . get ( self ) <EOL> if old_observer is not None : <EOL> old_observer . owner = None <EOL> if tracer . traced_items : <EOL> observer = SubscriptionObserver ( owner , self . binding . name ) <EOL> storage [ self ] = observer <EOL> for obj , name in tracer . traced_items : <EOL> obj . observe ( name , observer ) <EOL> return result <EOL> class OpDelegate ( OpSubscribe ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( ) <EOL> def notify ( self , change ) : <EOL> """<STR_LIT>""" <EOL> owner = change [ '<STR_LIT:object>' ] <EOL> nonlocals = Nonlocals ( owner , None ) <EOL> inverter = StandardInverter ( nonlocals ) <EOL> overrides = { '<STR_LIT>' : nonlocals , '<STR_LIT>' : owner } <EOL> f_locals = self . get_locals ( owner ) <EOL> func = self . binding . auxfunc <EOL> scope = DynamicScope ( <EOL> owner , f_locals , overrides , func . func_globals , None <EOL> ) <EOL> call_func ( func , ( inverter , change . get ( '<STR_LIT:value>' ) ) , { } , scope ) <EOL> _warn_color_registry = { } <EOL> def _warn_color_binding ( binding ) : <EOL> msg = "<STR_LIT>" <EOL> msg += "<STR_LIT>" <EOL> d = { <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> } <EOL> msg = msg % d [ binding . name ] <EOL> lineno = binding . lineno <EOL> filename = binding . root ( ) . filename <EOL> reg = _warn_color_registry <EOL> import warnings <EOL> warnings . warn_explicit ( msg , FutureWarning , filename , lineno , '<STR_LIT>' , reg ) <EOL> def assert_d_member ( klass , binding , readable , writable ) : <EOL> """<STR_LIT>""" <EOL> members = klass . members ( ) <EOL> m = members . get ( binding . name ) <EOL> if m is None : <EOL> if binding . name == '<STR_LIT>' and '<STR_LIT>' in members : <EOL> _warn_color_binding ( binding ) <EOL> binding . name = '<STR_LIT>' <EOL> m = members [ '<STR_LIT>' ] <EOL> elif binding . name == '<STR_LIT>' and '<STR_LIT>' in members : <EOL> _warn_color_binding ( binding ) <EOL> binding . name = '<STR_LIT>' <EOL> m = members [ '<STR_LIT>' ] <EOL> if m is None or m . metadata is None or not m . metadata . get ( '<STR_LIT>' ) : <EOL> msg = "<STR_LIT>" % binding . name <EOL> root = binding . root ( ) <EOL> filename = root . filename <EOL> context = root . typename <EOL> lineno = binding . lineno <EOL> raise DeclarativeError ( msg , filename , context , lineno ) <EOL> if readable and not m . metadata . get ( '<STR_LIT>' ) : <EOL> msg = "<STR_LIT>" % binding . name <EOL> root = binding . root ( ) <EOL> filename = root . filename <EOL> context = root . typename <EOL> lineno = binding . lineno <EOL> raise DeclarativeError ( msg , filename , context , lineno ) <EOL> if writable and not m . metadata . get ( '<STR_LIT>' ) : <EOL> msg = "<STR_LIT>" % binding . name <EOL> root = binding . root ( ) <EOL> filename = root . filename <EOL> context = root . typename <EOL> lineno = binding . lineno <EOL> raise DeclarativeError ( msg , filename , context , lineno ) <EOL> return m <EOL> def bind_read_operator ( klass , binding , operator ) : <EOL> """<STR_LIT>""" <EOL> member = assert_d_member ( klass , binding , True , False ) <EOL> klass . notify_operators ( ) . setdefault ( binding . name , [ ] ) . append ( operator ) <EOL> if not member . has_observer ( '<STR_LIT>' ) : <EOL> clone = member . clone ( ) <EOL> clone . add_static_observer ( '<STR_LIT>' ) <EOL> klass . members ( ) [ binding . name ] = clone <EOL> setattr ( klass , binding . name , clone ) <EOL> def bind_write_operator ( klass , binding , operator ) : <EOL> """<STR_LIT>""" <EOL> member = assert_d_member ( klass , binding , False , True ) <EOL> klass . eval_operators ( ) [ binding . name ] = operator <EOL> mode = ( DefaultValue . ObjectMethod_Name , '<STR_LIT>' ) <EOL> if member . default_value_mode != mode : <EOL> clone = member . clone ( ) <EOL> clone . set_default_value_mode ( * mode ) <EOL> klass . members ( ) [ binding . name ] = clone <EOL> setattr ( klass , binding . name , clone ) <EOL> def add_operator_storage ( klass ) : <EOL> """<STR_LIT>""" <EOL> members = klass . members ( ) <EOL> if '<STR_LIT>' not in members : <EOL> m = Typed ( sortedmap , ( ) ) <EOL> m . set_name ( '<STR_LIT>' ) <EOL> m . set_index ( len ( members ) ) <EOL> members [ '<STR_LIT>' ] = m <EOL> def op_simple ( klass , binding ) : <EOL> """<STR_LIT>""" <EOL> bind_write_operator ( klass , binding , OpSimple ( binding ) ) <EOL> def op_notify ( klass , binding ) : <EOL> """<STR_LIT>""" <EOL> bind_read_operator ( klass , binding , OpNotify ( binding ) ) <EOL> def op_update ( klass , binding ) : <EOL> """<STR_LIT>""" <EOL> bind_read_operator ( klass , binding , OpUpdate ( binding ) ) <EOL> def op_subscribe ( klass , binding ) : <EOL> """<STR_LIT>""" <EOL> bind_write_operator ( klass , binding , OpSubscribe ( binding ) ) <EOL> add_operator_storage ( klass ) <EOL> def op_delegate ( klass , binding ) : <EOL> """<STR_LIT>""" <EOL> operator = OpDelegate ( binding ) <EOL> bind_read_operator ( klass , binding , operator ) <EOL> bind_write_operator ( klass , binding , operator ) <EOL> add_operator_storage ( klass ) <EOL> DEFAULT_OPERATORS = { <EOL> '<STR_LIT:=>' : op_simple , <EOL> '<STR_LIT>' : op_notify , <EOL> '<STR_LIT>' : op_update , <EOL> '<STR_LIT>' : op_subscribe , <EOL> '<STR_LIT>' : op_delegate , <EOL> } <EOL> __operator_stack = [ ] <EOL> @ contextmanager <EOL> def operator_context ( ops , union = False ) : <EOL> """<STR_LIT>""" <EOL> if union : <EOL> new = dict ( __get_operators ( ) ) <EOL> new . update ( ops ) <EOL> ops = new <EOL> __operator_stack . append ( ops ) <EOL> yield <EOL> __operator_stack . pop ( ) <EOL> def __get_default_operators ( ) : <EOL> """<STR_LIT>""" <EOL> return DEFAULT_OPERATORS <EOL> def __set_default_operators ( ops ) : <EOL> """<STR_LIT>""" <EOL> global DEFAULT_OPERATORS <EOL> DEFAULT_OPERATORS = ops <EOL> def __get_operators ( ) : <EOL> """<STR_LIT>""" <EOL> if __operator_stack : <EOL> return __operator_stack [ - <NUM_LIT:1> ] <EOL> return __get_default_operators ( ) </s>
