""" mahali_data_service.py $Id: mahali_data_service.py 476 2015-08-13 21:20:35Z flind $ This service convert performs conversion of raw GPS TEC data to RINEX. """ # # #------------------------------------------------------------ # imports #------------------------------------------------------------ # import time import sys import os import io import optparse import string import json import redis import subprocess # 'system' call import uuid # unique id import daemon # for running as a daemon from mahali_common import * # #------------------------------------------------------------------- # class ExceptionString(Exception): """ Simple exception handling string """ def __str__(self): return repr(self.args[0]) # #------------------------------------------------------------------- # #------------------------------------------------------------------- def data_sync(config, logQ): if not config['mahali-data-service']['data_sync']: logQ.logDebug('rinex conversion attempt with data_sync disabled!') return raw_data_path = config['mahali-data-service']['raw_data_path'] rinex_data_path = config['mahali-data-service']['rinex_data_path'] instruments = config['mahali-instrument']['instrument_models'] sync_cnt = 0 for inst in instruments: print inst cfg_name = 'mahali-instrument-%s' % (inst) cfg_path = config[cfg_name]['path'] cfg_command = config[cfg_name]['command'] cfg_python = config[cfg_name]['python'] cfg_path = fixup_path(cfg_path) cfg_call = cfg_path + cfg_command if options.debug: logQ.logDebug('data_sync source %s to dest %s' % (raw_data_path, rinex_data_path)) sync_start = time.time() if options.debug and options.verbose: logQ.logDebug('data_sync command %s %s %s %s' % (cfg_python, cfg_call, raw_data_path, rinex_data_path)) if options.simulate: logQ.logInfo('simulate data sync call.') else: subprocess.check_call( [cfg_python, cfg_call, "-s %s" % (raw_data_path), "-d %s" % (rinex_data_path) ]) sync_cnt += 1 if options.debug: sync_complete = time.time() sync_time = sync_complete - sync_start logQ.logDebug('data sync of %ld instruments complete in time %s seconds.' % (sync_cnt, sync_time)) return sync_cnt def parse_command_line(): parser = optparse.OptionParser() parser.add_option("-v", "--verbose",action="store_true", dest="verbose", default=False,help="prints debug output and additional detail.") parser.add_option("-d", "--debug",action="store_true", dest="debug", default=False,help="run in debug mode and not service context.") parser.add_option("-s", "--simulate",action="store_true", dest="simulate", default=False,help="simulate cloud sync commands.") 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) # end parse_command_line # #------------------------------------------------------------------- # def mahali_data_service(options): start_time = time.time() # create redis interface redis_db = redis.StrictRedis(host='localhost', port=6379, db=0) if not mahali_register_service(redis_db, 'mahali-data-service'): sys.exit() pubsub = redis_db.pubsub() # listener # create logging instance logQ = redisLoggerQueue('data-service',redis_db, options) # channel subcriptions, hardware static for now pubsub.psubscribe('mahali-data-command') mahali_cfg = mahaliConfiguration('mahali-gps-array',options.CPATH, redis_db, logQ) config = mahali_cfg.load('mahali-data-config') # log startup logQ.logInfo('mahali-data-service startup') trg_count = 0 sync_count = 0 try: for ocmsg in pubsub.listen(): # blocking call, runs forever if options.debug and options.verbose: if ocmsg: print ocmsg if ocmsg['type'] == 'psubscribe': continue try: jmsg = ocmsg['data'] except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with message separation for %s." % (exp_str, str(ocmsg)) logQ.logError(emsg) continue msg = None # deserialize message try: msg = deserialize_json(jmsg) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with message deserialization for %s." % (exp_str, str(jmsg)) logQ.logError(emsg) continue if msg and msg['event_type'] == 'command' and msg['event'] == 'trigger': if (msg['event_data'] == 'data-sync'): trg_count += 1 if options.debug: logQ.logDebug('data sync event %s' % (msg)) try: sync_count += data_sync(config,logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s in data sync" % (exp_str) logQ.logError(emsg) continue elif (msg['event_data'] == 'clock-sync'): if options.debug: logQ.logDebug('clock sync event %s' % (msg)) try: pass #clock_sync(config,logQ) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s in clock sync" % (exp_str) logQ.logError(emsg) continue # mirror state active_time = start_time - time.time() state_info = {'start_time':start_time, 'active_time':active_time, 'update_time':time.time(), 'trigger_count':trg_count, 'sync_count': sync_count} mahali_set_object(redis_db,logQ,'mahali-cloud-service-state', state_info) except KeyboardInterrupt: lmsg = 'exiting on keyboard interrupt' if logQ: logQ.logInfo(lmsg) else: print lmsg except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with mahali data service" % (exp_str) if logQ: logQ.logError(emsg) else: print emsg # 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-data-service'): sys.exit() # deregister with redis redis_db.connection_pool.disconnect() # end main_program # #------------------------------------------------------------------- # # MAIN # #------------------------------------------------------------------- # if __name__ == '__main__': # parse command line options options,args = parse_command_line() if options.debug: DEBUG = options.debug if options.foreground: print "Mahali data service in DEBUG mode" mahali_data_service(options) # direct call else: with daemon.DaemonContext(): mahali_data_service(options) # called through daemonizer