from abc import ABC, abstractmethod
import copy
from enum import Enum
import logging
import requests
from crispy_forms.helper import FormHelper
from crispy_forms.layout import ButtonHolder, Layout, Submit, Div, HTML
from django import forms
from django.conf import settings
from django.contrib.auth.models import Group
from django.core.exceptions import ImproperlyConfigured
from django.core.files.base import ContentFile
from django.utils.module_loading import import_string
from tom_targets.models import Target
logger = logging.getLogger(__name__)
[docs]
class CredentialStatus(Enum):
"""
Enum representing the status of facility credentials.
This enum is used to track the state of credentials throughout the facility lifecycle,
providing clear information about whether credentials are available, where they came from,
and any validation issues.
"""
NOT_INITIALIZED = "not_initialized" # set_user() hasn't been called yet
NO_PROFILE = "no_profile" # User has no Profile (raises ImproperlyConfigured)
PROFILE_EMPTY = "profile_empty" # Profile exists but credentials empty
USING_DEFAULTS = "using_defaults" # Using settings.FACILITIES defaults
USING_USER_CREDS = "using_user_creds" # Using user's Profile credentials
VALIDATION_FAILED_AUTH = "validation_failed_auth" # Credentials failed (401/403)
VALIDATION_FAILED_NETWORK = "validation_failed_network" # Network/server issue
DEFAULT_FACILITY_CLASSES = [
'tom_observations.facilities.lco.LCOFacility',
'tom_observations.facilities.gemini.GEMFacility',
'tom_observations.facilities.soar.SOARFacility',
'tom_observations.facilities.blanco.BLANCOFacility',
]
try:
AUTO_THUMBNAILS = settings.AUTO_THUMBNAILS
except AttributeError:
AUTO_THUMBNAILS = False
def get_service_classes():
try:
TOM_FACILITY_CLASSES = settings.TOM_FACILITY_CLASSES
except AttributeError:
TOM_FACILITY_CLASSES = DEFAULT_FACILITY_CLASSES
service_choices = {}
for service in TOM_FACILITY_CLASSES:
try:
clazz = import_string(service)
except (ImportError, AttributeError) as e:
raise ImportError(f'Could not import {service}: {e}')
service_choices[clazz.name] = clazz
return service_choices
def get_service_class(name):
available_classes = get_service_classes()
try:
return available_classes[name]
except KeyError:
raise ImportError('Could not a find a facility with that name. Did you add it to TOM_FACILITY_CLASSES?')
# This aliasing exists to support backwards compatibility
GenericObservationForm = BaseRoboticObservationForm
[docs]
class BaseObservationFacility(ABC):
"""
This is the class that is responsible for defining the base facility class.
This form is meant to be subclassed by more specific BaseFacility classes that represent a
form for a particular type of facility. For implementing your own form, please look to
the other BaseObservationFacilities.
"""
name = 'BaseObservation'
observation_forms = {}
is_redirect = False
button_label = ""
button_tooltip = ""
def __init__(self):
self.user = None
self.credential_status = CredentialStatus.NOT_INITIALIZED
def set_user(self, user):
self.user = user
def _is_credential_empty(self, credential):
"""
Check if a credential is empty (None, empty string, or whitespace only).
Args:
credential: The credential value to check
Returns:
bool: True if credential is empty, False otherwise
"""
return credential is None or (isinstance(credential, str) and not credential.strip())
def _get_setting_credentials(self, facility_name, credential_keys):
"""
Safely get credentials from settings.FACILITIES.
Args:
facility_name (str): Name of the facility in settings.FACILITIES
credential_keys (list): List of credential key names to retrieve
Returns:
dict: Dictionary mapping credential keys to their values
Raises:
ImproperlyConfigured: If facility or required keys are missing from settings
"""
if not hasattr(settings, 'FACILITIES') or facility_name not in settings.FACILITIES:
raise ImproperlyConfigured(
f"No configuration found for '{facility_name}' in settings.FACILITIES. "
f"Please add default credentials to settings.FACILITIES['{facility_name}']."
)
facility_settings = settings.FACILITIES[facility_name]
credentials = {}
for key in credential_keys:
if key not in facility_settings:
raise ImproperlyConfigured(
f"Required credential key '{key}' not found in "
f"settings.FACILITIES['{facility_name}']. "
f"Please add '{key}' to the facility configuration."
)
credentials[key] = facility_settings[key]
return credentials
def _raise_no_profile_error(self, user, facility_name):
"""
Raise ImproperlyConfigured for missing user profile.
Args:
user: Django User instance
facility_name (str): Name of the facility
Raises:
ImproperlyConfigured: Always raises with informative message
"""
raise ImproperlyConfigured(
f"User '{user.username}' has no {facility_name}Profile configured. "
f"Please create a {facility_name}Profile for this user in the admin interface."
)
def _raise_no_defaults_error(self, user, facility_name):
"""
Raise ImproperlyConfigured when default credentials are needed but missing.
Args:
user: Django User instance
facility_name (str): Name of the facility
Raises:
ImproperlyConfigured: Always raises with informative message
"""
raise ImproperlyConfigured(
f"User '{user.username}' has no credentials configured and no default credentials "
f"found in settings.FACILITIES['{facility_name}']. "
f"Please configure either user credentials in {facility_name}Profile or "
f"default credentials in settings."
)
def all_data_products(self, observation_record):
from tom_dataproducts.models import DataProduct
products = {'saved': [], 'unsaved': []}
for product in self.data_products(observation_record.observation_id):
try:
dp = DataProduct.objects.get(product_id=product['id'])
products['saved'].append(dp)
except DataProduct.DoesNotExist:
products['unsaved'].append(product)
# Obtain products uploaded manually by users
user_products = DataProduct.objects.filter(
observation_record_id=observation_record.id, product_id=None
)
for product in user_products:
products['saved'].append(product)
# Add any JPEG images created from DataProducts
image_products = DataProduct.objects.filter(
observation_record_id=observation_record.id, data_product_type='image_file'
)
for product in image_products:
products['saved'].append(product)
return products
[docs]
def get_facility_context_data(self, **kwargs):
"""
This method provides an opportunity for the Facility subclass to add additional
context data. It will be called by the OberservationCreateView.get_context_data()
method and the context dictionary passed to the template will be updated with the
returned facility context dictionary.
"""
return kwargs
# TODO: consider making submit_observation create ObservationRecords as well
[docs]
@abstractmethod
def submit_observation(self, observation_payload):
"""
This method takes in the serialized data from the form and actually
submits the observation to the remote api
"""
[docs]
@abstractmethod
def validate_observation(self, observation_payload):
"""
Same thing as submit_observation, but a dry run. You can
skip this in different modules by just using "pass"
Typically called by the ObservationForm.is_valid() method.
"""
[docs]
def get_flux_constant(self):
"""
Returns the astropy quantity that a facility uses for its spectral flux conversion.
"""
[docs]
def get_wavelength_units(self):
"""
Returns the astropy units that a facility uses for its spectral wavelengths
"""
[docs]
def is_fits_facility(self, header):
"""
Returns True if the FITS header is from this facility based on valid keywords and associated
values, False otherwise.
"""
return False
[docs]
def get_start_end_keywords(self):
"""
Returns the keywords representing the start and end of an observation window for a facility. Defaults to
``start`` and ``end``.
"""
return 'start', 'end'
[docs]
@abstractmethod
def get_terminal_observing_states(self):
"""
Returns the states for which an observation is not expected
to change.
"""
[docs]
@abstractmethod
def get_observing_sites(self):
"""
Return an iterable of dictionaries that contain the information
necessary to be used in the planning (visibility) tool. The
iterable should contain dictionaries each that contain sitecode,
latitude, longitude and elevation. This is the static information
about a site.
"""
[docs]
def get_facility_weather_urls(self):
"""
Returns a dictionary containing a URL for weather information
for each site in the Facility SITES. This is intended to be useful
in observation planning.
`facility_weather = {'code': 'XYZ', 'sites': [ site_dict, ... ]}`
where
`site_dict = {'code': 'XYZ', 'weather_url': 'http://path/to/weather'}`
"""
return {}
[docs]
def get_facility_status(self):
"""
Returns a dictionary describing the current availability of the Facility
telescopes. This is intended to be useful in observation planning.
The top-level (Facility) dictionary has a list of sites. Each site
is represented by a site dictionary which has a list of telescopes.
Each telescope has an identifier (code) and an status string.
The dictionary hierarchy is of the form:
`facility_dict = {'code': 'XYZ', 'sites': [ site_dict, ... ]}`
where
`site_dict = {'code': 'XYZ', 'telescopes': [ telescope_dict, ... ]}`
where
`telescope_dict = {'code': 'XYZ', 'status': 'AVAILABILITY'}`
See lco.py for a concrete implementation example.
"""
return {}
[docs]
def cancel_observation(self, observation_id):
"""
Takes an observation id and submits a request to the observatory that the observation be cancelled.
If the cancellation was successful, return True. Otherwise, return False.
"""
raise NotImplementedError('This facility has not implemented cancel observation.')
[docs]
@abstractmethod
def get_observation_url(self, observation_id):
"""
Takes an observation id and return the url for which a user
can view the observation at an external location. In this case,
we return a URL to the LCO observation portal's observation
record page.
"""
def get_date_obs_from_fits_header(self, header):
return None
[docs]
class BaseRoboticObservationFacility(BaseObservationFacility):
"""
The facility class contains all the logic specific to the facility it is
written for. Some methods are used only internally (starting with an
underscore) but some need to be implemented by all facility classes.
All facilities should inherit from this class which
provides some base functionality.
In order to make use of a facility class, add the path to
``TOM_FACILITY_CLASSES`` in your ``settings.py``.
This specific class is intended for use with robotic facilities, such as LCO, Gemini, and SOAR.
For an implementation example, please see
https://github.com/TOMToolkit/tom_base/blob/main/tom_observations/facilities/lco.py
"""
name = 'BaseRobotic' # rename in concrete subclasses
def update_observation_status(self, observation_id):
from tom_observations.models import ObservationRecord
records = ObservationRecord.objects.filter(observation_id=observation_id)
if not records:
raise Exception('No records exist for that observation id')
status = self.get_observation_status(observation_id)
for record in records:
record.status = status['state']
record.scheduled_start = status['scheduled_start']
record.scheduled_end = status['scheduled_end']
record.save()
def update_all_observation_statuses(self, target=None):
from tom_observations.models import ObservationRecord
failed_records = []
records = ObservationRecord.objects.filter(facility=self.name)
if target:
records = records.filter(target=target)
records = records.exclude(status__in=self.get_terminal_observing_states())
for record in records:
try:
self.update_observation_status(record.observation_id)
except Exception as e:
failed_records.append((record.observation_id, str(e)))
return failed_records
def save_data_products(self, observation_record, product_id=None):
from tom_dataproducts.models import DataProduct
from tom_dataproducts.utils import create_image_dataproduct
final_products = []
products = self.data_products(observation_record.observation_id, product_id)
for product in products:
dp, created = DataProduct.objects.get_or_create(
product_id=product['id'],
target=observation_record.target,
observation_record=observation_record,
)
if created:
product_data = requests.get(product['url']).content
dfile = ContentFile(product_data)
dp.data.save(product['filename'], dfile)
dp.save()
logger.info('Saved new dataproduct: {}'.format(dp.data))
if AUTO_THUMBNAILS:
create_image_dataproduct(dp)
dp.get_preview()
final_products.append(dp)
return final_products
[docs]
@abstractmethod
def get_observation_status(self, observation_id):
"""
Return the status for a single observation. observation_id should
be able to be used to retrieve the status from the external service.
"""
[docs]
@abstractmethod
def data_products(self, observation_id, product_id=None):
"""
Using an observation_id, retrieve a list of the data
products that belong to this observation. In this case,
the LCO module retrieves a list of frames from the LCO
data archive.
"""
# This aliasing exists to support backwards compatibility
GenericObservationFacility = BaseRoboticObservationFacility
[docs]
class BaseManualObservationFacility(BaseObservationFacility):
"""
The facility class contains all the logic specific to the facility it is
written for. Some methods are used only internally (starting with an
underscore) but some need to be implemented by all facility classes.
All facilities should inherit from this class which
provides some base functionality.
In order to make use of a facility class, add the path to
``TOM_FACILITY_CLASSES`` in your ``settings.py``.
This specific class is intended for use with classical-style manual facilities.
"""
name = 'BaseManual' # rename in concrete subclasses
[docs]
class BaseRedirectObservationFacility(BaseObservationFacility):
"""
A Redirect Facility is a facility which delegates observation creation to
an external website. A RedirectFacility does not need to provide a UI for
observation creation, meaning it does not include a Form. It does still
implement other methods inherited from BaseObservationFacility, such as those
related to fetching status, data products, etc.
Implementations will need to provide the redirect_url method. This constructs
a URL that the user will be redirected to. The arguments the target id and a
callback URL. The external website should be able to redirect the user back
to the callback URL in order to complete the observation creation process
on the TOM side, by fetching the observation either by API endpoint or
some other means, and storing it in the TOM database.
The passed in callback URL will contain the following query parameters:
- target_id: The ID of the target for which the observation was created.
- facility: The name of the facility that created the observation.
The remote facility is responsible for adding these query params:
- observation_id: The ID of the observation that was created.
"""
is_redirect = True
def redirect_url(self, target_id: str, callback_url: str):
raise NotImplementedError("Must implement redirect_url")
def get_template_form(self, observation_type):
raise TypeError("RedirectFacility does not provide a form for observation creation")
[docs]
def submit_observation(self, observation_payload):
raise TypeError("RedirectFacility does not submit observations on remote facilities")
[docs]
def validate_observation(self, observation_payload):
raise TypeError("RedirectFacility does no observation creation or validation")