Python logging 模块，debug() 实例源码

我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用logging.debug()。

项目：bitcoin-arbitrage 作者：ucfyao | 项目源码 | 文件源码

def init_logger(self, args):
        level = logging.INFO
        if args.verbose:
            level = logging.VERBOSE
        if args.debug:
            level = logging.DEBUG
        logging.basicConfig(format='%(asctime)s [%(levelname)s] %(message)s',
                            level=level)

        Rthandler = RotatingFileHandler('arbitrage.log', maxBytes=100*1024*1024,backupCount=10)
        Rthandler.setLevel(level)
        formatter = logging.Formatter('%(asctime)-12s [%(levelname)s] %(message)s')  
        Rthandler.setFormatter(formatter)
        logging.getLogger('').addHandler(Rthandler)

        logging.getLogger("requests").setLevel(logging.WARNING)
        logging.getLogger("urllib3").setLevel(logging.WARNING)

项目：PyPlanet 作者：PyPlanet | 项目源码 | 文件源码

def check(self, first_check=False):
        from pyplanet import __version__ as current_version

        logging.debug('Checking for new versions...')

        async with aiohttp.ClientSession() as session:
            async with session.get(self.url) as resp:
                for release in await resp.json():
                    if not release['draft'] and not release['prerelease']:
                        self.latest = release['tag_name']
                        break

                self.current = current_version
                logging.debug('Version check, your version: {}, online version: {}'.format(self.current, self.latest))

                if first_check and self.update_available:
                    logging.info('New version of PyPlanet available, consider updating: {}'.format(self.latest))
                    await self.instance.chat(
                        '\uf1e6 $FD4$oPy$369Planet$z$s$fff \uf0e7 new version available: v{}. Consider updating!'.format(self.latest)
                    )

项目：defuse_division 作者：lelandbatey | 项目源码 | 文件源码

def update_locallist(listb, refresh_lock):
    global UPDATE_LOCALSERVERS
    cached = dict()
    durable_duration = 5
    while UPDATE_LOCALSERVERS:
        toremove = []
        for item in cached:
            if cached[item] > durable_duration:
                toremove.append(item)
            cached[item] += 1
        for item in toremove:
            del cached[item]
        info = zeroconf_info()
        for item in info:
            cached[item] = 0
        if not UPDATE_LOCALSERVERS: break
        logging.debug('Acquiring refresh lock for updating list of local servers.')
        refresh_lock.acquire()
        listb.update_items(cached.keys())
        refresh_lock.release()
        logging.debug('Releasing refresh lock after updating locla server list')

项目：YellowFin_Pytorch 作者：JianGoForIt | 项目源码 | 文件源码

def grad_sparsity(self):
    global_state = self._global_state
    if self._iter == 0:
      global_state["sparsity_avg"] = 0.0
    non_zero_cnt = 0.0
    all_entry_cnt = 0.0
    for group in self._optimizer.param_groups:
      for p in group['params']:
        if p.grad is None:
          continue
        grad = p.grad.data
        grad_non_zero = grad.nonzero()
        if grad_non_zero.dim() > 0:
          non_zero_cnt += grad_non_zero.size()[0]
        all_entry_cnt += torch.numel(grad)
    beta = self._beta
    global_state["sparsity_avg"] = beta * global_state["sparsity_avg"] \
      + (1 - beta) * non_zero_cnt / float(all_entry_cnt)
    self._sparsity_avg = \
      global_state["sparsity_avg"] / self.zero_debias_factor()

    if self._verbose:
      logging.debug("sparsity %f, sparsity avg %f", non_zero_cnt / float(all_entry_cnt), self._sparsity_avg)

    return

项目：YellowFin_Pytorch 作者：JianGoForIt | 项目源码 | 文件源码

def grad_sparsity(self):
    global_state = self._global_state
    if self._iter == 0:
      global_state["sparsity_avg"] = 0.0
    non_zero_cnt = 0.0
    all_entry_cnt = 0.0
    for group in self._optimizer.param_groups:
      for p in group['params']:
        if p.grad is None:
          continue
        grad = p.grad.data
        grad_non_zero = grad.nonzero()
        if grad_non_zero.dim() > 0:
          non_zero_cnt += grad_non_zero.size()[0]
        all_entry_cnt += torch.numel(grad)
    beta = self._beta
    global_state["sparsity_avg"] = beta * global_state["sparsity_avg"] \
      + (1 - beta) * non_zero_cnt / float(all_entry_cnt)
    self._sparsity_avg = \
      global_state["sparsity_avg"] / self.zero_debias_factor()

    if DEBUG:
      logging.debug("sparsity %f, sparsity avg %f", non_zero_cnt / float(all_entry_cnt), self._sparsity_avg)

    return

项目：YellowFin_Pytorch 作者：JianGoForIt | 项目源码 | 文件源码

def get_cubic_root(self):
    # We have the equation x^2 D^2 + (1-x)^4 * C / h_min^2
    # where x = sqrt(mu).
    # We substitute x, which is sqrt(mu), with x = y + 1.
    # It gives y^3 + py = q
    # where p = (D^2 h_min^2)/(2*C) and q = -p.
    # We use the Vieta's substution to compute the root.
    # There is only one real solution y (which is in [0, 1] ).
    # http://mathworld.wolfram.com/VietasSubstitution.html
    # eps in the numerator is to prevent momentum = 1 in case of zero gradient
    p = (self._dist_to_opt + eps)**2 * (self._h_min + eps)**2 / 2 / (self._grad_var + eps)
    w3 = (-math.sqrt(p**2 + 4.0 / 27.0 * p**3) - p) / 2.0
    w = math.copysign(1.0, w3) * math.pow(math.fabs(w3), 1.0/3.0)
    y = w - p / 3.0 / (w + eps)
    x = y + 1

    if DEBUG:
      logging.debug("p %f, den %f", p, self._grad_var + eps)
      logging.debug("w3 %f ", w3)
      logging.debug("y %f, den %f", y, w + eps)

    return x

项目：tsproxy 作者：WPO-Foundation | 项目源码 | 文件源码

def run(self):
    global lock, background_activity_count
    try:
      logging.debug('[{0:d}] AsyncDNS - calling getaddrinfo for {1}:{2:d}'.format(self.client_id, self.hostname, self.port))
      addresses = socket.getaddrinfo(self.hostname, self.port)
      logging.info('[{0:d}] Resolving {1}:{2:d} Completed'.format(self.client_id, self.hostname, self.port))
    except:
      addresses = ()
      logging.info('[{0:d}] Resolving {1}:{2:d} Failed'.format(self.client_id, self.hostname, self.port))
    message = {'message': 'resolved', 'connection': self.client_id, 'addresses': addresses, 'localhost': self.is_localhost}
    self.result_pipe.SendMessage(message, False)
    lock.acquire()
    if background_activity_count > 0:
      background_activity_count -= 1
    lock.release()
    # open and close a local socket which will interrupt the long polling loop to process the message
    s = socket.socket()
    s.connect((server.ipaddr, server.port))
    s.close()


########################################################################################################################
#   TCP Client
########################################################################################################################

项目：zappa-django-utils 作者：Miserlou | 项目源码 | 文件源码

def close(self, *args, **kwargs):
        """
        Engine closed, copy file to DB
        """
        super(DatabaseWrapper, self).close(*args, **kwargs)

        signature_version = self.settings_dict.get("SIGNATURE_VERSION", "s3v4")
        s3 = boto3.resource('s3',
                config=botocore.client.Config(signature_version=signature_version))

        try:
            with open(self.settings_dict['NAME'], 'rb') as f:
                fb = f.read()
                bytesIO = BytesIO()
                bytesIO.write(fb)
                bytesIO.seek(0)

                s3_object = s3.Object(self.settings_dict['BUCKET'], self.settings_dict['REMOTE_NAME'])
                result = s3_object.put('rb', Body=bytesIO)

        except Exception as e:
            print(e)

        logging.debug("Saved to remote DB!")

项目：lydoc 作者：Cecca | 项目源码 | 文件源码

def function_definition(self, ast):
        logging.debug("Found function definition %s", ast.name)
        self.add_position_info(ast)
        # Convert the AST to a simple dict, so that the grako buffer
        # associated with the parseinfo can be released when it is no
        # longer needed. Otherwise, the buffer is kept in memory
        # until the reference to this AST is kept in memory. When
        # dealing with many files this can lead to an excessive
        # memory usage.
        ast = dict(ast)
        ast['type'] = 'function'
        # Strip the comment character from the beginning of the line
        docs = ast['documentation']
        if docs is not None:
            stripped = strip_leading_comments(docs)
            ast['documentation'] = stripped
        self.collected_elements.append(ast)
        return ast

项目：lydoc 作者：Cecca | 项目源码 | 文件源码

def name_definition(self, ast):
        logging.debug('Found name definition %s', ast.name)
        self.add_position_info(ast)
        # Convert the AST to a simple dict, so that the grako buffer
        # associated with the parseinfo can be released when it is no
        # longer needed. Otherwise, the buffer is kept in memory
        # until the reference to this AST is kept in memory. When
        # dealing with many files this can lead to an excessive
        # memory usage.
        ast = dict(ast)
        ast['type'] = 'name'
        # Strip the comment character from the beginning of the line
        docs = ast['documentation']
        if docs is not None:
            stripped = strip_leading_comments(docs)
            ast['documentation'] = stripped
        self.collected_elements.append(ast)
        return ast

项目：core-framework 作者：RedhawkSDR | 项目源码 | 文件源码

def stop(self):
        self._log.debug("%s.stop()", self.naming_service_name)
        if self.query_commandAlive() == True:
            for sig, timeout in self.STOP_SIGNALS:
                try:
                    os.kill(self._pid, sig)
                except OSError:
                    self._pid = None
                    return
                if timeout != None:
                    giveup_time = time.time() + timeout
                    while os.waitpid(self._pid, os.WNOHANG) == (0,0):
                        time.sleep(0.1)
                        if time.time() > giveup_time:
                            break
                else:
                    # Wait until there is a response
                    os.waitpid(self._pid, 0)
            self._pid = None

项目：core-framework 作者：RedhawkSDR | 项目源码 | 文件源码

def query(self, configProperties):
    self._log.debug("BasicUsesDevice.query(%s)", configProperties)
    if configProperties == []:
      rv = []
      for key in self.props.keys():
        val = self.props[key].value
        d = CF.DataType(id=key, value=val)
        rv.append(d)
    else:
      unknownProperties = []
      for prop in configProperties:
        try:
            prop.value = self.props[prop.id].value
        except KeyError:
            unknownProperties.append(prop)

      if len(unknownProperties) > 0:
        raise CF.UnknownProperties(unknownProperties)

      rv = configProperties
    self._log.debug("BasicUsesDevice.query() -> %s", rv)
    return rv

项目：core-framework 作者：RedhawkSDR | 项目源码 | 文件源码

def _allocateCapacity(self, propname, value):
        """Override this if you want if you don't want magic dispatch"""
        self._log.debug("_allocateCapacity(%s, %s)", propname, value)
        if self._allocationCallbacks.has_key(propname):
            return self._allocationCallbacks[propname]._allocate(value)

        modified_propname = ''
        for ch in propname:
            if ch.isalnum():
                modified_propname += ch
            else:
                modified_propname += '_'
        allocate = _getCallback(self, "allocate_%s" % modified_propname)
        if allocate:
            self._log.debug("using callback for _allocateCapacity()", )
            return allocate(value)
        else:
            self._log.debug("no callback for _allocateCapacity()", )
            return False

项目：tornado-ssdb-project 作者：ego008 | 项目源码 | 文件源码

def safe_shutdown(self, timeout):
        if self._shutting_down:
            return

        self._shutting_down = True
        logging.debug('Stopping http server.')
        self.stop()

        logging.debug('Will be shutdown within %s seconds ...', timeout)
        io_loop = tornado.ioloop.IOLoop.instance()

        deadline = time.time() + timeout

        def safe_stop_loop():
            now = time.time()
            if now < deadline and io_loop._callbacks:
                io_loop.add_timeout(now + 1, safe_stop_loop)
            else:
                io_loop.stop()
                logging.debug('Http server has been shutdown.')
        safe_stop_loop()