class_docstr: The server maintains the port and hands off the requests to workers
  
  example of use:
  
  context = zmq.Context()
  with KnimeBridgeServer(
      context,
      "tcp://127.0.0.1:4000", # server address
      "inproc://Notify        # kill notify address
      ) as server:
      notify_socket = context.socket(zmq.PUB)
      notify_socket.bind("inproc://Notify")
      ....
      notify_socket.send("Stop")
class_docstr_tok: ["The","server","maintains","the","port","and","hands","off","the","requests","to","workers","example","of","use",":","context","=","zmq.Context","(",")","with","KnimeBridgeServer","(","context",",","``","tcp",":","\/\/127.0.0.1:4000","''",",","#","server","address","``","inproc",":","\/\/Notify","#","kill","notify","address",")","as","server",":","notify_socket","=","context.socket","(","zmq.PUB",")","notify_socket.bind","(","``","inproc",":","\/\/Notify","''",")","....","notify_socket.send","(","``","Stop","''",")"]
docstr: Handle a run-group request message
docstr_tok: ["Handle","a","run-group","request","message"]
code: def run_group_request(self, session_id, message_type, message):
      """Handle a run-group request message"""
      pipeline = cellprofiler_core.pipeline.Pipeline()
      m = Measurements()
      image_group = m.hdf5_dict.hdf5_file.create_group("ImageData")
      if len(message) < 2:
          self.raise_cellprofiler_exception(
              session_id, "Missing run request sections"
          )
          return
      pipeline_txt = message.pop(0).bytes
      image_metadata = message.pop(0).bytes
      n_image_sets = None
      try:
          image_metadata = json.loads(image_metadata)
          channel_names = []
          for channel_name, channel_metadata in image_metadata:
              channel_names.append(channel_name)
              if len(message) < 1:
                  self.raise_cellprofiler_exception(
                      session_id,
                      "Missing binary data for channel %s"
                      % channel_name,
                  )
                  return None, None, None
              pixel_data = self.decode_image(
                  channel_metadata,
                  message.pop(0).bytes,
                  grouping_allowed=True,
              )
              if pixel_data.ndim < 3:
                  self.raise_cellprofiler_exception(
                      session_id,
                      "The image for channel %s does not have a Z or T dimension",
                  )
                  return
              if n_image_sets is None:
                  n_image_sets = pixel_data.shape[0]
              elif n_image_sets != pixel_data.shape[0]:
                  self.raise_cellprofiler_exception(
                      session_id,
                      "The images passed have different numbers of Z or T planes",
                  )
                  return
              image_group.create_dataset(channel_name, data=pixel_data)
      except Exception as e:
          self.raise_cellprofiler_exception(session_id, e)
          return None, None, None
      try:
          pipeline.loadtxt(StringIO(pipeline_txt))
      except Exception as e:
          LOGGER.warning(
              "Failed to load pipeline: sending pipeline exception"
          )
          self.raise_pipeline_exception(session_id, str(e))
          return
  
      image_numbers = numpy.arange(1, n_image_sets + 1)
      for image_number in image_numbers:
          m[
              "Image",
              GROUP_NUMBER,
              image_number,
          ] = 1
          m[
              "Image",
              GROUP_INDEX,
              image_number,
          ] = image_number
      input_modules, other_modules = self.split_pipeline(pipeline)
      workspace = cellprofiler_core.workspace.Workspace(
          pipeline, None, m, None, m, None
      )
      LOGGER.info("Preparing group")
      for module in other_modules:
          module.prepare_group(
              workspace,
              dict([("image_number", i) for i in image_numbers]),
              image_numbers,
          )
  
      for image_index in range(n_image_sets):
          object_set = cellprofiler_core.object.ObjectSet()
          m.next_image_set(image_index + 1)
          for channel_name in channel_names:
              dataset = image_group[channel_name]
              pixel_data = dataset[image_index]
              m.add(channel_name, Image(pixel_data))
  
          for module in other_modules:
              workspace = cellprofiler_core.workspace.Workspace(
                  pipeline, module, m, object_set, m, None
              )
              try:
                  LOGGER.info(
                      "Running module # %d: %s"
                      % (module.module_num, module.module_name)
                  )
                  pipeline.run_module(module, workspace)
                  if workspace.disposition in (
                      DISPOSITION_SKIP,
                      DISPOSITION_CANCEL,
                  ):
                      break
              except Exception as e:
                  msg = (
                      'Encountered error while running module, "%s": %s'
                      % (
                          module.module_name,
                          e,
                      )
                  )
                  LOGGER.warning(msg)
                  self.raise_cellprofiler_exception(session_id, msg)
                  return
          else:
              continue
          if workspace.disposition == DISPOSITION_CANCEL:
              break
      for module in other_modules:
          module.post_group(
              workspace,
              dict([("image_number", i) for i in image_numbers]),
          )
      LOGGER.info("Finished group")
  
      type_names, feature_dict = self.find_measurements(
          other_modules, pipeline
      )
  
      double_features = []
      double_data = []
      float_features = []
      float_data = []
      int_features = []
      int_data = []
      string_features = []
      string_data = []
      metadata = [
          double_features,
          float_features,
          int_features,
          string_features,
      ]
  
      for object_name, features in list(feature_dict.items()):
          df = []
          double_features.append((object_name, df))
          ff = []
          float_features.append((object_name, ff))
          intf = []
          int_features.append((object_name, intf))
          sf = []
          string_features.append((object_name, sf))
          if object_name == "Image":
              object_counts = [] * n_image_sets
          else:
              object_numbers = m[
                  object_name,
                  OBJECT_NUMBER,
                  image_numbers,
              ]
              object_counts = [len(x) for x in object_numbers]
          for feature, data_type in features:
              if data_type == "java.lang.String":
                  continue
              if not m.has_feature(object_name, feature):
                  data = numpy.zeros(numpy.sum(object_counts))
              else:
                  data = m[object_name, feature, image_numbers]
              temp = []
              for i, (di, count) in enumerate(zip(data, object_counts)):
                  if count == 0:
                      continue
                  di = numpy.atleast_1d(di)
                  if len(di) > count:
                      di = di[:count]
                  elif len(di) == count:
                      temp.append(di)
                  else:
                      temp += [di + numpy.zeros(len(di) - count)]
              if len(temp) > 0:
                  data = numpy.hstack(temp)
  
              if type_names[data_type] == "java.lang.Double":
                  df.append((feature, len(data)))
                  if len(data) > 0:
                      double_data.append(data.astype("<f8"))
              elif type_names[data_type] == "java.lang.Float":
                  ff.append((feature, len(data)))
                  if len(data) > 0:
                      float_data.append(data.astype("<f4"))
              elif type_names[data_type] == "java.lang.Integer":
                  intf.append((feature, len(data)))
                  if len(data) > 0:
                      int_data.append(data.astype("<i4"))
      data = numpy.hstack(
          [
              numpy.frombuffer(
                  numpy.ascontiguousarray(numpy.hstack(ditem)).data,
                  numpy.uint8,
              )
              for ditem in (double_data, float_data, int_data)
              if len(ditem) > 0
          ]
      )
      data = numpy.ascontiguousarray(data)
      self.socket.send_multipart(
          [
              zmq.Frame(session_id),
              zmq.Frame(),
              zmq.Frame(RUN_REPLY_1),
              zmq.Frame(json.dumps(metadata)),
              zmq.Frame(data),
          ]
      )
  
imports: import json
  import logging
  import threading
  import uuid
  from io import StringIO
  import numpy
  import zmq
  from cellprofiler_core.constants.measurement import GROUP_NUMBER, GROUP_INDEX, OBJECT_NUMBER, EXPERIMENT, MCA_AVAILABLE_POST_RUN, COLTYPE_FLOAT, COLTYPE_INTEGER, COLTYPE_VARCHAR, IMAGE_NUMBER
  from cellprofiler_core.constants.workspace import DISPOSITION_SKIP, DISPOSITION_CANCEL
  from cellprofiler_core.image import Image
  from cellprofiler_core.measurement import Measurements
  from cellprofiler_core.module import Module
  from cellprofiler_core.setting.text import ImageName
  import cellprofiler_core.object
  import cellprofiler_core.pipeline
  import cellprofiler_core.workspace
label_desc: Use image node_id 10 for calling the KnimeBridgeServer obj's underlying member method code with example usage: obj.run_group_request(session_id, message_type, message) and returns: None,  None,  None, None,  None,  None