""" mahali_control_service.py $Id: mahali_controller_service.py 512 2015-09-22 18:08:30Z rps $ Behavioral control service for mahali implemented using redis messaging and json objects. The essential behavior is : 1. On startup check that the relay service is running. If not wait for it to startup. 2. Once the relay service is running, check the system uptime. If it is more than the threshold we can turn on the unit. 3. Check the latest temperature reading, if we are in the acceptable range, turn on the unit and led. 4. Check the temperature reading. If the unit is over temperature, shutdown the unit and led, wait until cooler. 5. If the unit is over temperature implement a time and temperature hysteresis to avoid repeated on / off cycles. 6. Emit a heartbeat and mirror state information to redis. """ import time import os import sys import optparse import threading import traceback #import numpy import json import redis import daemon # for running as a daemon import ConfigParser from mahali_common import * import datetime from datetime import timedelta from pytz import timezone # #--------------------------------------------------------- # GLOBALS #--------------------------------------------------------- # g_pingthread = None # set when thread instantiated g_instrument_check_time = -1 # initialize to immediate # #--------------------------------------------------------- # # for explanation, see: http://stackoverflow.com/questions/279561/what-is-the-python-equivalent-of-static-variables-inside-a-function # def static_vars(**kwargs): def decorate(func): for k in kwargs: setattr(func, k, kwargs[k]) return func return decorate # #--------------------------------------------------------- # def parse_command_line(): parser = optparse.OptionParser() parser.add_option("-v", "--verbose",action="store_true", dest="verbose", default=False,help="Print status messages to stdout.") parser.add_option("-d", "--debug",action="store_true", dest="debug", default=False,help="Print debug messages to stdout.") parser.add_option("-c", "--config",dest="CPATH",help="Use configuration files in CPATH.") parser.add_option("-f", "--foreground",action="store_true",dest="foreground",help="Execute in foreground and not daemon context.") (options, args) = parser.parse_args() return (options, args) # helper methods for state change decisions def relay_service_active(redis_db): return redis_db.sismember('mahali-service-list','mahali-relay-service') def get_relay_service_state(redis_db): msg = redis_db.get('mahali-relay-service-state') obj = deserialize_json(msg) return obj ## State machine state methods def startup_state(redis_db, config, options, logQ): if options.debug or options.verbose: logQ.logDebug('enter startup state') if not relay_service_active(redis_db): return 'startup' if get_uptime() < float(config['mahali-control-service']['minimum_uptime']): return 'startup' return 'activate' def activate_state(redis_db, config, options, logQ): imsg = 'enter activate state' logQ.logInfo(imsg) if (options.debug or options.verbose): logQ.logDebug(imsg) obj = get_relay_service_state(redis_db) if float(obj['temperature']) > float(config['mahali-instrument']['instrument_temperature_limit']): config['mahali-control-service']['overtemp_time'] = time.time() return 'overtemp_shutdown' # now turn on the receiver mahali_send_event(redis_db,logQ,'mahali-relay-command','command','set-power-state','on') mahali_send_event(redis_db,logQ,'mahali-relay-command','command','set-led-state','on') # force the data service to clock synchronize the unit after GPS power activation config['mahali-control-service']['synchronize_time'] = time.time() mahali_send_event(redis_db,logQ,'mahali-rinex-command','command','trigger','clock-sync') return 'synchronize' # #----------------------------------------------------------------------------------- # @static_vars(sync_first_f=True) def synchronize_state(redis_db, config, options, logQ): # debug statement placed outside because of reentry due to 'instrument_active_delay' obj = get_relay_service_state(redis_db) if float(obj['temperature']) > float(config['mahali-instrument']['instrument_temperature_limit']): config['mahali-control-service']['overtemp_time'] = time.time() return 'overtemp_shutdown' time1 = time.time() - config['mahali-control-service']['synchronize_time'] time2 = config['mahali-instrument']['instrument_active_delay'] if ( time1 < time2): if (options.debug): if (synchronize_state.sync_first_f): # print only once synchronize_state.sync_first_f = False dmsg = "%d seconds until running"%(time2-time1) logQ.logDebug(dmsg) return 'synchronize' synchronize_state.sync_first_f = True # reset for next state transition return 'running' # #----------------------------------------------------------------------------------- # def receiver_standby_state(redis_db, config, md_config, options, logQ): # # the code may loop through here for a while # if options.debug or options.verbose: logQ.logDebug('enter receiver standby state') instrument_model = config['mahali-instrument']['instrument_models'] if (instrument_model=='trimble'): gps_type = md_config['mahali-instrument-trimble']['gps_type'] if (gps_type == 'NETR9'): # implement time hysteresis if (time.time() - config['mahali-control-service']['standby_time']) < md_config['mahali-instrument-trimble']['turnon_active_delay']: return 'receiver_standby' return 'activate' # #----------------------------------------------------------------------------------- # def receiver_shutdown_state(redis_db, config, options, logQ): imsg = 'turning off the GPS receiver relay' logQ.logInfo(imsg) if (options.debug or options.verbose): logQ.logDebug(imsg) # the receiver is to be turned off (for whatever reason)... mahali_send_event(redis_db,logQ,'mahali-relay-command','command','set-power-state','off') mahali_send_event(redis_db,logQ,'mahali-relay-command','command','set-led-state','off') return 'receiver_standby' # automatic restart, if intended # end receiver_shutdown_state # #----------------------------------------------------------------------------------- # def instrument_check_init(redis_db, config, md_config, options, logQ): # # Note: You may ask, why place instrument specific software in the controller? # Services per design do not provide feedback to the controller, so if a service runs into problems # the controller won't know. For feedback the controller has to have some instrument specific code for # problems that require change of state. global g_pingthread if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, entered connectivity checker initialization' logQ.logDebug(dmsg) im_list = config['mahali-instrument']['instrument_models'] instrument_model = im_list[0] if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, instrument model=%s'%(instrument_model) logQ.logDebug(dmsg) if (instrument_model=='trimble'): gps_type = md_config['mahali-instrument-trimble']['gps_type'] if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, gps type=%s'%(gps_type) logQ.logDebug(dmsg) if (gps_type == 'NETR9'): from mahali_common import pingThread # only import if needed ipaddr = md_config['mahali-instrument-trimble']['ipaddr'] revisit_rate = md_config['mahali-instrument-trimble']['ping_rate'] g_pingthread = pingThread.pingThread(ipaddr, revisit_rate, logQ) # thread object is global g_pingthread.start() imsg= 'mahali_control_system, instrument_check, started connectivity checker for %s receiver'%(gps_type) logQ.logInfo(imsg) if (options.debug or options.verbose): logQ.logDebug(imsg) # end instrument_check_init # #----------------------------------------------------------------------------------- # def instrument_checker_shutdown(redis_db, config, md_config, options, logQ): if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, entered instrument_checker_shutdown.' logQ.logDebug(dmsg) im_list = config['mahali-instrument']['instrument_models'] instrument_model = im_list[0] if (instrument_model=='trimble'): gps_type = md_config['mahali-instrument-trimble']['gps_type'] if (gps_type == 'NETR9'): if (g_pingthread != None): g_pingthread.shutdown() imsg= 'mahali_control_system, instrument_check, shutting down connectivity checker for %s receiver'%(gps_type) logQ.logInfo(imsg) if (options.debug or options.verbose): logQ.logDebug(imsg) # end instrument_checker_shutdown # #----------------------------------------------------------------------------------- # def instrument_check(redis_db, config, md_config, options, logQ): global g_pingthread return_state = 'running' checked_f = False if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, entered instrument_check.' logQ.logDebug(dmsg) im_list = config['mahali-instrument']['instrument_models'] instrument_model = im_list[0] if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, instrument model=%s'%(instrument_model) logQ.logDebug(dmsg) if (instrument_model=='trimble'): gps_type = md_config['mahali-instrument-trimble']['gps_type'] if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, gps type=%s'%(gps_type) logQ.logDebug(dmsg) if (gps_type == 'NETR9'): # # if trimble then ping internal controller, if no response then turn power off, save time for hysterisis # checked_f = True if (g_pingthread != None): if (not g_pingthread.isResponding()): emsg= 'mahali_control_system, instrument_check, ping of %s NOT responding.'%(gps_type) logQ.logError(emsg) config['mahali-control-service']['standby_time'] = time.time() return_state = 'receiver_shutdown' else: imsg= 'mahali_control_system, instrument_check, ping of %s is responding.'%(gps_type) logQ.logInfo(imsg) if (options.debug or options.verbose): logQ.logDebug(imsg) # end else responding else: emsg= 'mahali_control_system, instrument_check, ping thread for %s not initialized'%(gps_type) LogQ.logError(emsg) if (options.debug or options.verbose): logQ.logDebug(emsg) # end else not initialized if (not checked_f): if (options.debug or options.verbose): dmsg= 'mahali_control_system, instrument_check, nothing to check.' logQ.logDebug(dmsg) return return_state # end instrument_check # #----------------------------------------------------------------------------------- # # the static flags declared as "decorators" below are used to reduce the amount of # events that are sent or to retain state (for control or debugging) # @static_vars(utc_first_f=True) @static_vars(utc_relay_str = []) @static_vars(utc_relay_int = []) def running_state(redis_db, config, md_config, options, logQ): global g_instrument_check_time return_state = 'running' err_f = False utc_enable = False # debug statement exterior to this call obj = get_relay_service_state(redis_db) relay_state = obj['power_state'] if obj['temperature'] > config['mahali-instrument']['instrument_temperature_limit']: config['mahali-control-service']['overtemp_time'] = time.time() return 'overtemp_shutdown' im_list = config['mahali-instrument']['instrument_models'] instrument_model = im_list[0] if (instrument_model=='trimble'): # # the trimble receiver is on its own IP address and sometimes gives up # this can be recognized by not responding to a ping # gps_type = md_config['mahali-instrument-trimble']['gps_type'] if (gps_type == 'NETR9'): if time.time() - g_instrument_check_time > md_config['mahali-instrument-trimble']['ping_rate']: # # time to ping # if (relay_state.find('on') >=0): # # only ping if not intentionally off # note the ping thread will still send an event if not receiving packets # return_state = instrument_check(redis_db, config, md_config, options, logQ) g_instrument_check_time = time.time() # reinitialize running_state.utc_first_f = True # reinitialize running_state.utc_first_1_f = True running_state.utc_first_2_f = True if (return_state != 'running'): return return_state # exit early # endif model NETR9 # endif vendor trimble # # if utc enabled and date is sane # if (current time > utc turn off time) and (current time < utc turn on time) # and if the relay is not already off, then then turn the relay off. # else if the relay is not on, then turn the relay on on. # utc_enable = config['mahali-control-service']['utc_enable'] dtNowUTC = datetime.datetime.now(timezone('UTC')) # all times must be timezone aware if ((dtNowUTC.year >= 2015) and utc_enable): if (running_state.utc_first_f): # initialize only once running_state.utc_first_f = False # don't send event again if (options.debug): dmsg = "utc start/stop enabled" logQ.logDebug(dmsg) running_state.utc_relay_str = config['mahali-control-service']['utc_relay'] running_state.utc_relay_int = [] # # evaluate elements # if (len(running_state.utc_relay_str) < 24): emsg = "24 elements not found in utc_relay" logQ.logError(emsg) err_f = True if (not err_f): ii = 0 while (ii < 24) and (not err_f): try: datum = int(running_state.utc_relay_str[ii]) except Exception as eobj: emsg = "24 elements not found in utc_relay" logQ.logError(emsg) err_f = True else: if (datum == 0) or (datum == 1): running_state.utc_relay_int = running_state.utc_relay_int + [datum] else: emsg = "Non-zero or non-one element found in utc_relay" logQ.logError(emsg) err_f = True ii = ii + 1 # end while # end if not error # endif not utc_first_f if (len(running_state.utc_relay_int)==24): # all relay hours have been collected on_off = running_state.utc_relay_int[dtNowUTC.hour -1] if (on_off == 0): # turn off, if not already off if (relay_state.find('on') >=0): # # receiver shutdown but don't restart, stay in the running state # imsg = "utc turning off receiver relay but continuing in running state" logQ.logInfo(imsg) if (options.debug): logQ.logDebug(imsg) dummy_state = receiver_shutdown_state(redis_db, config, options, logQ) # # do not use the state returned by the above call # else: # turn on, if not already on if (relay_state.find('off') >= 0): # # relay is off, go to receiver startup, change state cycle through startup # imsg = "utc turning on receiver relay in running state, re-initializing" logQ.logInfo(imsg) if (options.debug): logQ.logDebug(imsg) return_state = activate_state(redis_db, config, options, logQ) # end if # end else # endif 24 hours of relay settings collected # endif utc enabled return return_state # end running_state # #----------------------------------------------------------------------------------- # def overtemp_shutdown_state(redis_db, config, options, logQ): if (options.debug or options.verbose): logQ.logDebug('enter overtemp shutdown state') # we are over temperature, shut off... mahali_send_event(redis_db,logQ,'mahali-relay-command','command','set-power-state','off') mahali_send_event(redis_db,logQ,'mahali-relay-command','command','set-led-state','off') return 'overtemp_standby' def overtemp_standby_state(redis_db, config, options, logQ): if options.debug and options.verbose: logQ.logDebug('enter overtemp standby state') obj = get_relay_service_state(redis_db) if options.debug and options.verbose: logQ.logDebug('overtemp %2.1f %2.1f %2.1f' % (float(obj['temperature']), float(config['mahali-instrument']['instrument_temperature_limit']), float(config['mahali-control-service']['overtemp_recovery_margin']))) if obj['temperature'] > config['mahali-instrument']['instrument_temperature_limit']: config['mahali-control-service']['overtemp_time'] = time.time() return 'overtemp_standby' # implement time hysteresis if (time.time() - config['mahali-control-service']['overtemp_time']) < config['mahali-control-service']['overtemp_recovery_time']: return 'overtemp_standby' # implement temperature hysteresis if (float(obj['temperature']) + float(config['mahali-control-service']['overtemp_recovery_margin'])) > float(config['mahali-instrument']['instrument_temperature_limit']): return 'overtemp_standby' return 'activate' # primary control service def mahali_control_service(options): """ This service implements a functional finite state machine to provide correct startup and operational behavior. """ start_time = time.time() # create redis interface redis_db = redis.StrictRedis(host='localhost', port=6379, db=0) # register with redis in the mahali service list if not mahali_register_service(redis_db, 'mahali-control-service'): sys.exit() # create logging instance logQ = redisLoggerQueue('control-service',redis_db, options) # log startup logQ.logInfo('mahali-control-service startup') op_state = 'startup' saved_state = None op_counter = 0 data_sync_time = -1 cloud_sync_time = -1 clock_sync_time = -1 # load configuration mahali_cfg = mahaliConfiguration('mahali-gps-array',options.CPATH, redis_db, logQ) config = mahali_cfg.load('mahali-control-config') # bad name because there are more than one configs md_config = mahali_cfg.load('mahali-data-config') if options.debug: logQ.logDebug('loaded config ' + str(config)) # primary service loop try: # mirror site configuration to info on startup # this ensures traceability as we move boxes around # for a given data collection logQ.logInfo({'mahali-controller-service':'startup', 'config':config}) instrument_check_init(redis_db, config, md_config, options, logQ) # provides health feedback from instrument while True: time.sleep(0.5) if options.debug and options.verbose and not options.foreground: logQ.logDebug('service loop at ' + str(time.gmtime(time.time()))) op_counter = op_counter + 1 # used as divide down for certain checks if op_state == 'startup': saved_state = op_state # save for reducing error reporting try: op_state = startup_state(redis_db, config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception # end if elif op_state == 'overtemp_shutdown': saved_state = op_state try: op_state = overtemp_shutdown_state(redis_db, config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception # end if elif op_state == 'overtemp_standby': saved_state = op_state try: op_state = overtemp_standby_state(redis_db, config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception # end if elif op_state == 'receiver_shutdown': saved_state = op_state try: op_state = receiver_shutdown_state(redis_db, config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception # end if elif op_state == 'receiver_standby': saved_state = op_state try: op_state = receiver_standby_state(redis_db, config, md_config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception # end if elif op_state == 'synchronize': # this state may reenter due to delay in synchronizing if (options.debug or options.verbose): if (saved_state != 'synchronize'): # only print once on entry logQ.logDebug('enter synchronize state') saved_state = op_state try: op_state = synchronize_state(redis_db, config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception #end if elif op_state == 'activate': try: op_state = activate_state(redis_db, config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception # end if elif op_state == 'running': # this state will reenter if (options.debug or options.verbose): if (saved_state != 'running'): # only print once on entry logQ.logDebug('enter running state') saved_state = op_state try: op_state = running_state(redis_db, config, md_config, options, logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with controller state %s." % (exp_str, repr(op_state)) logQ.logError(emsg) op_state = 'startup' # end exception # end if else: logQ.logError('mahali-control-service unknown operational state %s' % (op_state)) op_state = 'startup' if not relay_service_active(redis_db): op_state = 'startup' # report over temperature conditions # every 60 seconds or so to reduce message rate... if (op_state == 'overtemp_shutdown' or op_state == 'overtemp_standby') and (int(op_counter % 120) == 0): obj = get_relay_service_state(redis_db) mahali_send_event(redis_db,logQ,'mahali-control-status','status','over-temperature',obj['temperature']) # trigger clock sync if config['mahali-control-service']['clock_sync']: if time.time() - clock_sync_time > config['mahali-control-service']['clock_interval']: if options.debug: logQ.logDebug('do clock_sync check') clock_sync_time = time.time() mahali_send_event(redis_db,logQ,'mahali-data-command','command','trigger','clock-sync') # trigger rinex data translator if config['mahali-control-service']['data_sync']: if (time.time() - data_sync_time) > config['mahali-control-service']['data_interval']: if options.debug: logQ.logDebug('do data_sync check') data_sync_time = time.time() mahali_send_event(redis_db,logQ,'mahali-data-command','command','trigger','data-sync') # trigger owncloud data transfers if config['mahali-control-service']['cloud_sync']: if time.time() - cloud_sync_time > config['mahali-control-service']['cloud_interval']: cloud_sync_time = time.time() if options.debug: logQ.logDebug('do cloud_sync check') mahali_send_event(redis_db,logQ,'mahali-cloud-command','command','trigger','cloud-sync') # provide heartbeat telemetry and state mirroring to redis active_time = time.time() - start_time mahali_send_event(redis_db,logQ,'mahali-control-service-heartbeat','heartbeat','active',active_time) # mirror controller state up_time = get_uptime() state_info = {'start_time':start_time, 'active_time':active_time, 'update_time':time.time(), 'uptime':up_time, 'control_state':op_state} mahali_set_object(redis_db,logQ,'mahali-control-service-state', state_info) # end while except KeyboardInterrupt: logQ.logInfo("exiting on keyboard interrupt") except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with mahali-control-service." % (exp_str) if logQ: logQ.logError(emsg) else: print emsg # shutdown instrument specific stuff owned by the controller that has a separate lifespan (any threads?) instrument_checker_shutdown(redis_db, config, md_config, options, logQ) # publish shutdown if logQ: logQ.logInfo('shutdown') else: print 'shutdown' """ Note : When the service exits we do not by default shutdown the power relay. This is intended to avoid turning off a working device on an error. """ if not mahali_unregister_service(redis_db, 'mahali-control-service'): sys.exit() # deregister with redis redis_db.connection_pool.disconnect() if __name__ == '__main__': # parse command line options options, args = parse_command_line() if (options.foreground): print "Mahali Control Service in DEBUG mode" mahali_control_service(options) else: with daemon.DaemonContext(): mahali_control_service(options)