def option_handle():
"""?????????"""
parse_command_line()
if options.config:
if os.path.exists(options.config):
options.parse_config_file(options.config)
else:
print "can not find %s"%options.config
print "usage:python %s --help"%os.path.basename(__file__)
sys.exit(1)
if options.voicelist:
def _parse_voicelist():
return tornado.escape.json_decode(options.voicelist)
global ALL_VOICE
ALL_VOICE = _parse_voicelist()
logging.debug("conf voicelist: %s", ALL_VOICE)
if options.cache_dir:
mkdir_p(options.cache_dir)
python类config()的实例源码
def setup_handlers(web_app):
env_name = 'production'
config = config_for_env(env_name, web_app.settings['base_url'])
define('config', config)
settings = dict(
debug=True,
serve_traceback=True,
compiled_template_cache=False,
template_path=os.path.join(os.path.dirname(__file__), 'static/'),
static_path=os.path.join(os.path.dirname(__file__), 'static/'),
# Ensure static urls are prefixed with the base url too
static_url_prefix=config['URL'] + 'static/',
)
web_app.settings.update(settings)
socket_url = url_path_join(config['URL'], r'socket/(\S+)')
host_pattern = '.*'
route_pattern = url_path_join(config['URL'], '/interact')
web_app.add_handlers(host_pattern, [
(route_pattern, LandingHandler),
(route_pattern + '/', LandingHandler),
(socket_url, RequestHandler)
])
def get(self, args):
is_file_request = ('file_url' in args)
# branch name can be omitted for default value
is_git_request = ('repo' in args and 'path' in args)
valid_request = xor(is_file_request, is_git_request)
def server_extension_url(url):
return 'nbextensions/nbpuller/' + url
if not valid_request:
self.render('404.html', server_extension_url=server_extension_url,)
util.logger.info("rendering progress page")
# These config options are passed into the `openStatusSocket`
# JS function.
username = str(self.get_current_user())
try:
username = self.get_current_user().get('name')
except Exception as e:
pass
util.logger.info("Username: " + username)
socket_args = json.dumps({
'is_development': options.config['DEBUG'],
'base_url': options.config['URL'],
'username': username,
})
self.render(
"progress.html",
socket_args=socket_args,
server_extension_url=server_extension_url,
)
def get(self, args):
is_file_request = ('file' in args)
is_git_request = ('repo' in args and 'path' in args)
valid_request = xor(is_file_request, is_git_request)
if not valid_request:
self.render('404.html')
hubauth = HubAuth(options.config)
# authenticate() returns either a username as a string or a redirect
redirection = username = hubauth.authenticate(self.request)
util.logger.info("authenticate returned: {}".format(redirection))
is_redirect = (redirection.startswith('/') or
redirection.startswith('http'))
if is_redirect:
values = []
for k, v in args.items():
if not isinstance(v, str):
v = '&path='.join(v)
values.append('%s=%s' % (k, v))
util.logger.info("rendering landing page")
download_links = (util.generate_git_download_link(args)
if is_git_request
else [args['file']])
return self.render(
'landing.html',
authenticate_link=redirection,
download_links=download_links,
query='&'.join(values))
util.logger.info("rendering progress page")
# These config options are passed into the `openStatusSocket`
# JS function.
socket_args = json.dumps({
'is_development': options.config['DEBUG'],
'base_url': options.config['URL'],
'username': username,
})
self.render("progress.html", socket_args=socket_args)
def open(self, username, args):
util.logger.info('({}) Websocket connected'.format(username))
# We don't do validation since we assume that the LandingHandler did
# it, so this isn't very secure.
is_file_request = ('file' in args)
try:
if is_file_request:
message = yield thread_pool.submit(
download_file_and_redirect,
username=username,
file_url=args['file'],
config=options.config,
)
else:
message = yield thread_pool.submit(
pull_from_github,
username=username,
repo_name=args['repo'],
paths=args['path'],
config=options.config,
progress=Progress(username, self.write_message)
)
util.logger.info('Sent message: {}'.format(message))
self.write_message(message)
except Exception as e:
# If something bad happens, the client should see it
message = messages.error(str(e))
util.logger.error('Sent message: {}'.format(message))
self.write_message(message)
def get(self):
self.render("index.html", config=global_game_manager, registered=False)
def setup_config(config):
global_game_manager.define_rule(
config['max_round'], config['initial_stack'], config['small_blind'],
config['ante'], config['blind_structure']
)
for player in config['ai_players']:
global_game_manager.join_ai_player(player['name'], player['path'])
def start_server(config_path, port, speed):
global MODE_SPEED
with open(config_path, "rb") as f:
config = yaml.load(f)
setup_config(config)
MODE_SPEED = speed
app = Application()
app.listen(port)
tornado.ioloop.IOLoop.current().start()
def main():
tornado.options.parse_command_line()
start_server(options.config, options.port, options.speed)
def post(self):
feedback = 'wait'
user_id = self.get_argument('user_id', 0)
object_type = self.get_argument('object_type', 0)
group_type = self.get_argument('group_type', 0)
order_by = self.get_argument('order_by', 0)
tag = self.get_argument('tag', 0)
key = ' '.join((user_id, object_type, group_type, order_by, tag))
file_path = os.path.join(options.config['root_path'], 'data', key + '.json')
handling = options.handling
if not os.path.exists(file_path):
if key not in handling:
yield grab(user_id=user_id, object_type=object_type,
group_type=group_type, order_by=order_by, tag=tag)
options.handling.append(key)
else:
if key in handling:
options.handling.remove(key)
items = json.load(open(file_path, 'r'))
if time.strftime('%Y-%m-%d', time.localtime(time.time())) != items[0]:
yield grab(user_id=user_id, object_type=object_type,
group_type=group_type, order_by=order_by, tag=tag)
if len(items) <= 1:
feedback = '404'
else:
feedback = items
respon_json = escape.json_encode(feedback)
self.write(respon_json)
def main():
"""Main entry point for iotile-gateway."""
supervisor = None
logging.basicConfig(format='%(levelname)-.1s-%(asctime)-15s-%(module)-10s:%(lineno)-4s %(message)s')
log = logging.getLogger(__name__)
def quit_signal_handler(signum, frame): # pylint: disable=W0613
"""Signal handler to catch ^C and cleanly shut down."""
log.critical("In quit signal handler.")
if supervisor is not None:
log.critical("Calling stop on supervisor loop")
supervisor.stop_from_signal()
try:
parse_command_line()
args = {}
config_file = options.config
if config_file is not None:
try:
with open(config_file, "rb") as conf:
args = json.load(conf)
except IOError, exc:
raise ArgumentError("Could not open required config file", path=config_file, error=str(exc))
except ValueError, exc:
raise ArgumentError("Could not parse JSON from config file", path=config_file, error=str(exc))
except TypeError, exc:
raise ArgumentError("You must pass the path to a json config file", path=config_file)
signal.signal(signal.SIGINT, quit_signal_handler)
supervisor = IOTileSupervisor(args)
supervisor.start()
supervisor.wait()
except Exception: # pylint: disable=W0703
log.exception("Fatal error starting supervisor")
def main():
parse_command_line()
if options.config:
parse_config_file(options.config)
app = MainApplication()
app.listen(options.port)
tornado.ioloop.IOLoop.current().start()
def open(self, username, args):
util.logger.info('({}) Websocket connected'.format(username))
# We don't do validation since we assume that the LandingHandler did
# it. TODO: ENHANCE SECURITY
is_file_request = ('file_url' in args)
try:
if is_file_request:
message = yield thread_pool.submit(
download_file_and_redirect,
username=username,
file_url=args['file_url'],
config=options.config,
)
else:
if 'branch' not in args:
args['branch'] = Config.DEFAULT_BRANCH_NAME
if 'notebook_path' not in args:
args['notebook_path'] = ''
if 'domain' not in args:
args['domain'] = Config.DEFAULT_DOMAIN
if 'account' not in args:
args['account'] = Config.DEFAULT_GITHUB_ACCOUNT
message = yield thread_pool.submit(
pull_from_remote,
username=username,
repo_name=args['repo'],
domain=args['domain'],
account=args['account'],
branch_name=args['branch'],
paths=args['path'],
config=options.config,
notebook_path=args['notebook_path'],
progress=Progress(username, self.write_message)
)
if message['type'] == "ERROR":
util.logger.exception('Sent message: {}'.format(message))
else:
util.logger.info('Sent message: {}'.format(message))
self.write_message(message)
except Exception as e:
# If something bad happens, the client should see it
message = messages.error(str(e))
util.logger.exception('Sent message: {}'.format(message))
self.write_message(message)
def main():
"""Main entry point for iotile-gateway."""
gateway = None
logging.basicConfig(format='%(levelname)-.1s-%(asctime)-15s-%(module)-10s:%(lineno)-4s %(message)s')
log = logging.getLogger(__name__)
def quit_signal_handler(signum, frame): # pylint: disable=W0613
"""Signal handler to catch ^C and cleanly shut down."""
log.critical("In quit signal handler.")
if gateway is not None:
log.critical("Calling stop on gateway loop")
gateway.stop_from_signal()
try:
parse_command_line()
config_file = options.config
if config_file is None:
log.critical("You must pass a config file using --config=<path to file>")
return 1
try:
with open(config_file, "rb") as conf:
args = json.load(conf)
except IOError, exc:
raise ArgumentError("Could not open required config file", path=config_file, error=str(exc))
except ValueError, exc:
raise ArgumentError("Could not parse JSON from config file", path=config_file, error=str(exc))
except TypeError, exc:
raise ArgumentError("You must pass the path to a json config file", path=config_file)
signal.signal(signal.SIGINT, quit_signal_handler)
gateway = IOTileGateway(args)
gateway.start()
gateway.wait()
except Exception: # pylint: disable=W0703
log.exception("Fatal error starting gateway")
