#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
# @Date: 2019-08-06
# @Filename: actor.py
# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
from __future__ import annotations
import logging
import os
from typing import List, Optional, Union
import clu.parsers.click
from clu import BaseActor, Command, JSONActor
from clu.tools import ActorHandler
from sdsstools.logger import SDSSLogger
from basecam import CameraSystem, EventListener
from basecam.exceptions import CameraWarning
from . import commands
__all__ = ["BaseCameraActor", "CameraActor", "BasecamCommand"]
[docs]
class BaseCameraActor(BaseActor):
"""Base class for a camera CLU-like actor class.
Expands a `CLU <https://clu.readthedocs.io/en/latest/>`__ 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 `~clu.actor.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
`~clu.parsers.click.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,kwars
Arguments and keyword arguments to be passed to the actor class.
"""
def __init__(
self,
camera_system: CameraSystem,
*args,
default_cameras: Union[List[str], str, None] = None,
command_parser: clu.parsers.click.CluGroup | None = None,
schema: Optional[str] = "internal",
**kwargs,
):
self._check_is_subclass()
assert camera_system is not None
self.camera_system = camera_system
assert self.camera_system.camera_class
# Add actor to camera class context and all connected cameras.
self.camera_system.camera_class.fits_model.context.update({"__actor__": self})
for camera in self.camera_system.cameras:
camera.fits_model.context.update({"__actor__": self})
#: An `.EventListener` that can be used to wait or respond to events.
self.listener = EventListener()
self.camera_system.notifier.register_listener(self.listener)
self.log: SDSSLogger
self.parser = command_parser or commands.camera_parser
if schema == "internal":
schema = os.path.join(os.path.dirname(__file__), "schema.json")
super().__init__(*args, schema=schema, **kwargs)
# Add commands that depend on what mixins the base camera has
# been subclassed with.
if self.parser == commands.camera_parser:
self._add_optional_commands()
# Output camera_system log messages as keywords.
actor_handler = ActorHandler(
self,
code_mapping={logging.INFO: "d"},
filter_warnings=[CameraWarning, UserWarning],
)
actor_handler.setLevel(logging.INFO)
self.camera_system.logger.addHandler(actor_handler)
self.default_cameras = None
self.set_default_cameras(default_cameras)
def _check_is_subclass(self):
"""Checks if the object is a subclass of a CLU actor."""
error = "BaseCameraActor must be sub-classed along with a CLU actor class."
bases = self.__class__.__bases__
assert issubclass(self.__class__, BaseCameraActor), error
# Check that at least one of the bases is a sublass of BaseActor
for base in bases:
if base == BaseCameraActor:
continue
if issubclass(base, BaseActor):
return
raise RuntimeError(error)
def _add_optional_commands(self):
"""Adds commands and groups based on the mixins present."""
camera_class = self.camera_system.camera_class
assert camera_class is not None
for mixin in camera_class.__bases__:
mixin_name = mixin.__name__
if mixin_name in commands._MIXIN_TO_COMMANDS:
for command in commands._MIXIN_TO_COMMANDS[mixin_name]:
self.parser.add_command(command)
[docs]
def set_default_cameras(self, cameras: Union[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.
"""
if cameras is None:
self.default_cameras = None
return
if isinstance(cameras, str):
self.default_cameras = cameras.split(",")
elif isinstance(cameras, (list, tuple)):
self.default_cameras = list(cameras)
else:
raise ValueError(f"invalid data type for cameras={cameras!r}")
connected_cameras = [camera.name for camera in self.camera_system.cameras]
for camera in self.default_cameras:
if camera not in connected_cameras:
self.log.warning(
f"camera {camera!r} made default but is not connected."
)
[docs]
class CameraActor(BaseCameraActor, JSONActor):
"""A camera actor that replies with JSONs using `~clu.actor.JSONActor`."""
pass
BasecamCommand = Command[BaseCameraActor]