Adds CassiniIssNac/Wac IsisLabel NaifData Driver (#470) (3aff4592) · Commits · aflab / astrogeology / Ale

ale/base/data_isis.py

+21 −5

Original line number	Diff line number	Diff line
		@@ -62,7 +62,7 @@ def parse_table(table_label, data):
		'Real' : 'f'}

		# Parse the binary data
		fields = table_label.getlist('Field')
		fields = table_label.getall('Field')
		results = {field['Name']:[] for field in fields}
		offset = 0
		for record in range(table_label['Records']):
		@@ -200,7 +200,11 @@ class IsisSpice():
		Instrument pointing table
		"""
		if not hasattr(self, "_inst_pointing_table"):
		for table in self.label.getlist('Table'):
		tables = []
		if "Table" in self.label:
		tables = self.label.getall('Table')

		for table in tables:
		if table['Name'] == 'InstrumentPointing':
		binary_data = read_table_data(table, self._file)
		self._inst_pointing_table = parse_table(table, binary_data)
		@@ -220,7 +224,11 @@ class IsisSpice():
		Body orientation table
		"""
		if not hasattr(self, "_body_orientation_table"):
		for table in self.label.getlist('Table'):
		tables = []
		if "Table" in self.label:
		tables = self.label.getall('Table')

		for table in tables:
		if table['Name'] == 'BodyRotation':
		binary_data = read_table_data(table, self._file)
		self._body_orientation_table = parse_table(table, binary_data)
		@@ -240,7 +248,11 @@ class IsisSpice():
		Instrument position table
		"""
		if not hasattr(self, "_inst_position_table"):
		for table in self.label.getlist('Table'):
		tables = []
		if "Table" in self.label:
		tables = self.label.getall('Table')

		for table in tables:
		if table['Name'] == 'InstrumentPosition':
		binary_data = read_table_data(table, self._file)
		self._inst_position_table = parse_table(table, binary_data)
		@@ -260,7 +272,11 @@ class IsisSpice():
		Sun position table
		"""
		if not hasattr(self, "_sun_position_table"):
		for table in self.label.getlist('Table'):
		tables = []
		if "Table" in self.label:
		tables = self.label.getall('Table')

		for table in tables:
		if table['Name'] == 'SunPosition':
		binary_data = read_table_data(table, self._file)
		self._sun_position_table = parse_table(table, binary_data)

ale/drivers/co_drivers.py

+199 −0

Original line number	Diff line number	Diff line
		@@ -110,6 +110,188 @@ wac_filter_to_focal_length = {
		("T3","IRP90"):201.07
		}

		class CassiniIssIsisLabelNaifSpiceDriver(Framer, IsisLabel, NaifSpice, RadialDistortion, Driver):

		@property
		def instrument_id(self):
		"""
		Returns an instrument id for unquely identifying the instrument, but often
		also used to be piped into Spice Kernels to acquire instrument kernel (IK) NAIF IDs.
		Therefore they use the same NAIF ID asin bods2c calls. Expects instrument_id to be
		defined from a mixin class. This should return a string containing either 'ISSNA' or
		'ISSWA'

		Returns
		-------
		: str
		instrument id
		"""
		return id_lookup[super().instrument_id]

		@property
		def spacecraft_name(self):
		"""
		Spacecraft name used in various Spice calls to acquire
		ephemeris data.

		Returns
		-------
		: str
		Name of the spacecraft
		"""
		return 'CASSINI'

		@property
		def sensor_name(self):
		"""
		Returns the name of the instrument

		Returns
		-------
		: str
		Name of the sensor
		"""
		return name_lookup[super().instrument_id]

		@property
		def ephemeris_start_time(self):
		"""
		Returns the start and stop ephemeris times for the image.

		Returns
		-------
		: float
		start time
		"""
		return spice.str2et(self.utc_start_time.strftime("%Y-%m-%d %H:%M:%S.%f"))[0]

		@property
		def center_ephemeris_time(self):
		"""
		Returns the starting ephemeris time as the ssi framers center is the
		start.

		Returns
		-------
		: double
		Center ephemeris time for an image
		"""
		center_time = self.ephemeris_start_time + (self.exposure_duration / 2.0)
		return center_time

		@property
		def odtk(self):
		"""
		The radial distortion coeffs are not defined in the ik kernels, instead
		they are defined in the ISS Data User Guide (Knowles). Therefore, we
		manually specify the codes here.
		Expects instrument_id to be defined. This should be a string containing either
		CASSINI_ISS_WAC or CASSINI_ISIS_NAC

		Returns
		-------
		: list<float>
		radial distortion coefficients
		"""
		if self.instrument_id == 'CASSINI_ISS_WAC':
		# WAC
		return [0, float('-6.2e-5'), 0]
		elif self.instrument_id == 'CASSINI_ISS_NAC':
		# NAC
		return [0, float('-8e-6'), 0]

		@property
		def focal_length(self):
		"""
		NAC uses multiple filter pairs, each filter combination has a different focal length.
		NAIF's Cassini kernels do not contain focal lengths for NAC filters and
		so we aquired updated NAC filter data from ISIS's IAK kernel.

		"""
		# default focal defined by IAK kernel
		if not hasattr(self, "_focal_length"):
		try:
		default_focal_len = super(CassiniIssPds3LabelNaifSpiceDriver, self).focal_length
		except:
		default_focal_len = float(spice.gdpool('INS{}_DEFAULT_FOCAL_LENGTH'.format(self.ikid), 0, 2)[0])

		filters = tuple(self.label["IsisCube"]["BandBin"]['FilterName'].split("/"))

		if self.instrument_id == "CASSINI_ISS_NAC":
		self._focal_length = nac_filter_to_focal_length.get(filters, default_focal_len)

		elif self.instrument_id == "CASSINI_ISS_WAC":
		self._focal_length = wac_filter_to_focal_length.get(filters, default_focal_len)
		return self._focal_length

		@property
		def _original_naif_sensor_frame_id(self):
		"""
		Original sensor frame ID as defined in Cassini's IK kernel. This
		is the frame ID you want to default to for WAC. For NAC, this Frame ID
		sits between J2000 and an extra 180 rotation since NAC was mounted
		upside down.

		Returns
		-------
		: int
		sensor frame code from NAIF's IK kernel
		"""
		return self.ikid

		@property
		def sensor_frame_id(self):
		"""
		Overwrite sensor frame id to return fake frame ID for NAC representing a
		mounting point with a 180 degree rotation. ID was taken from ISIS's IAK
		kernel for Cassini. This is because NAC requires an extra rotation not
		in NAIF's Cassini kernels. Wac does not require an extra rotation so
		we simply return original sensor frame id for Wac.

		Returns
		-------
		: int
		NAIF's Wac sensor frame ID, or ALE's Nac sensor frame ID
		"""
		if self.instrument_id == "CASSINI_ISS_NAC":
		return 14082360
		elif self.instrument_id == "CASSINI_ISS_WAC":
		return 14082361

		@property
		def frame_chain(self):
		"""
		Construct the initial frame chain using the original sensor_frame_id
		obtained from the ikid. Then tack on the ISIS iak rotation.

		Returns
		-------
		: Object
		Custom Cassini ALE Frame Chain object for rotation computation and application
		"""
		if not hasattr(self, '_frame_chain'):

		try:
		# Call frinfo to check if the ISIS iak has been loaded with the
		# additional reference frame. Otherwise, Fail and add it manually
		_ = spice.frinfo(self.sensor_frame_id)
		self._frame_chain = super().frame_chain
		except spice.utils.exceptions.NotFoundError as e:
		self._frame_chain = FrameChain.from_spice(sensor_frame=self._original_naif_sensor_frame_id,
		target_frame=self.target_frame_id,
		center_ephemeris_time=self.center_ephemeris_time,
		ephemeris_times=self.ephemeris_time,)

		rotation = ConstantRotation([[0, 0, 1, 0]], self.sensor_frame_id, self._original_naif_sensor_frame_id)

		self._frame_chain.add_edge(rotation=rotation)

		return self._frame_chain

		@property
		def sensor_model_version(self):
		return 1

		class CassiniIssPds3LabelNaifSpiceDriver(Framer, Pds3Label, NaifSpice, RadialDistortion, Driver):
		"""
		Cassini mixin class for defining Spice calls.
		@@ -374,3 +556,20 @@ class CassiniIssIsisLabelIsisSpiceDriver(Framer, IsisLabel, IsisSpice, NoDistort
		: float
		"""
		return self.inst_position_table['SpkTableStartTime']

		@property
		def focal_length(self):
		"""
		The focal length of the instrument
		Expects naif_keywords to be defined. This should be a dict containing
		Naif keyworkds from the label.
		Expects ikid to be defined. This should be the integer Naif ID code
		for the instrument.

		Returns
		-------
		float :
		The focal length in millimeters
		"""
		filters = self.label["IsisCube"]["BandBin"]['FilterName'].split("/")
		return self.naif_keywords.get('INS{}_{}_{}_FOCAL_LENGTH'.format(self.ikid, filters[0], filters[1]), None)

ale/util.py

+11 −7

Original line number	Diff line number	Diff line
		@@ -559,7 +559,7 @@ def get_isis_mission_translations(isis_data):
		"""
		mission_translation_file = read_pvl(os.path.join(isis_data, "base", "translations", "MissionName2DataDir.trn"))
		# For some reason this file takes the form [value, key] for mission name -> data dir
		lookup = [l[::-1] for l in mission_translation_file["MissionName"].getlist("Translation")]
		lookup = [l[::-1] for l in mission_translation_file["MissionName"].getall("Translation")]
		return dict(lookup)


		@@ -693,15 +693,19 @@ def search_isis_db(dbobj, labelobj, isis_data):
		# Flag is set when kernels encapsulating the entire image time is found
		full_match = False

		for selection in dbobj.getlist("Selection"):
		files = selection.getlist("File")
		for selection in dbobj.getall("Selection"):
		files = selection.getall("File")
		if not files:
		raise Exception(f"No File found in {selection}")

		# selection criteria
		matches = selection.getlist("Match")
		times = selection.getlist("Time")
		matches = []
		if "Match" in selection:
		matches = selection.getall("Match")

		if not files:
		raise Exception(f"No File found in {selection}")
		times = []
		if "Time" in selection:
		times = selection.getall("Time")

		files = [path.join(*file) if isinstance(file, list) else file for file in files]

tests/pytests/data/isds/cassiniiss_isis_isd.json

+2 −2

Original line number	Diff line number	Diff line
		@@ -125,7 +125,7 @@
		"detector_sample_summing": 1,
		"detector_line_summing": 1,
		"focal_length_model": {
		"focal_length": null
		"focal_length": 2003.09
		},
		"detector_center": {
		"line": 512.5,

tests/pytests/test_cassini_drivers.py

+87 −3

Original line number	Diff line number	Diff line
		@@ -10,7 +10,7 @@ from unittest.mock import PropertyMock, patch
		import json
		from conftest import get_image_label, get_image_kernels, get_isd, convert_kernels, compare_dicts, get_table_data

		from ale.drivers.co_drivers import CassiniIssPds3LabelNaifSpiceDriver
		from ale.drivers.co_drivers import CassiniIssPds3LabelNaifSpiceDriver, CassiniIssIsisLabelNaifSpiceDriver

		@pytest.fixture()
		def test_kernels(scope="module", autouse=True):
		@@ -27,7 +27,8 @@ def test_load_pds(test_kernels):
		isd_str = ale.loads(label_file, props={'kernels': test_kernels})
		isd_obj = json.loads(isd_str)
		print(json.dumps(isd_obj, indent=2))
		assert compare_dicts(isd_obj, compare_dict) == []
		x = compare_dicts(isd_obj, compare_dict)
		assert x == []

		def test_load_isis():
		label_file = get_image_label("N1702360370_1", label_type="isis3")
		@@ -40,7 +41,18 @@ def test_load_isis():
		isd_str = ale.loads(label_file)
		isd_obj = json.loads(isd_str)
		print(json.dumps(isd_obj, indent=2))
		assert compare_dicts(isd_obj, compare_dict) == []
		x = compare_dicts(isd_obj, compare_dict)
		assert x == []

		def test_load_isis_naif(test_kernels):
		label_file = get_image_label("N1702360370_1")
		compare_dict = get_isd("cassiniiss")

		isd_str = ale.loads(label_file, props={"kernels": test_kernels})
		isd_obj = json.loads(isd_str)
		print(json.dumps(isd_obj, indent=2))
		x = compare_dicts(isd_obj, compare_dict)
		assert x == []

		# ========= Test cassini pds3label and naifspice driver =========
		class test_cassini_pds3_naif(unittest.TestCase):
		@@ -130,3 +142,75 @@ class test_cassini_pds3_naif(unittest.TestCase):
		frame_chain = self.driver.frame_chain
		assert len(frame_chain.nodes()) == 0
		from_spice.assert_called_with(center_ephemeris_time=2.4, ephemeris_times=[2.4], nadir=False, sensor_frame=14082360, target_frame=-800)

		# ========= Test cassini isislabel and naifspice driver =========
		class test_cassini_isis_naif(unittest.TestCase):

		def setUp(self):
		label = get_image_label("N1702360370_1", "isis3")
		self.driver = CassiniIssIsisLabelNaifSpiceDriver(label)

		def test_instrument_id(self):
		assert self.driver.instrument_id == "CASSINI_ISS_NAC"

		def test_spacecraft_name(self):
		assert self.driver.spacecraft_name == "CASSINI"

		def test_sensor_name(self):
		assert self.driver.sensor_name == "Imaging Science Subsystem Narrow Angle Camera"

		def test_ephemeris_start_time(self):
		with patch('ale.drivers.co_drivers.spice.str2et', return_value=[12345]) as str2et:
		assert self.driver.ephemeris_start_time == 12345
		str2et.assert_called_with('2011-12-12 05:02:19.773000')

		def test_center_ephemeris_time(self):
		with patch('ale.drivers.co_drivers.spice.str2et', return_value=[12345]) as str2et:
		print(self.driver.exposure_duration)
		assert self.driver.center_ephemeris_time == 12347.3
		str2et.assert_called_with('2011-12-12 05:02:19.773000')

		def test_odtk(self):
		assert self.driver.odtk == [0, -8e-06, 0]

		def test_focal_length(self):
		# This value isn't used for anything in the test, as it's only used for the
		# default focal length calculation if the filter can't be found.
		with patch('ale.drivers.co_drivers.spice.gdpool', return_value=[10.0]) as gdpool, \
		patch('ale.base.data_naif.spice.bods2c', return_value=-12345) as bods2c:
		assert self.driver.focal_length == 2003.09

		def test_sensor_model_version(self):
		assert self.driver.sensor_model_version == 1

		def test_sensor_frame_id(self):
		assert self.driver.sensor_frame_id == 14082360

		@patch('ale.transformation.FrameChain.from_spice', return_value=ale.transformation.FrameChain())
		def test_custom_frame_chain(self, from_spice):
		with patch('ale.drivers.co_drivers.spice.bods2c', return_value=-12345) as bods2c, \
		patch('ale.drivers.co_drivers.CassiniIssIsisLabelNaifSpiceDriver.target_frame_id', \
		new_callable=PropertyMock) as target_frame_id, \
		patch('ale.drivers.co_drivers.CassiniIssIsisLabelNaifSpiceDriver.ephemeris_start_time', \
		new_callable=PropertyMock) as ephemeris_start_time:
		ephemeris_start_time.return_value = .1
		target_frame_id.return_value = -800
		frame_chain = self.driver.frame_chain
		assert len(frame_chain.nodes()) == 2
		assert 14082360 in frame_chain.nodes()
		assert -12345 in frame_chain.nodes()
		from_spice.assert_called_with(center_ephemeris_time=2.4000000000000004, ephemeris_times=[2.4000000000000004], sensor_frame=-12345, target_frame=-800)

		@patch('ale.transformation.FrameChain.from_spice', return_value=ale.transformation.FrameChain())
		def test_custom_frame_chain_iak(self, from_spice):
		with patch('ale.drivers.co_drivers.spice.bods2c', return_value=-12345) as bods2c, \
		patch('ale.drivers.co_drivers.CassiniIssIsisLabelNaifSpiceDriver.target_frame_id', \
		new_callable=PropertyMock) as target_frame_id, \
		patch('ale.drivers.co_drivers.CassiniIssIsisLabelNaifSpiceDriver.ephemeris_start_time', \
		new_callable=PropertyMock) as ephemeris_start_time, \
		patch('ale.drivers.co_drivers.spice.frinfo', return_value=True) as frinfo:
		ephemeris_start_time.return_value = .1
		target_frame_id.return_value = -800
		frame_chain = self.driver.frame_chain
		assert len(frame_chain.nodes()) == 0
		from_spice.assert_called_with(center_ephemeris_time=2.4000000000000004, ephemeris_times=[2.4000000000000004], nadir=False, sensor_frame=14082360, target_frame=-800, )