[docs]
class Info(ABC):
def __init__(self, configuration_path=None):
"""
Initialize an Info object.
The information object contains multiple :class:`InfoBase` objects
pertaining to specific types of information. It is one of the most
high-level classes in the reduction, containing all the necessary
configuration options and information from scans, and should
therefore be used to direct the reduction process.
Parameters
----------
configuration_path : str, optional
An alternate directory path to the configuration tree to be
used during the reduction. The default is
<package>/data/configurations.
"""
self.name = None
self.scan = None
self.parallelism = None
self.parent = None
self.configuration = Configuration(
configuration_path=configuration_path)
self.instrument = InstrumentInfo()
self.astrometry = AstrometryInfo()
self.observation = ObservationInfo()
self.origin = OriginationInfo()
self.telescope = TelescopeInfo()
@property
def referenced_attributes(self):
"""
Return a set of attribute names that should be referenced during copy.
Returns
-------
set (str)
"""
return {'configuration', 'scan', 'parent'}
[docs]
def set_parent(self, owner):
"""
Set the owner of the information.
Parameters
----------
owner : object
Returns
-------
None
"""
self.parent = owner
[docs]
def copy(self):
"""
Create and return a copy of the Info object.
The 'applied_scan' and 'configuration' attributes are referenced. All
other attributes are copied using their `copy` method, or deepcopy.
Returns
-------
Info
"""
new = self.__class__()
referenced_only = self.referenced_attributes
for attribute, value in self.__dict__.items():
if attribute in referenced_only:
setattr(new, attribute, value)
elif hasattr(value, 'copy'):
setattr(new, attribute, value.copy())
else:
setattr(new, attribute, deepcopy(value))
# So that info configurations reference new.configuration
for info_value in new.available_info.values():
info_value.configuration = new.configuration
return new
[docs]
def unlink_configuration(self):
"""
Create and set a separate copy of the configuration.
Ensures that the configuration is not referenced so that any changes
that occur do not impact other reduction objects unintentionally.
Returns
-------
None
"""
self.configuration = self.configuration.copy()
for info in self.available_info.values():
info.configuration = self.configuration
[docs]
def unlink_scan(self):
"""
Create and set a separate copy of the scan.
Ensures that the scan associated with this information is not
referenced by any other reduction object in cases where we do not
want any updates to occur outside of the info scope.
Returns
-------
None
"""
self.scan = deepcopy(self.scan)
@property
def available_info(self):
"""
Return all available specialized information.
Returns all contained :class:`InfoBase` objects and their name as a
dictionary of the form {name (`str`): info (`InfoBase`)}.
Returns
-------
dict
"""
result = {}
for key, value in self.__dict__.items():
if isinstance(value, InfoBase):
result[key] = value
return result
@property
def config_path(self):
"""
Return the configuration path for this information.
Returns
-------
file_path : str
"""
if self.name is None:
return self.configuration.config_path
else:
return os.path.join(self.configuration.config_path, self.name)
[docs]
@classmethod
def instance_from_instrument_name(cls, name, configuration_path=None):
"""
Return an Info instance given an instrument name.
Parameters
----------
name : str
The name of the instrument
configuration_path : str, optional
An alternate directory path to the configuration tree to be
used during the reduction. The default is
<package>/data/configurations.
Returns
-------
Info
"""
return info_class_for(name)(
configuration_path=configuration_path)
@property
def size_unit(self):
"""
Return the size unit for the instrument.
Returns
-------
astropy.units.Unit
"""
return self.instrument.get_size_unit()
@property
def frequency(self):
"""
Return the instrument frequency.
Returns
-------
astropy.units.Quantity
"""
return self.instrument.frequency
@property
def integration_time(self):
"""
Return the instrument integration time.
Returns
-------
astropy.units.Quantity
The instrument integration time in seconds.
"""
return self.instrument.integration_time
@integration_time.setter
def integration_time(self, value):
"""
Set the instrument integration time.
Parameters
----------
value : units.Quantity
Returns
-------
None
"""
if not isinstance(value, units.Quantity):
raise ValueError(f"Integration time must be {units.Quantity}.")
self.instrument.integration_time = value.copy()
@property
def resolution(self):
"""
Return the instrument resolution (spatial size)
Returns
-------
astropy.units.Quantity
"""
return self.instrument.resolution
@resolution.setter
def resolution(self, value):
"""
Set the instrument resolution (spatial size).
Parameters
----------
value : astropy.units.Quantity
Returns
-------
None
"""
unit = self.size_unit
if isinstance(value, units.Quantity):
self.instrument.resolution = value.to(unit)
else:
self.instrument.resolution = value * unit
@property
def sampling_interval(self):
"""
Return the instrument sampling interval (time).
Returns
-------
astropy.units.Quantity
"""
return self.instrument.sampling_interval
@sampling_interval.setter
def sampling_interval(self, value):
"""
Set the instrument sampling interval (time).
Parameters
----------
value : units.Quantity
Returns
-------
None
"""
if not isinstance(value, units.Quantity):
raise ValueError(f"Sampling interval must be {units.Quantity}.")
self.instrument.sampling_interval = value.copy()
@property
def gain(self):
"""
Return the overall instrument gain.
Returns
-------
float
"""
return self.instrument.gain
@gain.setter
def gain(self, value):
"""
Set the overall instrument gain.
Parameters
----------
value : float
Returns
-------
None
"""
self.instrument.gain = value
@property
def telescope_name(self):
"""
Return the name of the telescope.
Returns
-------
name : str
"""
return self.telescope.get_telescope_name()
@property
def jansky_per_beam(self):
"""
Return the Jansky's per beam.
Returns
-------
astropy.units.Quantity
"""
return self.instrument.jansky_per_beam()
@property
def data_unit(self):
"""
Return the data unit of the channel data.
Returns
-------
astropy.units.Unit
"""
return self.instrument.get_data_unit()
@property
def kelvin(self):
"""
Return the instrument temperature in Kelvin.
Returns
-------
astropy.units.Quantity
"""
return self.instrument.kelvin()
@property
def point_size(self):
"""
Return the point size of the instrument.
Returns
-------
units.Quantity
"""
return self.instrument.get_point_size()
@property
def source_size(self):
"""
Return the source size of the observation.
Returns
-------
units.Quantity
"""
return self.instrument.get_source_size()
[docs]
def get_channel_class(self):
"""
Returns a Channels instance for a given instrument.
Returns
-------
Channels
"""
return channel_class_for(self.instrument.name)
[docs]
def get_channel_data_class(self):
"""
Return the appropriate ChannelData class for a given instrument.
Returns
-------
channel_data : class (ChannelData)
"""
return channel_data_class_for(self.instrument.name)
[docs]
def get_channel_group_class(self):
"""
Returns the appropriate ChannelGroup class for a given instrument.
Returns
-------
class (ChannelGroup)
"""
return channel_group_class_for(self.instrument.name)
[docs]
def get_scan_class(self):
"""
Returns the appropriate Scan class for a given instrument.
Returns
-------
class (ChannelGroup)
"""
return get_scan_class(self.instrument.name)
[docs]
def get_channels_instance(self):
"""
Return a Channels instance for this information.
Returns
-------
Channels
"""
channel_class = self.get_channel_class()
return channel_class(info=self)
[docs]
def get_source_model_instance(self, scans, reduction=None):
"""
Return the source model applicable to the channel type.
Parameters
----------
scans : list (Scan)
A list of scans for which to create the source model.
reduction : Reduction, optional
The reduction to which the model will belong.
Returns
-------
Map
"""
source_type = self.configuration.get_string('source.type')
if source_type is None:
return None
if source_type == 'skydip':
return SkyDip(info=self, reduction=reduction)
elif source_type == 'map':
return AstroIntensityMap(info=self, reduction=reduction)
elif source_type == 'cube':
return SpectralCube(info=self, reduction=reduction)
elif source_type == 'null':
return None
else:
return None
[docs]
def validate_configuration_registration(self):
"""
Ensure that all configuration files are registered.
Returns
-------
None
"""
config_files = self.configuration.config_files
if config_files is not None:
for config_file in config_files:
self.register_config_file(config_file)
[docs]
def register_config_file(self, filename): # pragma: no cover
"""
Register that a configuration file has been read.
Parameters
----------
filename : str
Returns
-------
None
"""
pass
[docs]
def set_date_options(self, mjd):
"""
Set the configuration options for a given date (in MJD).
Parameters
----------
mjd : float
Returns
-------
None
"""
self.configuration.set_date(float(mjd), validate=True)
[docs]
def set_mjd_options(self, mjd):
"""
Set the configuration options for a given date (in MJD).
Parameters
----------
mjd : float
Returns
-------
None
"""
self.set_date_options(mjd)
[docs]
def set_serial_options(self, serial):
"""
Set the configuration options for a given serial number.
Parameters
----------
serial : int or str
Returns
-------
None
"""
self.configuration.set_serial(serial, validate=True)
[docs]
def set_object_options(self, source_name):
"""
Set the configuration object options for a given source.
Parameters
----------
source_name : str
Returns
-------
None
"""
self.configuration.set_object(source_name, validate=True)
[docs]
def read_configuration(self, configuration_file='default.cfg',
validate=True):
"""
Read and apply a configuration file.
Parameters
----------
configuration_file : str, optional
Path to, or name of, a configuration file.
validate : bool, optional
If `True` (default), validate information read from the
configuration file.
Returns
-------
None
"""
if self.configuration.instrument_name is None:
self.configuration.set_instrument(self.name)
self.configuration.read_configuration(
configuration_file, validate=validate)
self.register_config_file(configuration_file)
self.validate_configuration_registration()
[docs]
def apply_configuration(self):
"""
Apply a configuration to the information.
This should be once the FITS information from a scan file has been
applied to the information via `parse_header`.
Returns
-------
None
"""
if not isinstance(self.configuration, Configuration):
raise ValueError(f"Configuration must be a "
f"{Configuration} instance.")
if self.configuration.instrument_name is None:
self.configuration.set_instrument(self.name)
for info in self.available_info.values():
info.set_configuration(self.configuration)
[docs]
def validate(self):
"""
Validate all information following a read of scans/integrations.
At this point the astrometry can be verified.
Returns
-------
None
"""
for info in self.available_info.values():
info.validate()
[docs]
def validate_scan(self, scan):
"""
Validate a scan.
Parameters
----------
scan : Scan
Returns
-------
None
"""
for info in self.available_info.values():
info.validate_scan(scan)
[docs]
def add_history(self, header, scans=None): # pragma: no cover
"""
Add HISTORY messages to a FITS header.
Parameters
----------
header : astropy.io.fits.header.Header
The header to update with HISTORY messages.
scans : list (Scan), optional
A list of scans to add HISTORY messages from if necessary.
Returns
-------
None
"""
pass
[docs]
def validate_scans(self, scans):
"""
Validate a list of scans specific to the instrument
Parameters
----------
scans : list (Scan)
A list of scans.
Returns
-------
None
"""
if scans is None:
return
for i, scan in enumerate(scans):
if scan is None:
continue
if not scan.is_valid():
continue
if self.configuration.get_bool('jackknife.alternate'):
log.info("JACKKNIFE: Alternating scans.")
if scan.size == 0:
return
if i % 2 == 0:
for integration in scan.integrations:
integration.gain *= -1.0
[docs]
@classmethod
def get_focus_string(cls, focus):
"""
Return a string representing the focus.
Parameters
----------
focus : InstantFocus
Returns
-------
str
"""
if focus is None:
return ' No instant focus.'
msg = ''
if focus.x is not None:
msg += f"\n Focus.dX --> {focus.x.to('mm')}"
if focus.y is not None:
msg += f"\n Focus.dY --> {focus.y.to('mm')}"
if focus.z is not None:
msg += f"\n Focus.dZ --> {focus.z.to('mm')}"
return msg
[docs]
def get_name(self):
"""
Return the name of the information.
Returns
-------
name : str
"""
if self.name is None:
return ''
return self.name
[docs]
def set_name(self, name):
"""
Set the name for the information.
Parameters
----------
name : str
Returns
-------
None
"""
self.name = str(name)
[docs]
def set_outpath(self):
"""
Set the output directory based on the configuration.
If the configuration path does not exist, it will be created if the
'outpath.create' option is set. Otherwise, an error will be raised.
Returns
-------
None
"""
self.configuration.set_outpath()