docstr: The default snakemake log handler.
  
  Prints the output to the console.
  
  Args:
      msg (dict):     the log message dictionary
code: def text_handler(self, msg):
      """The default snakemake log handler.
  
      Prints the output to the console.
  
      Args:
          msg (dict):     the log message dictionary
      """
      if self.is_quiet_about("all"):
          # do not log anything
          return
  
      def job_info(msg):
          def format_item(item, omit=None, valueformat=str):
              value = msg[item]
              if value != omit:
                  return f"    {item}: {valueformat(value)}"
  
          yield "{}{} {}:".format(
              "local" if msg["local"] else "",
              "checkpoint" if msg["is_checkpoint"] else "rule",
              msg["name"],
          )
          for item in ["input", "output", "log"]:
              fmt = format_item(item, omit=[], valueformat=", ".join)
              if fmt != None:
                  yield fmt
  
          singleitems = ["jobid", "benchmark"]
          if self.printreason:
              singleitems.append("reason")
          for item in singleitems:
              fmt = format_item(item, omit=None)
              if fmt != None:
                  yield fmt
  
          wildcards = format_wildcards(msg["wildcards"])
          if wildcards:
              yield f"    wildcards: {wildcards}"
  
          for item, omit in zip("priority threads".split(), [0, 1]):
              fmt = format_item(item, omit=omit)
              if fmt != None:
                  yield fmt
  
          resources = format_resources(msg["resources"])
          if resources:
              yield f"    resources: {resources}"
  
      def show_logs(logs):
          for f in logs:
              try:
                  content = open(f, "r").read()
              except FileNotFoundError:
                  yield f"Logfile {f} not found."
                  return
              lines = content.splitlines()
              logfile_header = f"Logfile {f}:"
              if not lines:
                  logfile_header += " empty file"
                  yield logfile_header
                  return
              yield logfile_header
              # take the length of the longest line, but limit to max 80
              max_len = min(
                  max(max(len(l) for l in lines), len(logfile_header)),
                  80,
              )
              yield "=" * max_len
              yield from lines
              yield "=" * max_len
  
      def indent(item):
          if msg.get("indent"):
              return f"    {item}"
          else:
              return item
  
      def timestamp():
          self.logger.info(indent(f"[{time.asctime()}]"))
  
      level = msg["level"]
  
      if level == "job_info" and not self.is_quiet_about("rules"):
          if not self.last_msg_was_job_info:
              self.logger.info("")
          timestamp()
          if msg["msg"] is not None:
              self.logger.info(
                  indent("Job {}: {}".format(msg["jobid"], msg["msg"]))
              )
              if self.printreason:
                  self.logger.info(
                      indent("Reason: {}".format(msg["reason"]))
                  )
          else:
              self.logger.info("\n".join(map(indent, job_info(msg))))
          if msg["is_checkpoint"]:
              self.logger.warning(
                  indent(
                      "DAG of jobs will be updated after completion."
                  )
              )
          if msg["is_handover"]:
              self.logger.warning(
                  "Handing over execution to foreign system..."
              )
          self.logger.info("")
  
          self.last_msg_was_job_info = True
      elif level == "group_info" and not self.is_quiet_about("rules"):
          timestamp()
          msg = "group job {} (jobs in lexicogr. order):".format(
              msg["groupid"]
          )
          if not self.last_msg_was_job_info:
              msg = "\n" + msg
          self.logger.info(msg)
      elif level == "job_error":
  
          def job_error():
              yield "Error in rule {}:".format(msg["name"])
              if msg["msg"]:
                  yield "    message: {}".format(msg["msg"])
              yield "    jobid: {}".format(msg["jobid"])
              if msg["input"]:
                  yield "    input: {}".format(", ".join(msg["input"]))
              if msg["output"]:
                  yield "    output: {}".format(
                      ", ".join(msg["output"])
                  )
              if msg["log"]:
                  yield "    log: {} (check log file(s) for error details)".format(
                      ", ".join(msg["log"])
                  )
              if msg["conda_env"]:
                  yield "    conda-env: {}".format(msg["conda_env"])
              if msg["shellcmd"]:
                  yield "    shell:\n        {}\n        (one of the commands exited with non-zero exit code; note that snakemake uses bash strict mode!)".format(
                      msg["shellcmd"]
                  )
  
              for item in msg["aux"].items():
                  yield "    {}: {}".format(*item)
  
              if self.show_failed_logs and msg["log"]:
                  yield from show_logs(msg["log"])
  
              yield ""
  
          timestamp()
          self.logger.error("\n".join(map(indent, job_error())))
      elif level == "group_error":
  
          def group_error():
              yield f"Error in group {msg['groupid']}:"
              if msg["msg"]:
                  yield f"    message: {msg['msg']}"
              if msg["aux_logs"]:
                  yield f"    log: {', '.join(msg['aux_logs'])} (check log file(s) for error details)"
              yield "    jobs:"
              for info in msg["job_error_info"]:
                  yield f"        rule {info['name']}:"
                  yield f"            jobid: {info['jobid']}"
                  if info["output"]:
                      yield f"            output: {', '.join(info['output'])}"
                  if info["log"]:
                      yield f"            log: {', '.join(info['log'])} (check log file(s) for error details)"
              logs = msg["aux_logs"] + [
                  f
                  for info in msg["job_error_info"]
                  for f in info["log"]
              ]
              if self.show_failed_logs and logs:
                  yield from show_logs(logs)
              yield ""
  
          timestamp()
          self.logger.error("\n".join(group_error()))
      else:
          if level == "info" and not self.is_quiet_about("progress"):
              self.logger.warning(msg["msg"])
          if level == "warning":
              self.logger.critical(msg["msg"])
          elif level == "error":
              self.logger.error(msg["msg"])
          elif level == "debug":
              self.logger.debug(msg["msg"])
          elif level == "resources_info" and not self.is_quiet_about(
              "progress"
          ):
              self.logger.warning(msg["msg"])
          elif level == "run_info" and not self.is_quiet_about(
              "progress"
          ):
              self.logger.warning(msg["msg"])
          elif level == "progress" and not self.is_quiet_about(
              "progress"
          ):
              done = msg["done"]
              total = msg["total"]
              self.logger.info(
                  "{} of {} steps ({}) done".format(
                      done, total, format_percentage(done, total)
                  )
              )
          elif level == "shellcmd":
              if self.printshellcmds:
                  self.logger.warning(indent(msg["msg"]))
          elif level == "job_finished" and not self.is_quiet_about(
              "progress"
          ):
              timestamp()
              self.logger.info("Finished job {}.".format(msg["jobid"]))
              pass
          elif level == "rule_info":
              self.logger.info(msg["name"])
              if msg["docstring"]:
                  self.logger.info("    " + msg["docstring"])
          elif level == "d3dag":
              print(
                  json.dumps(
                      {"nodes": msg["nodes"], "links": msg["edges"]}
                  )
              )
          elif level == "dag_debug":
              if self.debug_dag:
                  if "file" in msg:
                      self.logger.warning(
                          "file {file}:\n    {msg}\n{exception}".format(
                              file=msg["file"],
                              msg=msg["msg"],
                              exception=textwrap.indent(
                                  str(msg["exception"]), "    "
                              ),
                          )
                      )
                  else:
                      job = msg["job"]
                      self.logger.warning(
                          "{status} job {name}\n    wildcards: {wc}".format(
                              status=msg["status"],
                              name=job.rule.name,
                              wc=format_wildcards(job.wildcards),
                          )
                      )
  
          self.last_msg_was_job_info = False
  
imports: import logging
  import platform
  import time
  import datetime
  import sys
  import os
  import json
  import threading
  from functools import partial
  import inspect
  import textwrap
label_desc: Use image node_id 27 for calling the Logger obj's underlying member method code with example usage: obj.text_handler(msg) and returns: str, item, item, str, item, str, valueformat