python类getsizeof()的实例源码

def allocDelayProg(size, duration):
    """Python program as str that allocates object of the specified size
    in Gigabytes and then waits for the specified time

    size  - allocation size, bytes; None to skip allocation
    duration  - execution duration, sec; None to skip the sleep

    return  - Python program as str
    """
    return """from __future__ import print_function, division  # Required for stderr output, must be the first import
import sys
import time
import array

if {size} is not None:
    a = array.array('b', [0])
    asize = sys.getsizeof(a)
    # Note: allocate at least empty size, i.e. empty list
    buffer = a * int(max({size} - asize + 1, 0))
    #if _DEBUG_TRACE:
    # print(''.join(('  allocDelayProg(), allocated ', str(sys.getsizeof(buffer))
    #   , ' bytes for ', str({duration}),' sec')), file=sys.stderr)
if {duration} is not None:
    time.sleep({duration})
""".format(size=size, duration=duration)

def set_cache_entry(url, content, metadata):
    if sys.getsizeof(content) > MAX_PAYLOAD_SIZE_BYTES:
        logger.info(u"Not caching {} because payload is too large: {}".format(
            url, sys.getsizeof(content)))
        return
    hash_key = _build_hash_key(url)
    # logger.info(u"*** {} {}".format(url, hash_key))

    k = boto.s3.key.Key(requests_cache_bucket)
    k.key = hash_key
    k.set_contents_from_string(content)

    if metadata:
        k.set_remote_metadata(metadata, {}, True)

    # remote_key = requests_cache_bucket.get_key(hash_key)
    # logger.info(u"metadata: {}".format(remote_key.metadata))

    return

def _get_size(item, seen):
    known_types = {dict: lambda d: chain.from_iterable(d.items())}
    default_size = getsizeof(0)

    def size_walk(item):
        if id(item) in seen:
            return 0
        seen.add(id(item))
        s = getsizeof(item, default_size)
        for _type, fun in known_types.iteritems():
            if isinstance(item, _type):
                s += sum(map(size_walk, fun(item)))
                break
        return s

    return size_walk(item)

def memory_dump():
    import gc
    x = 0
    for obj in gc.get_objects():
        i = id(obj)
        size = sys.getsizeof(obj, 0)
        # referrers = [id(o) for o in gc.get_referrers(obj)]
        try:
            cls = str(obj.__class__)
        except:
            cls = "<no class>"
        if size > 1024 * 50:
            referents = set([id(o) for o in gc.get_referents(obj)])
            x += 1
            print(x, {'id': i,
                      'class': cls,
                      'size': size,
                      "ref": len(referents)})
            #if len(referents) < 2000:
                #print(obj)

def __init__(self, read_cache=False):
        self._cache = None
        self._backup = None
        if read_cache:
            path = os.path.join(options.cfg.cache_path, "coq_cache.db")
            if os.path.exists(path):
                try:
                    self._cache = pickle.load(open(path, "rb"))
                    s = "Using query cache (current size: {}, max size: {})."
                    if options.cfg.verbose:
                        s = s.format(self._cache.currsize, self._cache.maxsize)
                        logger.info(s)
                        print(s)
                except (IOError, ValueError, EOFError):
                    S = "Cannot read query cache, creating a new one (size: {})."
                    S = S.format(options.cfg.query_cache_size)
                    logger.warning(S)

        if self._cache is None:
            self._cache = cachetools.LFUCache(
                maxsize=options.cfg.query_cache_size,
                getsizeof=sys.getsizeof)

def add(self, key, x):
        # if enabled, cache data frame
        if options.cfg.use_cache:
            size = sys.getsizeof(x)

            # do not attempt to cache overly large data frames:
            if size > self._cache.maxsize:
                S = ("Query result too large for the query cache ({} MBytes"
                     "missing).")
                S = S.format((size - self._cache.maxsize) // (1024*1024))
                logger.warning(S)

            # remove items from cache if necessary:
            while self._cache.currsize + size > self._cache.maxsize:
                self._cache.popitem()

            # add data frame to cache
            self._cache[key] = x

            self._backup = None

def debug_primes():
    '''
    Generate an infinite stream of primes and log out the size of
    the sieve as it increases
    '''
    max_sieve_size = 0
    sieve_size = 0

    sieve = defaultdict(list)
    k = 2

    while True:
        k_factors = sieve.get(k)
        if k_factors:
            del sieve[k]
            for f in k_factors:
                sieve[f+k].append(f)
            sieve_size = getsizeof(sieve)
            if sieve_size > max_sieve_size:
                max_sieve_size = sieve_size
                print('{}\tincrease: {} bytes'.format(k, max_sieve_size))
        else:
            yield k
            sieve[k ** 2] = [k]
        k += 1

def test_dump_tx_size(self):
        # GIVEN
        tx = Transaction()
        tx.put_data("TEST")
        tx.transaction_type = TransactionStatus.confirmed

        tx_only_data = TransactionDataOnly()
        tx_only_data.data = "TEST"

        # WHEN
        dump_a = pickle.dumps(tx_only_data)
        dump_b = pickle.dumps(tx)

        # THEN
        logging.debug("size of tx_only_data: " + str(sys.getsizeof(dump_a)))
        logging.debug("size of tx: " + str(sys.getsizeof(dump_b)))

        self.assertLessEqual(sys.getsizeof(dump_a), sys.getsizeof(dump_b) * 1.5)

def test_errors(self):
        class BadSizeof(object):
            def __sizeof__(self):
                raise ValueError
        self.assertRaises(ValueError, sys.getsizeof, BadSizeof())

        class InvalidSizeof(object):
            def __sizeof__(self):
                return None
        self.assertRaises(TypeError, sys.getsizeof, InvalidSizeof())
        sentinel = ["sentinel"]
        self.assertIs(sys.getsizeof(InvalidSizeof(), sentinel), sentinel)

        class OverflowSizeof(long):
            def __sizeof__(self):
                return int(self)
        self.assertEqual(sys.getsizeof(OverflowSizeof(sys.maxsize)),
                         sys.maxsize + self.gc_headsize)
        with self.assertRaises(OverflowError):
            sys.getsizeof(OverflowSizeof(sys.maxsize + 1))
        with self.assertRaises(ValueError):
            sys.getsizeof(OverflowSizeof(-1))
        with self.assertRaises((ValueError, OverflowError)):
            sys.getsizeof(OverflowSizeof(-sys.maxsize - 1))

def test_errors(self):
        class BadSizeof(object):
            def __sizeof__(self):
                raise ValueError
        self.assertRaises(ValueError, sys.getsizeof, BadSizeof())

        class InvalidSizeof(object):
            def __sizeof__(self):
                return None
        self.assertRaises(TypeError, sys.getsizeof, InvalidSizeof())
        sentinel = ["sentinel"]
        self.assertIs(sys.getsizeof(InvalidSizeof(), sentinel), sentinel)

        class OverflowSizeof(long):
            def __sizeof__(self):
                return int(self)
        self.assertEqual(sys.getsizeof(OverflowSizeof(sys.maxsize)),
                         sys.maxsize + self.gc_headsize)
        with self.assertRaises(OverflowError):
            sys.getsizeof(OverflowSizeof(sys.maxsize + 1))
        with self.assertRaises(ValueError):
            sys.getsizeof(OverflowSizeof(-1))
        with self.assertRaises((ValueError, OverflowError)):
            sys.getsizeof(OverflowSizeof(-sys.maxsize - 1))

def clean_message(self):
        # Quick check if the message really came encrypted
        message = self.cleaned_data.get("message")
        lines = message.split("\r\n")

        if getsizeof(message) > MAX_MESSAGE_SIZE:
            self.add_error(
                "message",
                _('The message or file exceeds your allowed size limit.')
            )

        begin = "-----BEGIN PGP MESSAGE-----"
        end = "-----END PGP MESSAGE-----"

        try:
            if lines[0] != begin or lines[-1] != end:
                self.add_error("message", _('Invalid PGP message'))
        except IndexError:
            self.add_error("message", _('Invalid PGP message'))
        return message

def _alchemy_image_request(self, method_name, image_file=None,
                               image_url=None, params=None):
        if params is None:
            params = {}
        params['outputMode'] = 'json'
        params = _convert_boolean_values(params)
        headers = {}
        image_contents = None

        if image_file:
            params['imagePostMode'] = 'raw'
            image_contents = image_file.read()
            # headers['content-length'] = sys.getsizeof(image_contents)
            url = '/image/Image' + method_name
        elif image_url:
            params['imagePostMode'] = 'not-raw'
            params['url'] = image_url
            url = '/url/URL' + method_name
        else:
            raise WatsonInvalidArgument(
                'image_file or image_url must be specified')

        return self.request(method='POST', url=url, params=params,
                            data=image_contents, headers=headers,
                            accept_json=True)

def frame_profile(frame_idx, serial_data_path, pickle_path,
                  mol_types, coords, sys_type, assoc_sel_idxs, assoc_type, inx_type):


    inxs, system, assoc = profile_coords(mol_types, coords, sys_type, assoc_sel_idxs, assoc_type, inx_type)

    # data output
    inx_type.pdb_serial_output(inxs[inx_type], serial_data_path, delim=" ")

    # persistent storage
    with open(pickle_path, 'wb') as f:
        pickle.dump(inxs, f)

    print("--------------------------------------------------------------------------------")
    print("frame", frame_idx)
    print("----------------------------------------")
    print("size of inxs {}".format(sys.getsizeof(inxs)))
    print("size of system {}".format(sys.getsizeof(system)))
    print("size of assoc {}".format(sys.getsizeof(assoc)))
    if len(inxs[inx_type]) > 0:
        print(len(inxs[inx_type]), "intermolecular hydrogen bonds")
        for inx in inxs[inx_type]:
            inx.pp()
    else:
        print(0, "intermolecular hydrogen bonds")

def test_memory_usage(self):
        for o in self.objs:
            res = o.memory_usage()
            res_deep = o.memory_usage(deep=True)

            if (com.is_object_dtype(o) or (isinstance(o, Series) and
                                           com.is_object_dtype(o.index))):
                # if there are objects, only deep will pick them up
                self.assertTrue(res_deep > res)
            else:
                self.assertEqual(res, res_deep)

            if isinstance(o, Series):
                self.assertEqual(
                    (o.memory_usage(index=False) +
                        o.index.memory_usage()),
                    o.memory_usage(index=True)
                )

            # sys.getsizeof will call the .memory_usage with
            # deep=True, and add on some GC overhead
            diff = res_deep - sys.getsizeof(o)
            self.assertTrue(abs(diff) < 100)

def save(operations,path):
  print('Saving texthasher model...')

  s = pickle.dumps(operations, pickle.HIGHEST_PROTOCOL)
  z = sys.getsizeof(s)
  print('texthasher model occupies total size of {0:.2f} GB'.format(z/_1GB))

  # In case of huge model, split it into smaller chunks 
  # and save seperately
  if z>_1GB:
    i = 0
    for c in split_to_chunks(s,_1GB):
      i += 1
      save_chunk(path, i, c)
  else:
    # Small model, just save it with typical method
    with open(path + '.0', 'wb+') as f:
      pickle.dump(operations, f)

def measure_memory(cls, obj, seen=None):
        """Recursively finds size of objects"""
        size = sys.getsizeof(obj)
        if seen is None:
            seen = set()
        obj_id = id(obj)
        if obj_id in seen:
            return 0
        # Important mark as seen *before* entering recursion to gracefully handle
        # self-referential objects
        seen.add(obj_id)
        if isinstance(obj, dict):
            size += sum([cls.measure_memory(v, seen) for v in obj.values()])
            size += sum([cls.measure_memory(k, seen) for k in obj.keys()])
        elif hasattr(obj, '__dict__'):
            size += cls.measure_memory(obj.__dict__, seen)
        elif hasattr(obj, '__iter__') and not isinstance(obj, (str, bytes, bytearray)):
            size += sum([cls.measure_memory(i, seen) for i in obj])
        return size

def getrecsizeof(o, seen=None):
    if seen is None:
        seen = set()
    if id(o) in seen:
        return 0
    seen.add(id(o))
    c = sys.getsizeof(o)
    if isinstance(o, dict):
        for k, v in o.items():
            c += getrecsizeof(k, seen)
            c += getrecsizeof(v, seen)
    elif isinstance(o, str) or isinstance(o, bytes):
        pass
    elif isinstance(o, collections.Iterable):
        for e in o:
            c += getrecsizeof(e, seen)
    return c

def test_errors(self):
        class BadSizeof(object):
            def __sizeof__(self):
                raise ValueError
        self.assertRaises(ValueError, sys.getsizeof, BadSizeof())

        class InvalidSizeof(object):
            def __sizeof__(self):
                return None
        self.assertRaises(TypeError, sys.getsizeof, InvalidSizeof())
        sentinel = ["sentinel"]
        self.assertIs(sys.getsizeof(InvalidSizeof(), sentinel), sentinel)

        class OverflowSizeof(long):
            def __sizeof__(self):
                return int(self)
        self.assertEqual(sys.getsizeof(OverflowSizeof(sys.maxsize)),
                         sys.maxsize + self.gc_headsize)
        with self.assertRaises(OverflowError):
            sys.getsizeof(OverflowSizeof(sys.maxsize + 1))
        with self.assertRaises(ValueError):
            sys.getsizeof(OverflowSizeof(-1))
        with self.assertRaises((ValueError, OverflowError)):
            sys.getsizeof(OverflowSizeof(-sys.maxsize - 1))

def test_subclasses(self):
        # Verify that subclasses can share keys (per PEP 412)
        class A:
            pass
        class B(A):
            pass

        a, b = A(), B()
        self.assertEqual(sys.getsizeof(vars(a)), sys.getsizeof(vars(b)))
        self.assertLess(sys.getsizeof(vars(a)), sys.getsizeof({}))
        a.x, a.y, a.z, a.w = range(4)
        self.assertNotEqual(sys.getsizeof(vars(a)), sys.getsizeof(vars(b)))
        a2 = A()
        self.assertEqual(sys.getsizeof(vars(a)), sys.getsizeof(vars(a2)))
        self.assertLess(sys.getsizeof(vars(a)), sys.getsizeof({}))
        b.u, b.v, b.w, b.t = range(4)
        self.assertLess(sys.getsizeof(vars(b)), sys.getsizeof({}))

def test_pickler(self):
            basesize = support.calcobjsize('5P2n3i2n3iP')
            p = _pickle.Pickler(io.BytesIO())
            self.assertEqual(object.__sizeof__(p), basesize)
            MT_size = struct.calcsize('3nP0n')
            ME_size = struct.calcsize('Pn0P')
            check = self.check_sizeof
            check(p, basesize +
                MT_size + 8 * ME_size +  # Minimal memo table size.
                sys.getsizeof(b'x'*4096))  # Minimal write buffer size.
            for i in range(6):
                p.dump(chr(i))
            check(p, basesize +
                MT_size + 32 * ME_size +  # Size of memo table required to
                                          # save references to 6 objects.
                0)  # Write buffer is cleared after every dump().

def get_file_size(file):
    """
    Returns a string with the file size and highest rating.
    """
    byte_size = sys.getsizeof(file)
    kb_size = byte_size / 1024
    if kb_size == 0:
        byte_size = "%s Bytes" % (byte_size)
        return byte_size
    mb_size = kb_size / 1024
    if mb_size == 0:
        kb_size = "%s KB" % (kb_size)
        return kb_size
    gb_size = mb_size / 1024 % (mb_size)
    if gb_size == 0:
        mb_size = "%s MB" %(mb_size)
        return mb_size
    return "%s GB" % (gb_size)

def __call__(self, *args, **kwargs):
        key = hash(repr((args, kwargs)))
        if (not key in self.known_keys): #when an ammount of arguments can't be identified from keys
            value = self.func(*args, **kwargs) #compute function
            self.known_keys.append(key) #add the key to your list of keys
            self.known_values.append(value) #add the value to your list of values
            if self.size is not None:
                self.size -= sys.getsizeof(value) #the assigned space has decreased
                if (sys.getsizeof(self.known_values) > self.size): #free cache when size of values goes beyond the size limit
                    del self.known_keys
                    del self.known_values
                    del self.size          
                    self.known_keys = []
                    self.known_values = []
                    self.size = self.size_copy 
            return value
        else: #if you've already computed everything
            i = self.known_keys.index(key) #find your key and return your values
            return self.known_values[i]

def getsize(top_obj):
    visited = set()

    def inner(obj):
        obj_id = id(obj)
        if obj_id in visited:
            return 0
        visited.add(obj_id)
        size = sys.getsizeof(obj)
        if isinstance(obj, (str, bytes, Number, range, bytearray)):
            pass
        elif isinstance(obj, (tuple, list, Set, deque)):
            size += sum(inner(i) for i in obj)
        elif isinstance(obj, Mapping) or hasattr(obj, 'items'):
            size += sum(inner(k) + inner(v) for k, v in obj.items())
        if hasattr(obj, '__dict__'):
            size += inner(vars(obj))
        if hasattr(obj, '__slots__'):
            size += sum(
                inner(getattr(obj, s))
                for s in obj.__slots__ if hasattr(obj, s))
        return size

    return inner(top_obj)

def __sizeof__(self):
        sizeof = _sys.getsizeof
        n = len(self) + 1                       # number of links including root
        size = sizeof(self.__dict__)            # instance dictionary
        size += sizeof(self.__map) * 2          # internal dict and inherited dict
        size += sizeof(self.__hardroot) * n     # link objects
        size += sizeof(self.__root) * n         # proxy objects
        return size

def post(self, url = '/', data = {}, endpoint = "http://0.0.0.0:9000"):
        #try:
        data = dict(data)
        data['token'] = session['token']
        logger.info ("Docklet Request: user = %s data = %s, url = %s"%(session['username'], data, url))
        reqtype = url.split("/")[1]
        userreq = {
                'login',
                'external_login',
                'register',
                'user',
                'beans',
                'notification',
                'cloud',
                'settings'
                }
        if ":" not in endpoint:
            endpoint = "http://"+endpoint+":"+master_port
        if reqtype in userreq:
            result = requests.post(user_endpoint + url, data=data).json()
        else:
            result = requests.post(endpoint + url, data=data).json()
        # logger.info('response content: %s'%response.content)
        # result = response.json()
        if (result.get('success', None) == "false" and result.get('reason', None) == "Unauthorized Action"):
            abort(401)
        if (result.get('Unauthorized', None) == 'True'):
            session['401'] = 'Token Expired'
            abort(401)
        logstr = "Docklet Response: user = %s result = %s, url = %s" % (session['username'], result, url)
        if (sys.getsizeof(logstr) > 512):
            logstr = "Docklet Response: user = %s, url = %s"%(session['username'], url)
        logger.info(logstr)
        return result
        #except:
            #abort(500)

def deep_getsizeof(o, ids): 
    """Find the memory footprint of a Python object

    This is a recursive function that drills down a Python object graph
    like a dictionary holding nested dictionaries with lists of lists
    and tuples and sets.

    The sys.getsizeof function does a shallow size of only. It counts each
    object inside a container as pointer only regardless of how big it
    really is.

    :param o: the object
    :param ids:
    :return:
    """
    d = deep_getsizeof
    if id(o) in ids:
        return 0

    r = getsizeof(o)
    ids.add(id(o))

    if isinstance(o, str) or isinstance(0, unicode):
        return r

    if isinstance(o, Mapping):
        return r + sum(d(k, ids) + d(v, ids) for k, v in o.iteritems())

    if isinstance(o, Container):
        return r + sum(d(x, ids) for x in o)

    return r

def test_user_public_key(user_contract, web3):
    key = encrypt.createKey()
    binPubKey1 = key.publickey().exportKey('DER')
    print(sys.getsizeof(binPubKey1))
    user_contract.transact().setPubKey(binPubKey1)
    binPubKey2 = user_contract.call().userPubkey(web3.eth.accounts[0])
    print(binPubKey2)
    pubKeyObj = RSA.importKey(binPubKey2)
    enc_data = encrypt.encrypt(pubKeyObj, 'Hello there!')
    print(encrypt.decrypt(enc_data, key))

def _getreferents(unused):
        return ()  # sorry, no refs

# sys.getsizeof() new in Python 2.6, adjusted below

def test_clean_file_upload_form_invalid_data(self):
        t = workflows.create_instance.CustomizeAction(self.request, {})
        upload_str = b'\x81'
        files = {'script_upload':
                 self.SimpleFile('script_name',
                                 upload_str,
                                 sys.getsizeof(upload_str))}

        self.assertRaises(
            forms.ValidationError,
            t.clean_uploaded_files,
            'script',
            files)

def test_clean_file_upload_form_valid_data(self):
        t = workflows.create_instance.CustomizeAction(self.request, {})
        precleaned = 'user data'
        upload_str = 'user data'
        files = {'script_upload':
                 self.SimpleFile('script_name',
                                 upload_str,
                                 sys.getsizeof(upload_str))}

        cleaned = t.clean_uploaded_files('script', files)

        self.assertEqual(
            cleaned,
            precleaned)