DropsDriver

myClient

Source code in dod/DropsDriver.py

class myClient:
  def __init__(self, ip, port, supported_json="supported.json", reload=True, queue=None, **kwargs):
    # dto pipelines
    self.__queue__ = Queue()
    self.__queue_ready__ = Semaphore(value=0)
    # configure connection object
    self.__IP__ = ip
    self.__PORT__ = port
    self.conn = HTTPConnection(host=self.__IP__, port=self.__PORT__)
    self.transceiver = HTTPTransceiver(self.conn, self.__queue__, self.__queue_ready__)
    logger.info(f"Connected to ip: {ip} port: {port}")

    # configuration persitence, updating
    self.supported_ends_handler = SupportedEndsHandler(supported_json,self.conn)
    if (reload):
      self.supported_ends_handler.reload_all()

    # convinient member lambda for grabbing supported endpoitns
    self.supported_ends = lambda : self.supported_ends_handler.get_endpoints()
    pprint.pprint(self.supported_ends())

  def __del__(self):
      # close network connection
      self.conn.close()

  '''
  Middleware interaction defintions
  Here we define specific interactions that the machine is capable of fielding
  These actions are exposed as member functions of this driver class. 
  '''

  def middle_invocation_wrapper(func):
    '''
    Define a decorator function to adorn all of these 'high' middleware calls
    Logs what functions is being called and returns the response. 
    '''
    def inner(self, *args):
      logger.info(f"Invoking {func.__name__}")
      func(self, *args)
      resp = self.get_response()
      # TODO: Do something with None response globally
      # Check if we have a GUI (get_status) that needs to be cleared, May need
      # to happen per function basis
      return resp
    return inner

  @middle_invocation_wrapper
  def connect(self, user : str):
      """
        Required to send 'Do' requests
      """
      self.send(f"/DoD/Connect?ClientName={user}")

  @middle_invocation_wrapper
  def disconnect(self):
      """
        The client can end the connection with access to ‘Do’ requests.
        Clicking the button ‘Disable API Control’ on the UI has the same effect.
      """
      self.send("/DoD/Disconnect")

  @middle_invocation_wrapper
  def get_status(self):
      """
        Returns Robot Status
      """
      self.send("/DoD/get/Status")


  @middle_invocation_wrapper
  def move(self, position : str):
      """
        Moves the drive to a position taken from the list of 'PositionNames'
      """
      self.send(f"/DoD/do/Move?PositionName={position}")

  @middle_invocation_wrapper
  def get_position_names(self):
      """
        Returns the list of positions in DOD robot
      """
      self.send(f"/DoD/get/PositionNames")

  @middle_invocation_wrapper
  def get_task_names(self):
      """
        Returns the list of tasks that are stored in the Robot by name
      """
      self.send(f"/DoD/get/TaskNames")

  @middle_invocation_wrapper
  def get_current_positions(self):
      """
        Returns the name and properties of the last selected position,
        together with the real current position coordinates.
        (The drives can have been stepped away from the stored position or 
        they include small dispenser related offsets.)
      """
      self.send(f"/DoD/get/CurrentPosition")

  @middle_invocation_wrapper
  def execute_task(self, value : str):
      """
        Runs a task from the list of ‘TaskName’.
        This operation is safe in general.
        It simulates the analog action on the UI.
      """
      self.send(f"/DoD/do/ExecuteTask?TaskName={value}")

  @middle_invocation_wrapper
  def auto_drop(self):
      """
        Runs the particular task that is linked to the UI button.
        Its name is ‘AutoDropDetection’. In principalm this endpoint is not needed.
        ‘ExecuteTask’ can be used instead.
      """
      self.send(f"/DoD/do/AutoDrop")

  @middle_invocation_wrapper
  def move_to_interaction_point(self):
      """
        The moving to the predefined position of the interaction point corresponds
        to the use of the endpoint ‘Move’. But only with this endpoint the UI
        elemts for the dispensers’ position adjustment become visible on the UI.
        The request simulates the button (beam simbol) on the UI.
      """
      self.send(f"/DoD/do/InteractionPoint")

  @middle_invocation_wrapper
  def move_x(self, value : float):
      """
        The X drive can be sent to any coordinate (the value’s unit is µm)
        within the allowed range.

        NOTE: This does not include a Z move up to the safe height nor any
        other safety feature checking whether the move from the current position
        to the selected coordinate can lead to collision or breaking of a
        dispenser Tip.
      """
      self.send(f"/DoD/do/MoveX?X={value}")

    ## API V2
  @middle_invocation_wrapper
  def get_pulse_names(self):
      """
        Returns the list of available pulse shapes for the sciPULSE channels.
      """
      self.send(f"/DoD/get/PulseNames")

  @middle_invocation_wrapper
  def get_nozzle_status(self):
      """
        Returns the activated and selected nozzles an the parameters for all
        activated nozzles. The parameter ‘Trigger’ (true/false) is not linked
        to a nozzle.

        Note: Nozzle parameters
        ‘ID’ (number),
        ‘Volt’,
        ‘Pulse’ (name or number),
        ‘Freq’ ,
        ‘Volume’ appear as an array of strings (JSON).
      """
      self.send("/DoD/get/NozzleStatus")

  @middle_invocation_wrapper
  def select_nozzle(self, channel: str):
      """
        Set the selected nozzle for dispensing and task execution etc.
        Returns a reject if the channel value is not one of the
        ‘Activated Nozzles’ (see ‘NozzleStatus’).
      """
      self.send(f"/DoD/do/SelectNozzle?Channel={channel}")


  @middle_invocation_wrapper
  def dispensing(self, state: str):
      """
        Switches between the dispensing states
        ‘Trigger’ (includes ‘Stat Continuous Dispensing’),
        ‘Free’ (‘Continuous Dispensing’ without trigger) and
        ‘Off’. Returns a reject if the value is not one of the three strings.

        (Some tasks can set the state to ‘Off’ without restarting dispensing afterwards.)
      """
      self.send(f"/DoD/do/Dispensing?State={state}")


  @middle_invocation_wrapper
  def setLED(self, duration: int, delay : int):
      """
      Sets the two strobe LED parameters ‘Delay’ (0 to 6500) and Duration (1 to 65000).
      Returns a reject if one of the values is out of range.
      """
      self.send(f"/DoD/do/SetLED?Duration={duration}&Delay={delay}")

  @middle_invocation_wrapper
  def move_y(self, value : float):
      """
          Same as for Y
      """
      #TODO: handle Dialog, Caputure None, Check if dialog and clear
      self.send(f"/DoD/do/MoveY?Y={value}")

  @middle_invocation_wrapper
  def move_z(self, value : float):
      """
          Same as for Z
      """
      self.send(f"/DoD/do/MoveZ?Z={value}")

  @middle_invocation_wrapper
  def take_probe(self, channel : int, probe_well : str, volume : float):
      """
      This endpoint requires the presence of the task ‘ProbeUptake’ (attached).
      If that is not given, the return is not a reject, but nothing happens.
      The parameters are

        ‘Channel’ (number of nozzle, includes effect as ‘SelectNozzle’),
        ‘ProbeWell’ (e.g. A1), Volume (µL). Returns a reject  if ‘Channel’ is not among
        ‘Active Nozzles’, Volume is > 250 or
        ‘ProbeWell’ is not one of the allowed wells for the selected nozzle.

      """
      self.send(f"/DoD/do/TakeProbe?Channel={channel}&ProbeWell={probe_well}&Volume={volume}")

## API V3 ##

  @middle_invocation_wrapper
  def get_task_details(self, task_name):
    '''
        Returns the content of the tasks that is specified be ‘TaskName’.
    '''
    self.send(f"/DoD/get/TaskDetails?TaskName={task_name}")

  @middle_invocation_wrapper
  def get_drive_range(self):
    '''
      Returns the maximum range of each axis (X,Y,Z) in units of µm.
    '''
    self.send(f"/DoD/get/DriveRange")

  @middle_invocation_wrapper
  def set_nozzle_parameters(self,
                            active_nozzles : str,
                            selected_nozzles: str,
                            volts : int,
                            pulse : str,
                            frequency : int):
    '''
      Sets the list of activate nozzles and of the selected nozzles for
        operations that work with more than one nozzle.
      (This overwrites and is overwritten by SelectNozzle, which only sets one nozzle channel.)
      Both values are strings with commas separating the channel numbers (e.g. “1,2,3”).
      ‘Volt’ and ‘Freq’ are integers.
      ‘Pulse’ is read as a string. It is either a integer number or the pulse
        shape name for sciPULSE channels (from PulseNames).

    '''

    self.send(f"/DoD/do/SetNozzleParameters?Active={active_nozzles}&Selected={selected_nozzles}&Volt={volts}&Pulse={pulse}&Freq={frequency}")

  @middle_invocation_wrapper
  def stop_task(self):
    '''
            Stops the running task (and moves).
    '''
    self.send(f"/DoD/do/StopTask")

  @middle_invocation_wrapper
  def set_ip_offest(self):
    '''
            This endpoint uses the current coordinates for the currently selected
            nozzle (SelectNozzle) to set the IP position (Nozzle 1) or calculates
            and sets the IP offsets (Nozzles 2, …).
            (Note: The ‘Nozzle Offset’ values in the nozzle parameter table are 
            always considered.
            Changing these would require a readjustment of the IP offsets.)
            Offsets are rejected if they exceed a maximum of 2 mm.
            Thus, the selected nozzle must be moved to the IP (InteractionPoint)
            before requesting SetIPOffset.
    '''
    self.send(f"/DoD/do/InteractionPoint")

  @middle_invocation_wrapper
  def set_humidity(self, value):
    '''
        Sets the wanted relative humidity as %rH. Values are integer.
    '''
    self.send(f"/DoD/do/SetHumidity?rH={value}")

  @middle_invocation_wrapper
  def set_cooling_temp(self, temp):
    '''
            Sets the temperature of the cooling device.
            Besides the setting of a °C value (float),
                there is the option to send the string “dewpoint”,
                which enables an automatic adjustment.
    '''
    self.send(f"/DoD/do/SetCoolingTemp?Temp={temp}")

  @middle_invocation_wrapper
  def close_dialog(self, reference, selection):
    '''
            In situations of “Status” = “Dialog” the endpoint Status provides the
                dialog’s reference, message text and button labels.
            (If the response of ‘Button2’ is empty, there is only one selection 
            available, typically with the label ‘OK.).
            This endpoint allows to close the dialog by specifying the reference and t
                he selection (“1” or “2”).
            The reference is individual, incrementing integer for each occurrence
                of a dialog, it starts a “1” when the device is initialized.

            In case of more than one open dialogs “Status” reports the last one,
                which is typically the first to be closed.
            But the remote control can try to close the earlier dialog.
    '''
    self.send(f"/DoD/do/CloseDialog?Reference={reference}&Selection={selection}")

  @middle_invocation_wrapper
  def reset_error(self):
    '''
            This endpoint is needed if after closing all (error) dialogs the status 
            “Error” persists and “ErrorMessage” (header) is not “NA”.
    '''
    self.send(f"/DoD/do/ResetError")

  '''
  send transmits a formatted HTTP GET request
  it will not check the validity of request
  it will persist result in a place that can be read... TODO: actually do this
  '''
  def send(self, endpoint):
    self.transceiver.send(endpoint)
    return

  '''
  Pops most recent response from response queue
  '''
  def get_response(self):
    return self.transceiver.get_response()

auto_drop()

Runs the particular task that is linked to the UI button. Its name is ‘AutoDropDetection’. In principalm this endpoint is not needed. ‘ExecuteTask’ can be used instead.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def auto_drop(self):
    """
      Runs the particular task that is linked to the UI button.
      Its name is ‘AutoDropDetection’. In principalm this endpoint is not needed.
      ‘ExecuteTask’ can be used instead.
    """
    self.send(f"/DoD/do/AutoDrop")

close_dialog(reference, selection)

In situations of “Status” = “Dialog” the endpoint Status provides the dialog’s reference, message text and button labels. (If the response of ‘Button2’ is empty, there is only one selection available, typically with the label ‘OK.). This endpoint allows to close the dialog by specifying the reference and t he selection (“1” or “2”). The reference is individual, incrementing integer for each occurrence of a dialog, it starts a “1” when the device is initialized.

In case of more than one open dialogs “Status” reports the last one, which is typically the first to be closed. But the remote control can try to close the earlier dialog.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def close_dialog(self, reference, selection):
  '''
          In situations of “Status” = “Dialog” the endpoint Status provides the
              dialog’s reference, message text and button labels.
          (If the response of ‘Button2’ is empty, there is only one selection 
          available, typically with the label ‘OK.).
          This endpoint allows to close the dialog by specifying the reference and t
              he selection (“1” or “2”).
          The reference is individual, incrementing integer for each occurrence
              of a dialog, it starts a “1” when the device is initialized.

          In case of more than one open dialogs “Status” reports the last one,
              which is typically the first to be closed.
          But the remote control can try to close the earlier dialog.
  '''
  self.send(f"/DoD/do/CloseDialog?Reference={reference}&Selection={selection}")

connect(user)

Required to send 'Do' requests

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def connect(self, user : str):
    """
      Required to send 'Do' requests
    """
    self.send(f"/DoD/Connect?ClientName={user}")

disconnect()

The client can end the connection with access to ‘Do’ requests. Clicking the button ‘Disable API Control’ on the UI has the same effect.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def disconnect(self):
    """
      The client can end the connection with access to ‘Do’ requests.
      Clicking the button ‘Disable API Control’ on the UI has the same effect.
    """
    self.send("/DoD/Disconnect")

dispensing(state)

Switches between the dispensing states ‘Trigger’ (includes ‘Stat Continuous Dispensing’), ‘Free’ (‘Continuous Dispensing’ without trigger) and ‘Off’. Returns a reject if the value is not one of the three strings.

(Some tasks can set the state to ‘Off’ without restarting dispensing afterwards.)

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def dispensing(self, state: str):
    """
      Switches between the dispensing states
      ‘Trigger’ (includes ‘Stat Continuous Dispensing’),
      ‘Free’ (‘Continuous Dispensing’ without trigger) and
      ‘Off’. Returns a reject if the value is not one of the three strings.

      (Some tasks can set the state to ‘Off’ without restarting dispensing afterwards.)
    """
    self.send(f"/DoD/do/Dispensing?State={state}")

execute_task(value)

Runs a task from the list of ‘TaskName’. This operation is safe in general. It simulates the analog action on the UI.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def execute_task(self, value : str):
    """
      Runs a task from the list of ‘TaskName’.
      This operation is safe in general.
      It simulates the analog action on the UI.
    """
    self.send(f"/DoD/do/ExecuteTask?TaskName={value}")

get_current_positions()

Returns the name and properties of the last selected position, together with the real current position coordinates. (The drives can have been stepped away from the stored position or they include small dispenser related offsets.)

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_current_positions(self):
    """
      Returns the name and properties of the last selected position,
      together with the real current position coordinates.
      (The drives can have been stepped away from the stored position or 
      they include small dispenser related offsets.)
    """
    self.send(f"/DoD/get/CurrentPosition")

get_drive_range()

Returns the maximum range of each axis (X,Y,Z) in units of µm.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_drive_range(self):
  '''
    Returns the maximum range of each axis (X,Y,Z) in units of µm.
  '''
  self.send(f"/DoD/get/DriveRange")

get_nozzle_status()

Returns the activated and selected nozzles an the parameters for all activated nozzles. The parameter ‘Trigger’ (true/false) is not linked to a nozzle.

Note: Nozzle parameters ‘ID’ (number), ‘Volt’, ‘Pulse’ (name or number), ‘Freq’ , ‘Volume’ appear as an array of strings (JSON).

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_nozzle_status(self):
    """
      Returns the activated and selected nozzles an the parameters for all
      activated nozzles. The parameter ‘Trigger’ (true/false) is not linked
      to a nozzle.

      Note: Nozzle parameters
      ‘ID’ (number),
      ‘Volt’,
      ‘Pulse’ (name or number),
      ‘Freq’ ,
      ‘Volume’ appear as an array of strings (JSON).
    """
    self.send("/DoD/get/NozzleStatus")

get_position_names()

Returns the list of positions in DOD robot

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_position_names(self):
    """
      Returns the list of positions in DOD robot
    """
    self.send(f"/DoD/get/PositionNames")

get_pulse_names()

Returns the list of available pulse shapes for the sciPULSE channels.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_pulse_names(self):
    """
      Returns the list of available pulse shapes for the sciPULSE channels.
    """
    self.send(f"/DoD/get/PulseNames")

get_status()

Returns Robot Status

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_status(self):
    """
      Returns Robot Status
    """
    self.send("/DoD/get/Status")

get_task_details(task_name)

Returns the content of the tasks that is specified be ‘TaskName’.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_task_details(self, task_name):
  '''
      Returns the content of the tasks that is specified be ‘TaskName’.
  '''
  self.send(f"/DoD/get/TaskDetails?TaskName={task_name}")

get_task_names()

Returns the list of tasks that are stored in the Robot by name

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def get_task_names(self):
    """
      Returns the list of tasks that are stored in the Robot by name
    """
    self.send(f"/DoD/get/TaskNames")

middle_invocation_wrapper(func)

Define a decorator function to adorn all of these 'high' middleware calls Logs what functions is being called and returns the response.

Source code in dod/DropsDriver.py

def middle_invocation_wrapper(func):
  '''
  Define a decorator function to adorn all of these 'high' middleware calls
  Logs what functions is being called and returns the response. 
  '''
  def inner(self, *args):
    logger.info(f"Invoking {func.__name__}")
    func(self, *args)
    resp = self.get_response()
    # TODO: Do something with None response globally
    # Check if we have a GUI (get_status) that needs to be cleared, May need
    # to happen per function basis
    return resp
  return inner

move(position)

Moves the drive to a position taken from the list of 'PositionNames'

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def move(self, position : str):
    """
      Moves the drive to a position taken from the list of 'PositionNames'
    """
    self.send(f"/DoD/do/Move?PositionName={position}")

move_to_interaction_point()

The moving to the predefined position of the interaction point corresponds to the use of the endpoint ‘Move’. But only with this endpoint the UI elemts for the dispensers’ position adjustment become visible on the UI. The request simulates the button (beam simbol) on the UI.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def move_to_interaction_point(self):
    """
      The moving to the predefined position of the interaction point corresponds
      to the use of the endpoint ‘Move’. But only with this endpoint the UI
      elemts for the dispensers’ position adjustment become visible on the UI.
      The request simulates the button (beam simbol) on the UI.
    """
    self.send(f"/DoD/do/InteractionPoint")

move_x(value)

The X drive can be sent to any coordinate (the value’s unit is µm) within the allowed range.

NOTE: This does not include a Z move up to the safe height nor any other safety feature checking whether the move from the current position to the selected coordinate can lead to collision or breaking of a dispenser Tip.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def move_x(self, value : float):
    """
      The X drive can be sent to any coordinate (the value’s unit is µm)
      within the allowed range.

      NOTE: This does not include a Z move up to the safe height nor any
      other safety feature checking whether the move from the current position
      to the selected coordinate can lead to collision or breaking of a
      dispenser Tip.
    """
    self.send(f"/DoD/do/MoveX?X={value}")

move_y(value)

Same as for Y

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def move_y(self, value : float):
    """
        Same as for Y
    """
    #TODO: handle Dialog, Caputure None, Check if dialog and clear
    self.send(f"/DoD/do/MoveY?Y={value}")

move_z(value)

Same as for Z

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def move_z(self, value : float):
    """
        Same as for Z
    """
    self.send(f"/DoD/do/MoveZ?Z={value}")

reset_error()

This endpoint is needed if after closing all (error) dialogs the status “Error” persists and “ErrorMessage” (header) is not “NA”.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def reset_error(self):
  '''
          This endpoint is needed if after closing all (error) dialogs the status 
          “Error” persists and “ErrorMessage” (header) is not “NA”.
  '''
  self.send(f"/DoD/do/ResetError")

select_nozzle(channel)

Set the selected nozzle for dispensing and task execution etc. Returns a reject if the channel value is not one of the ‘Activated Nozzles’ (see ‘NozzleStatus’).

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def select_nozzle(self, channel: str):
    """
      Set the selected nozzle for dispensing and task execution etc.
      Returns a reject if the channel value is not one of the
      ‘Activated Nozzles’ (see ‘NozzleStatus’).
    """
    self.send(f"/DoD/do/SelectNozzle?Channel={channel}")

setLED(duration, delay)

Sets the two strobe LED parameters ‘Delay’ (0 to 6500) and Duration (1 to 65000). Returns a reject if one of the values is out of range.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def setLED(self, duration: int, delay : int):
    """
    Sets the two strobe LED parameters ‘Delay’ (0 to 6500) and Duration (1 to 65000).
    Returns a reject if one of the values is out of range.
    """
    self.send(f"/DoD/do/SetLED?Duration={duration}&Delay={delay}")

set_cooling_temp(temp)

Sets the temperature of the cooling device. Besides the setting of a °C value (float), there is the option to send the string “dewpoint”, which enables an automatic adjustment.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def set_cooling_temp(self, temp):
  '''
          Sets the temperature of the cooling device.
          Besides the setting of a °C value (float),
              there is the option to send the string “dewpoint”,
              which enables an automatic adjustment.
  '''
  self.send(f"/DoD/do/SetCoolingTemp?Temp={temp}")

set_humidity(value)

Sets the wanted relative humidity as %rH. Values are integer.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def set_humidity(self, value):
  '''
      Sets the wanted relative humidity as %rH. Values are integer.
  '''
  self.send(f"/DoD/do/SetHumidity?rH={value}")

set_ip_offest()

This endpoint uses the current coordinates for the currently selected nozzle (SelectNozzle) to set the IP position (Nozzle 1) or calculates and sets the IP offsets (Nozzles 2, …). (Note: The ‘Nozzle Offset’ values in the nozzle parameter table are always considered. Changing these would require a readjustment of the IP offsets.) Offsets are rejected if they exceed a maximum of 2 mm. Thus, the selected nozzle must be moved to the IP (InteractionPoint) before requesting SetIPOffset.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def set_ip_offest(self):
  '''
          This endpoint uses the current coordinates for the currently selected
          nozzle (SelectNozzle) to set the IP position (Nozzle 1) or calculates
          and sets the IP offsets (Nozzles 2, …).
          (Note: The ‘Nozzle Offset’ values in the nozzle parameter table are 
          always considered.
          Changing these would require a readjustment of the IP offsets.)
          Offsets are rejected if they exceed a maximum of 2 mm.
          Thus, the selected nozzle must be moved to the IP (InteractionPoint)
          before requesting SetIPOffset.
  '''
  self.send(f"/DoD/do/InteractionPoint")

set_nozzle_parameters(active_nozzles, selected_nozzles, volts, pulse, frequency)

Sets the list of activate nozzles and of the selected nozzles for operations that work with more than one nozzle. (This overwrites and is overwritten by SelectNozzle, which only sets one nozzle channel.) Both values are strings with commas separating the channel numbers (e.g. “1,2,3”). ‘Volt’ and ‘Freq’ are integers. ‘Pulse’ is read as a string. It is either a integer number or the pulse shape name for sciPULSE channels (from PulseNames).

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def set_nozzle_parameters(self,
                          active_nozzles : str,
                          selected_nozzles: str,
                          volts : int,
                          pulse : str,
                          frequency : int):
  '''
    Sets the list of activate nozzles and of the selected nozzles for
      operations that work with more than one nozzle.
    (This overwrites and is overwritten by SelectNozzle, which only sets one nozzle channel.)
    Both values are strings with commas separating the channel numbers (e.g. “1,2,3”).
    ‘Volt’ and ‘Freq’ are integers.
    ‘Pulse’ is read as a string. It is either a integer number or the pulse
      shape name for sciPULSE channels (from PulseNames).

  '''

  self.send(f"/DoD/do/SetNozzleParameters?Active={active_nozzles}&Selected={selected_nozzles}&Volt={volts}&Pulse={pulse}&Freq={frequency}")

stop_task()

Stops the running task (and moves).

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def stop_task(self):
  '''
          Stops the running task (and moves).
  '''
  self.send(f"/DoD/do/StopTask")

take_probe(channel, probe_well, volume)

This endpoint requires the presence of the task ‘ProbeUptake’ (attached). If that is not given, the return is not a reject, but nothing happens. The parameters are

‘Channel’ (number of nozzle, includes effect as ‘SelectNozzle’), ‘ProbeWell’ (e.g. A1), Volume (µL). Returns a reject if ‘Channel’ is not among ‘Active Nozzles’, Volume is > 250 or ‘ProbeWell’ is not one of the allowed wells for the selected nozzle.

Source code in dod/DropsDriver.py

@middle_invocation_wrapper
def take_probe(self, channel : int, probe_well : str, volume : float):
    """
    This endpoint requires the presence of the task ‘ProbeUptake’ (attached).
    If that is not given, the return is not a reject, but nothing happens.
    The parameters are

      ‘Channel’ (number of nozzle, includes effect as ‘SelectNozzle’),
      ‘ProbeWell’ (e.g. A1), Volume (µL). Returns a reject  if ‘Channel’ is not among
      ‘Active Nozzles’, Volume is > 250 or
      ‘ProbeWell’ is not one of the allowed wells for the selected nozzle.

    """
    self.send(f"/DoD/do/TakeProbe?Channel={channel}&ProbeWell={probe_well}&Volume={volume}")