API Reference

Core classes

class gort.gort.Gort(*args, override_overwatcher=None, config_file=None, **kwargs)

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.

• allow_overwatcher – If the Overwatcher is running an exception will be raised to prevent GORT running commands that may interfere with it. If allow_overwatcher=True the exception will be suppressed.

add_actor(actor, device=None)

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

Adds a new device or device set to Gort.

Return type:

TypeVar(DevType, bound= GortDeviceSet | GortDevice)

async cleanup(readout=True, turn_lamps_off=True, home_telescopes=False)

Executes the shutdown sequence.

async emergency_shutdown()

Parks and closes the telescopes.

async execute_recipe(recipe, **kwargs)

Executes a recipe.

Parameters:

• recipe (str) – The name of the recipe to execute.

• kwargs – Arguments to be passed to the recipe.

async init() → Self

Initialises the client and all devices.

Return type:

Self

async observe(n_tile_positions=None, adjust_focus=True, show_progress=None, disable_tile_on_error=True, wait_if_no_tiles=60.0)

Runs a fully automatic science observing loop.

async restart_kubernetes_deployments(deployment)

Restarts a Kubernetes deployment.

async run_script(script)

Runs a script.

run_script_sync(script)

Runs a script.

async shutdown(**kwargs)

Executes the shutdown sequence.

async startup(**kwargs)

Executes the startup sequence.

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

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.

• verbosity (str) – The level of console logging verbosity. One of the standard logging levels.

• 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.

asyncio_exception_handler(loop, context)

Handle an uncaught asyncio exception and reports it.

exception_handler(log, exc_type, exc_value, exc_traceback)

A custom exception handler that logs exceptions to file.

get_log_path()

Returns the path of the log file. None if not logging to file.

async init() → Self

Initialises the client.

Returns:

The same instance of GortClient after initialisation.

Return type:

Self

async notify_event(event, payload={})

Emits an event notification.

set_verbosity(verbosity=None, log=None)

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.

• log (SDSSLogger | None) – The logger to set the verbosity for. If None, the internal logger will be used.

property connected

Returns True if the client is connected.

Devices

class gort.devices.enclosure.Enclosure(gort, name, actor, **kwargs)

Bases: GortDevice

Class representing the LVM enclosure.

async allowed_to_move()

Checks if the dome is allowed to move.

async close(park_telescopes=True, force=False, mode='normal')

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.

• mode (Literal['normal', 'overcurrent']) – In 'normal' mode the dome will be closed using a combination of rotary encoder and limit switches. With 'overcurrent' the dome is closed until the VFD detects an overcurrent condition. 'overcurrent' should only be used when the normal mode has failed.

async get_door_status()

Returns the status of the door and lock.

async is_closed()

Returns True if the enclosure is closed.

async is_local()

Returns True if the enclosure is in local mode.

async is_open()

Returns True if the enclosure is open.

async open(park_telescopes=True)

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()

Restarts the lvmecp deployment.

async status(get_registers=False)

Retrieves the status of the power outlet.

async stop()

Stop the enclosure dome.

class gort.devices.enclosure.Light(enclosure, name)

Bases: object

An enclosure light.

async off()

Turns on the light.

async on()

Turns on the light.

async status() → bool

Returns a boolean with the light status.

Return type:

bool

async toggle()

Turns on the light.

class gort.devices.enclosure.Lights(enclosure)

Bases: object

Controls the enclosure lights.

async dome_all_off()

Turns off all the lights in the dome.

class gort.devices.ag.AG(gort, name, actor, **device_data)

Bases: GortDevice

Class representing an AG camera.

async check_camera(ping=True, status=True)

Checks that the AG cameras are responding.

Parameters:

• ping (bool) – Whether to ping the cameras.

• status (bool) – Whether to check the camera status.

async expose(exposure_time=5.0, flavour='object', **kwargs)

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 is_idle()

Returns True if all the cameras are idle.

async reconnect()

Reconnect the AG cameras.

async status()

Returns the status of the AG.

property n_cameras

The number of AG cameras for this telescope.

class gort.devices.ag.AGSet(gort, data, **kwargs)

Bases: GortDeviceSet[AG]

A set of auto-guiders.

async are_idle()

Returns True if all the cameras are idle.

async check_cameras(allow_power_cycle=True)

Checks that all cameras are responding.

Parameters:

allow_power_cycle (bool) – Whether to allow power cycling the cameras if they are not responding.

async expose(exposure_time=5.0, flavour='object', **kwargs)

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()

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 power_cycle()

Power cycles all the cameras.

async reconnect()

Reconnects all the AG cameras.

Returns:

A list of cameras available after reconnecting.

Return type:

cameras

property n_cameras

The number of AG cameras in the set.

class gort.devices.guider.Guider(gort, name, actor, **kwargs)

Bases: GortDevice

Class representing a guider.

async adjust_focus()

Adjusts the focus position based on the current bench temperature.

async apply_corrections(enable=True)

Enable/disable corrections being applied to the axes.

async expose(*args, continuous=False, **kwargs)

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)

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 get_focus_info()

Returns the guider focus information.

async guide(ra=None, dec=None, exposure_time=5.0, pixel=None, monitor=True, output_monitor_data=True, **guide_kwargs)

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.

• output_monitor_data (bool) – Whether to output the monitor data to the log.

• 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, monitor=True)

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).

monitor

Start the guider monitor task. Data collected during the monitoring can be access as a polars.DataFrame from Guider.guider_monitor.get_dataframe().

async set_pixel(pixel=None)

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

Stops the guide loop.

Parameters:

wait_until_stopped – Blocks until the guider is idle.

Return type:

None

async update_status()

Returns the guider status.

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

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)

Bases: GortDeviceSet[Guider]

A set of telescope guiders.

async adjust_focus()

Adjusts the focus position based on the current bench temperature.

async apply_corrections(enable=True)

Enable/disable corrections being applied to the axes.

async expose(*args, continuous=False, **kwargs)

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)

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)

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)

Guides at a given position, sleeping between exposures.

See Guider.monitor for details.

async stop()

Stops the guide loop on all telescopes.

async take_darks()

Takes AG darks.

async wait_until_guiding(names=None, guide_tolerance=None, timeout=None)

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)

Bases: GortDevice

Class representing a networked power switch.

async all_off()

Turns off all the outlets.

async off(outlet)

Turns an outlet on.

async on(outlet)

Turns an outlet on.

async status(outlet=None)

Retrieves the status of the power outlet.

class gort.devices.nps.NPSSet(gort, data, **kwargs)

Bases: GortDeviceSet[NPS]

A set of networked power switches.

class gort.devices.telemetry.Telemetry(gort, name, actor, **kwargs)

Bases: GortDevice

Telemetry sensors.

async status()

Retrieves the status of sensors.

class gort.devices.telemetry.TelemetrySet(gort, data, **kwargs)

Bases: GortDeviceSet[Telemetry]

A set of telemetry sensors.

class gort.devices.spec.IEB(gort, name, actor)

Bases: GortDevice

A class representing an Instrument Electronics Box.

async close(devices)

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)

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)

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)

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)

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)

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()

Returns the status of the IEB.

class gort.devices.spec.Spectrograph(gort, name, actor, **kwargs)

Bases: GortDevice

Class representing an LVM spectrograph functionality.

async abort()

Aborts an ongoing exposure.

async expose(**kwargs)

Exposes the spectrograph.

async initialise()

Initialises the spectrograph and flashes the ACF configuration file.

async is_errored()

Returns True if the spectrograph is in error.

async is_exposing()

Returns True if the spectrograph is exposing.

async is_idle()

Returns True if the spectrograph is idle and ready to expose.

async is_reading()

Returns True if the spectrograph is idle and ready to expose.

async reset(full=False)

Resets the spectrograph to a valid state.

async status(simple=False)

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)

Bases: GortDeviceSet[Spectrograph]

A set of LVM spectrographs.

async abort()

Aborts an ongoing exposure.

async are_errored()

Returns True if any of the spectrographs are errored.

async are_exposing()

Returns True if any of the spectrographs are exposing.

async are_idle()

Returns True if all the spectrographs are idle and ready to expose.

async are_reading()

Returns True if any of the spectrographs are reading.

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]

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:

Exposure | list[Exposure]

get_expno()

Returns the next exposure sequence number.

async initialise()

Initialises the spectrographs and flashes the ACF configuration file.

async reset(full=False)

Reset the spectrographs.

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

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)

Bases: MoTanDevice

A device representing the fibre mask in the spectrophotometric telescope.

async home()

Homes the fibre selector.

list_positions() → list[str]

Returns a list of valid positions.

Return type:

list[str]

async move_relative(steps)

Move the mask a number of motor steps relative to the current position.

async move_to_position(position, rehome=False)

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.

• rehome (bool) – Home the fibre selector before moving to the new position.

SLEW_DELAY: float = 0

Artificial delay introduced to prevent all motors to slew at the same time.

class gort.devices.telescope.Focuser(gort, name, actor)

Bases: MoTanDevice

A device representing a focuser.

async home()

Homes the focuser.

Parameters:

restore_position – Whether to restore the previous focuser position after homing.

async move(dts)

Move the focuser to a position in DT.

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)

Bases: MoTanDevice

A device representing a K-mirror.

async home()

Homes the k-mirror.

async move(degs)

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()

Park the k-mirror at 90 degrees.

async slew(ra, dec, offset_angle=0.0, stop_degs_before=0.0)

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.

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)

Bases: GortDevice

A TwiceAsNice device.

async check_reachable()

Checks if the device is reachable or issues an error.

async is_moving()

Is the device moving.

async is_reachable()

Is the device reachable?

async run_command(command, *args, n_retries=3, delay=1, **kwargs) → ActorReply

Runs a MoTan command with retries.

Return type:

ActorReply

async slew_delay()

Sleeps the SLEW_DELAY amount.

async status()

Returns the status of the device.

async stop(force=False)

Stop the K-mirror movement.

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)

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)

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)

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)

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()

Determines the initial state of the telescope.

async initialise(home=None)

Connects to the telescope and initialises the axes.

Parameters:

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

async is_parked()

Is the telescope parked?

async is_ready()

Checks if the telescope is ready to be moved.

async offset(ra=None, dec=None, axis0=None, axis1=None)

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)

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()

Retrieves the status of the telescope.

async stop()

Stops the mount.

class gort.devices.telescope.TelescopeSet(gort, data)

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)

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)

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)

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)

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()

Initialise all telescopes.

async park(disable=True, use_pw_park=False, alt_az=None, kmirror=True, force=False)

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()

Restarts the lvmpwi and lvmtan deployments and re-homes.

async restart_lvmtan()

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()

Stops all the mounts.

Exposure

class gort.exposure.Exposure(gort, exp_no=None, flavour='object', object='', specs=None)

Bases: Future[Exposure]

A class representing an exposure taken by a SpectrographSet.

Parameters:

• gort (Gort) – 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 static write_to_db(files, table_name)

Records the exposures in gortdb.exposure.

async expose(exposure_time=None, header=None, async_readout=False, show_progress=None, object=None, raise_on_error=True)

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()

Returns the files written by the exposure.

async start_timer(exposure_time, readout_time=55)

Starts the rich timer.

stop_timer()

Cancels the timer.

async verify_files()

Checks that the files have been written and have the right contents.

Observer

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

Bases: object

A class to handle tile observations.

Parameters:

• gort (Gort) – The instance of Gort used to communicate with the devices.

• tile (Tile | None) – 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=None, telescopes=['sci', 'skye', 'skyw', 'spec'])

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 | None) – The maximum time allowed for acquisition. In case of timeout the acquired fields are evaluated and an exception is raised if the acquisition failed. If None, uses the default value from the configuration file.

• telescopes (list[str]) – The list of telescopes to acquire. By default all telescopes are acquired.

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, exposure_starts_callback=None)

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.

• exposure_starts_callback (Optional[Callable[[Any], Coroutine]]) – A callback to be called when the exposure starts.

Returns:

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

Return type:

exposures

async finish_observation(keep_guiding=False)

Finishes the observation, stops the guiders, etc.

get_running_stage()

Returns the running stage.

is_running()

Returns True if the observer is running.

async observe_tile(tile=None, ra=None, dec=None, pa=0.0, use_scheduler=True, dither_position=None, exposure_time=900.0, n_exposures=1, async_readout=False, keep_guiding=False, skip_slew_when_acquired=True, guide_tolerance=None, acquisition_timeout=None, show_progress=None, run_cleanup=True, adjust_focus=True, cleanup_on_interrupt=True) → tuple[bool, list[Exposure]]

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.

• dither_position (int | None) – The dither position to use. If not provided, uses the tile default dither position or zero.

• 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.

• skip_slew_when_acquired (bool) – If the tile has been acquired and is guiding, skips the slew, modifies the dither position, waits until the new position has been acquired, and starts exposing.

• 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.

• acquisition_timeout (float | None) – The maximum time allowed for acquisition. In case of timeout the acquired fields are evaluated and an exception is raised if the acquisition failed. If None, uses the default value from the configuration file.

• 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_interrupt (bool) – If True, registers a signal handler to catch interrupts and run the cleanup routine.

Return type:

tuple[bool, list[Exposure]]

async register_exposure(exposure, nexp=1, tile_id=None, dither_position=0, delay=0.0)

Registers the exposure in the database.

register_overhead(name)

Measures and registers and overhead.

reset(tile=None, on_interrupt=None, reset_stages=True)

Resets the observer.

async set_dither_position(dither)

Reacquire science telescope for a new dither position.

set_tile(tile)

Sets the current tile. Implies reset.

async slew(telescopes=['sci', 'skye', 'skyw', 'spec'])

Slew to the telescope fields.

write_to_log(message, level='debug', header=None, event=None, extra_payload={})

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.

• event (Event | None) – If specified, emits an event of this type.

• extra_payload (dict[str, Any]) – Additional payload to include in the event.

property has_standards

Returns True if standards will be observed.

property tile

Returns the current tile being observed.

Tile

class gort.tile.Coordinates(ra, dec, pa=None, centre_on_fibre=None)

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)

Returns the current altitude of the target.

is_observable()

Determines whether a target is observable.

set_mf_pixel(fibre_name=None, xz=None)

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)

Bases: Coordinates

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

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

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=[])

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)

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.

get_dither_radec()

Returns the dithered RA/Dec coordinates.

class gort.tile.SkyCoordinates(*args, name=None, **kwargs)

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)

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)

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.

classmethod from_coordinates(ra, dec, pa=0.0, sky_coords=None, spec_coords=None, **kwargs)

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)

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.

async disable(note=None)

Disables the tile for observation.

set_dither_position(dither)

Sets the full frame pixel for the science IFU to the dither position.

set_sci_coords(sci_coords) → ScienceCoordinates

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]

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]

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)

Bases: object

Base class for recipes.

async recipe(*args, **kwargs)

The recipe. Must be overridden by the subclass.

class gort.recipes.operations.CleanupRecipe(gort)

Bases: BaseRecipe

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

async recipe(readout=True, turn_lamps_off=True, home_telescopes=False)

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.

• home_telescopes (bool) – If True, homes the telescopes after stopping the guiders.

class gort.recipes.operations.ShutdownRecipe(gort)

Bases: BaseRecipe

Closes the telescope for the night.

async recipe(park_telescopes=True, additional_close=False, disable_overwatcher=False, show_message=True)

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. If the dome fails to close with park_telescopes=True, it will try again without parking the telescopes.

• 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.

• disable_overwatcher (bool) – If True, disables the Overwatcher.

• show_message (bool) – If True, shows a message with instructions on how to confirm the dome is closed.

class gort.recipes.operations.StartupRecipe(gort)

Bases: BaseRecipe

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

async recipe(open_enclosure=True, confirm_open=True, focus=True)

Runs the startup sequence.

Parameters:

• gort – The Gort instance to use.

• 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.

Tools

class gort.tools.GuiderMonitor(gort, actor=None)

Bases: object

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

get_dataframe() → DataFrame | None

Returns the collected data as a dataframe.

Return type:

DataFrame | None

reset()

Resets the internal state.

start_monitoring()

Starts monitoring the guider outputs.

stop_monitoring()

Stops monitoring the guider outputs.

to_header()

Returns a header with pointing and guiding information.

class gort.tools.LogNamespace(logger, header='') → None

Bases: object

A namespace for log messages.

class gort.tools.OverheadDict

Bases: TypedDict

TypedDict for recording overhead information.

async gort.tools.add_night_log_comment(comment, category='other')

Adds a comment to the night log.

gort.tools.angular_separation(lon1, lat1, lon2, lat2)

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.async_noop(*args, **kwargs)

A no-op coroutine.

async gort.tools.cancel_task(task)

Safely cancels a task.

gort.tools.check_overwatcher_not_running(coro)

Decorator that fails a coroutine if the overwatcher is running.

gort.tools.decap(string, add_period=False) → str

Decapitalises the first letter of a string.

Return type:

str

gort.tools.ensure_period(string)

Ensures that a string ends with a period.

gort.tools.get_by_source_id(source_id) → dict | None

Returns Gaia DR3 information for a source ID.

Return type:

dict | None

async gort.tools.get_calibrators(tile_id=None, ra=None, dec=None)

Get calibrators for a tile_id or science pointing.

gort.tools.get_calibrators_sync(tile_id=None, ra=None, dec=None) → dict

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]

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:

list[str]

gort.tools.get_db_connection()

Returns a DB connection from the configuration file parameters.

async gort.tools.get_ephemeris_summary(sjd=None) → dict

Returns the ephemeris summary from lvmapi.

Return type:

dict

gort.tools.get_ephemeris_summary_sync(sjd=None) → dict

Returns the ephemeris summary from lvmapi.

Return type:

dict

async gort.tools.get_exposure_list(mjd=None, as_dataframe=True) → dict | DataFrame

Overloads:

• mjd (int | None), as_dataframe (bool) → polars.DataFrame

• mjd (int | None), as_dataframe (bool) → dict

Retrieves the list of exposures for a given MJD from the LVM API.

gort.tools.get_gort_client(override_overwatcher=None)

Returns a GORT client.

async gort.tools.get_lvmapi_route(route, params={}, timeout=5, **kwargs)

Gets an lvmapi route.

gort.tools.get_md5sum(file)

Returns the MD5 checksum for a file.

gort.tools.get_md5sum_file(file)

Returns the path to the MD5 file for the spectro files.

gort.tools.get_md5sum_from_spectro(file)

Returns the MD5 checksum for a file from the spectro checksum file.

async gort.tools.get_next_tile_id() → dict

Retrieves the next tile_id from the scheduler API.

Return type:

dict

gort.tools.get_next_tile_id_sync() → dict

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)

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)

Converts a string to a valid variable name.

gort.tools.handle_signals(signals, callback=None, cancel=True)

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)

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 (Sequence[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 (Optional[Sequence[str]]) – 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).

async gort.tools.is_actor_running(actor, client=None)

Returns True if the actor is running.

This assumes that the actor implements a ping command.

gort.tools.is_interactive()

Returns True is we are in an interactive session.

gort.tools.is_notebook() → bool

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

async gort.tools.kubernetes_list_deployments()

Retrieves the Kubernetes deployments from the API.

async gort.tools.kubernetes_restart_deployment(name)

Restarts a Kubernetes deployment via the API.

gort.tools.mark_exposure_bad(tile_id, dither_position=0)

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)

Moves the fibre mask in the spectrophotometric telescope at intervals.

Parameters:

• gort (Gort) – 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 (Union[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.

async gort.tools.overwatcher_is_running(client=None)

Returns True if the overwatcher is running.

async gort.tools.ping_host(host)

Pings a host.

gort.tools.record_overheads(payload)

Decorator that records the overhead of a function.

gort.tools.redis_client_async() → AsyncGenerator[Redis, None]

Returns a Redis connection from the configuration file parameters.

Return type:

AsyncGenerator[Redis, None]

gort.tools.redis_client_sync() → Generator[redis.Redis, None]

Returns a Redis connection from the configuration file parameters.

async gort.tools.register_observation(payload)

Registers an observation with the scheduler.

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

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.

async gort.tools.run_lvmapi_task(route, params={}, wait=True, check_interval=3, timeout=None, **kwargs)

Runs an LVM API task route.

Task routes immediately start the task in the background and return a task ID. The API can then be queried until the task is completed.

Parameters:

• route (str) – The route to the task.

• params (dict) – The parameters to pass to the route.

• wait (bool) – If True, waits until the task is completed and returns the result.

• check_interval (float) – The interval at which to check the task status. Ignored if wait=False.

• timeout (float | None) – The timeout for the task execution. Ignored if wait=False.

Returns:

The task ID if wait=False or the result of the task if wait=True.

Return type:

result

async gort.tools.try_or_pass(coro) → T | None

Decorator that wraps a coroutine in a try/except block.

Return type:

Optional[TypeVar(T, bound= Any)]

Transformations

class gort.transforms.HomTrans(entries)

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)

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)

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)

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()

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)

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)

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)

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:

fa

gort.transforms.calculate_position_angle(ra, dec, obstime)

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:

ph

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

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:

tuple[float, float]

gort.transforms.fibre_to_master_frame(fibre_name)

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]

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:

tuple[float, float]

Raises:

ValueError – If the pixel is out of bounds.

gort.transforms.radec_sexagesimal_to_decimal(ra, dec, ra_is_hours=True)

Converts a string of sexagesimal RA and Dec to decimal.

gort.transforms.read_fibermap(path=None, force_cache=False) → DataFrame

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:

DataFrame

gort.transforms.wrap_pa_hex(pa)

Wraps a position angle to the range -30 to 30 degrees.

Parameters:

pa (float) – The position angle to wrap. Always in degrees in the range 0 to infinity.

Returns:

The wrapped position angle in the range -30 to 30 degrees.

Return type:

pa_wrap

gort.transforms.xy_to_radec_offset(xpmm, ypmm)

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=None, payload={}) → None

Bases: GortDeviceError

AG-related error.

exception gort.exceptions.GortDeviceError(message=None, error_code=None, payload={}) → None

Bases: GortError

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

exception gort.exceptions.GortEnclosureError(message=None, error_code=None, payload={}) → None

Bases: GortDeviceError

Enclosure-related error.

exception gort.exceptions.GortError(message=None, error_code=None, payload={})

Bases: Exception

A custom core GortError exception.

exception gort.exceptions.GortGuiderError(message=None, error_code=None, payload={}) → None

Bases: GortDeviceError

Guider-related error.

exception gort.exceptions.GortNPSError(message=None, error_code=None, payload={}) → None

Bases: GortDeviceError

NPS-related error.

exception gort.exceptions.GortNotImplemented(message=None)

Bases: GortError

A custom exception for not yet implemented features.

exception gort.exceptions.GortObserverCancelledError(message=None, error_code=None, payload={})

Bases: GortObserverError

An error issued when the observer is cancelled.

exception gort.exceptions.GortObserverError(message=None, error_code=None, payload={})

Bases: GortError

An error associated with the Observer.

exception gort.exceptions.GortSpecError(message=None, error_code=None, payload={}) → None

Bases: GortDeviceError

Spectrograph-related error.

exception gort.exceptions.GortTelescopeError(message=None, error_code=None, payload={}) → None

Bases: GortDeviceError

Telescope-related error.

exception gort.exceptions.GortTimeout(message=None, error_code=None, payload={})

Bases: GortError

Raised if a timeout occurs.

exception gort.exceptions.GortTimeoutError(message, command=None, remote_command=None)

Bases: GortError

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

exception gort.exceptions.GortUserWarning

Bases: UserWarning, GortWarning

The primary warning class.

exception gort.exceptions.GortWarning

Bases: Warning

Base warning for Gort.

exception gort.exceptions.InvalidRemoteCommand(message, command, remote_command)

Bases: GortError

A command that does not exist or cannot be parsed.

exception gort.exceptions.OverwatcherError(message=None, error_code=None, payload={})

Bases: GortError

An error in the overwatcher.

exception gort.exceptions.RemoteCommandError(message, command, remote_command, reply=None)

Bases: GortError

An error in a remote command to an actor.

exception gort.exceptions.TileError(message=None, error_code=None, payload={})

Bases: GortError

An error associated with a Tile.

exception gort.exceptions.TroubleshooterCriticalError(*args, close_dome=False, **kwargs)

Bases: OverwatcherError

A critical error in the troubleshooter that will shut down the system.

exception gort.exceptions.TroubleshooterTimeoutError(message=None, error_code=None, payload={})

Bases: OverwatcherError

The troubleshooter timed out while running a recipe.