Skip to content

changers#

zen_api_utils.changers #

Functions:

get_filter_by_position #

get_filter_by_position(filters: FilterWheelServiceGetFiltersResponse, position: int) -> FilterData | None

Return the filter that matches the given position from the filters list. Args: filters (FilterWheelServiceGetFiltersResponse): An object containing a list of filters. position (int): The position of the filter to retrieve. Returns: FilterData | None: The filter at the specified position if found, otherwise None.

get_used_filterwheel_positions #

get_used_filterwheel_positions(filters: FilterWheelServiceGetFiltersResponse) -> List[int]

Generates a list of used filter positions from a collection of filter data. Args: filters (FilterWheelServiceGetFiltersResponse): An object containing a list of filters with their respective positions. Returns: List[int]: A list of all positions currently in use by the filters.

get_reflector_by_position #

get_reflector_by_position(reflectors: ReflectorChangerServiceGetReflectorsResponse, position: int) -> ReflectorData | None

Return the reflector that matches the given position from the reflectors list. Args: reflectors (ReflectorChangerServiceGetReflectorsResponse): An object containing a list of reflectors. position (int): The position of the reflector to retrieve. Returns: ReflectorData | None: The reflector at the specified position if found, otherwise None.

get_used_reflector_positions #

get_used_reflector_positions(reflectors: ReflectorChangerServiceGetReflectorsResponse) -> List[int]

Generates a list of used reflector positions from a collection of reflector data. Args: reflectors (ReflectorChangerServiceGetReflectorsResponse): An object containing a list of reflectors with their respective positions. Returns: List[int]: A list of all positions currently in use by the reflectors.

get_objective_by_position #

get_objective_by_position(objectives: ObjectiveChangerServiceGetObjectivesResponse, position: int) -> ObjectiveData | None

Return the objective that matches the given position from the objectives list. Args: objectives (ObjectiveChangerServiceGetObjectivesResponse): An object containing a list of objectives. position (int): The position of the objective to retrieve. Returns: ObjectiveData | None: The objective at the specified position if found, otherwise None.

get_optovar_by_position #

get_optovar_by_position(optovars: OptovarServiceGetOptovarsResponse, position: int) -> OptovarData | None

Return the optovar that matches the given position from the objectives list. Args: objectives (OptovarServiceGetOptovarsResponse): An object containing a list of optovars. position (int): The position of the optovar to retrieve. Returns: OptovarData | None: The optovar at the specified position if found, otherwise None.

get_used_objective_positions #

get_used_objective_positions(objectives: ObjectiveChangerServiceGetObjectivesResponse) -> List[int]

Generates a list of used objective positions from a collection of objective data. Args: objectives (ObjectiveChangerServiceGetObjectivesResponse): An object containing a list of objectives with their respective positions. Returns: List[int]: A list of all positions currently in use by the objectives.

get_used_optovar_positions #

get_used_optovar_positions(optovars: OptovarServiceGetOptovarsResponse) -> List[int]

Generates a list of used optovar positions from a collection of optovar data. Args: objectives (OptovarServiceGetOptovarsResponse): An object containing a list of optovars with their respective positions. Returns: List[int]: A list of all positions currently in use by the optovars.

move_changers async #

move_changers(channel: Channel, metadata: list[tuple[str, str]], objectives: ObjectiveChangerServiceGetObjectivesResponse | None = None, optovars: OptovarServiceGetOptovarsResponse | None = None, reflectors: ReflectorChangerServiceGetReflectorsResponse | None = None, filters: FilterWheelServiceGetFiltersResponse | None = None, objective_position: int | None = None, optovar_position: int | None = None, reflector_position: int | None = None, filterwheel_position: int | None = None) -> dict[str, Any]

Move microscope changers to specified positions based on keyword arguments. This function is robust and handles situations where not all services are available.

Parameters:

  • channel (Channel) –

    gRPC channel for communication

  • metadata (list[tuple[str, str]]) –

    gRPC metadata for authentication

  • objectives (ObjectiveChangerServiceGetObjectivesResponse | None, default: None ) –

    Available objectives data (optional, will be fetched if service available)

  • optovars (OptovarServiceGetOptovarsResponse | None, default: None ) –

    Available optovars data (optional, will be fetched if service available)

  • reflectors (ReflectorChangerServiceGetReflectorsResponse | None, default: None ) –

    Available reflectors data (optional, will be fetched if service available)

  • filters (FilterWheelServiceGetFiltersResponse | None, default: None ) –

    Available filters data (optional, will be fetched if service available)

  • objective_position (int, default: None ) –

    Target position for objective changer

  • optovar_position (int, default: None ) –

    Target position for optovar changer

  • reflector_position (int, default: None ) –

    Target position for reflector changer

  • filterwheel_position (int, default: None ) –

    Target position for filterwheel changer

Returns:

  • dict ( dict[str, Any] ) –

    Dictionary containing the results of each movement operation with current positions. Includes 'skipped' entries for unavailable services and 'warnings' for service failures.

Note
  • If a service cannot be initialized, the operation will be skipped with a warning
  • Only successfully initialized services will be used for movement operations
  • The function continues execution even if some services fail to initialize
Example

This will attempt to move all requested changers, skipping unavailable services#

results = await move_changers( ... channel=channel, ... metadata=metadata, ... objective_position=2, ... optovar_position=1, ... reflector_position=3, ... filterwheel_position=2 ... ) print(f"Successful moves: {[k for k, v in results.items() if v and v.get('status') == 'success']}") print(f"Skipped services: {results['skipped']}")

move_changers_adv async #

move_changers_adv(objchanger_service: ObjectiveChangerServiceStub | None = None, optovar_service: OptovarServiceStub | None = None, reflector_service: ReflectorChangerServiceStub | None = None, filterwheel_service: FilterWheelServiceStub | None = None, objectives: ObjectiveChangerServiceGetObjectivesResponse | None = None, optovars: OptovarServiceGetOptovarsResponse | None = None, reflectors: ReflectorChangerServiceGetReflectorsResponse | None = None, filters: FilterWheelServiceGetFiltersResponse | None = None, objective_position: int | None = None, optovar_position: int | None = None, reflector_position: int | None = None, filterwheel_position: int | None = None) -> dict[str, Any]

Advanced version of move_changers that accepts service stubs directly instead of channel & metadata. This is more efficient when you already have service stubs initialized and want to avoid redundant service stub creation. The function is robust and handles situations where not all services are available.

Parameters:

  • objchanger_service (ObjectiveChangerServiceStub, default: None ) –

    Pre-initialized objective changer service

  • optovar_service (OptovarServiceStub, default: None ) –

    Pre-initialized optovar service

  • reflector_service (ReflectorChangerServiceStub, default: None ) –

    Pre-initialized reflector service

  • filterwheel_service (FilterWheelServiceStub, default: None ) –

    Pre-initialized filterwheel service

  • objectives (ObjectiveChangerServiceGetObjectivesResponse | None, default: None ) –

    Available objectives data (optional, will be fetched if None and service available)

  • optovars (OptovarServiceGetOptovarsResponse | None, default: None ) –

    Available optovars data (optional, will be fetched if None and service available)

  • reflectors (ReflectorChangerServiceGetReflectorsResponse | None, default: None ) –

    Available reflectors data (optional, will be fetched if None and service available)

  • filters (FilterWheelServiceGetFiltersResponse | None, default: None ) –

    Available filters data (optional, will be fetched if None and service available)

  • objective_position (int, default: None ) –

    Target position for objective changer

  • optovar_position (int, default: None ) –

    Target position for optovar changer

  • reflector_position (int, default: None ) –

    Target position for reflector changer

  • filterwheel_position (int, default: None ) –

    Target position for filterwheel changer

Returns:

  • dict ( dict[str, Any] ) –

    Dictionary containing the results of each movement operation with current positions. Includes 'skipped' entries for unavailable services and 'warnings' for missing services.

Note
  • If a position is specified but the corresponding service is not available, the operation will be skipped
  • Only available services will be used for movement operations
  • The function continues execution even if some services are unavailable
Example

Only objective and optovar services available#

objchanger_service = ObjectiveChangerServiceStub(channel=channel, metadata=metadata) optovar_service = OptovarServiceStub(channel=channel, metadata=metadata)

This will move objective and optovar, skip reflector and filterwheel#

results = await move_changers_adv( ... objchanger_service=objchanger_service, ... optovar_service=optovar_service, ... objective_position=2, ... optovar_position=1, ... reflector_position=3, # Will be skipped (no service) ... filterwheel_position=2 # Will be skipped (no service) ... )