API

Controller

class archon.controller.ArchonCommand(command_string, command_id, controller=None, expected_replies=1, timeout=None)

Bases: Future

Tracks the status and replies to a command sent to the Archon.

ArchonCommand is a Future and can be awaited, at which point the command will have completed or failed.

Parameters:

• command_string (str) – The command to send to the Archon. Will be converted to uppercase.

• command_id (int) – The command id to associate with this message.

• controller – The controller that is running this command.

• expected_replies (Optional[int]) – How many replies to expect from the controller before the command is done.

• timeout (Optional[float]) – Time without receiving a reply after which the command will be timed out. None disables the timeout.

async get_replies()

Yields an asynchronous generator of replies as they are produced.

Return type:

AsyncGenerator[ArchonCommandReply, None]

process_reply(reply)

Processes a new reply to this command.

The Archon can reply to a command of the form >xxCOMMAND (where xx is a 2-digit hexadecimal) with ?xx to indicate failure or <xxRESPONSE. In the latter case the RESPONSE ends with a newline. The Archon can also reply with <xx:bbbbb...bbbb with the : indicating that what follows is a binary string with 1024 characters. In this case the reply does not end with a newline.

Parameters:

reply (bytes) – The received reply, as bytes.

Return type:

ArchonCommandReply | None

succeeded()

Reports the command success status.

Returns True if the command succeeded, or False if it failed, timed out, or if the command is not yet done.

property raw

Returns the raw command sent to the Archon (without the newline).

replies: list[ArchonCommandReply]

List of replies received for this command.

Type:

List of str or bytes

status

The status of the command.

Type:

.ArchonCommandStatus

class archon.controller.ArchonCommandReply(raw_reply, command)

Bases: object

A reply received from the Archon to a given command.

When str(archon_command_reply) is called, the reply (without the reply code or command id) is returned, except when the reply is binary in which case an error is raised.

Parameters:

• raw_reply (bytes) – The raw reply received from the Archon.

• command (ArchonCommand) – The command associated with the reply.

Raises:

.ArchonError – Raised if the reply cannot be parsed.

class archon.controller.ArchonCommandStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Status of an Archon command.

DONE = 1

FAILED = 2

RUNNING = 3

TIMEDOUT = 4

class archon.controller.ArchonController(name, host, port=4242, config=None)

Bases: Device

Talks to an Archon controller over TCP/IP.

Parameters:

• name (str) – A name identifying this controller.

• host (str) – The hostname of the Archon.

• port (int) – The port on which the Archon listens to incoming connections. Defaults to 4242.

• config (dict | None) – Configuration data. Otherwise uses default configuration.

async abort(readout=False)

Aborts the current exposure.

If readout=False, does not trigger a readout immediately after aborting. Aborting does not flush the charge.

Parameters:

readout (bool) –

async expose(exposure_time=1, readout=True)

Integrates the CCD for exposure_time seconds.

Returns immediately once the exposure has begun. If readout=False, does not trigger a readout immediately after the integration finishes. The returned Task waits until the integration is done and, if readout, checks that the readout has started.

Parameters:

• exposure_time (float) –

• readout (bool) –

Return type:

Task

async fetch(buffer_no: int = -1, notifier: Callable[[str], None] | None = None, is_retry: bool = False, *, return_buffer: Literal[False]) → ndarray

async fetch(buffer_no: int = -1, notifier: Callable[[str], None] | None = None, is_retry: bool = False, *, return_buffer: Literal[True]) → tuple[ndarray, int]

async fetch(buffer_no: int = -1, notifier: Callable[[str], None] | None = None, is_retry: bool = False, return_buffer: bool = False) → ndarray

Fetches a frame buffer and returns a Numpy array.

Parameters:

• buffer_no – The frame buffer number to read. Use -1 to read the most recently complete frame.

• notifier – A callback that receives a message with the current operation. Useful when fetch is called by the actor to report progress to the users.

• return_buffer – If True, returns the buffer number returned.

• is_retry – Internal keyword to handle retries. If the buffer fetch does not match the expected size the code will automatically retry fetching the buffer once. If that also fails it will pad the buffer with zeros to the expected size.

Returns:

data – If return_buffer=False, returns the fetched data as a Numpy array. If return_buffer=True returns a tuple with the Numpy array and the buffer number.

async flush(count=2, wait_for=None)

Resets and flushes the detector. Blocks until flushing completes.

Parameters:

• count (int) –

• wait_for (float | None) –

async get_device_status(update_power_bits=True)

Returns a dictionary with the output of the STATUS command.

Parameters:

update_power_bits (bool) –

Return type:

dict[str, Any]

async get_frame()

Returns the frame information.

All the returned values in the dictionary are integers in decimal representation.

Return type:

dict[str, int]

async get_system()

Returns a dictionary with the output of the SYSTEM command.

Return type:

dict[str, Any]

async power(mode=None)

Handles power to the CCD(s). Sets the power status bit.

Parameters:

mode (bool | None) – If None, returns True if the array is currently powered, False otherwise. If True, powers n the array; if False powers if off.

Returns:

state (ArchonPower) – The power state as an ArchonPower flag.

async process_message(line)

Processes a message from the Archon and associates it with its command.

Parameters:

line (bytes) –

Return type:

None

async read_config(save=False)

Reads the configuration from the controller.

Parameters:

save (str | bool) – Save the configuration to a file. If save=True, the configuration will be saved to ~/archon_<controller_name>.acf, or set save to the path of the file to save.

Return type:

tuple[ConfigParser, list[str]]

async readout(force=False, block=True, delay=0, wait_for=None, notifier=None, idle_after=True)

Reads the detector into a buffer.

If force, triggers the readout routine regardless of the detector expected state. If block, blocks until the buffer has been fully written. Otherwise returns immediately. A delay can be passed to slow down the readout by as many seconds (useful for creating photon transfer frames).

Parameters:

• force (bool) –

• block (bool) –

• delay (int) –

• wait_for (float | None) –

• notifier (Callable[[str], None] | None) –

• idle_after (bool) –

async reset(autoflush=True, release_timing=True, update_status=True)

Resets timing and discards current exposures.

async reset_window()

Resets the exposure window.

async send_and_wait(command_string, raise_error=True, **kwargs)

Sends a command to the controller and waits for it to complete.

Parameters:

• command_string (str) – The command to send to the Archon. Will be converted to uppercase.

• raise_error (bool) – Whether to raise an error if the command fails. If False, a warning will be issued.

• kwargs – Other arguments to be sent to send_command.

send_command(command_string, command_id=None, **kwargs)

Sends a command to the Archon.

Parameters:

• command_string (str) – The command to send to the Archon. Will be converted to uppercase.

• command_id (int | None) – The command id to associate with this message. If not provided, a sequential, autogenerated one will be used.

• kwargs – Other keyword arguments to pass to ArchonCommand.

Return type:

ArchonCommand

async send_many(cmd_strs, max_chunk=100, timeout=None)

Sends many commands and waits until they are all done.

If any command fails or times out, cancels any future command. Returns a list of done commands and a list failed commands (empty if all the commands have succeeded). Note that done+pending can be fewer than the length of cmd_strs.

The order in which the commands are sent and done is not guaranteed. If that’s important, you should use send_command.

Parameters:

• cmd_strs (Iterable[str]) – List of command strings to send. The command ids are assigned automatically from available IDs in the pool.

• max_chunk – Maximum number of commands to send at once. After sending, waits until all the commands in the chunk are done. This does not guarantee that max_chunk of commands will be running at once, that depends on the available command ids in the pool.

• timeout (float | None) – Timeout for each single command.

Return type:

tuple[list[ArchonCommand], list[ArchonCommand]]

async set_autoflush(mode)

Enables or disables autoflushing.

Parameters:

mode (bool) –

set_binary_reply_size(size)

Sets the size of the binary buffers.

Parameters:

size (int) –

async set_param(param, value, force=False, silent=False)

Sets the parameter param to value value calling FASTLOADPARAM.

Parameters:

• param (str) –

• value (int) –

• force (bool) –

• silent (bool) –

Return type:

ArchonCommand | None

async set_window(lines=None, pixels=None, preskiplines=None, postskiplines=None, preskippixels=None, postskippixels=None, overscanlines=None, overscanpixels=None, hbin=None, vbin=None)

Sets the CCD window.

Parameters:

• lines (int | None) –

• pixels (int | None) –

• preskiplines (int | None) –

• postskiplines (int | None) –

• preskippixels (int | None) –

• postskippixels (int | None) –

• overscanlines (int | None) –

• overscanpixels (int | None) –

• hbin (int | None) –

• vbin (int | None) –

async start(reset=True, read_acf=True)

Starts the controller connection. If reset=True, resets the status.

Parameters:

• reset (bool) –

• read_acf (bool) –

async stop()

Stops the client and cancels the command tracker.

update_status(bits, mode='on', notify=True)

Updates the status bitmask, allowing to turn on, off, or toggle a bit.

Parameters:

bits (ControllerStatus | list[ControllerStatus]) –

async write_config(input, applyall=False, applymods=[], poweron=False, timeout=None, overrides={}, notifier=None)

Writes a configuration file to the contoller.

Parameters:

• input (str | PathLike[str]) – The path to the configuration file to load. It must be in INI format with a section called [CONFIG]. It can also be a string containing the configuration itself.

• applyall (bool) – Whether to run APPLYALL after successfully sending the configuration.

• applymods (list[str]) – A list of apply commands to send to modules (e.g., ['LOADTIMING', 'APPLYMOD2']).

• poweron (bool) – Whether to run POWERON after successfully sending the configuration. Requires applyall=True.

• timeout (float | None) – The amount of time to wait for each command to succeed. If None, reads the value from the configuration entry for timeouts.write_config_timeout.

• overrides (dict) – A dictionary with configuration lines to be overridden. Must be a mapping of keywords to replace, including the module name (e.g., MOD11/HEATERAP), to the new values.

• notifier (Callable[[str], None] | None) – A callback that receives a message with the current operation being performed. Useful when write_config is called by the actor to report progress to the users.

async write_line(keyword, value, mod=None, apply=True)

Write a single line to the controller, replacing the current configuration.

Parameters:

• keyword (str) – The config keyword to replace. If mod=None, must include the module name (e.g., MOD11/HEATERAP); otherwise the module is added from mod. Modules and module keywords can be separated by slashes or backlashes.

• value (int | float | str) – The value of the keyword.

• mod (str | None) – The name of the keyword module, e.g., MOD11.

• apply (bool | str) – Whether to re-apply the configuration for the module. If apply is a string, defines the command to be used to apply the new setting, e.g., APPLYCDS.

async yield_status()

Asynchronous generator yield the status of the controller.

Return type:

AsyncIterator[ControllerStatus]

property status: ControllerStatus

Returns the status of the controller as a ControllerStatus enum type.

class archon.controller.ControllerStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Flag

Status of the Archon controller.

get_flags()

Returns the the flags that compose the bit.

ACTIVE = 116

ERROR = 128

ERRORED = 1152

EXPOSING = 4

FETCHING = 32

FLUSHING = 64

IDLE = 2

POWERBAD = 1024

POWEROFF = 512

POWERON = 256

READING = 16

READOUT_PENDING = 8

UNKNOWN = 1

class archon.controller.ModType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Module type codes.

AD = 2

ADF = 13

ADLN = 15

ADX = 14

DRIVER = 1

HEATER = 5

HEATERX = 11

HS = 7

HVBIAS = 4

HVXBIAS = 8

LVBIAS = 3

LVDS = 10

LVXBIAS = 9

NONE = 0

UNKNOWN = 16

XVBIAS = 12

Actor

class archon.actor.actor.ArchonActor(*args, controllers=(), run_recovery_on_start=True, **kwargs)

Bases: ArchonBaseActor, AMQPActor

Archon actor based on the AMQP protocol.

Parameters:

• controllers (tuple[ArchonController, ...]) –

• run_recovery_on_start (bool) –

class archon.actor.actor.ArchonBaseActor(*args, controllers=(), run_recovery_on_start=True, **kwargs)

Bases: BaseActor

Archon controller base actor.

This class is intended to be subclassed with a specific actor class (normally AMQPActor or LegacyActor).

Parameters:

• controllers (tuple[ArchonController, ...]) – The list of ArchonController instances to manage.

• run_recovery_on_start (bool) –

CONTROLLER_CLASS

alias of ArchonController

DELEGATE_CLASS

alias of ExposureDelegate

classmethod from_config(config, *args, **kwargs)

Creates an actor from a configuration file.

async start()

Start the actor and connect the controllers.

async stop()

Shuts down all the remaining tasks.

BASE_CONFIG: ClassVar[str | Dict | None] = None

controllers

A mapping of controller name to controller.

Type:

dict[str, ArchonController]

is_legacy: bool = False

parser: ClassVar[click.Group] = <CluGroup command-parser>

Parameters:

• args (Any) –

• kwargs (Any) –

Return type:

Any

class archon.actor.delegate.ExposeData(exposure_time, flavour, controllers, start_time=<factory>, end_time=None, mjd=0, exposure_no=0, header=<factory>, delay_readout=0, window_mode=None, window_params=<factory>)

Bases: object

Data about the ongoing exposure.

Parameters:

• exposure_time (float) –

• flavour (str) –

• controllers (list[ArchonController]) –

• start_time (Time) –

• end_time (Time | None) –

• mjd (int) –

• exposure_no (int) –

• header (Dict[str, Any]) –

• delay_readout (int) –

• window_mode (str | None) –

• window_params (dict) –

controllers: list[ArchonController]

delay_readout: int = 0

end_time: Time | None = None

exposure_no: int = 0

exposure_time: float

flavour: str

header: Dict[str, Any]

mjd: int = 0

start_time: Time

window_mode: str | None = None

window_params: dict

class archon.actor.delegate.ExposureDelegate(actor)

Bases: Generic[Actor_co]

Handles the exposure workflow.

Parameters:

actor (Actor_co) –

async build_base_header(controller, ccd_name)

Returns the basic header of the FITS file.

Parameters:

• controller (ArchonController) –

• ccd_name (str) –

Return type:

dict[str, list]

async check_expose()

Performs a series of checks to confirm we can expose.

Return type:

bool

async expose(command, controllers, flavour='object', exposure_time=1.0, readout=True, window_mode=None, window_params={}, seqno=None, **readout_params)

Parameters:

• command (Command[Actor_co]) –

• controllers (List[ArchonController]) –

• flavour (str) –

• exposure_time (float | None) –

• readout (bool) –

• window_mode (str | None) –

• window_params (dict) –

• seqno (int | None) –

Return type:

bool

async expose_cotasks()

Tasks that will be executed concurrently with readout.

There is no guarantee that this coroutine will be waited or that it will complete before the shutter closes and the readout begins. To ensure that the expose tasks have completed, await the task in self._expose_cotasks.

fail(message=None)

Fails a command.

Parameters:

message (str | None) –

async fetch_data(controller)

Fetches the buffer and compiles the header.

Parameters:

controller (ArchonController) –

async post_process(fdata)

Custom post-processing.

This routine can be overridden to perform custom post-processing of the fetched data. It is called with the data, header, and other metadata for each CCD. It must modify the data in place and return None.

Parameters:

fdata (FetchDataDict) – A dictionary of fetched data with the data array, a header dictionary, the controller and ccd name, and the filename to which the data will be saved.

async pre_expose(controllers)

A routine that runs before integration begins.

Parameters:

controllers (List[ArchonController]) –

async readout(command, extra_header={}, delay_readout=0, write=True)

Reads the exposure, fetches the buffer, and writes to disk.

Parameters:

• command (Command[Actor_co]) –

• delay_readout (int) –

• write (bool) –

async readout_cotasks()

Tasks that will be executed concurrently with readout.

This routine can be overridden to run processes that do not need to wait until post_process. For example, reading out sensors and telescope data can happen here to save time.

reset()

Resets the exposure delegate.

async shutter(open=False)

Operate the shutter.

Return type:

bool

async static write_to_disk(ccd_data, excluded_cameras=[], write_async=True, write_engine='astropy')

Writes ccd data to disk.

Parameters:

• ccd_data (FetchDataDict) –

• excluded_cameras (list[str]) –

• write_async (bool) –

• write_engine (str) –

Return type:

str | None

property command

Returns the current command.

class archon.actor.delegate.FetchDataDict

Bases: TypedDict

Dictionary of fetched data.

buffer: int

ccd: str

controller: str

data: DataArray

exposure_no: int

filename: str

header: dict[str, list]

archon.actor.tools.check_controller(command, controller)

Performs sanity check in the controller.

Outputs error messages if a problem is found. Return False if the controller is not in a valid state.

Parameters:

• command (Command) –

• controller (ArchonController) –

Return type:

bool

archon.actor.tools.controller(f)

Parameters:

f (FC) –

Return type:

FC

archon.actor.tools.error_controller(command, controller, message)

Issues a error_controller message.

Parameters:

• command (Command) –

• controller (ArchonController) –

• message (str) –

archon.actor.tools.get_schema()

Returns the default JSONschema for Archon actors.

archon.actor.tools.open_with_lock(filename, mode='r')

Opens a file and adds an advisory lock on it.

Parameters:

• filename (PathLike) – The path to the file to open.

• mode (str) – The mode in which the file will be open.

Raises:

BlockingIOError – If the file is already locked.

Return type:

Generator[IO[Any], None, None]

archon.actor.tools.parallel_controllers(check=True)

A decarator that executes the same command for multiple controllers.

When decorated with parallel_controllers, the command gets an additional option controllers which allows to pass a list of controllers. If not controllers are passed, all the available controllers are used.

The callback is called for each one of the selected controllers as the second argument (after the command itself). All the controller callbacks are executed concurrently as tasks. If one of the tasks raises an exception, all the other tasks are immediately cancelled and the command is failed.

And example of use is

@parser.command()
@click.argument("archon_command", metavar="COMMAND", type=str)
@parallel_controllers()
async def talk(command, controller, archon_command):
    ...

where controller receives an instance of ArchonController. Within the callback you should not fail or finish the command; instead use error() or warning(). Your replies to the users must indicate to what controller they refer. If you want to force all the concurrent tasks to fail, raise an exception.