class_docstr: The Runner manages two threads (per instance) and all of the
  workers (per class, i.e., singletons).
  
  The two threads run interface() and jobserver(), below.
  
  interface() is responsible grouping jobs for dispatch, tracking job
  progress, integrating measurements returned from workers.
  
  jobserver() is a lightweight thread that serves jobs and receives their
  requests, acting as a switchboard between workers, interface(), and
  whatever event_listener is present (via post_event()).
  
  workers are stored in AnalysisRunner.workers[], and are separate processes.
  They are expected to exit if their stdin() closes, e.g., if the parent
  process goes away.
  
  interface() and jobserver() communicate via Queues and using condition
  variables to get each other's attention.  zmqrequest is used to communicate
  between jobserver() and workers[].
class_docstr_tok: ["The","Runner","manages","two","threads","(","per","instance",")","and","all","of","the","workers","(","per","class",",","i.e.",",","singletons",")",".","The","two","threads","run","interface","(",")","and","jobserver","(",")",",","below",".","interface","(",")","is","responsible","grouping","jobs","for","dispatch",",","tracking","job","progress",",","integrating","measurements","returned","from","workers",".","jobserver","(",")","is","a","lightweight","thread","that","serves","jobs","and","receives","their","requests",",","acting","as","a","switchboard","between","workers",",","interface","(",")",",","and","whatever","event_listener","is","present","(","via","post_event","(",")",")",".","workers","are","stored","in","AnalysisRunner.workers","[","]",",","and","are","separate","processes",".","They","are","expected","to","exit","if","their","stdin","(",")","closes",",","e.g.",",","if","the","parent","process","goes","away",".","interface","(",")","and","jobserver","(",")","communicate","via","Queues","and","using","condition","variables","to","get","each","other","'s","attention",".","zmqrequest","is","used","to","communicate","between","jobserver","(",")","and","workers","[","]","."]
docstr: Top-half thread for running an analysis.  Sets up grouping for jobs,
  deals with returned measurements, reports status periodically.
  
  start_signal- signal this semaphore when jobs are ready.
  image_set_start - beginning image set number to process
  image_set_end - last image set number to process
  overwrite - whether to recompute imagesets that already have data in initial_measurements.
code: def interface(
      self,
      start_signal,
      image_set_start=1,
      image_set_end=None,
      overwrite=True,
  ):
      """Top-half thread for running an analysis.  Sets up grouping for jobs,
      deals with returned measurements, reports status periodically.
  
      start_signal- signal this semaphore when jobs are ready.
      image_set_start - beginning image set number to process
      image_set_end - last image set number to process
      overwrite - whether to recompute imagesets that already have data in initial_measurements.
      """
      posted_analysis_started = False
      acknowledged_thread_start = False
      measurements = None
      workspace = None
      try:
          # listen for pipeline events, and pass them upstream
          self.pipeline.add_listener(
              lambda pipe, evt: self.post_event(evt)
          )
  
          initial_measurements = None
          # Make a temporary measurements file.
          fd, filename = tempfile.mkstemp(
              ".h5", dir=get_temporary_directory()
          )
          try:
              fd = os.fdopen(fd, "wb")
              fd.write(self.initial_measurements_buf)
              fd.close()
              initial_measurements = Measurements(
                  filename=filename, mode="r"
              )
              measurements = Measurements(
                  image_set_start=None,
                  copy=initial_measurements,
                  mode="a",
              )
          finally:
              if initial_measurements is not None:
                  initial_measurements.close()
              os.unlink(filename)
  
          # The shared dicts are needed in jobserver()
          self.shared_dicts = [
              m.get_dictionary() for m in self.pipeline.modules()
          ]
          workspace = Workspace(
              self.pipeline,
              None,
              None,
              None,
              measurements,
              ImageSetList(),
          )
  
          if image_set_end is None:
              image_set_end = measurements.get_image_numbers()[-1]
          image_sets_to_process = list(
              filter(
                  lambda x: image_set_start <= x <= image_set_end,
                  measurements.get_image_numbers(),
              )
          )
  
          self.post_event(Started())
          posted_analysis_started = True
  
          # reset the status of every image set that needs to be processed
          has_groups = measurements.has_groups()
          if self.pipeline.requires_aggregation():
              overwrite = True
          if has_groups and not overwrite:
              if not measurements.has_feature("Image", self.STATUS):
                  overwrite = True
              else:
                  group_status = {}
                  for image_number in measurements.get_image_numbers():
                      group_number = measurements[
                          "Image",
                          "Group_Number",
                          image_number,
                      ]
                      status = measurements[
                          "Image",
                          self.STATUS,
                          image_number,
                      ]
                      if status != self.STATUS_DONE:
                          group_status[
                              group_number
                          ] = self.STATUS_UNPROCESSED
                      elif group_number not in group_status:
                          group_status[group_number] = self.STATUS_DONE
  
          new_image_sets_to_process = []
          for image_set_number in image_sets_to_process:
              needs_reset = False
              if (
                  overwrite
                  or (
                      not measurements.has_measurements(
                          "Image",
                          self.STATUS,
                          image_set_number,
                      )
                  )
                  or (
                      measurements[
                          "Image",
                          self.STATUS,
                          image_set_number,
                      ]
                      != self.STATUS_DONE
                  )
              ):
                  needs_reset = True
              elif has_groups:
                  group_number = measurements[
                      "Image",
                      "Group_Number",
                      image_set_number,
                  ]
                  if group_status[group_number] != self.STATUS_DONE:
                      needs_reset = True
              if needs_reset:
                  new_image_sets_to_process.append(image_set_number)
          measurements[
              "Image", self.STATUS, new_image_sets_to_process
          ] = [self.STATUS_UNPROCESSED] * len(new_image_sets_to_process)
          image_sets_to_process = new_image_sets_to_process
  
          # Find image groups.  These are written into measurements prior to
          # analysis.  Groups are processed as a single job.
          if has_groups or self.pipeline.requires_aggregation():
              worker_runs_post_group = True
              job_groups = {}
              for image_set_number in image_sets_to_process:
                  group_number = measurements[
                      "Image",
                      "Group_Number",
                      image_set_number,
                  ]
                  group_index = measurements[
                      "Image",
                      "Group_Index",
                      image_set_number,
                  ]
                  job_groups[group_number] = job_groups.get(
                      group_number, []
                  ) + [(group_index, image_set_number)]
              job_groups = [
                  [isn for _, isn in sorted(job_groups[group_number])]
                  for group_number in sorted(job_groups)
              ]
          else:
              worker_runs_post_group = False  # prepare_group will be run in worker, but post_group is below.
              job_groups = [
                  [image_set_number]
                  for image_set_number in image_sets_to_process
              ]
  
          # XXX - check that any constructed groups are complete, i.e.,
          # image_set_start and image_set_end shouldn't carve them up.
  
          if not worker_runs_post_group:
              # put the first job in the queue, then wait for the first image to
              # finish (see the check of self.finish_queue below) to post the rest.
              # This ensures that any shared data from the first imageset is
              # available to later imagesets.
              self.work_queue.put(
                  (job_groups[0], worker_runs_post_group, True)
              )
              waiting_for_first_imageset = True
              del job_groups[0]
          else:
              waiting_for_first_imageset = False
              for job in job_groups:
                  self.work_queue.put(
                      (job, worker_runs_post_group, False)
                  )
              job_groups = []
          start_signal.release()
          acknowledged_thread_start = True
  
          # We loop until every image is completed, or an outside event breaks the loop.
          while not self.cancelled:
              # gather measurements
              while not self.received_measurements_queue.empty():
                  (
                      image_numbers,
                      buf,
                  ) = self.received_measurements_queue.get()
                  image_numbers = [int(i) for i in image_numbers]
                  recd_measurements = load_measurements_from_buffer(buf)
                  self.copy_recieved_measurements(
                      recd_measurements, measurements, image_numbers
                  )
                  recd_measurements.close()
                  del recd_measurements
  
              # check for jobs in progress
              while not self.in_process_queue.empty():
                  image_set_numbers = self.in_process_queue.get()
                  for image_set_number in image_set_numbers:
                      measurements[
                          "Image",
                          self.STATUS,
                          int(image_set_number),
                      ] = self.STATUS_IN_PROCESS
  
              # check for finished jobs that haven't returned measurements, yet
              while not self.finished_queue.empty():
                  finished_req = self.finished_queue.get()
                  measurements[
                      "Image",
                      self.STATUS,
                      int(finished_req.image_set_number),
                  ] = self.STATUS_FINISHED_WAITING
                  if waiting_for_first_imageset:
                      assert isinstance(
                          finished_req,
                          anareply.ImageSetSuccessWithDictionary,
                      )
                      self.shared_dicts = finished_req.shared_dicts
                      waiting_for_first_imageset = False
                      assert len(self.shared_dicts) == len(
                          self.pipeline.modules()
                      )
                      for worker_dict, module in zip(
                          self.shared_dicts, self.pipeline.modules()
                      ):
                          # Apply those imported states to the master pipeline
                          module.get_dictionary().update(worker_dict)
                      # if we had jobs waiting for the first image set to finish,
                      # queue them now that the shared state is available.
                      for job in job_groups:
                          self.work_queue.put(
                              (job, worker_runs_post_group, False)
                          )
                  finished_req.reply(anareply.Ack())
  
              # check progress and report
              counts = collections.Counter(
                  measurements[
                      "Image",
                      self.STATUS,
                      image_set_number,
                  ]
                  for image_set_number in image_sets_to_process
              )
              self.post_event(Progress(counts))
  
              # Are we finished?
              if counts[self.STATUS_DONE] == len(image_sets_to_process):
                  last_image_number = measurements.get_image_numbers()[
                      -1
                  ]
                  measurements.image_set_number = last_image_number
                  if not worker_runs_post_group:
                      self.pipeline.post_group(workspace, {})
  
                  workspace = Workspace(
                      self.pipeline,
                      None,
                      None,
                      None,
                      measurements,
                      None,
                      None,
                  )
                  workspace.post_run_display_handler = (
                      self.post_run_display_handler
                  )
                  self.pipeline.post_run(workspace)
                  break
  
              measurements.flush()
              # not done, wait for more work
              with self.interface_work_cv:
                  while self.paused or (
                      (not self.cancelled)
                      and self.in_process_queue.empty()
                      and self.finished_queue.empty()
                      and self.received_measurements_queue.empty()
                  ):
                      self.interface_work_cv.wait()  # wait for a change of status or work to arrive
      except Exception as e:
          print(e)
      finally:
          # Note - the measurements file is owned by the queue consumer
          #        after this post_event.
          #
          if not acknowledged_thread_start:
              start_signal.release()
          if posted_analysis_started:
              was_cancelled = self.cancelled
              self.post_event(Finished(measurements, was_cancelled))
      self.analysis_id = False
  
imports: import collections
  import io
  import logging
  import os
  import queue
  import subprocess
  import sys
  import tempfile
  import threading
  import re
  import numpy
  import psutil
  from .event import Finished
  from .event import Paused
  from .event import Progress
  from .event import Resumed
  from .event import Started
  from .None import reply
  from .None import request
  from ..image import ImageSetList
  from ..measurement import Measurements
  from ..utilities.measurement import load_measurements_from_buffer
  from ..pipeline import dump
  from ..preferences import get_plugin_directory
  from ..preferences import get_conserve_memory
  from ..preferences import get_temporary_directory
  from ..preferences import get_always_continue
  from ..preferences import preferences_as_dict
  from ..utilities.analysis import close_all_on_exec, start_daemon_thread
  from ..utilities.analysis import find_analysis_worker_source
  from ..utilities.analysis import find_python
  from ..utilities.analysis import find_worker_env
  from ..utilities.zmq import start_boundary, Boundary
  from ..utilities.zmq.communicable.reply import Reply
  from ..workspace import Workspace
label_desc: Use image node_id 10 for calling the Runner obj's underlying member method code with example usage: obj.interface(start_signal, image_set_start, image_set_end, overwrite) without return types