<s> """<STR_LIT>""" <EOL> import logging <EOL> from collections import defaultdict <EOL> import time <EOL> from mpf . system . config import CaseInsensitiveDict <EOL> from mpf . system . timing import Timing <EOL> from mpf . system . utility_functions import Util <EOL> class SwitchController ( object ) : <EOL> """<STR_LIT>""" <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> def __init__ ( self , machine ) : <EOL> self . machine = machine <EOL> self . registered_switches = CaseInsensitiveDict ( ) <EOL> self . active_timed_switches = defaultdict ( list ) <EOL> self . switches = CaseInsensitiveDict ( ) <EOL> self . switch_event_active = ( <EOL> self . machine . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . switch_event_inactive = ( <EOL> self . machine . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . switch_tag_event = ( <EOL> self . machine . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . machine . events . add_handler ( '<STR_LIT>' , self . _tick , <NUM_LIT:1000> ) <EOL> self . machine . events . add_handler ( '<STR_LIT>' , <EOL> self . _initialize_switches , <EOL> <NUM_LIT:1000> ) <EOL> self . machine . events . add_handler ( '<STR_LIT>' , <EOL> self . log_active_switches ) <EOL> self . monitors = list ( ) <EOL> def _initialize_switches ( self ) : <EOL> self . update_switches_from_hw ( ) <EOL> for switch in self . machine . switches : <EOL> self . set_state ( switch . name , switch . state , reset_time = True ) <EOL> self . registered_switches [ switch . name + '<STR_LIT>' ] = list ( ) <EOL> self . registered_switches [ switch . name + '<STR_LIT>' ] = list ( ) <EOL> if self . machine . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> switch . activation_events . add ( <EOL> self . machine . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] . replace ( <EOL> '<STR_LIT:%>' , switch . name ) ) <EOL> switch . deactivation_events . add ( <EOL> self . machine . config [ '<STR_LIT>' ] [ <EOL> '<STR_LIT>' ] . replace ( <EOL> '<STR_LIT:%>' , switch . name ) ) <EOL> if '<STR_LIT>' in switch . config : <EOL> for event in Util . string_to_lowercase_list ( <EOL> switch . config [ '<STR_LIT>' ] ) : <EOL> if "<STR_LIT:|>" in event : <EOL> ev_name , ev_time = event . split ( "<STR_LIT:|>" ) <EOL> self . add_switch_handler ( <EOL> switch_name = switch . name , <EOL> callback = self . machine . events . post , <EOL> state = <NUM_LIT:1> , <EOL> ms = Timing . string_to_ms ( ev_time ) , <EOL> callback_kwargs = { '<STR_LIT>' : ev_name } <EOL> ) <EOL> else : <EOL> switch . activation_events . add ( event ) <EOL> if '<STR_LIT>' in switch . config : <EOL> for event in Util . string_to_lowercase_list ( <EOL> switch . config [ '<STR_LIT>' ] ) : <EOL> if "<STR_LIT:|>" in event : <EOL> ev_name , ev_time = event . split ( "<STR_LIT:|>" ) <EOL> self . add_switch_handler ( <EOL> switch_name = switch . name , <EOL> callback = self . machine . events . post , <EOL> state = <NUM_LIT:0> , <EOL> ms = Timing . string_to_ms ( ev_time ) , <EOL> callback_kwargs = { '<STR_LIT>' : ev_name } <EOL> ) <EOL> else : <EOL> switch . deactivation_events . add ( event ) <EOL> def update_switches_from_hw ( self ) : <EOL> """<STR_LIT>""" <EOL> platforms = set ( ) <EOL> switches = set ( ) <EOL> for switch in self . machine . switches : <EOL> platforms . add ( switch . platform ) <EOL> switches . add ( ( switch , switch . number ) ) <EOL> for platform in platforms : <EOL> switch_states = platform . get_hw_switch_states ( ) <EOL> for switch , number in switches : <EOL> switch . state = switch_states [ number ] ^ switch . invert <EOL> def verify_switches ( self ) : <EOL> """<STR_LIT>""" <EOL> current_states = dict ( ) <EOL> for switch in self . machine . switches : <EOL> current_states [ switch ] = switch . state <EOL> self . update_switches_from_hw ( ) <EOL> for switch in self . machine . switches : <EOL> if switch . state ^ switch . invert != current_states [ switch ] : <EOL> self . log . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , switch . name , <EOL> current_states [ switch ] , <EOL> switch . state ^ switch . invert ) <EOL> def is_state ( self , switch_name , state , ms = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> if self . switches [ switch_name ] [ '<STR_LIT:state>' ] == state : <EOL> if ms <= self . ms_since_change ( switch_name ) : <EOL> return True <EOL> else : <EOL> return False <EOL> else : <EOL> return False <EOL> def is_active ( self , switch_name , ms = None ) : <EOL> """<STR_LIT>""" <EOL> return self . is_state ( switch_name = switch_name , <EOL> state = <NUM_LIT:1> , <EOL> ms = ms ) <EOL> def is_inactive ( self , switch_name , ms = None ) : <EOL> """<STR_LIT>""" <EOL> return self . is_state ( switch_name = switch_name , <EOL> state = <NUM_LIT:0> , <EOL> ms = ms ) <EOL> def ms_since_change ( self , switch_name ) : <EOL> """<STR_LIT>""" <EOL> return ( time . time ( ) - self . switches [ switch_name ] [ '<STR_LIT:time>' ] ) * <NUM_LIT> <EOL> def secs_since_change ( self , switch_name ) : <EOL> """<STR_LIT>""" <EOL> return time . time ( ) - self . switches [ switch_name ] [ '<STR_LIT:time>' ] <EOL> def set_state ( self , switch_name , state = <NUM_LIT:1> , reset_time = False ) : <EOL> """<STR_LIT>""" <EOL> if reset_time : <EOL> timestamp = <NUM_LIT:1> <EOL> else : <EOL> timestamp = time . time ( ) <EOL> self . switches . update ( { switch_name : { '<STR_LIT:state>' : state , <EOL> '<STR_LIT:time>' : timestamp <EOL> } <EOL> } ) <EOL> def process_switch ( self , name = None , state = <NUM_LIT:1> , logical = False , num = None , <EOL> obj = None , debounced = True ) : <EOL> """<STR_LIT>""" <EOL> self . log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , name , state , logical , <EOL> num , obj , debounced ) <EOL> if num is not None : <EOL> for switch in self . machine . switches : <EOL> if switch . number == num : <EOL> name = switch . name <EOL> obj = switch <EOL> break <EOL> elif obj : <EOL> name = obj . name <EOL> elif name : <EOL> try : <EOL> obj = self . machine . switches [ name ] <EOL> except KeyError : <EOL> self . log . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , name ) <EOL> return <EOL> name = obj . name <EOL> if not obj : <EOL> self . log . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , num , name ) <EOL> return <EOL> if state : <EOL> state = <NUM_LIT:1> <EOL> else : <EOL> state = <NUM_LIT:0> <EOL> hw_state = state <EOL> if obj . invert : <EOL> if logical : <EOL> hw_state ^= <NUM_LIT:1> <EOL> else : <EOL> state ^= <NUM_LIT:1> <EOL> if not debounced : <EOL> if obj . config [ '<STR_LIT>' ] : <EOL> return <EOL> obj . hw_state = hw_state <EOL> if state and not self . _check_recycle_time ( obj , state ) : <EOL> return <EOL> obj . state = state <EOL> if state : <EOL> obj . recycle_clear_tick = Timing . tick + obj . recycle_ticks <EOL> if self . switches [ name ] [ '<STR_LIT:state>' ] == state : <EOL> if not obj . recycle_ticks : <EOL> self . log . info ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> name ) <EOL> return <EOL> self . log . info ( "<STR_LIT>" , name , state ) <EOL> self . set_state ( name , state ) <EOL> switch_key = str ( name ) + '<STR_LIT:->' + str ( state ) <EOL> if switch_key in self . registered_switches : <EOL> for entry in self . registered_switches [ switch_key ] : <EOL> if entry [ '<STR_LIT>' ] : <EOL> key = time . time ( ) + ( entry [ '<STR_LIT>' ] / <NUM_LIT> ) <EOL> value = { '<STR_LIT>' : str ( name ) + '<STR_LIT:->' + str ( state ) , <EOL> '<STR_LIT>' : entry [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT:state>' : state , <EOL> '<STR_LIT>' : entry [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : entry [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : entry [ '<STR_LIT>' ] } <EOL> self . active_timed_switches [ key ] . append ( value ) <EOL> self . log . debug ( <EOL> "<STR_LIT>" , <EOL> key , value ) <EOL> else : <EOL> if entry [ '<STR_LIT>' ] : <EOL> entry [ '<STR_LIT>' ] ( switch_name = name , state = state , ms = <NUM_LIT:0> , <EOL> ** entry [ '<STR_LIT>' ] ) <EOL> else : <EOL> entry [ '<STR_LIT>' ] ( ** entry [ '<STR_LIT>' ] ) <EOL> for k , v , in self . active_timed_switches . items ( ) : <EOL> for item in v : <EOL> if item [ '<STR_LIT>' ] == str ( name ) + '<STR_LIT:->' + str ( state ^ <NUM_LIT:1> ) : <EOL> if self . active_timed_switches [ k ] : <EOL> del self . active_timed_switches [ k ] <EOL> for monitor in self . monitors : <EOL> monitor ( name , state ) <EOL> self . _post_switch_events ( name , state ) <EOL> def add_monitor ( self , monitor ) : <EOL> if monitor not in self . monitors : <EOL> self . monitors . append ( monitor ) <EOL> def add_switch_handler ( self , switch_name , callback , state = <NUM_LIT:1> , ms = <NUM_LIT:0> , <EOL> return_info = False , callback_kwargs = None ) : <EOL> """<STR_LIT>""" <EOL> if not callback_kwargs : <EOL> callback_kwargs = dict ( ) <EOL> self . log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , switch_name , callback , <EOL> state , ms , <EOL> return_info , callback_kwargs ) <EOL> entry_val = { '<STR_LIT>' : ms , '<STR_LIT>' : callback , <EOL> '<STR_LIT>' : return_info , <EOL> '<STR_LIT>' : callback_kwargs } <EOL> entry_key = str ( switch_name ) + '<STR_LIT:->' + str ( state ) <EOL> self . registered_switches [ entry_key ] . append ( entry_val ) <EOL> if ms : <EOL> if state == <NUM_LIT:1> : <EOL> if self . is_active ( switch_name , <NUM_LIT:0> ) and ( <EOL> self . ms_since_change ( switch_name ) < ms ) : <EOL> key = ( <EOL> time . time ( ) + ( ( ms - self . ms_since_change ( switch_name ) ) <EOL> / <NUM_LIT> ) ) <EOL> value = { '<STR_LIT>' : entry_key , <EOL> '<STR_LIT>' : callback , <EOL> '<STR_LIT>' : switch_name , <EOL> '<STR_LIT:state>' : state , <EOL> '<STR_LIT>' : ms , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : return_info , <EOL> '<STR_LIT>' : callback_kwargs } <EOL> self . active_timed_switches [ key ] . append ( value ) <EOL> elif state == <NUM_LIT:0> : <EOL> if self . is_inactive ( switch_name , <NUM_LIT:0> ) and ( <EOL> self . ms_since_change ( switch_name ) < ms ) : <EOL> key = ( <EOL> time . time ( ) + ( ( ms - self . ms_since_change ( switch_name ) ) <EOL> / <NUM_LIT> ) ) <EOL> value = { '<STR_LIT>' : entry_key , <EOL> '<STR_LIT>' : callback , <EOL> '<STR_LIT>' : switch_name , <EOL> '<STR_LIT:state>' : state , <EOL> '<STR_LIT>' : ms , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : return_info , <EOL> '<STR_LIT>' : callback_kwargs } <EOL> self . active_timed_switches [ key ] . append ( value ) <EOL> return { '<STR_LIT>' : switch_name , <EOL> '<STR_LIT>' : callback , <EOL> '<STR_LIT:state>' : state , <EOL> '<STR_LIT>' : ms } <EOL> def remove_switch_handler ( self , switch_name , callback , state = <NUM_LIT:1> , ms = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> self . log . debug ( <EOL> "<STR_LIT>" , <EOL> switch_name , state , ms ) <EOL> entry_key = str ( switch_name ) + '<STR_LIT:->' + str ( state ) <EOL> if entry_key in self . registered_switches : <EOL> for index , settings in enumerate ( <EOL> self . registered_switches [ entry_key ] ) : <EOL> if ( settings [ '<STR_LIT>' ] == ms and <EOL> settings [ '<STR_LIT>' ] == callback ) : <EOL> self . registered_switches [ entry_key ] . remove ( settings ) <EOL> for timed_key , timed_entry in self . active_timed_switches . items ( ) : <EOL> for key , entry in enumerate ( timed_entry ) : <EOL> if entry [ '<STR_LIT>' ] == entry_key and entry [ '<STR_LIT>' ] == ms and entry [ '<STR_LIT>' ] == callback : <EOL> entry [ '<STR_LIT>' ] = True <EOL> def log_active_switches ( self ) : <EOL> """<STR_LIT>""" <EOL> self . log . info ( "<STR_LIT>" ) <EOL> for k , v in self . switches . iteritems ( ) : <EOL> if v [ '<STR_LIT:state>' ] : <EOL> self . log . info ( "<STR_LIT>" , k ) <EOL> def _check_recycle_time ( self , switch , state ) : <EOL> if Timing . tick >= switch . recycle_clear_tick : <EOL> return True <EOL> else : <EOL> if state : <EOL> switch . recycle_jitter_count += <NUM_LIT:1> <EOL> return False <EOL> def _post_switch_events ( self , switch_name , state ) : <EOL> """<STR_LIT>""" <EOL> if state == <NUM_LIT:1> : <EOL> for event in self . machine . switches [ switch_name ] . activation_events : <EOL> self . machine . events . post ( event ) <EOL> for tag in self . machine . switches [ switch_name ] . tags : <EOL> self . machine . events . post ( <EOL> self . switch_tag_event . replace ( '<STR_LIT:%>' , tag ) ) <EOL> elif state == <NUM_LIT:0> : <EOL> for event in ( <EOL> self . machine . switches [ switch_name ] . deactivation_events ) : <EOL> self . machine . events . post ( event ) <EOL> def _tick ( self ) : <EOL> """<STR_LIT>""" <EOL> for k in self . active_timed_switches . keys ( ) : <EOL> if k <= time . time ( ) : <EOL> for entry in self . active_timed_switches [ k ] : <EOL> if entry [ '<STR_LIT>' ] : <EOL> continue <EOL> self . log . debug ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , entry [ '<STR_LIT>' ] , <EOL> entry [ '<STR_LIT:state>' ] , entry [ '<STR_LIT>' ] ) <EOL> if entry [ '<STR_LIT>' ] : <EOL> entry [ '<STR_LIT>' ] ( switch_name = entry [ '<STR_LIT>' ] , <EOL> state = entry [ '<STR_LIT:state>' ] , <EOL> ms = entry [ '<STR_LIT>' ] , <EOL> ** entry [ '<STR_LIT>' ] ) <EOL> else : <EOL> entry [ '<STR_LIT>' ] ( ** entry [ '<STR_LIT>' ] ) <EOL> del self . active_timed_switches [ k ] </s>
