X7ROOT File Manager
Current Path:
/opt/cloudlinux/venv/lib/python3.11/site-packages/guppy/heapy
opt
/
cloudlinux
/
venv
/
lib
/
python3.11
/
site-packages
/
guppy
/
heapy
/
??
..
??
Classifiers.py
(45.16 KB)
??
Console.py
(843 B)
??
Doc.py
(7.81 KB)
??
ImpSet.py
(870 B)
??
Monitor.py
(14.5 KB)
??
OutputHandling.py
(11.52 KB)
??
Part.py
(23.25 KB)
??
Path.py
(14.19 KB)
??
Prof.py
(90.38 KB)
??
RM.py
(119 B)
??
RefPat.py
(18.18 KB)
??
Remote.py
(15.31 KB)
??
RemoteConstants.py
(209 B)
??
Spec.py
(48.59 KB)
??
Target.py
(205 B)
??
UniSet.py
(66.5 KB)
??
Use.py
(23.05 KB)
??
View.py
(19.05 KB)
??
__init__.py
(209 B)
??
__pycache__
??
heapyc.cpython-311-x86_64-linux-gnu.so
(154.55 KB)
??
pbhelp.py
(17.4 KB)
??
test
Editing: OutputHandling.py
import types import weakref # To restore the old-style class behavior that __getattr__ also affects special # methods. class _AttrProxy: _oh_proxied_classes = weakref.WeakSet({type, object}) # Don'r rely on _oh_add_proxy_attr to prime us, some clients use # setup_printing to setup __repr__ function. def __repr__(self): return self.__getattr__('__repr__')() @classmethod def _oh_add_proxy_attr(cls, attr): if not attr.startswith('__') or not attr.endswith('__'): return for scls in cls.__mro__: if scls is object: break if attr in scls.__dict__: return if attr in ('__new__', '__init__', '__getattr__', '__getattribute__', '__setattr__', '__delattr__', '__str__', ): return def closure(attr): def generated_function(self, *args, **kwds): func = self.__getattr__(attr) return func(*args, **kwds) return generated_function setattr(cls, attr, closure(attr)) @classmethod def _oh_add_proxy_class(cls, base): if base in cls._oh_proxied_classes: return for scls in base.__mro__: cls._oh_proxied_classes.add(scls) for attr, val in scls.__dict__.items(): if not isinstance(val, types.FunctionType): continue cls._oh_add_proxy_attr(attr) class OutputHandler: def __init__(self, mod, output_file): self.mod = mod self.output_file = output_file class OutputBuffer: def __init__(self, mod, opts=None): self.mod = mod self.strio = mod._root.io.StringIO() if opts is None: opts = {} self.opts = opts self.lines = [''] self.line_no = 0 def getopt(self, opt): return self.opts.get(opt) def getvalue(self): return '\n'.join(self.lines) def new_line(self): self.line_no += 1 self.lines.append('') def write(self, s): lines = s.split('\n') for line in lines[:-1]: self.write_seg(line) self.new_line() self.write_seg(lines[-1]) def write_seg(self, s): self.lines[self.line_no] += s class AllPrinter(_AttrProxy): _oh_next_lineno = None def __init__(self, printer): self._oh_printer = printer self._hiding_tag_ = printer._hiding_tag_ def __getattr__(self, attr): return self._oh_printer.getattr(self, attr) def __repr__(self): return self._oh_printer.get_str(self, True) def _oh_get_next_lineno(self): return 0 def _oh_get_start_lineno(self): return 0 def _oh_get_max_lines(self, max_lines): return None class MorePrinter(_AttrProxy): _oh_next_lineno = None def __init__(self, printer, previous): self._oh_printer = printer self._oh_previous = previous self._hiding_tag_ = printer._hiding_tag_ def __getattr__(self, attr): return self._oh_printer.getattr(self, attr) def _oh_get_next_lineno(self): next_lineno = self._oh_next_lineno if next_lineno is None: repr(self) next_lineno = self._oh_next_lineno return next_lineno def _oh_get_start_lineno(self): return self._oh_previous._oh_get_next_lineno() def _oh_get_max_lines(self, max_lines): return max_lines class Printer: def __init__(self, mod, client, handler, get_line_iter=None, max_more_lines=None, get_num_lines=None, get_label=None, get_row_header=None, get_more_msg=None, get_more_state_msg=None, get_empty_msg=None, stop_only_when_told=None ): try: handler = handler.printer.handler except AttributeError: pass if get_line_iter is None: get_line_iter = handler._oh_get_line_iter if max_more_lines is None: max_more_lines = mod.max_more_lines self.mod = mod self._hiding_tag_ = mod._hiding_tag_ self.client = client self.handler = handler self.get_line_iter = get_line_iter self.max_more_lines = max_more_lines if get_num_lines is None: get_num_lines = getattr( handler, '_oh_get_num_lines', None) if get_num_lines is not None: self.get_num_lines = get_num_lines if get_label is None: get_label = getattr( handler, '_oh_get_label', None) if get_label is not None: self.get_label = get_label if get_row_header is None: get_row_header = getattr( handler, '_oh_get_row_header', None) if get_row_header is not None: self.get_row_header = get_row_header if get_more_msg is None: get_more_msg = getattr( handler, '_oh_get_more_msg', None) if get_more_msg is not None: self.get_more_msg = get_more_msg if get_more_state_msg is None: get_more_state_msg = getattr( handler, '_oh_get_more_state_msg', None) if get_more_state_msg is not None: self.get_more_state_msg = get_more_state_msg if get_empty_msg is None: get_empty_msg = getattr( handler, '_oh_get_empty_msg', None) if get_empty_msg is not None: self.get_empty_msg = get_empty_msg self.stop_only_when_told = stop_only_when_told self.reset() def getattr(self, mp, attr): try: g = getattr(self, '_get_'+attr) except AttributeError: return getattr(self.client, attr) else: return g(mp) def line_at(self, idx): while idx >= len(self.lines_seen): try: li = next(self.line_iter) except StopIteration: raise IndexError else: if isinstance(li, tuple): cmd, line = li if cmd == 'STOP_AFTER': self.stop_linenos[len(self.lines_seen)] = 1 else: line = li self.lines_seen.append(line) return self.lines_seen[idx] def lines_from(self, idx=0): line_iter = self.line_iter if line_iter is None: line_iter = self.line_iter = self.get_line_iter() while 1: try: yield self.line_at(idx) except IndexError: return idx += 1 def _get_more(self, mp): return MorePrinter(self, mp) def _get_all(self, mp): return AllPrinter(self) def _oh_get_next_lineno(self): next_lineno = getattr(self, '_oh_next_lineno', None) if next_lineno is None: self.get_str_of_top() next_lineno = self._oh_next_lineno return next_lineno def _get_prev(self, mp): return mp._oh_previous def _oh_get_start_lineno(self): return 0 def _oh_get_max_lines(self, max_lines): return max_lines def _get_top(self, mp): return self.client def _get___repr__(self, mp): return lambda: self.get_str(mp, False) def get_label(self): return None def get_row_header(self): return None def get_str_of_top(self): return self.get_str(self, True) def get_more_state_msg(self, start_lineno, end_lineno): num_lines = self.get_num_lines() if num_lines is None: of_num_lines = '' else: of_num_lines = ' of %d' % num_lines return "Lines %d..%d%s. " % (start_lineno, end_lineno, of_num_lines) def get_more_msg(self, start_lineno, end_lineno): state_msg = self.get_more_state_msg(start_lineno, end_lineno) return "<%sType e.g. '_.more' for more.>" % (state_msg) def get_empty_msg(self): return None def get_num_lines(self): return None def get_str(self, printer, is_top): def f(): _hiding_tag_ = printer._hiding_tag_ start_lineno = printer._oh_get_start_lineno() m_max_lines = printer._oh_get_max_lines(self.max_more_lines) ob = self.mod.output_buffer() it = self.lines_from(start_lineno) if is_top: msg = self.get_label() if msg is not None: print(msg, file=ob) printer._oh_next_lineno = start_lineno try: nxt = next(it) except StopIteration: msg = self.get_empty_msg() if msg is not None: print(msg, file=ob) printer._oh_next_lineno = start_lineno else: msg = self.get_row_header() if msg is not None: print(msg, file=ob) it = self.mod.itertools.chain((nxt,), it) numlines = 0 lineno = start_lineno for line in it: if (m_max_lines is not None and numlines >= m_max_lines and ((not self.stop_only_when_told) or self.stop_linenos.get(lineno-1))): try: self.line_at(lineno+1) except IndexError: print(line, file=ob) lineno += 1 break else: print(self.get_more_msg(start_lineno, lineno-1), file=ob) break numlines += 1 print(line, file=ob) lineno += 1 printer._oh_next_lineno = lineno return ob.getvalue().rstrip() return self.mod.View.enter(lambda: f()) def reset(self): self.lines_seen = [] self.stop_linenos = {} self.line_iter = None class _GLUECLAMP_: _chgable_ = 'output_file', 'max_more_lines', _preload_ = ('_hiding_tag_',) _imports_ = ( '_parent:View', '_root:itertools', ) max_more_lines = 10 def _get__hiding_tag_(self): return self._parent.View._hiding_tag_ def _get_output_file(self): return self._root.sys.stdout def more_printer(self, client, handler=None, **kwds): if handler is None: handler = client printer = Printer(self, client, handler, **kwds) _AttrProxy._oh_add_proxy_class(client.__class__) return MorePrinter(printer, printer) def output_buffer(self): return OutputBuffer(self) def output_handler(self, output_file=None): if output_file is None: output_file = self.output_file return OutputHandler(self, output_file) def setup_printing(self, client, handler=None, **kwds): if handler is None: handler = client more = self.more_printer(client, handler, **kwds) printer = more._oh_printer client.more = more client.all = more.all client.printer = printer def reprfunc(self): return self.printer.get_str_of_top() # Don't recreate, messes with relheap, and this is not # under _hiding_tag_ cls = client.__class__ if '__repr__' not in cls.__dict__: cls.__repr__ = reprfunc def _get_stdout(self): return self._root.sys.stdout
Upload File
Create Folder