Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
directory = wandb.wandb_dir()
if not os.path.exists(directory):
raise ClickException('No wandb directory found at %s' % directory)
paths = glob.glob(directory+"/*run*")
dates = [datetime.datetime.strptime(p.split("-")[1],'%Y%m%d_%H%M%S') for p in paths]
since = datetime.datetime.utcnow() - datetime.timedelta(hours=keep)
bad_paths = [paths[i] for i, d, in enumerate(dates) if d < since]
if len(bad_paths) > 0:
click.echo("Found {} runs, {} are older than {} hours".format(len(paths), len(bad_paths), keep))
click.confirm(click.style(
"Are you sure you want to remove %i runs?" % len(bad_paths), bold=True), abort=True)
for path in bad_paths:
shutil.rmtree(path)
click.echo(click.style("Success!", fg="green"))
else:
click.echo(click.style("No runs older than %i hours found" % keep, fg="red"))
def fail(self, msg, *args, **kwargs):
self.clear_line()
msg = msg.format(*args, **kwargs)
click.echo(" {} {}".format(click.style("✗", fg="red"), msg))
self._done = True
# Click prompts don't like None, so we have to use an empty string instead
f_default_print = f_default
if f_default is None:
f_default = ''
f_default_print = 'None'
# Overwrite the default prompt for boolean
if isinstance(f_default, bool):
f_default_print = 'Y/n' if f_default else 'y/N'
# Prompt for a response
f_user = click.prompt(
"\n{}\n {} {}".format(
self.nxf_flag_help[flag],
click.style(flag, fg='blue'),
click.style('[{}]'.format(str(f_default_print)), fg='green')
),
default = f_default,
show_default = False
)
# Only save if we've changed the default
if f_user != f_default:
# Convert string bools to real bools
try:
f_user = f_user.strip('"').strip("'")
if f_user.lower() == 'true': f_user = True
if f_user.lower() == 'false': f_user = False
except AttributeError:
pass
self.nxf_flags[flag] = f_user
def show_version():
"""Show version and diagnostic details"""
click.echo(shell("uname -a")["stdout"].decode().rstrip("\n"))
click.echo(
click.style(" * arkOS server version: ", fg="yellow") +
config.get("enviro", "version", "Unknown")
)
click.echo(
click.style(" * Arch / Board: ", fg="yellow") +
config.get("enviro", "arch", "Unknown") + " / " +
config.get("enviro", "board", "Unknown")
)
indices = set(cfg["SEARCH_ELASTIC_COLLECTION_INDEX_MAPPING"].values())
indices.add(cfg['SEARCH_ELASTIC_DEFAULT_INDEX'])
for index in indices:
possible_indices = [index]
if holdingpen:
possible_indices.append(cfg['WORKFLOWS_HOLDING_PEN_ES_PREFIX'] + index)
if index_name is not None and index_name not in possible_indices:
continue
click.echo(click.style('Rebuilding {0}... '.format(index), fg='yellow'), nl=False)
mapping = {}
mapping_filename = index + ".json"
if mapping_filename in mappings:
mapping = json.load(open(mappings[mapping_filename], "r"))
recreate_index(index, mapping, rebuild=rebuild)
click.echo(click.style('OK', fg='green'))
if holdingpen:
# Create Holding Pen index
if mapping:
mapping['mappings']['record']['properties'].update(
cfg['WORKFLOWS_HOLDING_PEN_ES_PROPERTIES']
)
name = cfg['WORKFLOWS_HOLDING_PEN_ES_PREFIX'] + index
click.echo(click.style(
'Rebuilding {0}... '.format(name), fg='yellow'
), nl=False)
recreate_index(name, mapping, rebuild=rebuild)
click.echo(click.style('OK', fg='green'))
def execute_command(step_name, command, cwd='.', report_empty_err=True, debug=False, stream_output_in_debug=True, force_std_out=False):
assert isinstance(command, basestring)
pipe_output = (None if (debug and stream_output_in_debug) else subprocess.PIPE)
if force_std_out:
pipe_output = None
click.echo(click.style("---> %s " % step_name, fg='blue', bold=True), nl=debug)
if debug:
click.echo(click.style("cwd: %s" % cwd, bg='blue', bold=False))
click.echo(click.style("$ %s" % command, bg='blue', bold=False))
with spinner.spinner(disable=debug, color='blue', bold=True):
pipe = subprocess.Popen(
command,
cwd=cwd,
shell=True,
executable='/bin/bash',
stdout=pipe_output,
stderr=pipe_output)
out, err = pipe.communicate()
if not debug:
click.echo("\n", nl=False)
if debug and not stream_output_in_debug:
click.echo(out)
def msgout(msg, error=False, warning=False, quiet=False):
"""Output messages to stdout via click.echo if quiet=False"""
if not quiet:
if error:
click.echo(click.style(click.style(msg, fg='red', bold=True)))
elif warning:
click.echo(click.style(click.style(msg, fg='yellow', bold=True)))
else:
click.echo(msg)
if check_token():
click.echo('Updating changelog')
try:
log = generate_changelog(current_version, new_version)
post_changelog(
owner,
name,
new_version,
markdown_changelog(new_version, log, header=False)
)
except GitError:
click.echo(click.style('Posting changelog failed.', 'red'), err=True)
else:
click.echo(
click.style('Missing token: cannot post changelog', 'red'), err=True)
click.echo(click.style('New release published', 'green'))
else:
click.echo('Version failed, no release will be published.', err=True)
{"id": i, "hwid": s.id, "type": s.type_name, "temperature": t, "unit": unit}
for i, s, t in zip(count(start=1), sensors, temperatures)
]
click.echo(json.dumps(data, indent=4, sort_keys=True))
else:
click.echo(
"Got temperatures of {0} sensors:".format(
click.style(str(len(sensors)), bold=True)
)
)
for i, sensor, temperature in zip(count(start=1), sensors, temperatures):
click.echo(
" Sensor {0} ({1}) measured temperature: {2} {3}".format(
click.style(str(i), bold=True),
click.style(sensor.id, bold=True),
click.style(str(round(temperature, 2)), bold=True),
click.style(unit, bold=True),
)
:type index: int
:param index: The index for the given item, used with the
gh view [index] commend.
:type url: str
:param url: The url to view
"""
contents = self.generate_url_contents(url)
header = click.style('Viewing ' + url + '\n\n',
fg=self.config.clr_message)
contents = header + contents
contents += click.style(('\nView this article in a browser with'
' the -b/--browser flag.\n'),
fg=self.config.clr_message)
contents += click.style(('\nPress q to quit viewing this '
'article.\n'),
fg=self.config.clr_message)
if contents == '{"error":"Not Found"}\n':
click.secho('Invalid user/repo combination.')
return
color = None
if platform.system() == 'Windows':
color = True
click.echo_via_pager(contents, color)