camacq package

Control microscope through client server program.

Subpackages

• camacq.data package
• camacq.helper package
  • ensure_dict()
  • get_module()
  • has_at_least_one_key()
  • register_signals()
  • setup_one_module()
  • Submodules
  • camacq.helper.template module
    • get_env()
    • make_template()
    • render_template()
    • template_next_well_x()
    • template_next_well_xy()
    • template_next_well_y()
• camacq.plugins package
  • get_plugins()
  • setup_module()
  • Subpackages
    • camacq.plugins.api package
      • Api
        • Api.name
        • Api.send()
        • Api.send_many()
        • Api.start_imaging()
        • Api.stop_imaging()
      • CommandEvent
        • CommandEvent.command
        • CommandEvent.event_type
      • ImageEvent
        • ImageEvent.channel_id
        • ImageEvent.event_type
        • ImageEvent.field_x
        • ImageEvent.field_y
        • ImageEvent.path
        • ImageEvent.plate_name
        • ImageEvent.well_x
        • ImageEvent.well_y
        • ImageEvent.z_slice_id
      • StartCommandEvent
        • StartCommandEvent.event_type
      • StopCommandEvent
        • StopCommandEvent.event_type
      • register_api()
      • setup_module()
      • validate_commands()
    • camacq.plugins.automations package
      • ActionSequence
        • ActionSequence.delay()
      • Automation
        • Automation.disable()
        • Automation.enable()
        • Automation.trigger()
      • TemplateAction
        • TemplateAction.render()
      • make_checker()
      • setup_module()
      • template_check()
      • Submodules
      • camacq.plugins.automations.event module
        • handle_trigger()
    • camacq.plugins.leica package
      • LeicaApi
        • LeicaApi.name
        • LeicaApi.receive()
        • LeicaApi.send()
        • LeicaApi.start_imaging()
        • LeicaApi.start_listen()
        • LeicaApi.stop_imaging()
      • LeicaCommandEvent
        • LeicaCommandEvent.command
        • LeicaCommandEvent.event_type
      • LeicaImageEvent
        • LeicaImageEvent.channel_id
        • LeicaImageEvent.event_type
        • LeicaImageEvent.field_x
        • LeicaImageEvent.field_y
        • LeicaImageEvent.job_id
        • LeicaImageEvent.path
        • LeicaImageEvent.plate_name
        • LeicaImageEvent.well_x
        • LeicaImageEvent.well_y
        • LeicaImageEvent.z_slice_id
      • LeicaStartCommandEvent
        • LeicaStartCommandEvent.event_type
      • LeicaStopCommandEvent
        • LeicaStopCommandEvent.event_type
      • setup_module()
      • Submodules
      • camacq.plugins.leica.command module
        • cam_com()
        • camstart_com()
        • camstop_com()
        • del_com()
        • enable_com()
        • gain_com()
        • start()
        • stop()
      • camacq.plugins.leica.helper module
        • find_image_path()
        • get_field()
        • get_imgs()
        • get_well()
      • camacq.plugins.leica.sample module
        • Channel
          • Channel.channel_id
          • Channel.change_event
          • Channel.images
          • Channel.name
          • Channel.values
        • ChannelEvent
          • ChannelEvent.channel_id
          • ChannelEvent.channel_name
          • ChannelEvent.event_type
        • Field
          • Field.field_x
          • Field.field_y
          • Field.change_event
          • Field.images
          • Field.name
          • Field.values
        • FieldEvent
          • FieldEvent.event_type
          • FieldEvent.field_img_ok
          • FieldEvent.field_x
          • FieldEvent.field_y
        • LeicaSample
          • LeicaSample.change_event
          • LeicaSample.image_event_type
          • LeicaSample.images
          • LeicaSample.name
          • LeicaSample.on_image()
          • LeicaSample.set_sample_schema
          • LeicaSample.values
        • LeicaSampleEvent
          • LeicaSampleEvent.event_type
        • Plate
          • Plate.plate_name
          • Plate.change_event
          • Plate.images
          • Plate.name
          • Plate.values
        • PlateEvent
          • PlateEvent.event_type
          • PlateEvent.plate_name
        • Well
          • Well.well_x
          • Well.well_y
          • Well.change_event
          • Well.images
          • Well.name
          • Well.values
        • WellEvent
          • WellEvent.event_type
          • WellEvent.well_img_ok
          • WellEvent.well_x
          • WellEvent.well_y
        • ZSlice
          • ZSlice.z_slice_id
          • ZSlice.change_event
          • ZSlice.images
          • ZSlice.name
          • ZSlice.values
        • ZSliceEvent
          • ZSliceEvent.event_type
          • ZSliceEvent.z_slice_id
        • next_well_xy()
        • setup_module()
    • camacq.plugins.sample package
      • Image
        • Image.change_event
        • Image.images
        • Image.name
        • Image.path
        • Image.values
      • ImageContainer
        • ImageContainer.change_event
        • ImageContainer.images
        • ImageContainer.name
        • ImageContainer.values
      • Sample
        • Sample.center
        • Sample.data
        • Sample.get_sample()
        • Sample.image_event_type
        • Sample.name
        • Sample.on_image()
        • Sample.set_sample()
        • Sample.set_sample_schema
      • SampleEvent
        • SampleEvent.container
        • SampleEvent.container_name
        • SampleEvent.event_type
        • SampleEvent.images
        • SampleEvent.values
      • SampleImageSetEvent
        • SampleImageSetEvent.event_type
      • Samples
      • get_matched_samples()
      • register_sample()
      • setup_module()
  • Submodules
  • camacq.plugins.rename_image module
    • rename_image()
    • setup_module()

Submodules

camacq.bootstrap module

Configure and set up control center.

async camacq.bootstrap.setup_dict(center: Center, config: dict[str, Any]) → None

Set up control center from config dict.

Parameters:

config (dict) – The config dict.

Returns:

Return the Center instance.

Return type:

Center instance

async camacq.bootstrap.setup_file(config_file: Path, cmd_args: dict[str, Any]) → Center

Set up control center from config file and command line args.

Parameters:

• config_file (str) – The path to the configuration YAML file.

• cmd_args (dict) – The dict with the command line arguments.

Returns:

Return the Center instance.

Return type:

Center instance

camacq.config module

Handle the config file.

camacq.config.create_default_config(config_dir: Path) → Path | None

Create a default config file in given configuration directory.

Parameters:

config_dir (pathlib.Path) – The path to the configuration directory.

Returns:

Return path to new configuration file if success, None if failed.

Return type:

pathlib.Path

camacq.config.ensure_config_exists(config_dir: Path) → Path | None

Ensure configuration file exists in the configuration directory.

Create a default configuration file if needed.

Parameters:

config_dir (pathlib.Path) – The path to the configuration directory.

Returns:

Return path to the configuration file.

Return type:

pathlib.Path

camacq.config.find_config_file(config_dir: Path) → Path | None

Find the configuration file in the configuration directory.

Parameters:

config_dir (pathlib.Path) – The path to the configuration directory.

Returns:

Return path to the configuration file if found, None if not found.

Return type:

pathlib.Path

camacq.config.get_default_config_dir() → Path

Get the default configuration directory based on OS.

Returns:

Return the path to the configuration directory.

Return type:

str

camacq.config.load_config_file(path: Path) → dict[str, Any]

Parse a YAML configuration file.

Parameters:

path (pathlib.Path) – The path to the configuration YAML file.

Returns:

Return a dict with the configuration contents.

Return type:

dict

camacq.const module

Store common constants.

camacq.control module

Control the microscope.

class camacq.control.Action(action_type: str, action_id: str, func: Callable[[...], Awaitable[None]], schema: Schema | Any)

Bases: object

Represent an action.

class camacq.control.ActionType

Bases: dotdict

Represent an action type.

class camacq.control.ActionsRegistry(center: Center)

Bases: object

Manage all registered actions.

property actions: dict[str, ActionType]

dict: Return dict of ActionTypes with all registered actions.

async call(action_type: str, action_id: str, **kwargs: Any) → None

Call an action with optional kwargs.

Parameters:

• action_type (str) – The name of the module where the action is registered.

• action_id (str) – The id of the action to call.

• **kwargs – Arbitrary keyword arguments. These will be passed to the action function when an action is called.

register(action_type: str, action_id: str, action_func: Callable[[...], Awaitable[None]], schema: Schema | Any) → None

Register an action.

Register actions per module.

Parameters:

• action_type (str) – The name of the action_type to register the action under.

• action_id (str) – The id of the action to register.

• action_func (voluptuous schema) – The function that should be called for the action.

• action_func – The voluptuous schema that should validate the parameters of the action call.

class camacq.control.CamAcqStartEvent(data: dict[str, Any] | None = None)

Bases: Event

An event fired when camacq has started.

event_type: ClassVar[str] = 'camacq_start_event'

class camacq.control.CamAcqStopEvent(data: dict[str, Any] | None = None)

Bases: Event

An event fired when camacq is about to stop.

event_type: ClassVar[str] = 'camacq_stop_event'

property exit_code: int | None

int: Return the plate instance of the event.

class camacq.control.Center(loop: AbstractEventLoop | None = None)

Bases: object

Represent a control center for the microscope.

Parameters:

loop (asyncio.EventLoop) – The event loop.

loop

Return the event loop.

Type:

asyncio.EventLoop

bus

Return the EventBus instance.

Type:

EventBus instance

samples

Return the Samples instance that holds all the Sample instances.

Type:

Samples instance

actions

Return the ActionsRegistry instance.

Type:

ActionsRegistry instance

data

Return dict that stores data from other modules than control.

Type:

dict

actions: ActionsRegistry

add_executor_job(func: ~collections.abc.Callable[[~_P], ~camacq.control._T], *args: ~typing.~_P, **kwargs: ~typing.~_P) → Future[_T]

Schedule a function to be run in the thread pool.

Return a task.

create_task(coro: Coroutine[Any, Any, _T]) → Task[_T]

Schedule a coroutine on the event loop.

Return a task.

data: dict[str, Any]

async end(code: int) → None

Prepare app for exit.

Parameters:

code (int) – Exit code to return when the app exits.

loop: AbstractEventLoop

samples: Samples

async start() → int

Start the app.

async wait_for() → None

Wait for all pending tasks.

camacq.control.loop_exception_handler(loop: AbstractEventLoop, context: dict[str, Any]) → None

Handle exceptions inside the event loop.

camacq.event module

Hold events.

class camacq.event.Event(data: dict[str, Any] | None = None)

Bases: object

A base event.

Parameters:

data (dict, optional) – The data of the event.

data: dict[str, Any]

Return the data of the event.

event_type: ClassVar[str] = 'base_event'

class camacq.event.EventBus(center: Center)

Bases: object

Representation of an eventbus.

Parameters:

center (Center instance) – The Center instance.

property event_types: list[str]

list: Return all registered event types.

async notify(event: Event) → None

Notify handlers that an event has fired.

Parameters:

event (Event instance) – An instance of Event or an instance of subclass of Event.

register(event_type: str, handler: EventHandler) → Callable[[], None]

Register event handler and return a function to remove it.

An event can be a message from the microscope API or an internal event.

Parameters:

• event_type (str) – A string representing the type of event.

• handler (callable) – A coroutine function that should accept two parameters, center and event. The first parameter is the Center instance, the second parameter is the Event instance that has fired.

Returns:

Return a function to remove the registered handler.

Return type:

callable

camacq.event.match_event(event: Event, **event_data: Any) → bool

Return True if event attributes match event_data.

camacq.exceptions module

Provide exceptions.

exception camacq.exceptions.CamAcqError

Bases: Exception

Represent the base camacq exception.

exception camacq.exceptions.MissingActionError(action_id: str)

Bases: CamAcqError

Represent a missing action error.

exception camacq.exceptions.MissingActionTypeError(action_type: str)

Bases: CamAcqError

Represent a missing action type error.

exception camacq.exceptions.SampleError

Bases: CamAcqError

Represent a sample error.

exception camacq.exceptions.TemplateError(exc: Exception)

Bases: CamAcqError

Represent a template error.

camacq.image module

Handle images.

class camacq.image.ImageData(path: str | None = None, data: ndarray[tuple[Any, ...], dtype[Any]] | None = None, metadata: dict[str, Any] | None = None)

Bases: object

Represent the data of an image with path, data, metadata and histogram.

Parameters:

• path (str) – Path to the image.

• data (numpy array) – A numpy array with the image data.

• metadata (dict) – The meta data of the image as a JSON dict.

path

The path to the image.

Type:

str

property data: ndarray[tuple[Any, ...], dtype[Any]]

numpy array: Return the data of the image.

Setter:

Set the data of the image.

property histogram: tuple[ndarray[tuple[Any, ...], dtype[Any]], ndarray[tuple[Any, ...], dtype[Any]]]

numpy array: Calculate and return image histogram.

property metadata: dict[str, Any]

str: Return metadata of image.

Setter:

Set the meta data of the image.

save(path: str | None = None, data: ndarray[tuple[Any, ...], dtype[Any]] | None = None, metadata: dict[str, Any] | None = None) → None

Save image with image data and optional meta data.

Parameters:

• path (str) – The path to the image.

• data (numpy array) – A numpy array with the image data.

• metadata (dict) – The meta data of the image as a JSON dict.

camacq.image.make_proj(images: dict[str, int]) → dict[int, ImageData]

Make a dict of max projections from a dict of channels and paths.

Each channel will make one max projection.

Parameters:

images (dict) – Dict of paths and channel ids.

Returns:

Return a dict of channels that map ImageData objects. Each image object have a max projection as data.

Return type:

dict

camacq.image.read_image(path: str) → ndarray[tuple[Any, ...], dtype[Any]] | None

Read a tif image and return the data.

Parameters:

path (str) – The path to the image.

Returns:

Return a numpy array with image data.

Return type:

numpy array

camacq.image.save_image(path: str, data: ndarray[tuple[Any, ...], dtype[Any]], description: str | None = None) → None

Save a tif image with image data and meta data.

Parameters:

• path (str) – The path to the image.

• data (numpy array) – A numpy array with the image data.

• description (str) – The description string of the image.

camacq.log module

Handle logging.

camacq.log.check_path(path: Path) → bool

Check that path to config exists and is writable for logging.

Parameters:

path (pathlib.Path) – The path to the log file or log directory.

Returns:

Return True if path exists and is writable.

Return type:

bool

camacq.log.enable_log(config: dict[str, Any]) → None

Enable logging.

Parameters:

config (dict) – The dict with the configuration.

camacq.util module

Host utils that are not aware of the implementation of camacq.

class camacq.util.dotdict

Bases: dict[str, Any]

Access to dictionary attributes with dot notation.