API Reference¶

Core classes¶

class gort.core.ActorReply(actor, command, replies=<factory>) → None[source]¶

Bases: object

A reply to an actor command.

actor¶: The RemoteActor that generated the reply.

command¶: The CLU command instance.

replies¶: A list of replies generated by the command.

flatten()[source]¶

Returns a flattened dictionary of replies.

Note that if a keyword has been output multiple times, only the last value is retained.

get(key)[source]¶: Returns the first occurrence of a keyword in the reply list.

class gort.core.RemoteActor(client, name, device=None)[source]¶

Bases: object

A programmatic representation of a remote actor.

async init() → Self[source]¶

Initialises the representation of the actor.

Return type:: Self

async refresh()[source]¶: Refresesh the command list.

async send_raw_command(*args, **kwargs)[source]¶

Sends a raw command to the actor.

The parameters are the same as CLU’s AMQPClient.send_command() with the exception of the consumer name, which is replaced with the current actor name.

class gort.core.RemoteCommand(remote_actor, model, parent=None)[source]¶

Bases: object

Representation of a remote command.

get_command_string(*args, **kwargs)[source]¶: Gets the command string for a set of arguments.

class gort.gort.Gort(*args, verbosity=None, **kwargs)[source]¶

Bases: GortClient

Gort’s robotic functionality.

Gort is subclass of GortClient that implements higher-level robotic functionality. This is the class a user will normally instantiate and interact with.

Parameters:

args – Arguments to pass to GortClient.
kwargs – Arguments to pass to GortClient.
verbosity (str | None) – The level of logging verbosity.

async cleanup(readout=True, turn_lamps_off=True)[source]¶: Executes the shutdown sequence.

async emergency_close()[source]¶: Parks and closes the telescopes.

async execute_recipe(recipe, **kwargs)[source]¶

Executes a recipe.

Parameters:

recipe (str) – The name of the recipe to execute.
kwargs – Arguments to be passed to the recipe.

async observe(n_tiles=None, adjust_focus=True, show_progress=None, disable_tile_on_error=True, wait_if_no_tiles=60.0)[source]¶: Runs a fully automatic science observing loop.

async observe_tile(tile=None, ra=None, dec=None, pa=0.0, use_scheduler=True, exposure_time=900.0, n_exposures=1, async_readout=True, keep_guiding=False, guide_tolerance=1.0, acquisition_timeout=180.0, show_progress=None, run_cleanup=True, adjust_focus=True, cleanup_on_interrrupt=True)[source]¶

Performs all the operations necessary to observe a tile.

Parameters:

tile (Tile | int | None) – The tile_id to observe, or a Tile object. If not provided, observes the next tile suggested by the scheduler (requires use_scheduler=True).
ra (float | None) – The RA and Dec where to point the science telescopes. The other telescopes are pointed to calibrators that fit the science pointing. Cannot be used with tile.
dec (float | None) – The RA and Dec where to point the science telescopes. The other telescopes are pointed to calibrators that fit the science pointing. Cannot be used with tile.
pa (float) – Position angle of the IFU. Defaults to PA=0.
use_scheduler (bool) – Whether to use the scheduler to determine the tile_id or select calibrators.
exposure_time (float) – The length of the exposure in seconds.
n_exposures (int) – Number of exposures to take while guiding.
async_readout (bool) – Whether to wait for the readout to complete or return as soon as the readout begins. If False, the exposure is registered but the observation is not finished. This should be True during normal science operations to allow the following acquisition to occur during readout.
keep_guiding (bool) – If True, keeps the guider running after the last exposure. This should be False during normal science operations.
guide_tolerance (float) – The guide tolerance in arcsec. A telescope will not be considered to be guiding if its separation to the commanded field is larger than this value.
acquisition_timeout (float) – The maximum time allowed for acquisition. In case of timeout the acquired fields are evaluated and an exception is raised if the acquisition failed.
show_progress (bool | None) – Displays a progress bar with the elapsed exposure time.
run_cleanup (bool) – Whether to run the cleanup routine.
adjust_focus (bool) – Adjusts the focuser positions based on temperature drift before starting the observation. This works best if the focus has been initially determined using a focus sweep.
cleanup_on_interrrupt (bool) – If True, registers a signal handler to catch interrupts and run the cleanup routine.

async run_script(script)[source]¶: Runs a script.

run_script_sync(script)[source]¶: Runs a script.

async shutdown(**kwargs)[source]¶: Executes the shutdown sequence.

async startup(**kwargs)[source]¶: Executes the startup sequence.

class gort.gort.GortClient(host='lvm-hub.lco.cl', port=5672, user='guest', password='guest', use_rich_output=True, log_file_path=None)[source]¶

Bases: AMQPClient

The main gort client, used to communicate with the actor system.

A subclass of AMQPClient with defaults for host and logging, it loads the GortDeviceSet and RemoteActor instances for the LVM system.

Parameters:

host (str) – The host on which the RabbitMQ exchange is running.
port (int) – The port on which the RabbitMQ exchange listens to connections.
user (str) – The user to connect to the exchange.
password (str) – The password to connect to the exchange.
use_rich_output (bool) – If True, uses rich to provide colourised tracebacks and prettier outputs.
log_file_path (Union[str, Path, Literal[False], None]) – The path where to save GORT’s log. File logs are always saved with DEBUG logging level. If None, a temporary file will be used whose path can be retrieved by calling get_log_path. If False, no file logging will happen.

add_actor(actor, device=None)[source]¶

Adds an actor to the programmatic API.

Parameters:

actor (str) – The name of the actor to add.
device (GortDevice | None) – A device associated with this actor.

add_device(class_, *args, **kwargs) → DevType[source]¶

Adds a new device or device set to Gort.

Return type:: TypeVar(DevType, bound= GortDeviceSet | GortDevice)

get_log_path()[source]¶: Returns the path of the log file. None if not logging to file.

async init() → Self[source]¶

Initialises the client.

Returns:: The same instance of GortClient after initialisation.
Return type:: object

set_verbosity(verbosity=None)[source]¶

Sets the level of verbosity to debug, info, or warning.

Parameters:: verbosity (str | int | None) – The level of verbosity. Can be a string level name, an integer, or None, in which case the default verbosity will be used.

property connected¶: Returns True if the client is connected.

class gort.gort.GortDevice(gort, name, actor)[source]¶

Bases: object

A gort-managed device.

Parameters:

gort (GortClient) – The GortClient instance.
name (str) – The name of the device.
actor (str) – The name of the actor used to interface with this device. The actor is added to the list of RemoteActor in the GortClient.

async init()[source]¶

Runs asynchronous tasks that must be executed on init.

If the device is part of a DeviceSet, this method is called by DeviceSet.init.

log_replies(reply, skip_debug=True)[source]¶: Outputs command replies.

write_to_log(message, level='debug', header=None)[source]¶

Writes a message to the log with a custom header.

Parameters:

message (str) – The message to log.
level (str) – The level to use for logging: 'debug', 'info', 'warning', or 'error'.
header (str | None) – The header to prepend to the message. By default uses the device name.

class gort.gort.GortDeviceSet(gort, data, **kwargs)[source]¶

Bases: dict[str, GortDeviceType], Generic[GortDeviceType]

A set to gort-managed devices.

Devices can be accessed as items of the GortDeviceSet dictionary or using dot notation, as attributes.

Parameters:

gort (GortClient) – The GortClient instance.
data (dict[str, dict]) – A mapping of device to device info. Each device must at least include an actor key with the actor to use to communicated with the device. Any other information is passed to the GortDevice on instantiation.
kwargs – Other keyword arguments to pass wo the device class.

async call_device_method(method, *args, **kwargs)[source]¶

Calls a method in each one of the devices.

Parameters:

method (Callable) – The method to call. This must be the abstract class method, not the method from an instantiated object.
args – Arguments to pass to the method.
kwargs – Arguments to pass to the method.

async init()[source]¶: Runs asynchronous tasks that must be executed on init.

async restart()[source]¶

Restarts the set deployments and resets all controllers.

Returns:: A boolean indicting if the entire restart procedure succeeded.
Return type:: result

async send_command_all(command, *args, devices=None, **kwargs)[source]¶

Sends a command to all the devices.

Parameters:

command (str) – The command to call.
args – Arguments to pass to the RemoteCommand.
kwargs – Arguments to pass to the RemoteCommand.

write_to_log(message, level='debug', header=None)[source]¶

Writes a message to the log with a custom header.

Parameters:

message (str) – The message to log.
level (str) – The level to use for logging: 'debug', 'info', 'warning', or 'error'.
header (str | None) – The header to prepend to the message. By default uses the class name.

Devices¶

class gort.devices.enclosure.Enclosure(gort, name, actor, **kwargs)[source]¶

Bases: GortDevice

Class representing the LVM enclosure.

async close(park_telescopes=True, force=False)[source]¶

Close the enclosure dome.

Parameters:

park_telescopes (bool) – Move the telescopes to the park position before closing the enclosure to prevent dust or other debris falling on them.
force (bool) – Tries to closes the dome even if the system believes it is already closed.

async get_door_status()[source]¶: Returns the status of the door and lock.

async is_closed()[source]¶: Returns True if the enclosure is closed.

async is_local()[source]¶: Returns True if the enclosure is in local mode.

async is_open()[source]¶: Returns True if the enclosure is open.

async open(park_telescopes=True)[source]¶

Open the enclosure dome.

Parameters:: park_telescopes (bool) – Move the telescopes to the park position before opening the enclosure to prevent dust or other debris falling on them.

async restart()[source]¶: Restarts the lvmecp deployment.

async status(get_registers=False)[source]¶: Retrieves the status of the power outlet.

async stop()[source]¶: Stop the enclosure dome.

class gort.devices.enclosure.Light(enclosure, name)[source]¶

Bases: object

An enclosure light.

async off()[source]¶: Turns on the light.

async on()[source]¶: Turns on the light.

async status() → bool[source]¶

Returns a boolean with the light status.

Return type:: bool

async toggle()[source]¶: Turns on the light.

class gort.devices.enclosure.Lights(enclosure)[source]¶

Bases: object

Controls the enclosure lights.

async dome_all_off()[source]¶: Turns off all the lights in the dome.

class gort.devices.ag.AG(gort, name, actor)[source]¶

Bases: GortDevice

Class representing an AG camera.

async expose(exposure_time=5.0, flavour='object', **kwargs)[source]¶

Exposes the cameras.

Parameters:

exposure_time (float) – The amount of time, in seconds, to expose the cameras.
flavour (str) – The type of image to take, one of 'bias', 'dark', 'flat', or 'object'.
kwargs – Any parameters to send to the lvmcam expose command.

async reconnect()[source]¶: Reconnect the AG cameras.

async status()[source]¶: Returns the status of the AG.

class gort.devices.ag.AGSet(gort, data, **kwargs)[source]¶

Bases: GortDeviceSet[AG]

A set of auto-guiders.

async expose(exposure_time=5.0, flavour='object', **kwargs)[source]¶

Exposes the cameras.

Parameters:

exposure_time (float) – The amount of time, in seconds, to expose the cameras.
flavour (str) – The type of image to take, one of 'bias', 'dark', 'flat', or 'object'.
kwargs – Any parameters to send to the lvmcam expose command.

async list_alive_cameras()[source]¶

Returns a list of cameras found alive and well.

Currently when a camera has disconnected, the status command still reports it as “online”, but it doesn’t report its temperature or other parameters.

async reconnect()[source]¶

Reconnects all the AG cameras.

Returns:: A list of cameras available after reconnecting.
Return type:: cameras

class gort.devices.guider.Guider(gort, name, actor, **kwargs)[source]¶

Bases: GortDevice

Class representing a guider.

async adjust_focus()[source]¶: Adjusts the focus position based on the current bench temperature.

async apply_corrections(enable=True)[source]¶: Enable/disable corrections being applied to the axes.

async expose(*args, continuous=False, **kwargs)[source]¶

Exposes this telescope cameras.

Parameters:

args – Arguments to be passed to the guider expose command.
kwargs – Arguments to be passed to the guider expose command.
continuous (bool) – Whether to expose the camera continuously. If False it takes a single exposure.

async focus(inplace=False, sweep=True, guess=None, step_size=0.2, steps=7, exposure_time=5.0)[source]¶

Focus the telescope.

Parameters:

inplace – If True, focuses the telescope where it is pointing at. Otherwise points to zenith.
sweep (bool) – Performs a focus sweep around the initial guess position to find the best focus. If False, the focus position is determined based on the current bench temperature.
guess (float | None) – The initial guess for the focuser position. If None, the initial guess is determined based on the current bench temperature.
step_size (float) – The size, in focuser units, of each step.
steps (int) – The total number of step points. Must be an odd number.
exposure_time (float) – The exposure time for each step.

async guide(ra=None, dec=None, exposure_time=5.0, pixel=None, monitor=True, **guide_kwargs)[source]¶

Starts the guide loop.

This command blocks until stop is called.

Parameters:

ra (float | None) – The coordinates to acquire. If None, the current telescope coordinates are used.
dec (float | None) – The coordinates to acquire. If None, the current telescope coordinates are used.
exposure_time (float) – The exposure time of the AG integrations.
pixel (tuple[float, float] | str | None) – The pixel on the master frame on which to guide. Defaults to the central pixel. This can also be the name of a known pixel position for this telescope, e.g., 'P1-1' for spec.
monitor (bool) – Whether to monitor the guide loop and output the average and last guide metrics every 30 seconds.
guide_kwargs – Other keyword arguments to pass to lvmguider guide. The includes the pa argument that if not provided is assumed to be zero.

async monitor(ra=None, dec=None, exposure_time=5.0, sleep=60)[source]¶

Guides at a given position, sleeping between exposures.

This is a convenience function mainly to monitor transparency during bad weather conditions. The telescope will be slewed to a given position (default to zenith) and guide with a low cadence. This results in the guider keywords, including transparency and FWHM, being output and the plots in Grafana being updated.

After cancelling the monitoring make sure to stop the guiders with the Guider.stop method.

Parameter¶

ra,dec: The coordinates to acquire. If None, the current zenith coordinates are used.
exposure_time: The exposure time of the AG integrations.
sleep: The time to sleep between exposures (seconds).

async set_pixel(pixel=None)[source]¶

Sets the master frame pixel on which to guide.

Parameters:: pixel (tuple[float, float] | str | None) – The pixel on the master frame on which to guide. Defaults to the central pixel. This can also be the name of a known pixel position for this telescope, e.g., 'P1-1' for spec.

async stop() → None[source]¶

Stops the guide loop.

Parameters:: wait_until_stopped – Blocks until the guider is idle.
Return type:: None

async wait_until_guiding(guide_tolerance=None, timeout=None) → tuple[bool, GuiderStatus, float | None, bool][source]¶

Waits until the guider has converged.

Parameters:

guide_tolerance (float | None) – The minimum separation, in arcsec, between the measured and desired positions that needs to be reached before returning. If None, waits until guiding (as opposed to acquisition) begins.
timeout (float | None) – Maximum time, in seconds, to wait before returning. If None, waits indefinitely. If the timeout is reached it does not raise an exception.

Return type:

tuple[bool, GuiderStatus, float | None, bool]

Returns:

reached – Whether the desired minimum separation was reached.
status – The current GS.
separation – The current separation.
timedout – True if the acquisition timed out.

property ag¶: Gets the AG device associated with this guider.

property telescope¶: Gets the Telescope device associated with this guider.

class gort.devices.guider.GuiderSet(gort, data, **kwargs)[source]¶

Bases: GortDeviceSet[Guider]

A set of telescope guiders.

async adjust_focus()[source]¶: Adjusts the focus position based on the current bench temperature.

async apply_corrections(enable=True)[source]¶: Enable/disable corrections being applied to the axes.

async expose(*args, continuous=False, **kwargs)[source]¶

Exposes all the cameras using the guider.

Parameters:

args – Arguments to be passed to Guider.expose.
kwargs – Arguments to be passed to Guider.expose.
continuous (bool) – Whether to expose the camera continuously. If False it takes a single exposure.

async focus(inplace=False, guess=None, step_size=0.5, steps=7, exposure_time=5.0)[source]¶

Focus all the telescopes.

Parameters:

inplace – If True, focuses the telescopes where they are pointing at. Otherwise points to zenith.
guess (float | dict[str, float] | None) – The initial guesses for focuser position. If None, an estimate based on the current bench temperatures is used. It can also be a float value, which will be used for all telescopes, or a mapping of telescope name to guess value. Missing values will default to the configuration value.
step_size (float) – The size, in focuser units, of each step.
steps (int) – The total number of step points. Must be an odd number.
exposure_time (float) – The exposure time for each step.

async guide(*args, **kwargs)[source]¶

Guide on all telescopes.

Parameters:

args – Arguments to be passed to Guider.guide.
kwargs – Arguments to be passed to Guider.guide.

async monitor(*args, **kwargs)[source]¶

Guides at a given position, sleeping between exposures.

See Guider.monitor for details.

async stop()[source]¶: Stops the guide loop on all telescopes.

async take_darks()[source]¶: Takes AG darks.

async wait_until_guiding(names=None, guide_tolerance=None, timeout=None)[source]¶

Waits until the guiders have converged.

Parameters:

names (list[str] | None) – List of telescopes to wait for convergence.
guide_tolerance (float | None) – The minimum separation, in arcsec, between the measured and desired positions that needs to be reached before returning. If None, waits until guiding (as opposed to acquisition) begins.
timeout (float | None) – Maximum time, in seconds, to wait before returning. If None, waits indefinitely. If the timeout is reached it does not raise an exception.

Returns:

A dictionary with the telescope names and a tuple indicating whether the desired minimum separation was reached. The current GuiderStatus, and the current separation for that telescope.

Return type:

status

class gort.devices.nps.NPS(gort, name, actor, **kwargs)[source]¶

Bases: GortDevice

Class representing a networked power switch.

async all_off()[source]¶: Turns off all the outlets.

async off(outlet)[source]¶: Turns an outlet on.

async on(outlet)[source]¶: Turns an outlet on.

async status(outlet=None)[source]¶: Retrieves the status of the power outlet.

class gort.devices.nps.NPSSet(gort, data, **kwargs)[source]¶

Bases: GortDeviceSet[NPS]

A set of networked power switches.

class gort.devices.telemetry.Telemetry(gort, name, actor, **kwargs)[source]¶

Bases: GortDevice

Telemetry sensors.

async status()[source]¶: Retrieves the status of sensors.

class gort.devices.telemetry.TelemetrySet(gort, data, **kwargs)[source]¶

Bases: GortDeviceSet[Telemetry]

A set of telemetry sensors.

class gort.devices.spec.IEB(gort, name, actor)[source]¶

Bases: GortDevice

A class representing an Instrument Electronics Box.

async close(devices)[source]¶

Closes a device or list of devices.

Parameters:: device – The device to act on. Either 'shutter', 'hartmann_left', or 'hartmann_right'. Can be a list of devices to close.

async do(devices, action)[source]¶

Performs an action on a device. Powers the device if needed.

Parameters:

device – The device to act on. Either 'shutter', 'hartmann_left', or 'hartmann_right'. Can be a list of devices to modify.
action (str) – The action to perform. Can be 'open', 'close', 'home', or 'init'.

async home(devices)[source]¶

Homes a device or list of devices.

Parameters:: device – The device to act on. Either 'shutter', 'hartmann_left', or 'hartmann_right'. Can be a list of devices to home.

async init(devices, home=True)[source]¶

Initialises a device or list of devices.

Parameters:

device – The device to act on. Either 'shutter', 'hartmann_left', or 'hartmann_right'. Can be a list of devices to initialise.
home (bool) – If True homes the devices after initialising them.

async open(devices)[source]¶

Opens a device or list of devices.

Parameters:: device – The device to act on. Either 'shutter', 'hartmann_left', or 'hartmann_right'. Can be a list of devices to open.

async power(devices, on=True)[source]¶

Powers on/off the shutter or Hartmann doors.

Parameters:

device – The device to power on/off. Either 'shutter', 'hartmann_left', or 'hartmann_right'. Can be a list of devices to modify.
on (bool) – If True powers on the device; otherwise powers it down.

async status()[source]¶: Returns the status of the IEB.

class gort.devices.spec.Spectrograph(gort, name, actor, **kwargs)[source]¶

Bases: GortDevice

Class representing an LVM spectrograph functionality.

async abort()[source]¶: Aborts an ongoing exposure.

async expose(**kwargs)[source]¶: Exposes the spectrograph.

async initialise()[source]¶: Initialises the spectrograph and flashes the ACF configuration file.

async is_exposing()[source]¶: Returns True if the spectrograph is exposing.

async is_idle()[source]¶: Returns True if the spectrograph is idle and ready to expose.

async is_reading()[source]¶: Returns True if the spectrograph is idle and ready to expose.

async reset(full=False)[source]¶: Resets the spectrograph to a valid state.

async status(simple=False)[source]¶

Retrieves the status of the telescope.

Parameters:: simple (bool) – If True returns a short version of the status.

class gort.devices.spec.SpectrographSet(gort, data, **kwargs)[source]¶

Bases: GortDeviceSet[Spectrograph]

A set of LVM spectrographs.

async abort()[source]¶: Aborts an ongoing exposure.

async are_idle()[source]¶: Returns True if all the spectrographs are idle and ready to expose.

async calibrate(sequence='normal', slew_telescopes=True, park_after=False, show_progress=None)[source]¶

Runs the calibration sequence.

Parameters:

sequence (str | dict) – The name calibration sequence to execute. It can also be a dictionary with the calibration sequence definition that follows the calibration schema.
slew_telescopes (bool) – Whether to move the telescopes to point to the FF screen.
park_after (bool) – Park the telescopes after a successful calibration sequence.
show_progress (bool | None) – Displays a progress bar with the elapsed exposure time.

async expose(exposure_time=None, flavour=None, header=None, show_progress=None, async_readout=False, count=1, object=None, specs=None, **kwargs) → Exposure | list[Exposure][source]¶

Exposes the spectrographs.

Parameters:

exposure_time (float | None) – The exposure time. If not set, assumes this must be a bias.
flavour (str | None) – The exposure type, either 'object', 'arc', 'flat', 'dark', or 'bias'
header (dict | None) – Additional data to add to the headers.
show_progress (bool | None) – Displays a progress bar with the elapsed exposure time. If None (the default), will show the progress bar only in interactive sessions.
async_readout (bool) – Returns after integration completes. Readout is initiated but handled asynchronously and can be await by awaiting the returned Exposure object.
count (int) – The number of exposures to take.
object (str | None) – A string that will be stored in the OBJECT header keyword.
specs (Optional[Sequence[str]]) – List the spectrographs to expose. Defaults to all.

Returns:

The numbers of the exposed frames. If count is greater than one, returns a list of exposures.

Return type:

exp_nos

get_calibration_sequence(sequence)[source]¶

Returns a dictionary with the configuration for a calibration sequence.

Parameters:: sequence (str) – The name calibration sequence.
Returns:: The calibration sequence dictionary. This dictionary can be altered and then passed to calibrate to execute the modified sequence. The returned dictionary if a deep copy of the original sequence; modifying it won’t modify the original sequence.
Return type:: sequence_dict

get_expno()[source]¶: Returns the next exposure sequence number.

async initialise()[source]¶: Initialises the spectrographs and flashes the ACF configuration file.

async reset(full=False)[source]¶: Reset the spectrographs.

async status(simple=False) → dict[str, dict][source]¶

Collects the status of each spectrograph.

Parameters:: simple (bool) – If True returns a short version of the status.
Return type:: dict[str, dict]

class gort.devices.telescope.FibSel(gort, name, actor)[source]¶

Bases: MoTanDevice

A device representing the fibre mask in the spectrophotometric telescope.

async home()[source]¶: Homes the fibre selector.

list_positions() → list[str][source]¶

Returns a list of valid positions.

Return type:: list[str]

async move_relative(steps)[source]¶: Move the mask a number of motor steps relative to the current position.

async move_to_position(position)[source]¶

Moves the spectrophotometric mask to the desired position.

Parameters:: position (str | int) – A position in the form PN-M where N=1,2 and M=1-12, in which case the mask will rotate to expose the fibre with that name. If position is a number, moves the mask to that value.

async status()[source]¶: Returns the status of the fibre selector.

SLEW_DELAY: ClassVar[float | dict[str, float]] = 0¶: Artificial delay introduced to prevent all motors to slew at the same time.

class gort.devices.telescope.Focuser(gort, name, actor)[source]¶

Bases: MoTanDevice

A device representing a focuser.

async home(restore_position=True)[source]¶

Homes the focuser.

Parameters:: restore_position (bool) – Whether to restore the previous focuser position after homing.

async move(dts)[source]¶: Move the focuser to a position in DT.

async status()[source]¶: Returns the status of the focuser.

SLEW_DELAY: ClassVar[float | dict[str, float]] = 0¶: Artificial delay introduced to prevent all motors to slew at the same time.

class gort.devices.telescope.KMirror(gort, name, actor)[source]¶

Bases: MoTanDevice

A device representing a K-mirror.

async home()[source]¶: Homes the k-mirror.

async move(degs)[source]¶

Move the k-mirror to a position in degrees. Does NOT track after the move.

Parameters:: degs (float) – The position to which to move the k-mirror, in degrees.

async park()[source]¶: Park the k-mirror at 90 degrees.

async slew(ra, dec, offset_angle=0.0, stop_degs_before=0.0)[source]¶

Moves the mirror to the position for ra, dec and starts slewing.

Parameters:

ra (float) – Right ascension of the field to track, in degrees.
dec (float) – Declination of the field to track, in degrees.
offset_angle (float) – Derotation offset in degrees. This value is converted to the -180 to 180 deg range before sending it to the k-mirror.
stop_degs_before (float) – Number of degrees to stop before reaching the desired position angle. This has the effect of actually slewing to offset_angle-stop_degs_before. Useful if we want to be sure that positive offsets will be applied without backlash.

async status()[source]¶: Returns the status of the k-mirror.

SLEW_DELAY: ClassVar[float | dict[str, float]] = 0¶: Artificial delay introduced to prevent all motors to slew at the same time.

class gort.devices.telescope.MoTanDevice(gort, name, actor)[source]¶

Bases: GortDevice

A TwiceAsNice device.

async is_moving()[source]¶: Is the device moving.

async is_reachable()[source]¶: Is the device reachable?

async slew_delay()[source]¶: Sleeps the SLEW_DELAY amount.

SLEW_DELAY: ClassVar[float | dict[str, float]] = 0¶: Artificial delay introduced to prevent all motors to slew at the same time.

class gort.devices.telescope.Telescope(gort, name, actor, **kwargs)[source]¶

Bases: GortDevice

Class representing an LVM telescope functionality.

async goto_coordinates(ra=None, dec=None, pa=0.0, alt=None, az=None, kmirror=True, kmirror_kwargs={}, altaz_tracking=False, force=False, retry=True)[source]¶

Moves the telescope to a given RA/Dec or Alt/Az.

Parameters:

ra (float | None) – Right ascension coordinates to move to, in degrees.
dec (float | None) – Declination coordinates to move to, in degrees.
pa (float) – Position angle of the IFU. Defaults to PA=0.
alt (float | None) – Altitude coordinates to move to, in degrees.
az (float | None) – Azimuth coordinates to move to, in degrees.
kmirror (bool) – Whether to move the k-mirror into position. Only when the coordinates provided are RA/Dec.
kmirror_kwargs (dict) – Dictionary of keyword arguments to pass to KMirror.slew.
altaz_tracking (bool) – If True, starts tracking after moving to alt/az coordinates. By default the PWI won’t track with those coordinates.
force (bool) – Move the telescopes even if mode is local.
retry (bool) – Retry once if the coordinates are not reached.

async goto_named_position(name, altaz_tracking=False, force=False)[source]¶

Sends the telescope to a named position.

Parameters:

name (str) – The name of the position, e.g., 'zenith'.
altaz_tracking (bool) – Whether to start tracking after reaching the position, if the coordinates are alt/az.
force (bool) – Move the telescope even in local enclosure mode.

async home(home_telescope=True, home_km=True, home_focuser=False, home_fibsel=False)[source]¶

Initialises and homes the telescope.

Parameters:

home_telescope (bool) – Homes the telescope. Defaults to True.
home_km (bool) – Homes the K-mirror, if present. Defaults to True.
home_focuser (bool) – Homes the focuser. Defaults to False.
home_fibsel (bool) – Homes the fibre selector, if present. Defaults to False.

async init()[source]¶: Determines the initial state of the telescope.

async initialise(home=None)[source]¶

Connects to the telescope and initialises the axes.

Parameters:: home (bool | None) – If True, runs the homing routine after initialising.

async is_parked()[source]¶: Is the telescope parked?

async is_ready()[source]¶: Checks if the telescope is ready to be moved.

async offset(ra=None, dec=None, axis0=None, axis1=None)[source]¶

Apply an offset to the telescope axes.

Parameters:

ra (float | None) – Offset in RA, in arcsec.
dec (float | None) – Offset in Dec, in arcsec.
axis0 (float | None) – Offset in axis 0, in arcsec.
axis1 (float | None) – Offset in axis 1, in arcsec.

async park(disable=True, use_pw_park=False, alt_az=None, kmirror=True, force=False)[source]¶

Parks the telescope.

Parameters:

disable – Disables the axes after reaching the park position.
use_pw_park – Uses the internal park position stored in the PlaneWave mount.
alt_az (tuple[float, float] | None) – A tuple with the alt and az position at which to park the telescope. If not provided, defaults to the park named position.
kmirror (bool) – Whether to park the mirror at 90 degrees.
force (bool) – Moves the telescope even if the mode is local.

async status()[source]¶: Retrieves the status of the telescope.

async stop()[source]¶: Stops the mount.

class gort.devices.telescope.TelescopeSet(gort, data)[source]¶

Bases: GortDeviceSet[Telescope]

A representation of a set of telescopes.

async goto(sci=None, spec=None, skye=None, skyw=None, sci_km_stop_degs_before=0.0, force=False)[source]¶

Sends each telescope to a different position.

Parameters:

sci (tuple[float, float] | tuple[float, float, float] | None) – The RA and Dec where to slew the science telescope. A third value can be provided for the PA. Note that in this case a -1 factor is applied before sending it to the K-mirror.
spec (tuple[float, float] | None) – The RA and Dec where to slew the spectrophotometric telescope.
skye (tuple[float, float] | None) – The RA and Dec where to slew the skyE telescope.
skyw (tuple[float, float] | None) – The RA and Dec where to slew the skyW telescope.
sci_km_stop_degs_before (float) – The number of degrees before the desired position where to send the science k-mirror. Useful if we want to be sure that positive offsets will be applied without backlash.

async goto_coordinates_all(ra=None, dec=None, alt=None, az=None, kmirror=True, altaz_tracking=False, force=False)[source]¶

Moves all the telescopes to a given RA/Dec or Alt/Az.

Parameters:

ra (float | None) – Right ascension coordinates to move to.
dec (float | None) – Declination coordinates to move to.
alt (float | None) – Altitude coordinates to move to.
az (float | None) – Azimuth coordinates to move to.
kmirror (bool) – Whether to move the k-mirror into position. Only when the coordinates provided are RA/Dec.
altaz_tracking (bool) – If True, starts tracking after moving to alt/az coordinates. By defaul the PWI won’t track with those coordinates.
force (bool) – Move the telescopes even if mode is local.

async goto_named_position(name, altaz_tracking=False, force=False)[source]¶

Sends the telescopes to a named position.

Parameters:

name (str) – The name of the position, e.g., 'zenith'.
altaz_tracking (bool) – Whether to start tracking after reaching the position, if the coordinates are alt/az.
force (bool) – Move the telescopes even in local enclosure mode.

async home(home_telescopes=True, home_kms=True, home_focusers=False, home_fibsel=False)[source]¶

Initialises and homes the telescope.

Parameters:

home_telescopes (bool) – Homes the telescopes. Defaults to True.
home_kms (bool) – Homes the K-mirrors, if present. Defaults to True.
home_focusers (bool) – Homes the focusers. Defaults to False.
home_fibsel (bool) – Homes the fibre selector, if present. Defaults to False.

async initialise()[source]¶: Initialise all telescopes.

async park(disable=True, use_pw_park=False, alt_az=None, kmirror=True, force=False)[source]¶

Parks the telescopes.

Parameters:

disable – Disables the axes after reaching the park position.
use_pw_park – Uses the internal park position stored in the PlaneWave mounts.
alt_az (tuple[float, float] | None) – A tuple with the alt and az position at which to park the telescopes. If not provided, defaults to the park named position.
kmirror – Whether to park the mirrors at 90 degrees.
force – Moves the telescopes even if the mode is local.

async restart()[source]¶: Restarts the lvmpwi and lvmtan deployments and re-homes.

async restart_lvmtan()[source]¶

Restarts and rehomes Twice-As-Nice controller.

After the actor has been restarted the K-mirrors, focuser, and fibre selector are rehomed. The focuser positions are preserved.

async stop()[source]¶: Stops all the mounts.

Exposure¶

class gort.exposure.Exposure(gort, exp_no=None, flavour='object', object='', specs=None)[source]¶

Bases: Future[Exposure]

A class representing an exposure taken by a SpectrographSet.

Parameters:

gort (Gort | GortClient) – A Gort instance to communicate with the actors.
exp_no (int | None) – The exposure sequence number. If None, the next valid sequence number will be used.
flavour (str | None) – The image type. Defaults to 'object'.
object (str | None) – The object name to be added to the header.
specs (Optional[Sequence[str]]) – List the spectrographs to expose. Defaults to all.

hooks¶

A dictionary of hooks to call in specific steps of the exposure. Each hook must be a list of coroutines to call in that specific situation. All coroutines for a given hook are called concurrently and depending on the hook they may be scheduled as a task and not awaited. Available hooks are: - 'pre-readout' which is called with the header before readout begins; the coroutine can modify the header in place or perform any tasks that should be complete at the end of integration. - 'post-readout' called as a task (not awaited) after the readout is complete. Receives the Exposure object.

To add a coroutine to a hook

async def update_header(header):
    header.update({'KEY': 1})

exp = Exposure(g)
exp.hooks['pre-readout'].append(update_header)

async expose(exposure_time=None, header=None, async_readout=False, show_progress=None, object=None, raise_on_error=True)[source]¶

Exposes the spectrograph.

Parameters:

exposure_time (float | None) – The exposure time.
header (dict | None) – A dictionary with the extra header values..
async_readout (bool) – Returns after integration completes. Readout is initiated but handled asynchronously and can be await by awaiting the returned Exposure object.
show_progress (bool | None) – Displays a progress bar with the elapsed exposure time. If None (the default), will show the progress bar only in interactive sessions.
object (str | None) – The object name to be passed to the header.
raise_on_error (bool) – Whether to raise an error when the exposure is marked as errored.

get_files()[source]¶: Returns the files written by the exposure.

async start_timer(exposure_time, readout_time=55)[source]¶: Starts the rich timer.

stop_timer()[source]¶: Cancels the timer.

verify_files()[source]¶: Checks that the files have been written and have the right contents.

Observer¶

class gort.observer.GortObserver(gort, tile, mask_positions_pattern='P1-*', on_interrupt=None)[source]¶

Bases: object

A class to handle tile observations.

Parameters:

gort (Gort) – The instance of Gort used to communicate with the devices.
tile (Tile) – The Tile with the information about the observation.
mask_positions_pattern (str) – The spec fibre mask positions to use.
on_interrupt (Optional[Callable]) – Callback to be called when the observation is interrupted.

async acquire(guide_tolerance=None, timeout=180)[source]¶

Acquires the field in all the telescopes. Blocks until then.

Parameters:

guide_tolerance (float | None) – The guide tolerance in arcsec. A telescope will not be considered to be guiding if its separation to the commanded field is larger than this value. If None, default values from the configuration file are used.
timeout (float) – The maximum time allowed for acquisition. In case of timeout the acquired fields are evaluated and an exception is raised if the acquisition failed.

Raises:

GortObserverError – If the acquisition failed or the minimum required telescopes are not guiding.

async expose(exposure_time=900.0, show_progress=None, count=1, async_readout=False, keep_guiding=True, object=None, dither_position=None)[source]¶

Starts exposing the spectrographs.

Parameters:

exposure_time (float) – The length of the exposure in seconds.
show_progress (bool | None) – Displays a progress bar with the elapsed exposure time.
count (int) – Number of exposures. If iterate_over_standards=True, a full sequence of standards will be observed during each exposure.
async_readout (bool) – Whether to wait for the readout to complete or return as soon as the readout begins. If False, the exposure is registered but the observation is not finished.
keep_guiding (bool) – If True, keeps the guider running after the last exposure.
object (str | None) – The object name to be added to the header.
dither_position (int | None) – The dither position. If None, uses the first dither position in the tile. Only relevant for exposure registration.

Returns:

Either a single Exposure or a list of exposures if count>1.

Return type:

exposures

async finish_observation()[source]¶: Finishes the observation, stops the guiders, etc.

async register_exposure(exposure, tile_id=None, dither_position=0)[source]¶: Registers the exposure in the database.

register_overhead(name)[source]¶: Measures and registers and overhead.

async set_dither_position(dither)[source]¶: Reacquire science telescope for a new dither position.

async slew()[source]¶: Slew to the telescope fields.

write_to_log(message, level='debug', header=None)[source]¶

Writes a message to the log with a custom header.

Parameters:

message (str) – The message to log.
level (str) – The level to use for logging: 'debug', 'info', 'warning', or 'error'.
header (str | None) – The header to prepend to the message. By default uses the class name.

property has_standards¶: Returns True if standards will be observed.

Tile¶

class gort.tile.Coordinates(ra, dec, pa=None, centre_on_fibre=None)[source]¶

Bases: object

Basic coordinates class.

Parameters:

ra (float) – The RA coordinate, in degrees. FK5 frame at the epoch of observation.
dec (float) – The Dec coordinate, in degrees.
pa (float | None) – Position angle of the IFU. Defaults to PA=0.
centre_on_fibre (str | None) – The name of the fibre on which to centre the target, with the format <ifulabel>-<finufu>. By default, acquires the target on the central fibre of the science IFU.

calculate_altitude(time=None)[source]¶: Returns the current altitude of the target.

is_observable()[source]¶: Determines whether a target is observable.

set_mf_pixel(fibre_name=None, xz=None)[source]¶

Calculates and sets the master frame pixel on which to centre the target.

If neither fibre_name or xz are passed, resets to centring the target on the central fibre of the IFU.

Parameters:

fibre_name (str | None) – The fibre to which to centre the target, with the format <ifulabel>-<finifu>.
xz (tuple[float, float] | None) – The coordinates, in master frame pixels, on which to centre the target.

Returns:

A tuple with the x and z coordinates of the pixel in the master frame, or None if resetting to the central fibre.

Return type:

pixel

class gort.tile.QuerableCoordinates(ra, dec, pa=None, centre_on_fibre=None)[source]¶

Bases: Coordinates

A class of coordinates that can be retrieved from the database.

classmethod from_science_coordinates(sci_coords, exclude_coordinates=[], exclude_invisible=True)[source]¶

Retrieves a valid and observable position from the database.

Parameters:

sci_coords (ScienceCoordinates) – The science coordinates. The position selected will be the closest to these coordinates.
exclude_coordinates (Sequence[tuple[float, float]]) – A list of RA/Dec coordinates to exclude. No region closer than one degree to these coordinates will be selected.
exclude_invisible (bool) – Exclude targets that are too low.

verify_and_replace(exclude_coordinates=[])[source]¶

Verifies that the coordinates are visible and if not, replaces them.

Parameters:: exclude_coordinates (Sequence[tuple[float, float]]) – A list of RA/Dec coordinates to exclude. No region closer than one degree to these coordinates will be selected.

class gort.tile.ScienceCoordinates(ra, dec, pa=None, centre_on_fibre=None)[source]¶

Bases: Coordinates

A science position.

Parameters:

ra (float) – The RA coordinate, in degrees. FK5 frame at the epoch of observation.
dec (float) – The Dec coordinate, in degrees.
pa (float | None) – Position angle of the IFU. Defaults to PA=0.
centre_on_fibre (str | None) – The name of the fibre on which to centre the target, with the format <ifulabel>-<finufu>. By default, acquires the target on the central fibre of the science IFU.

class gort.tile.SkyCoordinates(*args, name=None, **kwargs)[source]¶

Bases: QuerableCoordinates

A sky position.

In addition to the QuerableCoordinates arguments the class accepts a name identifier.

class gort.tile.StandardCoordinates(ra=None, dec=None, source_id=None, **kwargs)[source]¶

Bases: QuerableCoordinates

A standard position.

In addition to the QuerableCoordinates arguments the class accepts a source_id Gaia identifier.

class gort.tile.Tile(sci_coords, sky_coords=None, spec_coords=None, dither_positions=0, object=None, allow_replacement=True)[source]¶

Bases: dict[str, Coordinates | Sequence[Coordinates] | None]

A representation of a science pointing with associated calibrators.

This class is most usually initialised from a classmethod like from_scheduler.

Parameters:

sci_coords (ScienceCoordinates) – The science telescope pointing.
sky_coords (dict[str, SkyCoordinates] | dict[str, tuple[float, float]] | None) – A dictionary of skye and skyw coordinates.
spec_coords (Optional[Sequence[StandardCoordinates | tuple[float, float] | int | dict]]) – A list of coordinates to observe with the spectrophotometric telescope.
dither_position – The dither position(s) to obseve.
object (str | None) – The name of the object.
allow_replacement (bool) – If True, allows the replacement of empty, invalid or low altitude sky and standard targets.

async disable()[source]¶: Disables the tile for observation.

async enable()[source]¶: Enables the tile for observation.

classmethod from_coordinates(ra, dec, pa=0.0, sky_coords=None, spec_coords=None, **kwargs)[source]¶

Creates an instance from coordinates, allowing autocompletion.

Parameters:

ra (float) – The science telescope pointing.
dec (float) – The science telescope pointing.
pa (float) – Position angle of the science IFU. Defaults to PA=0.
sky_coords (dict[str, SkyCoordinates] | dict[str, tuple[float, float]] | None) – A dictionary of skye and skyw coordinates. If None, autocompleted from the closest available regions.
spec_coords (Optional[Sequence[StandardCoordinates | tuple[float, float] | int | dict]]) – A list of coordinates to observe with the spectrophotometric telescope. If None, selects the 12 closest standard stars.
kwargs – Arguments to be passed to the initialiser.

classmethod from_scheduler(tile_id=None, ra=None, dec=None, pa=0.0, **kwargs)[source]¶

Creates a new instance of Tile with data from the scheduler.

Parameters:

tile_id (int | None) – The tile_id for which to create a new Tile. If None, and ra and dec are also None, the best tile_id selected by the scheduler will be used.
ra (float | None) – Right ascension coordinates of the science telescope pointing. Calibrators will be selected from the scheduler.
dec (float | None) – Declination coordinates of the science telescope pointing.
pa (float) – Position angle of the science IFU. Defaults to PA=0.
kwargs – Arguments to be passed to the initialiser.

set_dither_position(dither)[source]¶: Sets the full frame pixel for the science IFU to the dither position.

set_sci_coords(sci_coords) → ScienceCoordinates[source]¶

Sets the science telescope coordinates.

Parameters:: sci_coords (ScienceCoordinates | tuple[float, float]) – A ScienceCoordinates object or a tuple with RA/Dec coordinates for the science telescope.
Return type:: ScienceCoordinates

set_sky_coords(sky_coords=None, allow_replacement=True) → dict[str, SkyCoordinates][source]¶

Sets the sky telescopes coordinates.

Parameters:

sky_coords (dict[str, SkyCoordinates] | dict[str, tuple[float, float]] | None) – A dictionary of skye and skyw coordinates. Each value must be a SkyCoordinates object or a tuple of RA/Dec coordinates.
allow_replacement (bool) – If True, allows the replacement of empty, invalid or low altitude targets.

Return type:

dict[str, SkyCoordinates]

set_spec_coords(spec_coords=None, reject_invisible=True) → Sequence[StandardCoordinates][source]¶

Sets the spec telescope coordinates.

Parameters:

spec_coords (Optional[Sequence[StandardCoordinates | tuple[float, float] | int | dict]]) – A list of coordinates to observe with the spectrophotometric telescope.
reject_invisible (bool) – Skip targets that are not visible now.

Return type:

Sequence[StandardCoordinates]

property sci_coords¶: Returns the science coordinates.

property sky_coords: dict[str, SkyCoordinates]¶: Returns the sky coordinates.

property spec_coords¶: Returns the Spec coordinates.

Recipes¶

class gort.recipes.base.BaseRecipe(gort)[source]¶

Bases: object

Base class for recipes.

async recipe(*args, **kwargs)[source]¶: The recipe. Must be overridden by the subclass.

class gort.recipes.calibration.BiasSequence(gort)[source]¶

Bases: BaseRecipe

Takes a sequence of bias frames.

async recipe(count=7)[source]¶

Takes a sequence of bias frames.

Parameters:: count (int) – The number of bias frames to take.

class gort.recipes.calibration.CalibrationRecipe(gort)[source]¶

Bases: BaseRecipe

Runs a calibration sequence.

get_calibration_sequence(sequence)[source]¶

Returns a dictionary with the configuration for a calibration sequence.

Parameters:: sequence (str) – The name calibration sequence.
Returns:: The calibration sequence dictionary. This dictionary can be altered and then passed to calibrate to execute the modified sequence. The returned dictionary if a deep copy of the original sequence; modifying it won’t modify the original sequence.
Return type:: sequence_dict

async recipe(sequence='normal', slew_telescopes=True, park_after=False, show_progress=None)[source]¶

Runs the calibration sequence.

Parameters:

sequence (str | dict) – The name calibration sequence to execute. It can also be a dictionary with the calibration sequence definition that follows the calibration schema.
slew_telescopes (bool) – Whether to move the telescopes to point to the FF screen.
park_after (bool) – Park the telescopes after a successful calibration sequence.
show_progress (bool | None) – Displays a progress bar with the elapsed exposure time.

class gort.recipes.calibration.QuickCals(gort)[source]¶

Bases: BaseRecipe

Runs a quick calibration sequence.

async recipe()[source]¶: Runs the calibration sequence.

class gort.recipes.calibration.TwilightFlats(gort)[source]¶

Bases: BaseRecipe

Takes a sequence of twilight flats.

async goto_fibre_position(n_fibre, secondary=False)[source]¶: Moves the mask to a fibre position.

async recipe(wait=True, secondary=False)[source]¶

Takes a sequence of twilight flats.

Based on K. Kreckel’s code.

class gort.recipes.operations.CleanupRecipe(gort)[source]¶

Bases: BaseRecipe

Stops guiders, aborts exposures, and makes sure the system is ready to go.

async recipe(readout=True, turn_lamps_off=True)[source]¶

Runs the cleanup recipe.

Parameters:

readout (bool) – If the spectrographs are idle and with a readout pending, reads the spectrographs.
turn_lamps_off (bool) – If True, turns off the dome lights and calibration lamps.

class gort.recipes.operations.ShutdownRecipe(gort)[source]¶

Bases: BaseRecipe

Closes the telescope for the night.

async recipe(park_telescopes=True, additional_close=True)[source]¶

Shutdown the telescope, closes the dome, etc.

Parameters:

park_telescopes (bool) – Park telescopes (and disables axes). Set to False if only closing for a brief period of time.
additional_close (bool) – Issues an additional close command after the dome is closed. This is a temporary solution to make sure the dome is closed while we investigate the issue with the dome not fully closing sometimes.

class gort.recipes.operations.StartupRecipe(gort)[source]¶

Bases: BaseRecipe

Starts the telescopes, runs the calibration sequence, and opens the enclosure.

async recipe(calibration_sequence=False, open_enclosure=True, confirm_open=True, focus=True)[source]¶

Runs the startup sequence.

Parameters:

gort – The Gort instance to use.
calibration_sequence (Union[str, None, Literal[False]]) – The name of the calibration sequence to use. If None, uses the default sequence from the configuration. If False, skips the calibration sequence.
open_enclosure (bool) – Whether to open the enclosure.
confirm_open (bool) – If True, asks the user to confirm opening the enclosure.
focus (bool) – Whether to focus after the enclosure has open.

Kubernetes¶

class gort.kubernetes.Kubernetes(log=None)[source]¶

Bases: object

Interface with the Kubernetes cluster.

apply_from_file(name)[source]¶

Applies a YAML file.

Parameters:: name (str | Path) – The full path to the file to apply. If the path is relative, the file will be searched in the directory for YAML files defined in the configuration.

delete_deployment(deployment)[source]¶

Deletes resources from a YAML file.

Parameters:: deployment (str) – The deployment to delete.

get_deployment_namespace(deployment)[source]¶: Returns the namespace of a deployment.

get_yaml_file(name)[source]¶: Finds and returns the contents of a Kubernetes YAML file.

list_deployments()[source]¶: Returns a list of deployments.

list_namespaces()[source]¶: Returns a list of namespaces.

restart_deployment(deployment, from_file=True)[source]¶

Restarts a deployment.

If the deployment is running, does a rollout restart. Otherwise looks for the deployment file and applies it.

Tools¶

class gort.tools.GuiderMonitor(gort, actor=None)[source]¶

Bases: object

A tool to monitor guider outputs and store them in a dataframe.

get_dataframe() → DataFrame | None[source]¶

Returns the collected data as a dataframe.

Return type:: DataFrame | None

reset()[source]¶: Resets the internal state.

start_monitoring()[source]¶: Starts monitoring the guider outputs.

stop_monitoring()[source]¶: Stops monitoring the guider outputs.

to_header()[source]¶: Returns a header with pointing and guiding information.

gort.tools.angular_separation(lon1, lat1, lon2, lat2)[source]¶

A wrapper around astropy’s angular_separation.

Returns the separation between two sets of coordinates. All units must be degrees and the returned values is also the separation in degrees.

async gort.tools.cancel_task(task)[source]¶: Safely cancels a task.

async gort.tools.ds9_agcam_monitor(amqp_client, cameras=None, replace_path_prefix=None, **kwargs)[source]¶: Shows guider images in DS9.

async gort.tools.ds9_display_frames(files, ds9=None, order=['sci.west', 'sci.east', 'skye.west', 'skye.east', 'skyw.west', 'skyw.east', 'spec.east'], ds9_target='DS9:*', show_all_frames=True, preserve_frames=True, clear_frames=False, adjust_zoom=True, adjust_scale=True, show_tiles=True)[source]¶: Displays a series of images in DS9.

async gort.tools.get_calibrators(tile_id=None, ra=None, dec=None)[source]¶: Get calibrators for a tile_id or science pointing.

gort.tools.get_calibrators_sync(tile_id=None, ra=None, dec=None) → dict[source]¶

Get calibrators for a tile_id or science pointing. Synchronous version.

Return type:: dict

gort.tools.get_ccd_frame_path(frame_id, sjd=None, cameras=None, spectro_path='/data/spectro') → list[str][source]¶

Returns the paths for the files for a spectrograph frame.

Parameters:

frame_id (int) – The spectrograph frame for which the paths are searched.
mjd – The SJD in which the frames where taken. If not provided, all the directories under spectro_path are searched.
cameras (str | list[str] | None) – The cameras to be returned. If None, all cameras found are returned.
spectro_path – The path to the spectro directory where spectrograph files are stored under an SJD structure.

Returns:

The list of paths to CCD frames that match frame_id.

Return type:

paths

gort.tools.get_db_connection()[source]¶: Returns a DB connection from the configuration file parameters.

gort.tools.get_md5sum(file)[source]¶: Returns the MD5 checksum for a file.

gort.tools.get_md5sum_file(file)[source]¶: Returns the path to the MD5 file for the spectro files.

gort.tools.get_md5sum_from_spectro(file)[source]¶: Returns the MD5 checksum for a file from the spectro checksum file.

async gort.tools.get_next_tile_id() → dict[source]¶

Retrieves the next tile_id from the scheduler API.

Return type:: dict

gort.tools.get_next_tile_id_sync() → dict[source]¶

Retrieves the next tile_id from the scheduler API. Synchronous version.

Return type:: dict

gort.tools.get_temporary_file_path(*args, create_parents=False, **kwargs)[source]¶

Returns a valid path to a temporary file.

args and kwargs are directly passed to tempfile.NamedTemporaryFile. If create_parents, the parent directories are created if they don’t exist.

gort.tools.get_valid_variable_name(var_name)[source]¶: Converts a string to a valid variable name.

gort.tools.handle_signals(signals, callback=None, cancel=True)[source]¶

Runs a callback when a signal is received during the execution of a task.

This function is meant to decorate coroutines. If a signal matching signals is received, the callback is run and the coroutine (which is executed as a task) is cancelled if cancel=True.

gort.tools.insert_to_database(table_name, payload, columns=None)[source]¶

Inserts data into the database.

Parameters:

table_name (str) – The table in the database where to insert the data. Can be in the format schema.table_name.
payload (list[dict[str, Any]]) – The data to ingest, as a list of dictionaries in which each dictionary is a mapping of column name in table to the value to ingest.
columns (list[str] | None) – A list of table columns. If not passed, the column names are inferred from the first element in the payload. In this case you must ensure that all the elements in the payload contain entries for all the columns (use None to fill missing data).

gort.tools.is_interactive()[source]¶: Returns True is we are in an interactive session.

gort.tools.is_notebook() → bool[source]¶

Returns True if the code is run inside a Jupyter Notebook.

https://stackoverflow.com/questions/15411967/how-can-i-check-if-code-is-executed-in-the-ipython-notebook

Return type:: bool

gort.tools.mark_exposure_bad(tile_id, dither_position=0)[source]¶: Marks a registered tile/dither as bad.

async gort.tools.move_mask_interval(gort, positions='P1-*', order_by_steps=False, total_time=None, time_per_position=None, notifier=None)[source]¶

Moves the fibre mask in the spectrophotometric telescope at intervals.

Parameters:

gort (GortClient) – The instance of Gort to communicate with the actor system.
positions (str | list[str]) – The positions to iterate over. It can be a string in which case it will be treated as a regular expression and any mask position that matches the value will be iterated, in alphabetic order. Alternative it can be a list of positions to move to which will be executed in that order.
order_by_steps (bool) – If True, the positions are iterated in order of smaller to larger number of step motors.
total_time (float | None) – The total time to spend iterating over positions, in seconds. Each position will be visited for an equal amount of time. The time required to move the mask will not be taken into account, which means the total execution time will be longer than total_time.
time_per_position (float | None) – The time to spend on each mask position, in seconds. The total execution time will be len(positions)*total_time+overhead where overhead is the time required to move the mask between positions.
notifier (Callable[[str], None] | Callable[[str], Coroutine] | None) – A function or coroutine to call every time a new position is reached. If it’s a coroutine, it is scheduled as a task. If it is a normal callback it should run quickly to not perceptibly affect the total execution time.

gort.tools.parse_agcam_filename(file_)[source]¶: Returns the type of an agcam file in the form (telescope, camera).

async gort.tools.register_observation(payload)[source]¶: Registers an observation with the scheduler.

async gort.tools.run_in_executor(fn, *args, catch_warnings=False, executor='thread', **kwargs)[source]¶

Runs a function in an executor.

In addition to streamlining the use of the executor, this function catches any warning issued during the execution and reissues them after the executor is done. This is important when using the actor log handler since inside the executor there is no loop that CLU can use to output the warnings.

In general, note that the function must not try to do anything with the actor since they run on different loops.

Transformations¶

class gort.transforms.HomTrans(entries)[source]¶

Bases: object

A single affine coordinate transformation.

Represented internally by a 4x4 matrix as a projective.

From https://github.com/sdss/lvmtipo/blob/main/python/lvmtipo/homtrans.py

apply(rhs)[source]¶

Apply self transformation to a vector of coordinates.

Parameters:: rhs (ndarray) – The vector. If it has only the standard 3 coordinates, a virtual 1 is appended before applying the transformation.
Returns:: A vector of 3 (standard, projected) Cartesian coordinates.
Return type:: vector

multiply(rhs)[source]¶

Multiplies by another transformation.

Parameters:: rhs (HomTrans | ndarray) – The transformation to the right of the multiplication sign. So rhs is applied before this transformation.
Returns:: The homogeneous transformation which is the (non-communtative) product of self with rhs, representing the consecutive application of rhs, then self.
Return type:: hom_trans

class gort.transforms.Mirror(normal, disttoorg)[source]¶

Bases: object

A flat mirror.

This represents an infintely large flat plane. The internal representation is the surface normal and the standard equation that the dot product of points on the surface by the surface normal equals the distance (of the plane to the origin of coordinates).

From https://github.com/sdss/lvmtipo/blob/main/python/lvmtipo/mirror.py

Parameters:

normal (ndarray) – The 3 Cartesian coordinates of the surface normal. It must have nonzero length, but does not need to be normalized to unit length.
disttoorg (float) – The distance of the mirror to the origin of coordinates. As in usual geometry, the distance is the shortest distance of the origin to the infinitely extended mirror plane.

to_hom_trans()[source]¶: The homogeneous transformation that represents the reflection of rays off this mirror surface.

class gort.transforms.Siderostat(zenang=90.0, azang=0.0, medSign=-1, m1m2dist=240.0, om1_off_ang=118.969, om2_off_ang=-169.752)[source]¶

Bases: object

A siderostat of 2 mirrors.

Adapted from https://github.com/sdss/lvmtipo/blob/main/python/lvmtipo/siderostat.py

Parameters:

zenang (float) – Zenith angle of the direction of the exit beam (degrees) in the range 0..180. Default is the design value of the LVMT: horizontal
azang (float) – Azimuth angle of the direction of the exit beam (degrees) in the range -180..360 degrees, N=0, E=90. Ought to be zero for the LCO LVMT where the FP is north of the siderostat and 180 for the MPIA test setup where the FP is south of the siderostat. The default is the angle for LCO.
medSign (int) – Sign of the meridian flip design of the mechanics. Must be either +1 or -1. Default is the LCO LVMT design as build (in newer but not the older documentation).
m1m2dist (float) – Distance between the centers of M1 and M2 in millimeter. The default value is taken from LVM-0098_Sky Baffle Design of 2022-04-18 by subracting the 84 and 60 cm distance of the output pupils to M1 and M2.
om2_off_ang (float | Angle) – The offset angle which aligns PW motor angle of the M2 axis, which is ax0 of the PW GUI, to the angles of the manuscript.
om1_off_ang (float | Angle) – The offset angle which aligns PW motor angle of the M1 axis, which is ax1 of the PW GUI, to the angles of the manuscript, degrees. om1_off_ang and om2_off_ang are either angles in degrees or both astropy.coordinates.Angle.

field_angle(target, time=None)[source]¶

Determines the field angle (direction to NCP).

Parameters:

target (SkyCoord) – Sidereal target in ra/dec.
time (Time | str | None) – Time of the observation.

Returns:

Field angle (direction to NCP) in degrees.

Return type:

field_angle

gort.transforms.calculate_field_angle(ra, dec, obstime=None)[source]¶

Returns the field angle for a set of coordinates.

Parameters:

ra (float) – The RA of the centre of the field.
dec (float) – The Dec of the centre of the field.
obstime (str | Time | None) – The time of the observation. Either an ISOT string or an astropy time object.

Returns:

The field angle. See Siderostat.field_angle for details.

Return type:

gort.transforms.calculate_position_angle(ra, dec, obstime)[source]¶

Calculates the position angle seen for a set of coordinates.

Parameters:

ra (float) – The RA of the centre of the field.
dec (float) – The Dec of the centre of the field.
obstime (Time | str) – The time of the observation. Either an ISOT string or an astropy time object.

Returns:

The position angle. In an image this is the angle between North and the y direction, with positive angles being CCW.

Return type:

gort.transforms.fibre_slew_coordinates(ra, dec, fibre_name, derotated=True) → tuple[float, float][source]¶

Determines the slew coordinates for a fibre.

This function provides approximate slew coordinates to place a target with coordinates ra, dec on a given fibre. The preferred way to use it is to slew the telescope using the coordinates returned by this function but then guide on the original target coordinates and use the appropriate pixel on the master frame to guide on the fibre.

Parameters:

ra (float) – The RA and Dec of the target to which to slew.
dec (float) – The RA and Dec of the target to which to slew.
fibre_name (str) – The fibre to which to slew the target, with the format <ifulabel>-<finifu>.
derotated (bool) – Whether a k-mirror is derotating the field. If False, the field rotation for the current time will be used.

Returns:

A tuple of RA/Dec coordinates to slew, which should place the target near the desired fibre.

Return type:

slew_coordinates

gort.transforms.fibre_to_master_frame(fibre_name)[source]¶

Returns the xz coordinates in the master frame of a named fibres.

Parameters:: fibre_name (str) – The fibre for which to calculate the master frame coordinates, with the format <ifulabel>-<finifu>.
Returns:: A tuple with the x and z coordinates of the pixel in the master frame.
Return type:: pixel
Raises:: ValueError – If the pixel is out of bounds.

gort.transforms.offset_to_master_frame_pixel(xmm=None, ymm=None, ra=None, dec=None) → tuple[float, float][source]¶

Determines the pixel on master frame coordinates for an offset.

Parameters:

xmm (float | None) – The x offset, in mm, with respect to the central fibre in the IFU.
ymm (float | None) – The y offset, in mm, with respect to the central fibre in the IFU.
ra (float | None) – The offset in RA, in arcsec. See xy_to_radec_offset for the caveats on this calculations.
dec (float | None) – The offset in declination, in arcsec.

Returns:

A tuple with the x and z coordinates of the pixel in the master frame.

Return type:

pixel

Raises:

ValueError – If the pixel is out of bounds.

gort.transforms.radec_sexagesimal_to_decimal(ra, dec, ra_is_hours=True)[source]¶: Converts a string of sexagesimal RA and Dec to decimal.

gort.transforms.read_fibermap(path=None, force_cache=False) → DataFrame[source]¶

Reads the fibermap file.

Parameters:

path (str | Path | None) – Path to the fibermap file. If None uses the path from the configuration file.
force_cache (bool) – If True, forces a re-read of the fibermap file; otherwise reads it from the cache if available.

Returns:

A Polars DataFrame with the fibermap data.

Return type:

fibermap

gort.transforms.xy_to_radec_offset(xpmm, ypmm)[source]¶

Converts offsets in the IFU to approximate RA/Dec offsets.

Warning

This is an approximate conversion that assumes the IFU is perfectly aligned with the AG cameras in the focal plane and that the field de-rotation is perfect. It should only be used to determine initial offsets for blind telescope slews.

Parameters:

xpmm (float) – The x offset with respect to the central IFU fibre, in mm, as defined in the fibermap.
ypmm (float) – As xpmm for the y offset.

Returns:

RA/Dec offset in arcsec as a tuple. See the warning above for caveats.

Return type:

radec

Exceptions¶

exception gort.exceptions.GortAGError(message=None, error_code=0) → None[source]¶

Bases: GortDeviceError

AG-related error.

exception gort.exceptions.GortDeviceError(message=None, error_code=0) → None[source]¶

Bases: GortError

A device error, which appends the name of the device to the error message.

exception gort.exceptions.GortEnclosureError(message=None, error_code=0) → None[source]¶

Bases: GortDeviceError

Enclosure-related error.

exception gort.exceptions.GortError(message=None, error_code=0, payload={})[source]¶

Bases: Exception

A custom core GortError exception

exception gort.exceptions.GortGuiderError(message=None, error_code=0) → None[source]¶

Bases: GortDeviceError

Guider-related error.

exception gort.exceptions.GortNPSError(message=None, error_code=0) → None[source]¶

Bases: GortDeviceError

NPS-related error.

exception gort.exceptions.GortNotImplemented(message=None)[source]¶

Bases: GortError

A custom exception for not yet implemented features.

exception gort.exceptions.GortObserverError(message=None, error_code=0, payload={})[source]¶

Bases: GortError

An error associated with the Observer.

exception gort.exceptions.GortSpecError(message=None, error_code=0) → None[source]¶

Bases: GortDeviceError

Spectrograph-related error.

exception gort.exceptions.GortTelescopeError(message=None, error_code=0) → None[source]¶

Bases: GortDeviceError

Telescope-related error.

exception gort.exceptions.GortTimeout(message=None, error_code=0, payload={})[source]¶

Bases: GortError

Raised if a timeout occurs.

exception gort.exceptions.GortTimeoutError(message, command=None, remote_command=None)[source]¶

Bases: GortError

A timeout error, potentially associated with a timed out remote command.

exception gort.exceptions.GortUserWarning[source]¶

Bases: UserWarning, GortWarning

The primary warning class.

exception gort.exceptions.GortWarning[source]¶

Bases: Warning

Base warning for Gort.

exception gort.exceptions.RemoteCommandError(message, command, remote_command)[source]¶

Bases: GortError

An error in a remote command to an actor.

exception gort.exceptions.TileError(message=None, error_code=0, payload={})[source]¶

Bases: GortError

An error associated with a Tile.

class gort.exceptions.ErrorCodes(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]¶

Bases: Enum

List of error codes.