Source code for ibeis.main_module

# -*- coding: utf-8 -*-
"""
This module defines the entry point into the IBEIS system
ibeis.opendb and ibeis.main are the main entry points
"""
from __future__ import absolute_import, division, print_function
from six.moves import builtins
import sys
import multiprocessing

PREINIT_MULTIPROCESSING_POOLS = '--preinit' in sys.argv
QUIET = '--quiet' in sys.argv
NOT_QUIET = not QUIET
USE_GUI = '--gui' in sys.argv or '--nogui' not in sys.argv

try:
    profile = getattr(builtins, 'profile')
except AttributeError:
[docs] def profile(func): return func
def _on_ctrl_c(signal, frame): proc_name = multiprocessing.current_process().name print('[ibeis.main_module] Caught ctrl+c in %s' % (proc_name,)) try: _close_parallel() except Exception as ex: print('Something very bad happened' + repr(ex)) finally: print('[ibeis.main_module] sys.exit(0)') sys.exit(0) #----------------------- # private init functions def _init_signals(): import signal signal.signal(signal.SIGINT, _on_ctrl_c) def _reset_signals(): import signal signal.signal(signal.SIGINT, signal.SIG_DFL) # reset ctrl+c behavior def _parse_args(): from ibeis import params params.parse_args() #@profile def _init_matplotlib(): from plottool import __MPL_INIT__ __MPL_INIT__.init_matplotlib() #@profile def _init_gui(activate=True): import guitool if NOT_QUIET: print('[main] _init_gui()') guitool.ensure_qtapp() #USE_OLD_BACKEND = '--old-backend' in sys.argv #if USE_OLD_BACKEND: from ibeis.gui import guiback back = guiback.MainWindowBackend() #else: # from ibeis.gui import newgui # back = newgui.IBEISGuiWidget() if activate: guitool.activate_qwindow(back.mainwin) return back #@profile def _init_ibeis(dbdir=None, verbose=None, use_cache=True, web=None, **kwargs): """ Private function that calls code to create an ibeis controller """ import utool as ut from ibeis import params from ibeis.control import IBEISControl if verbose is None: verbose = ut.VERBOSE if verbose and NOT_QUIET: print('[main] _init_ibeis()') # Use command line dbdir unless user specifies it if dbdir is None: ibs = None ut.printWARN('[main!] WARNING args.dbdir is None') else: kwargs = kwargs.copy() request_dbversion = kwargs.pop('request_dbversion', None) asproxy = kwargs.pop('asproxy', None) ibs = IBEISControl.request_IBEISController( dbdir=dbdir, use_cache=use_cache, request_dbversion=request_dbversion, asproxy=asproxy) if web is None: web = ut.get_argflag(('--webapp', '--webapi', '--web', '--browser'), help_='automatically launch the web app / web api') #web = params.args.webapp if web: from ibeis.web import app port = params.args.webport app.start_from_ibeis(ibs, port=port, **kwargs) return ibs def __import_parallel_modules(): # Import any modules which parallel process will use here # so they are accessable when the program forks #from utool import util_sysreq #util_sysreq.ensure_in_pythonpath('hesaff') #util_sysreq.ensure_in_pythonpath('pyrf') #util_sysreq.ensure_in_pythonpath('code') #import pyhesaff # NOQA #import pyrf # NOQA from ibeis import core_annots # NOQA #.algo.preproc import preproc_chip # NOQA def _init_parallel(): import utool as ut if ut.VERBOSE: print('_init_parallel') from utool import util_parallel from ibeis import params __import_parallel_modules() util_parallel.set_num_procs(params.args.num_procs) if PREINIT_MULTIPROCESSING_POOLS: util_parallel.init_pool(params.args.num_procs) def _close_parallel(): #if ut.VERBOSE: # print('_close_parallel') try: from utool import util_parallel util_parallel.close_pool(terminate=True) except Exception as ex: import utool as ut ut.printex(ex, 'error closing parallel') raise def _init_numpy(): import utool as ut import numpy as np if ut.VERBOSE: print('_init_numpy') error_options = ['ignore', 'warn', 'raise', 'call', 'print', 'log'] on_err = error_options[0] #np.seterr(divide='ignore', invalid='ignore') numpy_err = { 'divide': on_err, 'over': on_err, 'under': on_err, 'invalid': on_err, } #numpy_print = { # 'precision': 8, # 'threshold': 500, # 'edgeitems': 3, # 'linewidth': 200, # default 75 # 'suppress': False, # 'nanstr': 'nan', # 'formatter': None, #} np.seterr(**numpy_err) #np.set_printoptions(**numpy_print) #----------------------- # private loop functions def _guitool_loop(main_locals, ipy=False): import guitool from ibeis import params print('[main] guitool loop') back = main_locals.get('back', None) if back is not None: loop_freq = params.args.loop_freq ipy = ipy or params.args.cmd guitool.qtapp_loop(qwin=back.mainwin, ipy=ipy, frequency=loop_freq, init_signals=False) if ipy: # If we're in IPython, the qtapp loop won't block, so we need to refresh back.refresh_state() else: if NOT_QUIET: print('WARNING: back was not expected to be None')
[docs]def set_newfile_permissions(): r""" sets this processes default permission bits when creating new files CommandLine: python -m ibeis.main_module --test-set_newfile_permissions Example: >>> # ENABLE_DOCTEST >>> from ibeis.main_module import * # NOQA >>> import os >>> import utool as ut >>> # write before umask >>> ut.delete('tempfile1.txt') >>> ut.write_to('tempfile1.txt', 'foo') >>> stat_result1 = os.stat('tempfile1.txt') >>> # apply umask >>> set_newfile_permissions() >>> ut.delete('tempfile2.txt') >>> ut.write_to('tempfile2.txt', 'foo') >>> stat_result2 = os.stat('tempfile2.txt') >>> # verify results >>> print('old masked all bits = %o' % (stat_result1.st_mode)) >>> print('new masked all bits = %o' % (stat_result2.st_mode)) """ import os #import stat # Set umask so all files written will be group read and writable # To get the permissions we want subtract what you want from 0o0666 because # umask subtracts the mask you give it. #mask = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IWGRP | stat.S_IROTH #mask = 0o000 # most permissive umask mask = 0o000 # most permissive umask prev_mask = os.umask(mask) return prev_mask #print('prev_mask = %o' % (prev_mask,)) #print('new_mask = %o' % (mask,)) #@profile
[docs]def main(gui=True, dbdir=None, defaultdb='cache', allow_newdir=False, db=None, delete_ibsdir=False, **kwargs): """ Program entry point Inits the system environment, an IBEISControl, and a GUI if requested Args: gui (bool): (default=True) If gui is False a gui instance will not be created dbdir (None): full directory of a database to load db (None): name of database to load relative to the workdir allow_newdir (bool): (default=False) if False an error is raised if a a new database is created defaultdb (str): codename of database to load if db and dbdir is None. a value of 'cache' will open the last database opened with the GUI. Returns: dict: main_locals """ set_newfile_permissions() from ibeis.init import main_commands from ibeis.init import sysres # Display a visible intro message msg1 = ''' _____ ....... _______ _____ _______ | |_____| |______ | |______ ..|.. |.....| |______s__|__ ______| ''' msg2 = ''' _____ ______ _______ _____ _______ | |_____] |______ | |______ __|__ |_____] |______ __|__ ______| ''' if NOT_QUIET: print(msg2 if '--myway' not in sys.argv else msg1) # Init the only two main system api handles ibs = None back = None if NOT_QUIET: print('[main] ibeis.main_module.main()') _preload() DIAGNOSTICS = NOT_QUIET if DIAGNOSTICS: import os import utool as ut import ibeis print('[main] MAIN DIAGNOSTICS') print('[main] * username = %r' % (ut.get_user_name())) print('[main] * ibeis.__version__ = %r' % (ibeis.__version__,)) print('[main] * computername = %r' % (ut.get_computer_name())) print('[main] * cwd = %r' % (os.getcwd(),)) print('[main] * sys.argv = %r' % (sys.argv,)) # Parse directory to be loaded from command line args # and explicit kwargs dbdir = sysres.get_args_dbdir(defaultdb, allow_newdir, db, dbdir, cache_priority=False) if delete_ibsdir is True: from ibeis.other import ibsfuncs assert allow_newdir, 'must be making new directory if you are deleting everything!' ibsfuncs.delete_ibeis_database(dbdir) # Execute preload commands main_commands.preload_commands(dbdir, **kwargs) # PRELOAD CMDS try: # Build IBEIS Control object ibs = _init_ibeis(dbdir) if gui and USE_GUI: back = _init_gui(activate=kwargs.get('activate', True)) back.connect_ibeis_control(ibs) except Exception as ex: print('[main()] IBEIS LOAD imageseted exception: %s %s' % (type(ex), ex)) raise main_commands.postload_commands(ibs, back) # POSTLOAD CMDS main_locals = {'ibs': ibs, 'back': back} return main_locals
[docs]def opendb_in_background(*args, **kwargs): """ Starts a web server in the background """ import utool as ut import time sec = kwargs.pop('wait', 0) if sec != 0: print('waiting %s seconds for startup' % (sec,)) proc = ut.spawn_background_process(opendb, *args, **kwargs) if sec != 0: time.sleep(sec) # wait for process to initialize return proc
[docs]def opendb_bg_web(*args, **kwargs): """ Wrapper around opendb_in_background, returns a nice web_ibs object to execute web calls using normal python-like syntax Accespts domain and port as kwargs Kwargs: port, domain CommandLine: python -m ibeis.main_module opendb_bg_web Example: >>> # DISABLE_DOCTEST >>> from ibeis.main_module import * # NOQA >>> print('Opening a web_ibs') >>> web_ibs = opendb_bg_web() >>> print('SUCESS Opened a web_ibs!') >>> print(web_ibs) >>> print('Now kill the web_ibs') >>> web_ibs.terminate2() """ import utool as ut kwargs = kwargs.copy() domain = kwargs.pop('domain', ut.get_argval('--domain', type_=str, default=None)) port = kwargs.pop('port', 5000) if 'wait' in kwargs: print('NOTE: No need to specify wait param anymore. ' 'This is automatically taken care of.') if domain is None: # Requesting a local test server _kw = dict(web=True, browser=False) _kw.update(kwargs) web_ibs = opendb_in_background(*args, **_kw) else: # Using a remote controller, no need to spin up anything web_ibs = ut.DynStruct() web_ibs.terminate2 = lambda: None # Augment web instance with usefull test functions if domain is None: domain = 'http://127.0.1.1' if not domain.startswith('http://'): domain = 'http://' + domain baseurl = domain + ':' + str(port) web_ibs.domain = domain web_ibs.port = port web_ibs.baseurl = baseurl def send_ibeis_request(suffix, type_='post', **kwargs): """ Posts a request to a url suffix """ import requests import utool as ut if not suffix.endswith('/'): raise Exception('YOU PROBABLY WANT A / AT THE END OF YOUR URL') payload = ut.map_dict_vals(ut.to_json, kwargs) if type_ == 'post': resp = requests.post(baseurl + suffix, data=payload) json_content = resp._content elif type_ == 'get': resp = requests.get(baseurl + suffix, data=payload) json_content = resp.content try: content = ut.from_json(json_content) except ValueError: raise Exception('Expected JSON string but got json_content=%r' % (json_content,)) else: # print('content = %r' % (content,)) if content['status']['code'] != 200: print(content['status']['message']) raise Exception(content['status']['message']) request_response = content['response'] return request_response def wait_for_results(jobid, timeout=None, delays=[1, 3, 10]): """ Waits for results from an engine """ for _ in ut.delayed_retry_gen(delays): print('Waiting for jobid = %s' % (jobid,)) status_response = web_ibs.send_ibeis_request('/api/engine/job/status/', jobid=jobid) if status_response['jobstatus'] == 'completed': break return status_response def read_engine_results(jobid): result_response = web_ibs.send_ibeis_request('/api/engine/job/result/', jobid=jobid) return result_response def send_request_and_wait(suffix, type_='post', timeout=None, **kwargs): jobid = web_ibs.send_ibeis_request(suffix, type_=type_, **kwargs) status_response = web_ibs.wait_for_results(jobid, timeout) # NOQA result_response = web_ibs.read_engine_results(jobid) #>>> cmdict = ut.from_json(result_response['json_result'])[0] return result_response web_ibs.send_ibeis_request = send_ibeis_request web_ibs.wait_for_results = wait_for_results web_ibs.read_engine_results = read_engine_results web_ibs.send_request_and_wait = send_request_and_wait def wait_until_started(): """ waits until the web server responds to a request """ import requests for count in ut.delayed_retry_gen([1], timeout=15): if ut.VERBOSE: print('Waiting for server to be up. count=%r' % (count,)) try: web_ibs.send_ibeis_request('/api/test/heartbeat/', type_='get') break except requests.ConnectionError: pass wait_until_started() return web_ibs
[docs]def opendb(db=None, dbdir=None, defaultdb='cache', allow_newdir=False, delete_ibsdir=False, verbose=False, use_cache=True, web=None, **kwargs): """ main without the preload (except for option to delete database before opening) Args: db (str): database name in your workdir used only if dbdir is None dbdir (None): full database path defaultdb (str): dbdir search stratagy when db is None and dbdir is None allow_newdir (bool): (default=True) if True errors when opening a nonexisting database delete_ibsdir (bool): BE CAREFUL! (default=False) if True deletes the entire verbose (bool): verbosity flag web (bool): starts webserver if True (default=param specification) use_cache (bool): if True will try to return a previously loaded controller Returns: ibeis.IBEISController: ibs Example: >>> # ENABLE_DOCTEST >>> from ibeis.main_module import * # NOQA >>> db = None >>> dbdir = None >>> defaultdb = 'cache' >>> allow_newdir = False >>> delete_ibsdir = False >>> verbose = False >>> use_cache = True >>> ibs = opendb(db, dbdir, defaultdb, allow_newdir, delete_ibsdir, >>> verbose, use_cache) >>> result = str(ibs) >>> print(result) """ from ibeis.init import sysres from ibeis.other import ibsfuncs dbdir = sysres.get_args_dbdir(defaultdb, allow_newdir, db, dbdir, cache_priority=False) if delete_ibsdir is True: assert allow_newdir, ( 'must be making new directory if you are deleting everything!') ibsfuncs.delete_ibeis_database(dbdir) ibs = _init_ibeis(dbdir, verbose=verbose, use_cache=use_cache, web=web, **kwargs) return ibs
[docs]def start(*args, **kwargs): """ alias for main() """ # + main.__doc__ return main(*args, **kwargs)
[docs]def test_main(gui=True, dbdir=None, defaultdb='cache', allow_newdir=False, db=None): """ alias for main() """ # + main.__doc__ from ibeis.init import sysres _preload() dbdir = sysres.get_args_dbdir(defaultdb, allow_newdir, db, dbdir, cache_priority=False) ibs = _init_ibeis(dbdir) return ibs #@profile
def _preload(mpl=True, par=True, logging=True): """ Sets up python environment """ import utool as ut #from ibeis.init import main_helpers from ibeis import params if multiprocessing.current_process().name != 'MainProcess': return if ut.VERBOSE: print('[ibies] _preload') _parse_args() # mpl backends if logging and not params.args.nologging: # Log in the configured ibeis log dir (which is maintained by utool) # fix this to be easier to figure out where the logs actually are ut.start_logging(appname='ibeis') if mpl: _init_matplotlib() # numpy print settings _init_numpy() # parallel servent processes if par: _init_parallel() # ctrl+c _init_signals() # inject colored exceptions ut.util_inject.inject_colored_exceptions() # register type aliases for debugging #main_helpers.register_utool_aliases() #return params.args #@profile
[docs]def main_loop(main_locals, rungui=True, ipy=False, persist=True): """ Runs the qt loop if the GUI was initialized and returns an executable string for embedding an IPython terminal if requested. If rungui is False the gui will not loop even if back has been created the main locals dict must be callsed main_locals in the scope you call this function in. Args: main_locals (dict_): rungui (bool): ipy (bool): persist (bool): Returns: str: execstr """ print('[main] ibeis.main_module.main_loop()') from ibeis import params import utool as ut #print('current process = %r' % (multiprocessing.current_process().name,)) #== 'MainProcess': if rungui and not params.args.nogui: try: _guitool_loop(main_locals, ipy=ipy) except Exception as ex: ut.printex(ex, 'error in main_loop') raise #if not persist or params.args.cmd: # main_close() # Put locals in the exec namespace ipycmd_execstr = ut.ipython_execstr() locals_execstr = ut.execstr_dict(main_locals, 'main_locals') execstr = locals_execstr + '\n' + ipycmd_execstr return execstr
[docs]def main_close(main_locals=None): #import utool as ut #if ut.VERBOSE: # print('main_close') _close_parallel() _reset_signals() #if __name__ == '__main__': # multiprocessing.freeze_support()
if __name__ == '__main__': """ CommandLine: python -m ibeis.main_module python -m ibeis.main_module --allexamples python -m ibeis.main_module --allexamples --noface --nosrc """ multiprocessing.freeze_support() # for win32 import utool as ut # NOQA ut.doctest_funcs()