PVL header being populated with point size information (f1b81fbd) · Commits · aflab / astrogeology / Autocnet

autocnet/control/control.py

+20 −3

Original line number	Diff line number	Diff line
		@@ -30,6 +30,23 @@ class C(pd.DataFrame):

		creationdate : str
		The date that this control network was created.

		Examples
		--------
		This example illustrates the manual creation of a pandas dataframe with
		a multi-index (created from a list of tuples).

		>>> ids = ['pt1','pt1', 'pt1', 'pt2', 'pt2']
		>>> ptype = [2,2,2,2,2]
		>>> serials = ['a', 'b', 'c', 'b', 'c']
		>>> mtype = [2,2,2,2,2]
		>>> multi_index = pd.MultiIndex.from_tuples(list(zip(ids, ptype, serials, mtype)),\
		names=['Id', 'Type', 'Serial Number', 'Measure Type'])
		>>> columns = ['Random Number']
		>>> data_length = 5
		>>> data = np.random.randn(data_length)
		>>> C = control.C(data, index=multi_index, columns=columns)

		"""
		def __init__(self, args, *kwargs):
		super(C, self).__init__(args, *kwargs)
		@@ -44,13 +61,13 @@ class C(pd.DataFrame):
		@property
		def n(self):
		if not getattr(self, '_n', None):
		self._n = 100
		self._n = len(self.index.levels[0])
		return self._n

		@property
		def m(self):
		if not getattr(self, '_m', None):
		self._m = 500
		self._m = len(self)
		return self._m

		@property

autocnet/control/tests/test_control.py

+19 −4

Original line number	Diff line number	Diff line
		@@ -3,6 +3,9 @@ import sys
		from time import gmtime, strftime
		import unittest

		import numpy as np
		import pandas as pd

		sys.path.insert(0, os.path.abspath('..'))

		from autocnet.control import control
		@@ -11,17 +14,29 @@ from autocnet.control import control
		class TestC(unittest.TestCase):

		def setUp(self):
		self.C = control.C()
		ids = ['pt1','pt1', 'pt1', 'pt2', 'pt2']
		ptype = [2,2,2,2,2]
		serials = ['a', 'b', 'c', 'b', 'c']
		mtype = [2,2,2,2,2]

		multi_index = pd.MultiIndex.from_tuples(list(zip(ids, ptype, serials, mtype)),
		names=['Id', 'Type', 'Serial Number', 'Measure Type'])


		columns = ['Random Number']
		self.data_length = 5
		data = np.random.randn(self.data_length)

		self.C = control.C(data, index=multi_index, columns=columns)

		def test_n_point(self):
		self.assertEqual(self.C.n, 100)
		self.assertEqual(self.C.n,2)

		def test_n_measures(self):
		self.assertEqual(self.C.m, 500)
		self.assertEqual(self.C.m, self.data_length)

		def test_modified_date(self):
		self.assertEqual(self.C.modifieddate, 'Not modified')

		def test_creation_date(self):

		self.assertEqual(self.C.creationdate, strftime("%Y-%m-%d %H:%M:%S", gmtime()))
		No newline at end of file

autocnet/fileio/io_controlnetwork.py

+53 −14

Original line number	Diff line number	Diff line
		@@ -58,15 +58,20 @@ def to_isis(path, C, mode='w', version=VERSION,

		if isinstance(path, str):
		with IsisStore(path, mode) as store:
		point_messages, point_sizes = store.create_points(C)
		points_bytes = sum(point_sizes)
		#store.write()
		buffer_header, buffer_header_size = store.create_buffer_header(C, networkid,
		targetname,
		description,
		username)
		username,
		point_sizes)

		print(buffer_header_size)
		store.write(buffer_header,HEADERSTARTBYTE)

		header = store.create_pvl_header(C, version, headerstartbyte, networkid,
		targetname, description, username, buffer_header_size)
		targetname, description, username,
		buffer_header_size, points_bytes)


		store.write(header)
		@@ -101,8 +106,41 @@ class IsisStore(object):
		self._handle.seek(offset)
		self._handle.write(data)

		def create_points(self, cnet):
		point_sizes = []
		point_messages = []

		for point_id in cnet.index.levels[0]:

		# Instantiate the proto spec
		point_spec = cnf.ControlPointFileEntryV0002()

		# Get the subset of the dataframe
		point = cnet.loc[point_id]

		point_spec.id = point_id
		point_spec.type = 2 # Hard coded to free

		# A single extend call is cheaper than many add calls to pack points
		measure_iterable = []

		for m in point.iterrows():
		measure_spec = point_spec.Measure()
		serialnumber = m[0][1]
		mtype = m[0][2]
		measure_spec.serialnumber = serialnumber
		measure_spec.type = mtype
		measure_iterable.append(measure_spec)
		point_spec.measures.extend(measure_iterable)

		point_message = point_spec.SerializeToString()
		point_sizes.append(sys.getsizeof(point_message))
		point_messages.append(point_message)

		return point_messages, point_sizes

		def create_buffer_header(self, cnet, networkid, targetname,
		description, username):
		description, username, point_sizes):
		"""
		Create the Google Protocol Buffer header using the
		protobuf spec.
		@@ -125,16 +163,15 @@ class IsisStore(object):
		raw_header_message.targetName = targetname
		raw_header_message.userName = username


		raw_header_message.pointMessageSizes.extend(range(10))
		print(dir(raw_header_message))
		raw_header_message.pointMessageSizes.extend(point_sizes)
		header_message = raw_header_message.SerializeToString()
		header_message_size = sys.getsizeof(header_message)

		return header_message, header_message_size

		def create_pvl_header(self, cnet, version, headerstartbyte,
		networkid, targetname, description, username,
		buffer_header_size):
		buffer_header_size, points_bytes):
		"""
		Create the PVL header object
		Parameters
		@@ -142,6 +179,9 @@ class IsisStore(object):
		cnet : C
		A control net object

		points_bytes : int
		The total number of bytes all points require

		Returns
		-------

		@@ -149,16 +189,15 @@ class IsisStore(object):

		encoder = pvl.encoder.IsisCubeLabelEncoder

		headerbytes = buffer_header_size
		pointsstartbyte = HEADERSTARTBYTE + buffer_header_size
		pointsbytes = 1
		header_bytes = buffer_header_size
		points_start_byte = HEADERSTARTBYTE + buffer_header_size

		header = pvl.PVLModule([
		('Protobuffer',
		{'Core':{'HeaderStartByte':headerstartbyte,
		'HeaderBytes':headerbytes,
		'PointsStartByte':pointsstartbyte,
		'PointsBytes':pointsbytes}}),
		'HeaderBytes':header_bytes,
		'PointsStartByte':points_start_byte,
		'PointsBytes':points_bytes}}),
		('ControlNetworkInfo',pvl.PVLGroup([
		('NetworkId', networkid),
		('TargetName', targetname),

autocnet/fileio/tests/test_io_controlnetwork.py

+40 −13

Original line number	Diff line number	Diff line
		import os
		from time import gmtime, strftime
		import unittest
		from unittest.mock import Mock
		from unittest.mock import MagicMock
		import sys
		sys.path.insert(0, os.path.abspath('..'))

		import numpy as np
		import pandas as pd
		import pvl

		from .. import io_controlnetwork
		@@ -17,19 +19,29 @@ class TestWriteIsisControlNetwork(unittest.TestCase):

		def setUp(self):
		"""
		The C object is mocked in, we do not want to test it here
		Not 100% sure how to mock in the DF without creating lots of methods...
		"""
		cnet = Mock(spec=C)
		cnet.n = 75
		cnet.m = 621
		ids = ['pt1','pt1', 'pt1', 'pt2', 'pt2']
		ptype = [2,2,2,2,2]
		serials = ['a', 'b', 'c', 'b', 'c']
		mtype = [2,2,2,2,2]

		multi_index = pd.MultiIndex.from_tuples(list(zip(ids, ptype, serials, mtype)),
		names=['Id', 'Type', 'Serial Number', 'Measure Type'])

		columns = ['Random Number']
		self.data_length = 5
		data = np.random.randn(self.data_length)

		self.creation_time = strftime("%Y-%m-%d %H:%M:%S", gmtime())
		cnet.creationdate = self.creation_time
		self.modified = 'Not modified'
		cnet.modifieddate = self.modified
		cnet = C(data, index=multi_index, columns=columns)

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



		def test_create_buffer_header(self):
		header_message_size = 124
		header_message_size = 116
		with open('test.net', 'rb') as f:
		f.seek(io_controlnetwork.HEADERSTARTBYTE)
		raw_header_message = f.read(header_message_size)
		@@ -44,17 +56,32 @@ class TestWriteIsisControlNetwork(unittest.TestCase):
		self.assertEqual(self.creation_time,
		header_protocol.created)
		self.assertEqual('None', header_protocol.description)
		self.assertEqual(self.modified, header_protocol.lastModified)
		self.assertEqual('Not modified', header_protocol.lastModified)

		#Repeating
		self.assertEqual(list(range(10)), header_protocol.pointMessageSizes)
		self.assertEqual([64, 56], header_protocol.pointMessageSizes)

		def test_create_point(self):
		with open('test.net', 'rb') as f:
		point_protocol = cnf.ControlPointFileEntryV0002()
		#self.assertTrue(False)

		def test_create_pvl_header(self):
		pvl_header = pvl.load('test.net')

		npoints = find_in_dict(pvl_header, 'NumberOfPoints')
		self.assertEqual(75, npoints)
		self.assertEqual(2, npoints)

		mpoints = find_in_dict(pvl_header, 'NumberOfMeasures')
		self.assertEqual(621, mpoints)
		self.assertEqual(5, mpoints)

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

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



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