text: #!/usr/bin/env python
  
  """
  The main entrypoint for plaster.
  
  Note that this is typically run through the ./p helper instead of directly.
  """
  
  import json
  import logging
  import time
  import arrow
  from munch import Munch
  from plaster.run.job import JobResult
  from plaster.run.run import RunExecutor
  from plaster.tools.assets import assets
  from plaster.tools.utils import utils
  from plaster.tools.utils.build_utils import out_of_date
  from plaster.tools.zap import zap
  from plaster.tools.zlog import zlog
  from plaster.tools.zlog.zlog import tell, spy
  
  from plumbum import cli, local
  
  log = logging.getLogger(__name__)
  
  
  class RunApp(cli.Application):
      """
      Run plaster components as directed by a plaster_run.yaml file in the given path.
  
      If the specified src_dir is not in ./jobs_folder then you will be
      asked if you want to create a symbolic link within jobs_folder
      to the specified src_dir.
      """
  
      limit = cli.SwitchAttr(
          ["--limit"],
          str,
          help="Comma separated list of targets to build (skipping upstream requirements)",
      )
      force = cli.Flag(
          "--force",
          default=False,
          help="If True, targets will be rebuilt unconditionally",
      )
      clean = cli.Flag(
          "--clean", default=False, help="If True, the targets will cleaned but not built"
      )
      cpu_limit = cli.SwitchAttr(
          "--cpu_limit", int, default=-1, help="Set to limit to n_cpus"
      )
      skip_reports = cli.Flag(
          "--skip_reports", default=False, help="Do not run the report notebook(s)"
      )
      reports_only = cli.Flag(
          "--reports_only",
          default=False,
          help="run only report (may want to combine with --force)",
      )
      debug_mode = cli.Flag("--debug_mode", default=False, help="Easier debug tracing")
      continue_on_error = cli.Flag("--continue_on_error", default=False)
  
      n_fields_limit = cli.SwitchAttr(
          "--n_fields_limit", int, help="Set to limit on number of fields to process"
      )
      no_progress = cli.Flag("--no_progress", default=False, help="Do not print progress")
      job = cli.SwitchAttr("--job", str, default=None, help="job folder")
  
      def run_ipynb(self, ipynb_path):
          # Note: the timeout has been set to 8 hours to facilitate reports for
          # huge jobs (e.g. 100+ runs).
          log.info(f"Executing report notebook {ipynb_path}")
          local["jupyter"](
              "nbconvert",
              "--to",
              "html",
              "--execute",
              ipynb_path,
              "--ExecutePreprocessor.timeout=28800",
              "--no-input",
          )
  
      def main(self, job_folder=None):
          switches = utils.plumbum_switches(self)
  
          if job_folder is None:
              job_folder = self.job
  
          job_folder = assets.validate_job_folder(job_folder)
          # At this point job_folder is a plumbum path
  
          # Add a new handler so we get PER-RUN log files into the job folder
          per_run_log_path = job_folder / f"{int(time.time()):06x}.log"
          formatter = zlog.ColorfulFormatter(
              "%(name)s %(asctime)s %(levelname)s %(message)s %(filename)s %(lineno)d"
          )
          handler = logging.StreamHandler(open(per_run_log_path, "w"))
          handler.setFormatter(formatter)
          zlog.add_handler(handler)
  
          tell(f"Trapping run logs into {per_run_log_path}")
  
          if job_folder is None:
              log.error(f"No job_folder was specified")
              return 1
  
          tell(
              f"Plaster run {job_folder} limit={self.limit} started at {arrow.utcnow().format()}"
          )
  
          if not job_folder.exists():
              log.error(f"Unable to find the path {job_folder}")
              return 1
  
          # Load the job_uuid if available, evntually this will be nice for logging
          job_uuid = None
          job_yaml = job_folder / "job_manifest.yaml"
          if job_yaml.exists():
              job_manifest = utils.yaml_load_munch(job_yaml)
              job_uuid = job_manifest.uuid
  
          # Find all the plaster_run.yaml files. They might be in run subfolders
          found = list(job_folder.walk(filter=lambda p: p.name == "plaster_run.yaml"))
          run_dirs = [p.dirname for p in found]
  
          if len(run_dirs) == 0:
              log.error(
                  "Plaster: Nothing to do because no run_dirs have plaster_run.yaml files"
              )
              return 1
  
          def run_reports():
              report_paths = [job_folder / "report.ipynb"] + (
                  job_folder / "_reports" // "*.ipynb"
              )
  
              for report_src_path in report_paths:
                  report_dst_path = report_src_path.with_suffix(".html")
                  if report_src_path.exists() and (
                      self.force or out_of_date(report_src_path, report_dst_path)
                  ):
                      tell(f"Running report {report_src_path}")
                      self.run_ipynb(report_src_path)
  
          if self.reports_only:
              run_reports()
              return 0
  
          # A normal run where all happens in this process
          failure_count = 0
          for run_dir_i, run_dir in enumerate(sorted(run_dirs)):
              zlog.metrics(
                  f"Starting run subdirectory {run_dir}. {run_dir_i + 1} of {len(run_dirs)}",
                  log=log,
                  _type="plaster_start",
                  run_dir=run_dir,
                  run_dir_i=run_dir_i,
                  run_dir_n=len(run_dirs),
                  **switches,
              )
  
              try:
                  with zap.Context(
                      cpu_limit=self.cpu_limit,
                      mode="debug" if self.debug_mode else None,
                      allow_inner_parallelism=True,
                  ):
                      # allow_inner_parallelism=True needs to be true so that each task such as sigproc_v2
                      # can allocate parallel jobs to each field.
  
                      run = RunExecutor(run_dir).load()
                      if "_erisyon" in run.config:
                          zlog.metrics(
                              "run metrics",
                              log=log,
                              _type="erisyon_block",
                              **run.config._erisyon,
                          )
  
                      failure_count += run.execute(
                          force=self.force,
                          limit=self.limit.split(",") if self.limit else None,
                          clean=self.clean,
                          n_fields_limit=self.n_fields_limit,
                          no_progress=self.no_progress,
                      )
              except Exception as e:
                  failure_count += 1
                  if not self.continue_on_error:
                      raise e
  
          if failure_count == 0 and self.limit is None and not self.clean:
              # WRITE job_info.yaml with metadata used by the indexer
              n_runs = len(run_dirs)
              job_info = Munch(n_runs=n_runs, job_uuid=job_uuid)
              if n_runs == 1:
                  job = JobResult(job_folder=job_folder)
                  tsv_data = {}
                  try:
                      tsv_data = job.runs[0].ims_import.tsv_data
                  except:
                      pass
  
                  nd2_metadata = {}
                  try:
                      nd2_metadata = job.runs[0].ims_import._nd2_metadata
                  except:
                      pass
  
                  job_info.update(tsv_data=tsv_data, nd2_metadata=nd2_metadata)
  
              utils.yaml_save(job_folder / "job_info.yaml", job_info)
  
              # RUN reports if not skipped
              if not self.skip_reports:
                  run_reports()
  
          return failure_count
  
  
  if __name__ == "__main__":
      # This is ONLY executed if you do not "main.py" because
      # main.py imports this file as a subcommand
      RunApp.run()
  