# -*- 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()