def main():
colorama.init()
import ai.backend.client.cli.config # noqa
import ai.backend.client.cli.run # noqa
import ai.backend.client.cli.proxy # noqa
import ai.backend.client.cli.admin # noqa
import ai.backend.client.cli.admin.keypairs # noqa
import ai.backend.client.cli.admin.sessions # noqa
import ai.backend.client.cli.admin.agents # noqa
import ai.backend.client.cli.admin.vfolders # noqa
import ai.backend.client.cli.vfolder # noqa
import ai.backend.client.cli.ps # noqa
if len(sys.argv) <= 1:
global_argparser.print_help()
return
mode = Path(sys.argv[0]).stem
if mode == '__main__':
pass
elif mode == 'lcc':
sys.argv.insert(1, 'c')
sys.argv.insert(1, 'run')
elif mode == 'lpython':
sys.argv.insert(1, 'python')
sys.argv.insert(1, 'run')
args = global_argparser.parse_args()
if hasattr(args, 'function'):
args.function(args)
else:
print_fail('The command is not specified or unrecognized.')
python类init()的实例源码
def _lazy_colorama_init():
"""
Lazily init colorama if necessary, not to screw up stdout is debug not
enabled.
This version of the function does nothing.
"""
pass
def __init__(self):
colorama.init()
self.white = colorama.Fore.WHITE
self.red = colorama.Fore.RED
self.blue = colorama.Fore.BLUE
self.green = colorama.Fore.GREEN
self.yellow = colorama.Fore.YELLOW # yellow
self.lightwhite = colorama.Fore.LIGHTWHITE_EX
self.lightred = colorama.Fore.LIGHTRED_EX
self.lightblue = colorama.Fore.LIGHTBLUE_EX
self.lightgreen = colorama.Fore.LIGHTGREEN_EX
self.lightyellow = colorama.Fore.LIGHTYELLOW_EX
def __init__(self, parent):
"""
NOTE: This could be implemented
as decorators...
"""
self.parent = parent
if COLORAMA:
colorama.init(autoreset = True)
def __init__(self, parent):
"""
NOTE: This could be implemented
as decorators...
"""
self.parent = parent
if COLORAMA:
colorama.init(autoreset = True)
def __init__(self, parent):
"""
NOTE: This could be implemented
as decorators...
"""
self.parent = parent
if COLORAMA:
colorama.init(autoreset = True)
def _RunInternal(parser, output_directory=None):
colorama.init()
parser.set_defaults(output_directory=output_directory)
from_wrapper_script = bool(output_directory)
args = _ParseArgs(parser, from_wrapper_script)
run_tests_helper.SetLogLevel(args.verbose_count)
args.command.ProcessArgs(args)
args.command.Run()
# Incremental install depends on the cache being cleared when uninstalling.
if args.command.name != 'uninstall':
_SaveDeviceCaches(args.command.devices, output_directory)
# TODO(agrieve): Remove =None from target_cpu on or after October 2017.
# It exists only so that stale wrapper scripts continue to work.
def __init__(self):
# `autoreset` allows us to not have to sent reset sequences for every
# string. `strip` lets us preserve color when redirecting.
colorama.init(autoreset=True, strip=False)
def _install_coreos(self, node_id, provision_config_id, network, password):
self.info('Installing CoreOS...')
# boot in to the provision configuration
self.info('Booting into Provision configuration profile...')
self.request('linode.boot', params={'LinodeID': node_id, 'ConfigID': provision_config_id})
# connect to the server via ssh
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
while True:
try:
ssh.connect(str(network['public']), username='root', password=password, allow_agent=False, look_for_keys=False)
break
except:
continue
# copy the cloud config
self.info('Pushing cloud config...')
cloud_config_template = string.Template(self._get_cloud_config())
cloud_config = cloud_config_template.safe_substitute(public_ipv4=network['public'], private_ipv4=network['private'], gateway=network['gateway'],
hostname=network['hostname'])
sftp = ssh.open_sftp()
sftp.open('cloud-config.yaml', 'w').write(cloud_config)
self.info('Installing...')
commands = [
'wget https://raw.githubusercontent.com/coreos/init/master/bin/coreos-install -O $HOME/coreos-install',
'chmod +x $HOME/coreos-install',
'$HOME/coreos-install -d /dev/sdb -C ' + self.coreos_channel + ' -V ' + self.coreos_version + ' -c $HOME/cloud-config.yaml -t /dev/shm'
]
for command in commands:
stdin, stdout, stderr = ssh.exec_command(command)
stdout.channel.recv_exit_status()
print stdout.read()
ssh.close()
def main():
colorama.init()
parser = argparse.ArgumentParser(description='Apply a "Security Group" to a Deis cluster')
parser.add_argument('--private-key', required=True, type=file, dest='private_key', help='Cluster SSH Private Key')
parser.add_argument('--private', action='store_true', dest='private', help='Only allow access to the cluster from the private network')
parser.add_argument('--discovery-url', dest='discovery_url', help='Etcd discovery url')
parser.add_argument('--hosts', nargs='+', dest='hosts', help='The IP addresses of the hosts to apply rules to')
args = parser.parse_args()
nodes = get_nodes_from_args(args)
hosts = args.hosts if args.hosts is not None else nodes
node_ips = []
for ip in nodes:
if validate_ip_address(ip):
node_ips.append(ip)
else:
log_warning('Invalid IP will not be added to security group: ' + ip)
if not len(node_ips) > 0:
raise ValueError('No valid IP addresses in security group.')
host_ips = []
for ip in hosts:
if validate_ip_address(ip):
host_ips.append(ip)
else:
log_warning('Host has invalid IP address: ' + ip)
if not len(host_ips) > 0:
raise ValueError('No valid host addresses.')
log_info('Generating iptables rules...')
rules = get_firewall_contents(node_ips, args.private)
log_success('Generated rules:')
log_debug(rules)
log_info('Applying rules...')
apply_rules_to_all(host_ips, rules, args.private_key)
log_success('Done!')
def _install_coreos(self, node_id, provision_config_id, network, password):
self.info('Installing CoreOS...')
# boot in to the provision configuration
self.info('Booting into Provision configuration profile...')
self.request('linode.boot', params={'LinodeID': node_id, 'ConfigID': provision_config_id})
# connect to the server via ssh
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
while True:
try:
ssh.connect(str(network['public']), username='root', password=password, allow_agent=False, look_for_keys=False)
break
except:
continue
# copy the cloud config
self.info('Pushing cloud config...')
cloud_config_template = string.Template(self._get_cloud_config())
cloud_config = cloud_config_template.safe_substitute(public_ipv4=network['public'], private_ipv4=network['private'], gateway=network['gateway'],
hostname=network['hostname'])
sftp = ssh.open_sftp()
sftp.open('cloud-config.yaml', 'w').write(cloud_config)
self.info('Installing...')
commands = [
'wget https://raw.githubusercontent.com/coreos/init/master/bin/coreos-install -O $HOME/coreos-install',
'chmod +x $HOME/coreos-install',
'$HOME/coreos-install -d /dev/sdb -C ' + self.coreos_channel + ' -V ' + self.coreos_version + ' -c $HOME/cloud-config.yaml -t /dev/shm'
]
for command in commands:
stdin, stdout, stderr = ssh.exec_command(command)
stdout.channel.recv_exit_status()
print stdout.read()
ssh.close()
def init_color(no_color: bool) -> None:
global success_color, info_hl, info_color, warning_hl, warning_color, error_hl, error_color, highlight_color, \
color_reset
if no_color:
success_color = info_hl = info_color = warning_hl = warning_color = error_hl = error_color = highlight_color =\
color_reset = ""
else:
colorama.init()
def _colorama(*args, **kwargs):
"""Temporarily enable colorama."""
colorama.init(*args, **kwargs)
try:
yield
finally:
colorama.deinit()
def before_process_document(self, document):
"""Initialize ANSI colors if color is True or 'auto'.
If color=='auto', detect if sys.stdout is interactive
(terminal, not redirected to a file).
"""
super().before_process_document(document)
if self.color == 'auto':
self.color = sys.stdout.isatty()
if self.color:
colorama.init()
if self.print_doc_meta:
for key, value in sorted(document.meta.items()):
print('%s = %s' % (key, value))
def _lazy_colorama_init():
"""
Lazily init colorama if necessary, not to screw up stdout is debug not
enabled.
This version of the function does nothing.
"""
pass
def showHelp():
if sys.platform=="win32":
try:
import colorama
colorama.init()
except:
pass
prefix=styles["prefix"](rsjoin(" ", ("python -m", endMainRx.sub("", __spec__.name))))
print(
rsjoin("\n", (
styles["cli"](rsjoin(" ", (prefix, cliArg))) + styles["help"](" - "+par[1])
for cliArg, par in singleArgLUT.items()
))
)
def main():
# colorama works cross-platform to color text output in CLI
colorama.init()
parser = argparse.ArgumentParser(description='''
aptos is a tool for validating client-submitted data using the
JSON Schema vocabulary and converts JSON Schema documents into
different data-interchange formats.
''', usage='%(prog)s [arguments] SCHEMA', epilog='''
More information on JSON Schema: http://json-schema.org/''')
subparsers = parser.add_subparsers(title='Arguments')
validation = subparsers.add_parser(
'validate', help='Validate a JSON instance')
validation.add_argument(
'-instance', type=str, default=json.dumps({}),
help='JSON document being validated')
validation.set_defaults(func=validate)
conversion = subparsers.add_parser(
'convert', help='''
Convert a JSON Schema into a different data-interchange format''')
conversion.add_argument(
'-format', type=str, choices=['avro'], help='data-interchange format')
conversion.set_defaults(func=convert)
parser.add_argument(
'schema', type=str, help='JSON document containing the description')
arguments = parser.parse_args()
arguments.func(arguments)
def init():
pick_da_useragent()
atexit.register(save_data)
load_data()
def main():
init()
run()
def initialize_git_repo(current):
os.chdir(current)
result = subprocess.Popen(['git', 'init'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
res1 = result.communicate()[0]
subprocess.Popen(['git', 'add', '.'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE).wait()
subprocess.Popen(['git', 'commit', '-m', 'First commit'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE).wait()
subprocess.Popen(['git', 'tag', '-a', '0.1', '-m', 'First alpha version'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE).wait()
print(Fore.MAGENTA + " [+] " + str(res1, 'utf-8') + Fore.RESET)
