class_docstr: An analysis worker processing work at a given address
  
      
class_docstr_tok: ["An","analysis","worker","processing","work","at","a","given","address"]
docstr: Handle a work request to its completion
  
  job - request.Work
docstr_tok: ["Handle","a","work","request","to","its","completion","job","-","request.Work"]
code: def do_job(self, job):
      """Handle a work request to its completion
  
      job - request.Work
      """
      import cellprofiler_core.pipeline as cpp
  
      job_measurements = []
      try:
          send_dictionary = job.wants_dictionary
  
          LOGGER.info("Starting job")
          # Fetch the pipeline and preferences for this analysis if we don't have it
          current_pipeline = self.pipeline
          current_preferences = self.preferences
          if not current_pipeline:
              LOGGER.debug("Fetching pipeline and preferences")
              rep = self.send(
                  PipelinePreferences(self.current_analysis_id)
              )
              LOGGER.debug("Received pipeline and preferences response")
              preferences_dict = rep.preferences
              # update preferences to match remote values
              set_preferences_from_dict(preferences_dict)
  
              LOGGER.debug("Loading pipeline")
  
              current_pipeline = cpp.Pipeline()
              pipeline_chunks = rep.pipeline_blob.tolist()
              pipeline_io = io.StringIO("".join(pipeline_chunks))
              current_pipeline.loadtxt(pipeline_io, raise_on_error=True)
  
              LOGGER.debug("Pipeline loaded")
              current_pipeline.add_listener(
                  self.pipeline_listener.handle_event
              )
              current_preferences = rep.preferences
              self.pipeline = current_pipeline
              self.pipeline.calculate_last_image_uses()
              self.preferences = current_preferences
          else:
              # update preferences to match remote values
              set_preferences_from_dict(current_preferences)
  
          # Reset the listener's state
          self.pipeline_listener.reset()
          LOGGER.debug("Getting initial measurements")
          # Fetch the path to the intial measurements if needed.
  
          if self.initial_measurements is None:
              LOGGER.debug("Sending initial measurements request")
              rep = self.send(
                  InitialMeasurements(self.current_analysis_id)
              )
              LOGGER.debug("Got initial measurements")
              self.initial_measurements = load_measurements_from_buffer(
                  rep.buf
              )
          else:
              LOGGER.debug("Has initial measurements")
          # Make a copy of the measurements for writing during this job
          current_measurements = Measurements(
              copy=self.initial_measurements
          )
          all_measurements.add(current_measurements)
          job_measurements.append(current_measurements)
  
          successful_image_set_numbers = []
          image_set_numbers = job.image_set_numbers
          worker_runs_post_group = job.worker_runs_post_group
          LOGGER.info(
              "Doing job: " + ",".join(map(str, image_set_numbers))
          )
  
          self.pipeline_listener.image_set_number = image_set_numbers[0]
  
          if not worker_runs_post_group:
              # Get the shared state from the first imageset in this run.
              shared_dicts = self.send(
                  SharedDictionary(self.current_analysis_id)
              ).dictionaries
              assert len(shared_dicts) == len(
                  current_pipeline.modules()
              )
              for module, new_dict in zip(
                  current_pipeline.modules(), shared_dicts
              ):
                  module.set_dictionary_for_worker(new_dict)
  
          # Run prepare group if this is the first image in the group.  We do
          # this here (even if there's no grouping in the pipeline) to ensure
          # that any changes to the modules' shared state dictionaries get
          # propagated correctly.
          should_process = True
          if (
              current_measurements[
                  "Image", "Group_Index", image_set_numbers[0]
              ]
              == 1
          ):
              workspace = Workspace(
                  current_pipeline,
                  None,
                  None,
                  None,
                  current_measurements,
                  None,
                  None,
              )
              if not current_pipeline.prepare_group(
                  workspace,
                  current_measurements.get_grouping_keys(),
                  image_set_numbers,
              ):
                  # exception handled elsewhere, possibly cancelling this run.
                  should_process = False
              del workspace
  
          # process the images
          if should_process:
              abort = False
              for image_set_number in image_set_numbers:
                  try:
                      self.pipeline_listener.image_set_number = (
                          image_set_number
                      )
                      last_workspace = current_pipeline.run_image_set(
                          current_measurements,
                          image_set_number,
                          self.interaction_handler,
                          self.display_handler,
                          self.cancel_handler,
                      )
                      if self.pipeline_listener.should_abort:
                          abort = True
                          break
                      elif self.pipeline_listener.should_skip:
                          # Report skipped image sets as successful so that
                          # analysis can complete.
                          # Report their measurements because some modules
                          # may have provided measurements before skipping.
                          pass
                      successful_image_set_numbers.append(
                          image_set_number
                      )
                      # Send an indication that the image set finished successfully.
                      if send_dictionary:
                          # The jobserver would like a copy of our modules'
                          # run_state dictionaries.
                          dicts = [
                              m.get_dictionary_for_worker()
                              for m in current_pipeline.modules()
                          ]
                          req = ImageSetSuccessWithDictionary(
                              self.current_analysis_id,
                              image_set_number=image_set_number,
                              shared_dicts=dicts,
                          )
                      else:
                          req = ImageSetSuccess(
                              self.current_analysis_id,
                              image_set_number=image_set_number,
                          )
                      rep = self.send(req)
                  except CancelledException:
                      LOGGER.info("Aborting job after cancellation")
                      abort = True
                  except Exception as e:
                      try:
                          LOGGER.error(
                              "Error in pipeline", exc_info=True
                          )
                          if (
                              self.handle_exception(
                                  image_set_number=image_set_number
                              )
                              == ED_STOP
                          ):
                              abort = True
                              break
                      except:
                          LOGGER.error(
                              "Error in handling of pipeline exception",
                              exc_info=True,
                          )
                          # this is bad.  We can't handle nested exceptions
                          # remotely so we just fail on this run.
                          abort = True
  
              if abort:
                  current_measurements.close()
                  job_measurements.remove(current_measurements)
                  return
  
              if worker_runs_post_group:
                  last_workspace.interaction_handler = (
                      self.interaction_handler
                  )
                  last_workspace.cancel_handler = self.cancel_handler
                  last_workspace.post_group_display_handler = (
                      self.post_group_display_handler
                  )
                  # There might be an exception in this call, but it will be
                  # handled elsewhere, and there's nothing we can do for it
                  # here.
                  current_pipeline.post_group(
                      last_workspace,
                      current_measurements.get_grouping_keys(),
                  )
                  del last_workspace
  
          # send measurements back to server
          req = MeasurementsReport(
              self.current_analysis_id,
              buf=current_measurements.file_contents(),
              image_set_numbers=image_set_numbers,
          )
          rep = self.send(req)
  
      except CancelledException:
          # Main thread received shutdown signal
          raise
  
      except Exception:
          LOGGER.error("Error in worker", exc_info=True)
          if self.handle_exception() == ED_STOP:
              raise CancelledException(
                  "Cancelling after user-requested stop"
              )
      finally:
          # Clean up any measurements owned by us
          for m in job_measurements:
              m.close()
  
imports: import io
  import logging
  import sys
  import time
  import traceback
  import zmq
  from ._pipeline_event_listener import PipelineEventListener
  from ..analysis.reply import ImageSetSuccess, ServerExited
  from ..analysis.reply import ImageSetSuccessWithDictionary
  from ..analysis.reply import NoWork
  from ..analysis.request import AnalysisCancel, Display, DisplayPostGroup, OmeroLogin
  from ..analysis.request import DebugComplete
  from ..analysis.request import DebugWaiting
  from ..analysis.request import ExceptionReport
  from ..analysis.request import InitialMeasurements
  from ..analysis.request import Interaction
  from ..analysis.request import MeasurementsReport
  from ..analysis.request import PipelinePreferences
  from ..analysis.request import SharedDictionary
  from ..analysis.request import Work
  from ..constants.worker import ED_STOP
  from ..constants.worker import NOTIFY_STOP
  from ..constants.worker import all_measurements
  from ..measurement import Measurements
  from ..utilities.measurement import load_measurements_from_buffer
  from ..pipeline import CancelledException
  from ..preferences import get_awt_headless
  from ..preferences import set_preferences_from_dict
  from ..utilities.zmq.communicable.reply.upstream_exit import UpstreamExit
  from ..workspace import Workspace
label_desc: Use image node_id 7 for calling the Worker obj's underlying member method code with example usage: obj.do_job(job) without return types