API Reference

Camera

class basecam.camera.CameraSystem(camera_class: Type[_T_BaseCamera] | None = None, camera_config: str | Path | Dict[str, Any] | None = None, include: List[Any] | None = None, exclude: List[Any] | None = None, logger: SDSSLogger | None = None, log_header: str | None = None, log_file: str | Path | None = None, verbose: bool | int | None = False)

Bases: LoggerMixIn, Generic[_T_BaseCamera]

A base class for the camera system.

Provides an abstract class for the camera system, including camera connection/disconnection event handling, adding and removing cameras, etc.

While the instance does not handle the loop, it assumes that an event loop is running elsewhere, for example via asyncio.loop.run_forever.

Parameters:

• camera_class (Type[basecam.camera._T_BaseCamera] | None) – The subclass of BaseCamera to use with this camera system.

• camera_config – A dictionary with the configuration parameters for the multiple cameras that can be present in the system, or the path to a YAML file. Refer to the documentation for details on the accepted format.

• include – List of camera UIDs that can be connected.

• exclude – List of camera UIDs that will be ignored.

• logger (sdsstools.logger.SDSSLogger) – The logger instance to use. If None, a new logger will be created.

• log_header – A string to be prefixed to each message logged.

• log_file – The path to which to log. If set, the log will be saved to that file path.

• verbose – Logging level for the console handler If False, only warnings and errors will be logged to the console.

async add_camera(name: str | None = None, uid: str | None = None, force: bool = False, autoconnect: bool = True, **kwargs) → BaseCamera

Adds a new camera instance to cameras.

The configuration values (if any) found in the configuration that match the name or uid of the camera will be used. These parameters can be overridden by providing additional keyword values for the corresponding parameters.

Parameters:

• name – The name of the camera.

• uid – The unique identifier for the camera.

• force – Forces the camera to stay in the CameraSystem list even if it does not appear in the system camera list.

• autoconnect – Whether to autoconnect the camera.

• kwargs – Other arguments to be passed to BaseCamera during initialisation. These parameters override the default configuration where applicable.

Returns:

camera – The new camera instance.

async disconnect()

Shuts down the system.

get_camera(name: str | None = None, uid: str | None = None) → BaseCamera | Literal[False]

Gets a camera matching a name or UID.

If only one camera is connected and the method is called without arguments, returns the camera.

Parameters:

• name – The name of the camera.

• uid – The unique identifier for the camera.

Returns:

camera – The connected camera (must be one of the cameras in cameras) whose name or UID matches the input parameters. Returns False if the camera was not found.

get_camera_config(name: str | None = None, uid: str | None = None) → Dict[str, Any] | None

Gets camera parameters from the configuration.

Parameters:

• name – The name of the camera.

• uid – The unique identifier of the camera.

Returns:

camera_params – A dictionary with the camera parameters. If the camera is not present in the configuration, returns None.

abstractmethod list_available_cameras() → List[str]

Lists the connected cameras as reported by the camera system.

Returns:

connected_cameras – A list of unique identifiers of cameras connected to the system.

on_camera_connected(uid: str) → Task[BaseCamera]

Event handler for a newly connected camera.

Parameters:

uid – The unique identifier for the camera.

Returns:

task – The task calling add_camera.

on_camera_disconnected(uid: str) → Task[None]

Event handler for a camera that was disconnected.

Parameters:

uid – The unique identifier for the camera.

Returns:

task – The task calling remove_camera.

async remove_camera(name: str | None = None, uid: str | None = None)

Removes a camera, cancelling any ongoing process.

Parameters:

• name – The name of the camera.

• uid – The unique identifier for the camera.

setup()

Setup custom camera system.

To be overridden by the subclass if needed. Must return self.

async start_camera_poller(interval: float = 1.0) → _T_CameraSystem

Monitors changes in the camera list.

Issues calls to list_available_cameras on an interval and compares the connected cameras with those in cameras. New found cameras are added and cameras not present are cleanly removed.

This poller should not be initiated if the camera API already provides a framework to detect camera events. In that case the event handler should be wrapped to call on_camera_connected and on_camera_disconnected.

Similarly, if you prefer to handle camera connections manually, avoid starting this poller.

Parameters:

interval – Interval, in seconds, on which the connected cameras are polled for changes.

async stop_camera_poller()

Stops the camera poller if it is running.

cameras: List[_T_BaseCamera]

The list of cameras being handled.

Type:

list

notifier

Notifies of CameraSystemEvent and CameraEvent events.

Type:

.EventNotifier

class basecam.camera.BaseCamera(uid: str, camera_system: CameraSystem, name: str | None = None, force: bool = False, image_namer: ImageNamer | dict | None = None, camera_params={})

Bases: LoggerMixIn

A base class for wrapping a camera API in a standard implementation.

Instantiating the BaseCamera class does not open the camera and makes it ready to be used. To do that, call and await connect.

Parameters:

• uid – The unique identifier for the camera.

• camera_system – The camera system handling this camera.

• name – The name of the camera. If not set, the uid will be used.

• force – Forces the camera to stay in the CameraSystem list even if it does not appear in the system camera list.

• image_namer – An instance of ImageNamer used to sequentially assign predefined names to new exposure images, or a dictionary of parameters to be passed to ImageNamer to create a new instance. If not set, creates an image namer with format {camera.name}-{num:04d}.fits.

• camera_params – Parameters used to define how to connect to the camera, its geometry, initialisation parameters, etc. The format of the parameters must follow the structure of the configuration file.

Variables:

• connected (bool) – Whether the camera is open and connected.

• fits_model (.FITSModel) – An instance of FITSModel defining the data model of the images taken by the camera. If not defined, a basic model will be used.

• image_namer (.ImageNamer) – And instance of ImageNamer to determine the default file path for new exposures. If not provided, uses '{camera.name}-{num:04d}.fits' where camera.name is the name of the camera, and num is a sequential counter.

abstractmethod async _connect_internal(**conn_params)

Internal method to connect the camera.

Must raise CameraConnectionError if the connection fails.

async _disconnect_internal()

Internal method to disconnect a camera.

Must raise a CameraConnectionError if the shutdown fails.

abstractmethod async _expose_internal(exposure: Exposure, **kwargs) → Exposure

Internal method to handle camera exposures.

This method handles the entire process of exposing and reading the camera and return an array or FITS object with the exposed frame. If necessary, it must handle the correct operation of the shutter before and after the exposure.

The method receives an Exposure instance in which the exposure time, image type, and other parameters have been set by expose. Additional parameters can be passed via the kwargs arguments. The camera instance can be accessed via the Exposure.camera attribute.

The method is responsible for adding any relevant attributes in the exposure instance. The time of the start of the exposure is initially set just before _expose_internal is called, but if necessary it must be updated when the camera is actually commanded to expose (or, if flushing occurs, when the integration starts). Finally, it must set Exposure.data with the image as a Numpy array.

Parameters:

• exposure – The exposure being taken.

• kwargs – Other keyword arguments to configure the exposure.

_get_basic_payload() → Dict[str, Any]

Returns a dictionary with basic payload for notifying events.

async _post_process_internal(exposure: Exposure, **kwargs) → Exposure

Performs post-process on an exposure.

This method can be overridden to perform additional processing on the already read exposure. It’s called by expose after _expose_internal is complete but before the image is returned to the user or written to disk.

The user is responsible for issuing any events. If an ExposureError is raised, the error will be propagated and the exposure will not be returned.

_status_internal() → Dict[str, Any]

Gets a dictionary with the status of the camera.

This method is intended to be overridden by the specific camera.

Returns:

status – A dictionary with status values from the camera (e.g., temperature, cooling status, firmware information, etc.)

async connect(force: bool = False, **kwargs) → _T_BaseCamera

Connects the camera and performs all the necessary setup.

Parameters:

• force – Forces the camera to reconnect even if it’s already connected.

• kwargs – A series of keyword arguments to be passed to the internal implementation of connect for a the camera. If provided, they override the connection settings in the configuration for this camera.

async disconnect() → bool

Shuts down the camera.

async expose(exptime: float, image_type: str = 'object', stack: int = 1, stack_function: ~typing.Callable[[...], ~numpy.ndarray] = <function median>, fits_model: ~basecam.models.fits.FITSModel | None = None, filename: str | None = None, num: int | None = None, write: bool = False, postprocess: bool = True, **kwargs) → Exposure

Exposes the camera.

This is the general method to expose the camera. It receives the exposure time and type of exposure, along with other necessary arguments, and returns an Exposure instance with the data and metadata for the image.

The Exposure object is created and populated by expose and passed to the parent mixins for the camera class. It is also passed to the internal expose method where the concrete implementation of the camera expose system happens.

Parameters:

• exptime – The exposure time for the image, in seconds.

• image_type – The type of image ({'bias', 'dark', 'object', 'flat'}).

• stack – Number of images to stack.

• stack_function – The function to apply to the several images taken to generate the final stacked image. Defaults to the median.

• fits_model – A FITSModel that can be used to override the default model for the camera.

• filename – The path where to write the image. If not given, a new name is automatically assigned based on the camera instance of ImageNamer.

• num – If specified, a num value to pass wot ImageNamer to define the sequence number of the exposure filename.

• write – If True, writes the image to disk immediately.

• postprocess – Whether to run the post-process stage. This argument is ignored if the subclass of BaseCamera does not override _post_process_internal.

• kwargs – Other keyword arguments to pass to the internal expose and post-process methods.

Returns:

exposure – An Exposure object containing the image data, exposure time, and header datamodel.

get_status(update: bool = False) → Dict[str, Any]

Returns a dictionary with the camera status values.

Parameters:

update – If True, retrieves the status from the camera; otherwise returns the last cached status.

notify(event: CameraEvent, extra_payload: Dict[str, Any] | None = None)

Notifies an event.

property status

Returns the cached status. Equivalent to get_status(update=False).

Mixins

class basecam.mixins.CoolerMixIn

Bases: object

Methods to control the cooling system of the camera.

abstractmethod async _get_temperature_internal()

Internal method to get the camera temperature.

If the camera can report multiple temperatures, this method must return the temperature that the cooler modifies. Other temperature can be reported in the status. Must raise CameraError if there is a problem.

abstractmethod async _set_temperature_internal(temperature)

Internal method to set the camera temperature.

This method should return immediately after setting the new temperature or raise CameraError if there is a problem.

async get_temperature()

Returns the temperature of the camera.

async set_temperature(temperature)

Sets a new temperature goal for the camera.

Emits a NEW_SET_POINT event when the new temperature is set. The coroutine blocks until the temperature has been reached (at which point it emits SET_POINT_REACHED) or until the set point changes.

Parameters:

temperature (float) – The goal temperature, in degrees Celsius.

class basecam.mixins.ExposureTypeMixIn

Bases: object

Methods to take exposures with different image_types.

async bias(**kwargs)

Take a bias image.

async dark(exp_time: float, **kwargs)

Take a dark image.

async flat(exp_time: float, **kwargs)

Take a flat image.

async object(exp_time: float, **kwargs)

Take a science image.

class basecam.mixins.ImageAreaMixIn

Bases: object

Allows to select the image area and binning.

abstractmethod async _get_binning_internal()

Internal method to return the binning.

abstractmethod async _get_image_area_internal()

Internal method to return the image area.

abstractmethod async _set_binning_internal(hbin, vbin)

Internal method to set the binning.

In case of error it must raise CameraError.

abstractmethod async _set_image_area_internal(area=None)

Internal method to set the image area.

If area=None must restore the full image area. In case of error, must raise CameraError.

async get_binning()

Returns the horizontal and vertical binning as (hbin, vbin).

async get_image_area()

Returns the imaging area as 1-indexed (x0, x1, y0, y1).

async set_binning(hbin=1, vbin=None)

Sets the binning.

Parameters:

• hbin (int) – Horizontal binning.

• vbin (int) – Vertical binning. If not provided, same as hbin.

async set_image_area(area=None)

Sets the image area.

Parameters:

area (tuple) – The image area to set as 1-indexed (x0, x1, y0, y1). If not provided, restores the full image area.

class basecam.mixins.ShutterMixIn

Bases: object

A mixin that provides manual control over the shutter.

abstractmethod async _get_shutter_internal()

Internal method to get the position of the shutter.

abstractmethod async _set_shutter_internal(shutter_open)

Internal method to set the position of the shutter.

async close_shutter()

Opens the shutter (alias for set_shutter(False)).

async get_shutter()

Gets the position of the shutter.

async open_shutter()

Opens the shutter (alias for set_shutter(True)).

async set_shutter(shutter, force=False)

Sets the position of the shutter.

Parameters:

• shutter (bool) – If True moves the shutter open, otherwise closes it.

• force (bool) – Normally, a call is made to get_shutter to determine if the shutter is already in the commanded position. If it is, the shutter is not commanded to move. force=True sends the move command regardless of the internal status of the shutter.

Exposure

class basecam.exposure.Exposure(camera: BaseCamera, filename: str | None = None, data: ndarray | None = None, fits_model: FITSModel | None = None, wcs: WCS | None = None)

Bases: object

Exposure class. Represents an exposure data and its metadata.

An Exposure is defined by the raw image taken by the camera, a series of attributes that define the exposure (exposure time, shutter, etc.) and a data model that is used to generate the FITS file for the exposure.

Parameters:

• camera – The instance of a subclass of BaseCamera that took this exposure.

• filename – The filename of the FITS image generated.

• data – The exposure raw data, as a 2D array.

• fits_model – The model to create the FITS image. If None, a single extension with a basic header will be used.

• wcs – The WCS object describing the astrometry of this exposure.

Variables:

• image_type – The type of image, one of bias, flat, dark, object.

• exptime (float) – The exposure time, in seconds, of a single integration.

• exptime_n (float) – The total exposure time, in seconds. If the image is stacked, this is the total time, i.e., the sum of the exposure time of each stacked image.

• stack (int) – Number of exposures stacked.

• stack_function – Name of the function used for stacking.

• filename – The path where to write the image.

• wcs (WCS) – The WCS object describing the astrometry of this exposure.

add_hdu(hdu: BinTableHDU | ImageHDU, index: int | None = None)

Adds an HDU to the list of extensions.

Parameters:

• hdu – The BinTableHDU or ImageHDU HDU to append.

• index – The index where the extension will be added. Extra HDUs are inserted in order after the FITS model has been generated. index=None appends the new HDU at the end of the list. Note that astropy.io.fits may change the final order of the extensions to ensure that a primary HDU remains as the first HDU.

to_hdu(context={}) → HDUList

Return an HDUList for the image.

Parameters:

context – A dictionary of arguments used to evaluate the FITS model.

Returns:

hdulist – A list of HDUs in which the FITS data model has been evaluated for this exposure.

async write(filename: str | None = None, context={}, overwrite=False, checksum=True, retry=True) → HDUList

Writes the image to disk.

Parameters:

• filename – The path where to write the file. If not provided, uses Exposure.filename.

• context – A dictionary of arguments used to evaluate the FITS model.

• overwrite – Whether to overwrite the image if it exists.

• checksum – When True adds both DATASUM and CHECKSUM cards to the headers of all HDUs written to the file.

• retry – If True and the image fails to write, tries again. This can be useful when writing to network volumen where failures are more frequent.

Returns:

hdulist – A list of HDUs in which the FITS data model has been evaluated for this exposure.

property obstime: Time

The time at the beginning of the observation.

It must be an astropy.time.Time object or a datetime in ISO format; in the latter case, UTC scale is assumed.

class basecam.exposure.ImageNamer(basename: str = '{camera.name}-{num:04d}.fits', dirname: str = '.', overwrite: bool = False, camera: BaseCamera | None = None, reset_sequence: bool = True)

Bases: object

Creates a new sequential filename for an image.

Parameters:

• basename – The basename of the image filenames. Must contain a placeholder num in the place where to insert the sequence number. For example, 'test-{num:04d}.fits' will produce image names test-0001.fits, test-0002.fits, etc. It’s also possible to use placeholders for camera values, e.g. {camera.name}-{num}.fits.

• dirname – The directory for the images. Can include an expression based on the date substitution which is a now object. For example: dirname='/data/{camera.uid}/{int(date.mjd)}'.

• overwrite – If True, the sequence will start at 1 regardless of the existing images. If False, the first element in the sequence will be selected to avoid colliding with any image already existing in the directory.

• camera – A BaseCamera instance. It can also be passed when calling the instance.

• reset_sequence – Resets the sequence number when the directory changes (for example when the MJD rolls over).

Examples

>>> namer = ImageNamer('{camera.name}-{num:04d}.fits', dirname='testdir')
>>> namer(camera=camera)
PosixPath('testdir/my_camera-0001.fits')
>>> namer(camera=camera)
PosixPath('testdir/my_camera-0002.fits')

get_dirname() → Path

Returns the evaluated dirname.

property basename: str

The image name pattern.

Models

class basecam.models.fits.Extension(data: Literal['raw'] | Literal['none'] | None | bool | ndarray = None, header_model: HeaderModel | None = None, name: str | None = None, compressed: bool | str = False, compression_params: dict[str, Any] = {})

Bases: object

A model for a FITS extension.

Parameters:

• data – The data for this FITS extension. Can be an array or a macro string indicating the type of data to store in the extension. Available macros are: 'raw' or None for the raw image, or 'none' for empty data.

• header_model – A HeaderModel for this extension.

• name – The name of the HDU.

• compressed – If False, the extension data will not be compressed. Otherwise, a string with one of the compression_type in CompImageHDU.

• compression_params – Additional parameters to be passed to CompImageHDU if the extension is compressed.

get_data(exposure: Exposure, primary: bool = False) → ndarray | None

Returns the data as a numpy array.

to_hdu(exposure: Exposure, primary: bool = False, context: Dict[str, Any] = {}) → ImageHDU | CompImageHDU | PrimaryHDU

Evaluates the extension as an HDU.

Parameters:

• exposure – The exposure for which we want to evaluate the extension.

• primary – Whether this is the primary extension.

• context – A dictionary of arguments used to evaluate the parameters in the extension.

Returns:

hdu – An ImageHDU with the data and header evaluated for exposure, or CompImageHDU if compressed=True.

class basecam.models.fits.FITSModel(extensions: List[Extension] | None = None, context: Dict[str, Any] = {})

Bases: list

A model representing a FITS image.

This model defines the data and header for each extension in a FITS image. The model can be evaluated for given Exposure, returning a fully formed astropy HDUList.

Parameters:

• extension – A list of HDU extensions defined as Extension objects. If none is defined, a single, basic extension is added.

• context – A default context to pass to the header model when evaluating it.

copy() → T

Return a shallow copy of the list.

to_hdu(exposure: Exposure, context: Dict[str, Any] = {}) → HDUList

Returns an HDUList from an exposure.

Parameters:

• exposure – The exposure for which the FITS model is evaluated.

• context – A dictionary of parameters used to fill the card replacement fields. Updates the default context.

Returns:

hdulist – A list of HDUs, each one define by its corresponding Extension instance. The first extension is created as a PrimaryHDU unless it is compressed, in which case an empty primary HDU is prepended.

class basecam.models.fits.HeaderModel(cards: List[Card | CardGroup | MacroCard | str | tuple | list | None] = [])

Bases: list

A model defining the header of an HDU.

Parameters:

cards – A list of Card, CardGroup, or MacroCard instances. It can also be a string with the name of a default card.

Examples

>>> header_model = HeaderModel([Card('TELESCOP', 'APO-2.5', 'The telescope'),
                                Card('OBSERVATORY', 'APO'),
                                'EXPTIME',
                                Card('camname', '{(camera).name}')])

append(card: Card | CardGroup | MacroCard | str | tuple | list | None)

Append object to the end of the list.

describe()

Returns a table-like representation of the header model.

insert(idx: int, card: Card | CardGroup | MacroCard | str | tuple | list | None)

Insert object before index.

to_header(exposure: Exposure, context: Dict[str, Any] = {}) → Header

Evaluates the header model for an exposure and returns a header.

Parameters:

• exposure – The exposure for which we want to evaluate the model.

• context – A dictionary of arguments used to evaluate the parameters in the model.

Returns:

header – A Header, created by evaluating the model for the input exposure.

class basecam.models.card.Card(name: str | Iterable[Any], *args, **kwargs)

Bases: object

A card representing a single entry in the header.

This class is similar to astropy’s Card with the difference that the value of the card does not need to be a literal and is evaluated when evaluate is called.

The value of the card can be:

• A Python literal.

• A callable, with a list of arguments, that is called when the card is evaluated.

• A string using Python’s Format String Syntax. As with normal strings, the replacement fields are surrounded by { and } and filled when evaluate is called. The replacement fields can point to an attribute, function, method, or property. In addition, it is possible to use the string __exposure__ and __camera__ to refer to the Exposure instance for which this card will be evaluated, and the instance of BaseCamera that took the image, respectively. For example, value='{__camera__.name}' will be replaced with the name of the camera, and value='{__exposure__.obstime}' will retrieve the time of the start of the exposure.

• A string to be evaluated using Python’s eval function. For example, value='__camera__.get_temperature()'.

If the name of the card is one of the Default Cards, the value and comment are automatically defined and do not need to be specified.

Parameters:

• name – The name of the card. Will be trimmed to 8 characters. Can be one of the Default Cards names, which defines the value and comment.

• value – A Python literal, callable, or string, as described above.

• comment – A short comment describing the card value.

• default – Default value to use if the card value contains placeholders that are not provided while the card is evaluated.

• type – The type of the card value. The value will be casted after processing.

• autocast – If True and type is not defined, tries to cast the value.

• fargs – If value is a callable, the arguments to pass to it when it gets called. The arguments are evaluated following the same rules as with the value before being passed to the callable.

• evaluate – If True, assumes the value is a string to be evaluated in realtime. The context is used as locals for the evaluation. For security, globals are not available.

• context – A dictionary of parameters used to fill the value replacement fields. Two values, __exposure__ and __camera__, are always defined. This context can be updated during the evaluation of the card.

evaluate(exposure: Exposure, context: Dict[str, Any] = {}) → EvaluatedCard

Evaluates the card.

Evaluates the card value, expanding its template parameters. If the value is a callable, calls the function at this time.

Parameters:

• exposure – The exposure for which we want to evaluate the card.

• context – A dictionary of arguments used to evaluate the parameters in the value. These argument update those defined when the Card was created.

Returns:

EvaluatedCard – A tuple with the name, evaluated value, and comment for this card.

set_exposure(exposure, context={})

Sets the current exposure and context, clearing it on exit.

class basecam.models.card.CardGroup(cards: Iterable[Card | Iterable | str], name: str | None = None, use_group_title: bool = True)

Bases: list

A group of Card instances.

A CardGroup is just a list of Cards that are grouped for convenience and for easy reuse.

Parameters:

• cards – A list of Card instances. Elements can also be a tuple of two or three elements (for name, value, and optionally a comment) or a string with a default card.

• name (str | None) – A name for the card group.

• use_group_title – Whether to prepend a COMMENT card with the group title when creating a header.

append(card: Card)

Append object to the end of the list.

evaluate(exposure, context={})

Evaluates all the cards.

Parameters:

• exposure (.Exposure) – The exposure for which we want to evaluate the cards.

• context (dict) – A dictionary of arguments used to evaluate the cards.

Returns:

list – A list of tuples with the name, evaluated value, and comment for each card.

insert(idx, card: Card)

Insert object before index.

to_header(exposure: Exposure, context: Dict[str, Any] = {}, use_group_title: bool = False) → Header

Evaluates all the cards and returns a header.

Parameters:

• exposure – The exposure for which we want to evaluate the cards.

• context – A dictionary of arguments used to evaluate the cards.

• use_group_title – Whether to prepend a COMMENT card with the group title when creating the header.

Returns:

header – A header composed from the cards in the group.

class basecam.models.card.DefaultCard(name: str | Iterable[Any], *args, **kwargs)

Bases: Card

class basecam.models.card.MacroCard(name: str | None = None, use_group_title: bool = False, **kwargs)

Bases: object

A macro that returns a list of keywords and values.

This is an abstract class in which the macro method must be overridden to return a list of keywords and values that can be used to create a header.

Parameters:

• name – A name for the macro group.

• use_group_title – Whether to prepend a COMMENT card with the macro title when creating a header.

• kwargs – Additional arguments for the macro.

evaluate(exposure: Exposure, context: Dict[str, Any] = {}) → List[tuple]

Evaluates the macro. Equivalent to calling macro directly.

Parameters:

• exposure – The exposure for which we want to evaluate the macro.

• context – A dictionary of parameters that can be used by the macro.

Returns:

tuples – A list of tuples with the format (keyword, value, comment) or (keyword, value).

abstractmethod macro(exposure: Exposure, context: Dict[str, Any] = {})

The macro.

Must return a list of item which can be tuples with the format (keyword, value, comment) or (keyword, value), Card instances, or CardGroup instances. Cards and card groups will be evaluated.

Parameters:

• exposure – The exposure for which we want to evaluate the macro.

• context – A dictionary of parameters that can be used by the macro.

to_header(exposure: Exposure, context: Dict[str, Any] = {}, use_group_title=False) → Header

Evaluates the macro and returns a header.

Parameters:

• exposure – The exposure for which we want to evaluate the macro.

• context – A dictionary of arguments used to evaluate the macro.

• use_group_title – Whether to prepend a COMMENT card with the macro title when creating the header.

Returns:

header – A header composed from the cards in the macro.

class basecam.models.card.WCSCards(name: str | None = None, use_group_title: bool = False, **kwargs)

Bases: MacroCard

A macro that adds WCS header information.

To use, just add WCSCards() to the header model and make sure the Exposure.wcs is set. If Exposure.wcs=None, a default WCS header will be added.

macro(exposure: Exposure, context: Dict[str, Any] = {})

The macro.

Must return a list of item which can be tuples with the format (keyword, value, comment) or (keyword, value), Card instances, or CardGroup instances. Cards and card groups will be evaluated.

Parameters:

• exposure – The exposure for which we want to evaluate the macro.

• context – A dictionary of parameters that can be used by the macro.

basecam.models.card.DEFAULT_CARDS: Dict[str, DefaultCard] = {'BASECAMV': <DefaultCard (name='BASECAMV', value='0.8.2')>, 'CAMNAME': <DefaultCard (name='CAMNAME', value='{__camera__.name}')>, 'CAMUID': <DefaultCard (name='CAMUID', value='{__camera__.uid}')>, 'EXPTIME': <DefaultCard (name='EXPTIME', value='{__exposure__.exptime}')>, 'EXPTIMEN': <DefaultCard (name='EXPTIMEN', value='{__exposure__.exptime_n}')>, 'IMAGETYP': <DefaultCard (name='IMAGETYP', value='{__exposure__.image_type}')>, 'OBSTIME': <DefaultCard (name='OBSTIME', value='{__exposure__.obstime.tai}')>, 'STACK': <DefaultCard (name='STACK', value='{__exposure__.stack}')>, 'STACKFUN': <DefaultCard (name='STACKFUN', value='{__exposure__.stack_function.__name__}')>, 'VCAM': <DefaultCard (name='VCAM', value='{__camera__.__version__}')>}

Default cards

basecam.models.builtin.basic_fits_model = [<Extension (name='PRIMARY', compressed=False)>]

A basic FITS model for uncompressed images. Includes a single extension with the raw data and a basic_header_model.

basecam.models.builtin.basic_fz_fits_model = [<Extension (name='PRIMARY', compressed=RICE_1)>]

A compressed, basic FITS model. Similar to basic_fits_model but uses RICE_1 compression.

basecam.models.builtin.basic_header_model = [<DefaultCard (name='VCAM', value='{__camera__.__version__}')>, <DefaultCard (name='BASECAMV', value='0.8.2')>, <DefaultCard (name='CAMNAME', value='{__camera__.name}')>, <DefaultCard (name='CAMUID', value='{__camera__.uid}')>, <DefaultCard (name='IMAGETYP', value='{__exposure__.image_type}')>, <DefaultCard (name='EXPTIME', value='{__exposure__.exptime}')>, <DefaultCard (name='EXPTIMEN', value='{__exposure__.exptime_n}')>, <DefaultCard (name='STACK', value='{__exposure__.stack}')>, <DefaultCard (name='STACKFUN', value='{__exposure__.stack_function.__name__}')>, <Card (name='TIMESYS', value='TAI')>, <Card (name='DATE-OBS', value='{__exposure__.obstime.tai.isot}')>]

A basic header model with camera and exposure information.

Notifier

class basecam.notifier.EventListener(filter_events=None, autostart=True)

Bases: Queue

An event queue with callbacks.

Parameters:

• filter_events (list) – A list of enum values of which to be notified. If None, all events will be notified.

• autostart (bool) – Whether to start the listener as soon as the object is created.

register_callback(callback)

Registers a callback to be called when an event is read.

Parameters:

callback – A function or coroutine function to be called. The callback receives the event (an enumeration value) as the first argument and the payload associated with that event as a dictionary. If the callback is a coroutine, it is scheduled as a task.

remove_callback(callback)

De-registers a callback.

async start_listening()

Starts the listener task. The queue will be initially purged.

async stop_listening()

Stops the listener task.

async wait_for(events, timeout=None)

Blocks until a certain event happens.

Parameters:

• events – The event to wait for. It can be a list of events, in which case it returns when any of the events has been seen.

• timeout (float or None) – Timeout in seconds. If None, blocks until the event is received.

Returns:

result (bool) – Returns a set of events received that intersects with the events that were being waited for. Normally this is a single event, the first one to be seen, but it can be more than one if multiple events that were being waited for happen at the same time. Returns False if the routine timed out before receiving the event.

class basecam.notifier.EventNotifier

Bases: object

A registry of clients to be notified of events.

Allows to register a listener queue in which to announce events.

notify(event, payload={})

Sends an event to all listeners.

Parameters:

• event (Enum) – An enumeration value belonging to the event_class.

• payload (dict) – A dictionary with the information associated with the event.

register_listener(listener)

Registers a listener.

Parameters:

listener (.EventListener) – An EventListener instance to which to send events for processing.

remove_listener(listener)

Removes a listener.

Utils

class basecam.utils.LoggerMixIn

Bases: object

A mixin to provide easy logging with a header.

log(message, level=10, use_header=True)

Logs a message with a header.

class basecam.utils.Poller(name, callback, delay=1)

Bases: object

A task that runs a callback periodically.

Parameters:

• name (str) – The name of the poller.

• callback (function or coroutine) – A function or coroutine to call periodically.

• delay (float) – Initial delay between calls to the callback.

• loop (event loop) – The event loop to which to attach the task.

async call_now()

Calls the callback immediately.

async poller()

The polling loop.

async set_delay(delay=None, immediate=False)

Sets the delay for polling.

Parameters:

• delay (float) – The delay between calls to the callback. If None, restores the original delay.

• immediate (bool) – If True, stops the currently running task and sets the new delay. Otherwise waits for the current task to complete.

start(delay=None)

Starts the poller.

Parameters:

delay (float) – The delay between calls to the callback. If not specified, restores the original delay used when the class was instantiated.

async stop()

Cancel the poller.

property running

Returns True if the poller is running.

async basecam.utils.cancel_task(task)

Cleanly cancels a task.

async basecam.utils.gzip_async(file: Path | str, complevel=1)

Compresses a file with gzip asynchronously.

async basecam.utils.subprocess_run_async(*args, shell=False)

Runs a command asynchronously.

If shell=True the command will be executed through the shell. In that case the argument must be a single string with the full command. Otherwise, must receive a list of program arguments. Returns the output of stdout.

Actor

class basecam.actor.actor.BaseCameraActor(camera_system: CameraSystem, *args, default_cameras: List[str] | str | None = None, command_parser: CluGroup | None = None, schema: str | None = 'internal', **kwargs)

Bases: BaseActor

Base class for a camera CLU-like actor class.

Expands a CLU actor to receive commands, interact with the camera system, and reply to the commander. This base class needs to be subclassed along with the desired implementation of CLU BaseActor. For example

from clu.actor import AMQPActor

class MyCameraActor(BaseCameraActor, AMQPActor):
    pass

Parameters:

• camera_system – The camera system, already instantiated.

• default_cameras – A list of camera names or UIDs that define what cameras to use by default in most command.

• command_parser – The list of commands to use. It must be a command group deriving from CluGroup containing all the commands to use. If commands=None, uses the internal command set.

• schema – The path to the JSONSchema file with the actor datamodel. If "internal", uses the default basecam model; None disables model validation.

• args – Arguments and keyword arguments to be passed to the actor class.

• kwars – Arguments and keyword arguments to be passed to the actor class.

set_default_cameras(cameras: str | List[str] | None = None)

Sets the camera(s) that will be used by default.

These cameras will be used by default when a command is issued without listing the cameras to command.

Parameters:

cameras (str or list) – A list of camera names or a string with comma-separated camera names. If None, no cameras will be considered default and commands will fail if they do not specify a camera.

listener

An EventListener that can be used to wait or respond to events.

class basecam.actor.actor.CameraActor(camera_system: CameraSystem, *args, default_cameras: List[str] | str | None = None, command_parser: CluGroup | None = None, schema: str | None = 'internal', **kwargs)

Bases: BaseCameraActor, JSONActor

A camera actor that replies with JSONs using JSONActor.

basecam.actor.tools.get_cameras(command, cameras=None, check_cameras=True, fail_command=False)

A helper to determine what cameras to use.

Parameters:

• command – The command that wants to access the cameras.

• cameras (list) – The list of camera names passed to the command.

• check_cameras (bool) – If any of the cameras is not connected, returns False.

• fail_command (bool) – Fails the command before returning False.

Returns:

cameras (list of Camera instances) – A list of Camera instances that match the input cameras or, if cameras=None, the default cameras. If cameras=None and there are no default cameras defined, returns all the connected camera. If check_cameras=True and any of the selected cameras is not connected, the command is failed and returns False.

basecam.actor.tools.get_schema() → Dict[str, Any]

Returns the actor schema as a dictionary.

Exceptions

exception basecam.exceptions.CameraConnectionError(message='')

Bases: CameraError

An error to be raised if the camera fails to connect/disconnect.

exception basecam.exceptions.CameraError(message='')

Bases: Exception

A custom core exception

exception basecam.exceptions.CameraWarning(message, *args, **kwargs)

Bases: UserWarning

Base warning.

exception basecam.exceptions.CardError

Bases: FITSModelError

Error raised by a FITS Card.

exception basecam.exceptions.CardWarning

Bases: FITSModelWarning

Warning raised by a FITS Card.

exception basecam.exceptions.ExposureError

Bases: Exception

The exposure failed.

exception basecam.exceptions.ExposureWarning

Bases: UserWarning

Warning for exposures.

exception basecam.exceptions.FITSModelError

Bases: Exception

An error related to the FITS model.

exception basecam.exceptions.FITSModelWarning

Bases: UserWarning

A warnings related to the FITS model.

Events

class basecam.events.CameraSystemEvent(*values)

Bases: Enum

Enumeration of camera system events.

CAMERA_ADDED = 1

CAMERA_REMOVED = 2

class basecam.events.CameraEvent(*values)

Bases: Enum

Enumeration of camera events.

CAMERA_CONNECTED = 'connected'

CAMERA_CONNECT_FAILED = 'connect_failed'

CAMERA_DISCONNECTED = 'disconnected'

CAMERA_DISCONNECT_FAILED = 'disconnect_failed'

EXPOSURE_IDLE = 'idle'

EXPOSURE_FLUSHING = 'flushing'

EXPOSURE_INTEGRATING = 'integrating'

EXPOSURE_READING = 'reading'

EXPOSURE_READ = 'read'

EXPOSURE_DONE = 'done'

EXPOSURE_FAILED = 'failed'

EXPOSURE_WRITING = 'writing'

EXPOSURE_WRITTEN = 'written'

EXPOSURE_POST_PROCESSING = 'post_processing'

EXPOSURE_POST_PROCESS_DONE = 'post_process_done'

EXPOSURE_POST_PROCESS_FAILED = 'post_process_failed'

NEW_SET_POINT = 'new_set_point'

SET_POINT_REACHED = 'set_point_reached'