Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from typing import Any, Callable, Dict, List, NamedTuple, Tuple
from configargparse import Action
CArg = Tuple[List[str], Dict[str, Any]]
class Command(NamedTuple):
name: str
description: str
func: Callable
arguments: List[CArg]
class SetStoragePathAction(Action):
def __call__(self, parser, namespace, values, option_string=None):
setattr(namespace, self.dest, values)
default_storage_path = getattr(namespace, 'storage_path')
try:
default_storage_path = default_storage_path % values
except TypeError:
# storage_path already defined
pass
setattr(namespace, 'storage_path', default_storage_path)
COMMON_ARGS = [
(['--config'], {
'help': 'path to configuration file',
'metavar': 'FILE',
import configargparse
class Singleton(type):
instance = None
def __call__(cls, *args, **kwargs):
if not cls.instance:
cls.instance = super().__call__(*args, **kwargs)
return cls.instance
def reload(cls):
cls.instance = None
class VAction(configargparse.Action):
def __call__(self, parser, args, values, option_string=None):
if values is None:
values = "1"
try:
values = int(values)
except ValueError:
values = values.count("v") + 1
setattr(args, self.dest, values)
def add_verbose_argument(argparser, **kwargs):
return argparser.add_argument("-v", "--verbose", nargs="?", action=VAction, default=0, **kwargs)
def get_group_arg_names(group):
return [a.dest for a in group._group_actions]
"NOTSET": logging.NOTSET
}
CHOICES = list(LEVELS.keys()) + [str(_) for _ in LEVELS.values()]
def __init__(self, option_strings, dest, help=None, **kwargs):
super(StoreLoggingLevelAction, self).__init__(
option_strings, dest, help=help, **kwargs)
def __call__(self, parser, namespace, value, option_string=None):
# Get the key 'value' in the dict, or just use 'value'
level = StoreLoggingLevelAction.LEVELS.get(value, value)
setattr(namespace, self.dest, level)
class DeprecateAction(configargparse.Action):
""" Deprecate action """
def __init__(self, option_strings, dest, help=None, **kwargs):
super(DeprecateAction, self).__init__(option_strings, dest, nargs=0,
help=help, **kwargs)
def __call__(self, parser, namespace, values, flag_name):
help = self.help if self.help is not None else ""
msg = "Flag '%s' is deprecated. %s" % (flag_name, help)
raise configargparse.ArgumentTypeError(msg)
"NOTSET": logging.NOTSET
}
CHOICES = list(LEVELS.keys()) + [str(_) for _ in LEVELS.values()]
def __init__(self, option_strings, dest, help=None, **kwargs):
super(StoreLoggingLevelAction, self).__init__(
option_strings, dest, help=help, **kwargs)
def __call__(self, parser, namespace, value, option_string=None):
# Get the key 'value' in the dict, or just use 'value'
level = StoreLoggingLevelAction.LEVELS.get(value, value)
setattr(namespace, self.dest, level)
class DeprecateAction(configargparse.Action):
""" Deprecate action """
def __init__(self, option_strings, dest, help=None, **kwargs):
super(DeprecateAction, self).__init__(option_strings, dest, nargs=0,
help=help, **kwargs)
def __call__(self, parser, namespace, values, flag_name):
help = self.help if self.help is not None else ""
msg = "Flag '%s' is deprecated. %s" % (flag_name, help)
raise configargparse.ArgumentTypeError(msg)
if x < 0:
raise configargparse.ArgumentTypeError("A negative number was supplied")
return x
def commitpoint_type(x):
x = int(x)
if x < const.MIN_COMMIT_POINT:
raise configargparse.ArgumentTypeError(
"Minimum allowed commitpoint is: {}"
.format(const.MIN_COMMIT_POINT))
return x
class AppendDateTimeDirAction(configargparse.Action):
def __init__(self, option_strings, dest, nargs=None, **kwargs):
super(AppendDateTimeDirAction, self).__init__(option_strings,
dest, **kwargs)
def __call__(self, parser, namespace, values, option_string=None):
values = filesystem_utils.append_datetime_dir(values)
setattr(namespace, self.dest, values)
def _is_initsync(mode):
return (mode == const.INITSYNC or
mode == const.INITSYNCEXTRACT or
mode == const.INITSYNCAPPLY)
def _is_extract(mode):
group.add('--window', '-window', default='hamming',
help='Window type for spectrogram generation')
# Option most relevant to image input
group.add('--image_channel_size', '-image_channel_size',
type=int, default=3, choices=[3, 1],
help="Using grayscale image can training "
"model faster and smaller")
# Copyright 2016 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
class StoreLoggingLevelAction(configargparse.Action):
""" Convert string to logging level """
import logging
LEVELS = {
"CRITICAL": logging.CRITICAL,
"ERROR": logging.ERROR,
"WARNING": logging.WARNING,
"INFO": logging.INFO,
"DEBUG": logging.DEBUG,
"NOTSET": logging.NOTSET
}
CHOICES = list(LEVELS.keys()) + [str(_) for _ in LEVELS.values()]
def __init__(self, option_strings, dest, help=None, **kwargs):
super(StoreLoggingLevelAction, self).__init__(
option_strings, dest, help=help, **kwargs)
group.add('--window', '-window', default='hamming',
help='Window type for spectrogram generation')
# Option most relevant to image input
group.add('--image_channel_size', '-image_channel_size',
type=int, default=3, choices=[3, 1],
help="Using grayscale image can training "
"model faster and smaller")
# Copyright 2016 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
class StoreLoggingLevelAction(configargparse.Action):
""" Convert string to logging level """
import logging
LEVELS = {
"CRITICAL": logging.CRITICAL,
"ERROR": logging.ERROR,
"WARNING": logging.WARNING,
"INFO": logging.INFO,
"DEBUG": logging.DEBUG,
"NOTSET": logging.NOTSET
}
CHOICES = list(LEVELS.keys()) + [str(_) for _ in LEVELS.values()]
def __init__(self, option_strings, dest, help=None, **kwargs):
super(StoreLoggingLevelAction, self).__init__(
option_strings, dest, help=help, **kwargs)
""" SRU Implementation """
# flake8: noqa
import subprocess
import platform
import os
import re
import configargparse
import torch
import torch.nn as nn
from torch.autograd import Function
from collections import namedtuple
# For command-line option parsing
class CheckSRU(configargparse.Action):
def __init__(self, option_strings, dest, **kwargs):
super(CheckSRU, self).__init__(option_strings, dest, **kwargs)
def __call__(self, parser, namespace, values, option_string=None):
if values == 'SRU':
check_sru_requirement(abort=True)
# Check pass, set the args.
setattr(namespace, self.dest, values)
# This SRU version implements its own cuda-level optimization,
# so it requires that:
# 1. `cupy` and `pynvrtc` python package installed.
# 2. pytorch is built with cuda support.
# 3. library path set: export LD_LIBRARY_PATH=.
def check_sru_requirement(abort=False):
def make_action(additional_arg):
class customAction(configargparse.Action):
def __call__(self, parser, args, values, option_string=None):
json_obj = json.loads(values)['fields']
setattr(args, self.dest, json_obj)
return customAction
def make_action(additional_arg):
class customAction(configargparse.Action):
def __call__(self, parser, args, values, option_string=None):
print(additional_arg)
json_obj = json.loads(values)['fields']
setattr(args, self.dest, json_obj)
return customAction