Protobuf output is not a minimal valid cnet that cnet stats is able to parse (8269e4b3) · Commits · aflab / astrogeology / Autocnet

autocnet/fileio/io_controlnetwork.py

+1 −4

Original line number	Original line	Diff line number	Diff line
	@@ -65,11 +65,9 @@ def to_isis(path, C, mode='w', version=VERSION,
	description,		description,
	username,		username,
	point_sizes)		point_sizes)

	# Write the buffer header		# Write the buffer header
	store.write(buffer_header,HEADERSTARTBYTE)		store.write(buffer_header,HEADERSTARTBYTE)


	# Then write the points, so we know where to start writing, + 1 to avoid overwrite		# Then write the points, so we know where to start writing, + 1 to avoid overwrite
	point_start_offset = HEADERSTARTBYTE + buffer_header_size		point_start_offset = HEADERSTARTBYTE + buffer_header_size
	for i, point in enumerate(point_messages):		for i, point in enumerate(point_messages):
	@@ -82,8 +80,7 @@ def to_isis(path, C, mode='w', version=VERSION,


	store.write(header)		store.write(header)
	import os
	print(os.path.abspath(path))
	class IsisStore(object):		class IsisStore(object):
	"""		"""
	Class to manage IO of an ISIS3 control network (version 2).		Class to manage IO of an ISIS3 control network (version 2).

autocnet/fileio/tests/test_io_controlnetwork.py

+21 −12

Original line number	Original line	Diff line number	Diff line
	@@ -20,9 +20,19 @@ class TestWriteIsisControlNetwork(unittest.TestCase):
	"""		"""
	Not 100% sure how to mock in the DF without creating lots of methods...		Not 100% sure how to mock in the DF without creating lots of methods...
	"""		"""

			serial_times = {295: '1971-07-31T01:24:11.754',
			296: '1971-07-31T01:24:36.970',
			297: '1971-07-31T01:25:02.243',
			298: '1971-07-31T01:25:27.457',
			299: '1971-07-31T01:25:52.669',
			300: '1971-07-31T01:26:17.923'}
			self.serials = ['APOLLO15/METRIC/{}'.format(i) for i in serial_times.values()]

	ids = ['pt1','pt1', 'pt1', 'pt2', 'pt2']		ids = ['pt1','pt1', 'pt1', 'pt2', 'pt2']
	ptype = [2, 2, 2, 2, 2]		ptype = [2, 2, 2, 2, 2]
	serials = ['a', 'b', 'c', 'b', 'c']		serials = [self.serials[0], self.serials[1], self.serials[2],
			self.serials[2], self.serials[3]]
	mtype = [2, 2, 2, 2, 2]		mtype = [2, 2, 2, 2, 2]

	multi_index = pd.MultiIndex.from_tuples(list(zip(ids, ptype, serials, mtype)),		multi_index = pd.MultiIndex.from_tuples(list(zip(ids, ptype, serials, mtype)),
	@@ -37,8 +47,8 @@ class TestWriteIsisControlNetwork(unittest.TestCase):

	io_controlnetwork.to_isis('test.net', cnet, mode='wb', targetname='Moon')		io_controlnetwork.to_isis('test.net', cnet, mode='wb', targetname='Moon')

	self.header_message_size = 83		self.header_message_size = 84
	self.point_start_byte = 65619		self.point_start_byte = 65620

	def test_create_buffer_header(self):		def test_create_buffer_header(self):
	with open('test.net', 'rb') as f:		with open('test.net', 'rb') as f:
	@@ -58,21 +68,21 @@ class TestWriteIsisControlNetwork(unittest.TestCase):
	self.assertEqual('Not modified', header_protocol.lastModified)		self.assertEqual('Not modified', header_protocol.lastModified)

	#Repeating		#Repeating
	self.assertEqual([31, 23], header_protocol.pointMessageSizes)		self.assertEqual([145, 99], header_protocol.pointMessageSizes)

	def test_create_point(self):		def test_create_point(self):
	with open('test.net', 'rb') as f:		with open('test.net', 'rb') as f:

	with open('test.net', 'rb') as f:		with open('test.net', 'rb') as f:
	f.seek(self.point_start_byte)		f.seek(self.point_start_byte)
	for i, length in enumerate([31, 23]):		for i, length in enumerate([145, 99]):
	point_protocol = cnf.ControlPointFileEntryV0002()		point_protocol = cnf.ControlPointFileEntryV0002()
	raw_point = f.read(length)		raw_point = f.read(length)
	point_protocol.ParseFromString(raw_point)		point_protocol.ParseFromString(raw_point)
	self.assertEqual('pt{}'.format(i+1), point_protocol.id)		self.assertEqual('pt{}'.format(i+1), point_protocol.id)
	self.assertEqual(2, point_protocol.type)		self.assertEqual(2, point_protocol.type)
	for m in point_protocol.measures:		for m in point_protocol.measures:
	self.assertTrue(m.serialnumber in ['a', 'b', 'c'])		self.assertTrue(m.serialnumber in self.serials)
	self.assertEqual(2, m.type)		self.assertEqual(2, m.type)

	def test_create_pvl_header(self):		def test_create_pvl_header(self):
	@@ -85,11 +95,10 @@ class TestWriteIsisControlNetwork(unittest.TestCase):
	self.assertEqual(5, mpoints)		self.assertEqual(5, mpoints)

	points_bytes = find_in_dict(pvl_header, 'PointsBytes')		points_bytes = find_in_dict(pvl_header, 'PointsBytes')
	self.assertEqual(54, points_bytes)		self.assertEqual(244, points_bytes)

	points_start_byte = find_in_dict(pvl_header, 'PointsStartByte')		points_start_byte = find_in_dict(pvl_header, 'PointsStartByte')
	self.assertEqual(65619, points_start_byte)		self.assertEqual(65620, points_start_byte)

	def tearDown(self):		def tearDown(self):
	return		os.remove('test.net')
	#os.remove('test.net')		No newline at end of file
	No newline at end of file

autocnet/utils/utils.py

+22 −1

Original line number	Original line	Diff line number	Diff line
	import functools		import functools
	import logging
			import pvl


	def enum(sequential, *named):		def enum(sequential, *named):
	@@ -139,6 +140,26 @@ def find_in_dict(obj, key):
	if item is not None:		if item is not None:
	return item		return item

			def create_isis3_serial(in_cube):
			"""

			Parameters
			----------
			in_cube : str
			Path to the input data cube

			Returns
			-------
			serial : str
			The ISIS3 serial number
			"""
			header = pvl.load(in_cube)
			components = []

			spacecraft_name = find_in_dict(header, 'SpacecraftName')
			instrument_id = find_in_dict(header, 'InstrumentId')
			spacecraft_clock_count = find_in_dict(header, 'StartTime')

	# note that this decorator ignores **kwargs		# note that this decorator ignores **kwargs
	def memoize(obj):		def memoize(obj):
	cache = obj.cache = {}		cache = obj.cache = {}