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)[source]#
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 [source]#
Adds a new
camera
instance tocameras
.The configuration values (if any) found in the configuration that match the
name
oruid
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.
- get_camera(name: str | None = None, uid: str | None = None) BaseCamera | Literal[False] [source]#
Gets a camera matching a name or UID.
If only one camera is connected and the method is called without arguments, returns the camera.
- get_camera_config(name: str | None = None, uid: str | None = None) Dict[str, Any] | None [source]#
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
.
- abstract list_available_cameras() List[str] [source]#
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] [source]#
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] [source]#
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)[source]#
Removes a camera, cancelling any ongoing process.
- Parameters:
name – The name of the camera.
uid – The unique identifier for the camera.
- setup()[source]#
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 [source]#
Monitors changes in the camera list.
Issues calls to
list_available_cameras
on an interval and compares the connected cameras with those incameras
. 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
andon_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.
- notifier#
Notifies of
CameraSystemEvent
andCameraEvent
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={})[source]#
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 awaitconnect
.- 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 toImageNamer
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'
wherecamera.name
is the name of the camera, andnum
is a sequential counter.
- abstract async _connect_internal(**conn_params)[source]#
Internal method to connect the camera.
Must raise
CameraConnectionError
if the connection fails.
- async _disconnect_internal()[source]#
Internal method to disconnect a camera.
Must raise a
CameraConnectionError
if the shutdown fails.
- abstract async _expose_internal(exposure: Exposure, **kwargs) Exposure [source]#
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 byexpose
. Additional parameters can be passed via thekwargs
arguments. The camera instance can be accessed via theExposure.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 setExposure.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] [source]#
Returns a dictionary with basic payload for notifying events.
- async _post_process_internal(exposure: Exposure, **kwargs) Exposure [source]#
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] [source]#
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 [source]#
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 theconnection
settings in the configuration for this 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 [source]#
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 byexpose
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 wotImageNamer
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] [source]#
Returns a dictionary with the camera status values.
- Parameters:
update – If
True
, retrieves the status from the camera; otherwise returns the last cached status.
- property status#
Returns the cached status. Equivalent to
get_status(update=False)
.
Mixins#
- class basecam.mixins.CoolerMixIn[source]#
Bases:
object
Methods to control the cooling system of the camera.
- abstract async _get_temperature_internal()[source]#
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.
- abstract async _set_temperature_internal(temperature)[source]#
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 set_temperature(temperature)[source]#
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 emitsSET_POINT_REACHED
) or until the set point changes.- Parameters:
temperature (float) – The goal temperature, in degrees Celsius.
- class basecam.mixins.ExposureTypeMixIn[source]#
Bases:
object
Methods to take exposures with different image_types.
- class basecam.mixins.ImageAreaMixIn[source]#
Bases:
object
Allows to select the image area and binning.
- abstract async _set_binning_internal(hbin, vbin)[source]#
Internal method to set the binning.
In case of error it must raise
CameraError
.
- abstract async _set_image_area_internal(area=None)[source]#
Internal method to set the image area.
If
area=None
must restore the full image area. In case of error, must raiseCameraError
.
- class basecam.mixins.ShutterMixIn[source]#
Bases:
object
A mixin that provides manual control over the shutter.
- abstract async _set_shutter_internal(shutter_open)[source]#
Internal method to set the position of the shutter.
- async set_shutter(shutter, force=False)[source]#
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)[source]#
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. IfNone
, 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)[source]#
Adds an HDU to the list of extensions.
- Parameters:
hdu – The
BinTableHDU
orImageHDU
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 thatastropy.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 [source]#
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 [source]#
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 bothDATASUM
andCHECKSUM
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)[source]#
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 namestest-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 anow
object. For example:dirname='/data/{camera.uid}/{int(date.mjd)}'
.overwrite – If
True
, the sequence will start at 1 regardless of the existing images. IfFalse
, 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')
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] = {})[source]#
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'
orNone
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 thecompression_type
inCompImageHDU
.compression_params – Additional parameters to be passed to
CompImageHDU
if the extension is compressed.
- get_data(exposure: Exposure, primary: bool = False) ndarray | None [source]#
Returns the data as a numpy array.
- to_hdu(exposure: Exposure, primary: bool = False, context: Dict[str, Any] = {}) ImageHDU | CompImageHDU | PrimaryHDU [source]#
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 forexposure
, orCompImageHDU
ifcompressed=True
.
- class basecam.models.fits.FITSModel(extensions: List[Extension] | None = None, context: Dict[str, Any] = {})[source]#
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 astropyHDUList
.- 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.
- to_hdu(exposure: Exposure, context: Dict[str, Any] = {}) HDUList [source]#
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 aPrimaryHDU
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] = [])[source]#
Bases:
list
A model defining the header of an HDU.
- Parameters:
cards – A list of
Card
,CardGroup
, orMacroCard
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)[source]#
Append object to the end of the list.
- insert(idx: int, card: Card | CardGroup | MacroCard | str | tuple | list | None)[source]#
Insert object before index.
- to_header(exposure: Exposure, context: Dict[str, Any] = {}) Header [source]#
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)[source]#
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 whenevaluate
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 whenevaluate
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 theExposure
instance for which this card will be evaluated, and the instance ofBaseCamera
that took the image, respectively. For example,value='{__camera__.name}'
will be replaced with the name of the camera, andvalue='{__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
andtype
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 thevalue
before being passed to the callable.evaluate – If
True
, assumes the value is a string to be evaluated in realtime. The context is used aslocals
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 [source]#
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.
- class basecam.models.card.CardGroup(cards: Iterable[Card | Iterable | str], name: str | None = None, use_group_title: bool = True)[source]#
Bases:
list
A group of
Card
instances.A
CardGroup
is just a list ofCards
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.
- evaluate(exposure, context={})[source]#
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.
- to_header(exposure: Exposure, context: Dict[str, Any] = {}, use_group_title: bool = False) Header [source]#
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)[source]#
Bases:
Card
- class basecam.models.card.MacroCard(name: str | None = None, use_group_title: bool = False, **kwargs)[source]#
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] [source]#
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)
.
- abstract macro(exposure: Exposure, context: Dict[str, Any] = {})[source]#
The macro.
Must return a list of item which can be tuples with the format
(keyword, value, comment)
or(keyword, value)
,Card
instances, orCardGroup
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 [source]#
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)[source]#
Bases:
MacroCard
A macro that adds WCS header information.
To use, just add
WCSCards()
to the header model and make sure theExposure.wcs
is set. IfExposure.wcs=None
, a default WCS header will be added.- macro(exposure: Exposure, context: Dict[str, Any] = {})[source]#
The macro.
Must return a list of item which can be tuples with the format
(keyword, value, comment)
or(keyword, value)
,Card
instances, orCardGroup
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.1a0')>, '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 usesRICE_1
compression.
- basecam.models.builtin.basic_header_model = [<DefaultCard (name='VCAM', value='{__camera__.__version__}')>, <DefaultCard (name='BASECAMV', value='0.8.1a0')>, <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(loop=None, filter_events=None, autostart=True)[source]#
Bases:
Queue
An event queue with callbacks.
- Parameters:
- register_callback(callback)[source]#
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.
- async wait_for(events, timeout=None)[source]#
Blocks until a certain event happens.
- Parameters:
- Returns:
result (bool) – Returns a
set
of events received that intersects with theevents
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. ReturnsFalse
if the routine timed out before receiving the event.
- class basecam.notifier.EventNotifier[source]#
Bases:
object
A registry of clients to be notified of events.
Allows to register a listener queue in which to announce events.
- register_listener(listener)[source]#
Registers a listener.
- Parameters:
listener (.EventListener) – An
EventListener
instance to which to send events for processing.
Utils#
- class basecam.utils.LoggerMixIn[source]#
Bases:
object
A mixin to provide easy logging with a header.
- class basecam.utils.Poller(name, callback, delay=1, loop=None)[source]#
Bases:
object
A task that runs a callback periodically.
- Parameters:
- async basecam.utils.gzip_async(file: Path | str, complevel=1)[source]#
Compresses a file with gzip asynchronously.
- async basecam.utils.subprocess_run_async(*args, shell=False)[source]#
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)[source]#
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 examplefrom 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. Ifcommands=None
, uses the internal command set.schema – The path to the JSONSchema file with the actor datamodel. If
"internal"
, uses the defaultbasecam
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)[source]#
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.
- 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)[source]#
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)[source]#
A helper to determine what cameras to use.
- Parameters:
- Returns:
cameras (list of
Camera
instances) – A list ofCamera
instances that match the inputcameras
or, ifcameras=None
, the default cameras. Ifcameras=None
and there are no default cameras defined, returns all the connected camera. Ifcheck_cameras=True
and any of the selected cameras is not connected, the command is failed and returnsFalse
.
Exceptions#
- exception basecam.exceptions.CameraConnectionError(message='')[source]#
Bases:
CameraError
An error to be raised if the camera fails to connect/disconnect.
- exception basecam.exceptions.CameraError(message='')[source]#
Bases:
Exception
A custom core exception
- exception basecam.exceptions.CameraWarning(message, *args, **kwargs)[source]#
Bases:
UserWarning
Base warning.
- exception basecam.exceptions.CardError[source]#
Bases:
FITSModelError
Error raised by a FITS
Card
.
- exception basecam.exceptions.CardWarning[source]#
Bases:
FITSModelWarning
Warning raised by a FITS
Card
.
- exception basecam.exceptions.ExposureWarning[source]#
Bases:
UserWarning
Warning for exposures.
- exception basecam.exceptions.FITSModelError[source]#
Bases:
Exception
An error related to the FITS model.
- exception basecam.exceptions.FITSModelWarning[source]#
Bases:
UserWarning
A warnings related to the FITS model.
Events#
- class basecam.events.CameraSystemEvent(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#
Bases:
Enum
Enumeration of camera system events.
- CAMERA_ADDED = 1#
- CAMERA_REMOVED = 2#
- class basecam.events.CameraEvent(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#
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'#