Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import pysnooper.pycompat
def get_function_arguments(function, exclude=()):
try:
getfullargspec = inspect.getfullargspec
except AttributeError:
result = inspect.getargspec(function).args
else:
result = getfullargspec(function).args
for exclude_item in exclude:
result.remove(exclude_item)
return result
class _BaseEntry(pysnooper.pycompat.ABC):
def __init__(self, prefix=''):
self.prefix = prefix
@abc.abstractmethod
def check(self, s):
pass
def __repr__(self):
init_arguments = get_function_arguments(self.__init__,
exclude=('self',))
attributes = {
key: repr(getattr(self, key)) for key in init_arguments
if getattr(self, key) is not None
}
return '%s(%s)' % (
type(self).__name__,
def __snoop(*args, **kwargs):
if os_name != 'nt' and do_not_effect_on_linux: # 不要修改任何代码,自动就会不在linux上debug,一般linux是部署机器。
return func(*args, **kwargs)
else:
if line_can_click:
return _snoop_can_click(output, variables, depth, prefix)(func)(*args, **kwargs)
else:
return pysnooper.snoop(output, variables, depth, prefix)(func)(*args, **kwargs)
@pysnooper.snoop()
def schedule_lpt(jobs, num_bins):
"""This function implements the Longest Processing Time algorithm to get
a good division of labor for the multiprocessor scheduling problem.
Args:
jobs (dict or list): A dictionary with string 'key' specifying job
name and float 'value' specifying job weight (how long job should
run compared to other jobs). A list of tuples may also be passed
if it is equivalent to dict.items()
num_bins (int): Number of groups to split jobs into.
Returns:
List of lists: Each group (list) in list is a group of jobs that should
be run together on a single worker or instance and consists of
tuples as provided in jobs input.
def inner(cls: Type[T]) -> Type[T]:
methods = inspect.getmembers(cls, predicate=inspect.isfunction)
for name, method in methods:
snooper_method = pysnooper.snoop(
output,
watch,
watch_explode,
depth,
prefix,
overwrite,
thread_info,
custom_repr,
max_variable_length,
)(method)
setattr(cls, name, snooper_method)
return cls
def compute_repr(self, x):
orig_repr_func = pysnooper.utils.get_repr_function(x, self.orig_custom_repr)
if torch.is_tensor(x):
return self.tensor_format(x)
elif isinstance(x, numpy.ndarray):
return self.numpy_format(x)
elif self.is_return_types(x):
return self.return_types_repr(x)
elif orig_repr_func is not repr:
return orig_repr_func(x)
elif isinstance(x, (list, tuple)):
content = ''
for i in x:
if content != '':
content += ', '
content += self.compute_repr(i)
if isinstance(x, tuple) and len(x) == 1:
content += ','
def truncate():
with open(six.text_type(output), 'w') as output_file:
pass
elif callable(output):
def write(s):
with open(six.text_type(output), 'a') as output_file:
output_file.write(s)
def truncate():
def patch_snooper_max_variable_length(max_lenth=100000):
"""
提供一个猴子补丁,三方包默认是变量最大显示100个字母,对于我这种经常debug对接方json的,要加到10万才能显示一个josn。
最好是放在name = main中去执行此补丁,否则由于模块是单例的永远只导入一次,会改变其他地方的运行表现。
:param max_lenth:
:return:
"""
from pysnooper import tracer
tracer.MAX_VARIABLE_LENGTH = max_lenth
return prefix + properties_str + suffix
default_format = TensorFormat()
class NumpyFormat:
def __call__(self, x):
return 'ndarray<{}, {}>'.format(x.shape, x.dtype.name)
default_numpy_format = NumpyFormat()
class TorchSnooper(pysnooper.tracer.Tracer):
def __init__(self, *args, tensor_format=default_format, numpy_format=default_numpy_format, **kwargs):
self.orig_custom_repr = kwargs['custom_repr'] if 'custom_repr' in kwargs else ()
custom_repr = (lambda x: True, self.compute_repr)
kwargs['custom_repr'] = (custom_repr,)
super(TorchSnooper, self).__init__(*args, **kwargs)
self.tensor_format = tensor_format
self.numpy_format = numpy_format
@staticmethod
def is_return_types(x):
return type(x).__module__ == 'torch.return_types'
def return_types_repr(self, x):
if type(x).__name__ in {'max', 'min', 'median', 'mode', 'sort', 'topk', 'kthvalue'}:
return type(x).__name__ + '(values=' + self.tensor_format(x.values) + ', indices=' + self.tensor_format(x.indices) + ')'
import pysnooper
import yaml
from .format import PythonVersion
from .model.pydantic import (
BaseModel,
CustomRootType,
DataModelField,
dump_resolve_reference_action,
)
from .parser.base import Parser
from .version import version as __version__
T = TypeVar('T')
pysnooper.tracer.DISABLED = True
DEFAULT_BASE_CLASS: str = 'pydantic.BaseModel'
def enable_debug_message() -> None: # pragma: no cover
pysnooper.tracer.DISABLED = False
def snooper_to_methods( # type: ignore
output=None,
watch=(),
watch_explode=(),
depth=1,
prefix='',
overwrite=False,
thread_info=False,