本文整理汇总了Python中mi.dataset.parser.wfp_eng__stc_imodem.WfpEngStcImodemParser类的典型用法代码示例。如果您正苦于以下问题:Python WfpEngStcImodemParser类的具体用法?Python WfpEngStcImodemParser怎么用?Python WfpEngStcImodemParser使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了WfpEngStcImodemParser类的13个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_simple_telemetered
def test_simple_telemetered(self):
"""
Read test data and pull out data particles one at a time.
Assert that the results are those we expected.
"""
file_path = os.path.join(RESOURCE_PATH, "simple.dat")
with open(file_path, "rb") as stream_handle:
parser = WfpEngStcImodemParser(
self._telem_config, None, stream_handle, lambda state, ingested: None, lambda data: None
)
result = parser.get_records(5)
self.assert_particles(result, "simple_telem.yml", RESOURCE_PATH)
开发者ID:emilyhahn,项目名称:mi-dataset,代码行数:15,代码来源:test_wfp_eng__stc_imodem.py
示例2: test_long_stream_recovered
def test_long_stream_recovered(self):
"""
Test a long stream of data
"""
file_path = os.path.join(RESOURCE_PATH, "E0000039.DAT")
with open(file_path, "rb") as stream_handle:
parser = WfpEngStcImodemParser(
self._recov_config, None, stream_handle, lambda state, ingested: None, lambda data: None
)
# start with the start time record
result = parser.get_records(100)
self.assert_particles(result, "E0000039_recov.yml", RESOURCE_PATH)
开发者ID:emilyhahn,项目名称:mi-dataset,代码行数:15,代码来源:test_wfp_eng__stc_imodem.py
示例3: test_bad_data_telemetered
def test_bad_data_telemetered(self):
"""
Ensure that missing data causes us to miss records
TODO: This test should be improved if we come up with a more accurate regex for the data sample
"""
file_path = os.path.join(RESOURCE_PATH, "bad_data.dat")
with open(file_path, "rb") as stream_handle:
parser = WfpEngStcImodemParser(
self._telem_config, None, stream_handle, lambda state, ingested: None, lambda data: None
)
result = parser.get_records(5)
self.assert_particles(result, "bad_data_telem.yml", RESOURCE_PATH)
开发者ID:emilyhahn,项目名称:mi-dataset,代码行数:15,代码来源:test_wfp_eng__stc_imodem.py
示例4: test_get_many_telemetered
def test_get_many_telemetered(self):
"""
Read test data and pull out multiple data particles at one time.
Assert that the results are those we expected.
"""
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA_SHORT)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED), self.start_state, self.stream_handle,
self.state_callback, self.pub_callback)
# start with the start time record
result = self.parser.get_records(1)
self.assert_result(result, 24, self.particle_a_start_time_telem, False)
result = self.parser.get_records(4)
self.assertEqual(result, [self.particle_a_eng_telem,
self.particle_b_eng_telem,
self.particle_c_eng_telem,
self.particle_d_eng_telem])
self.assertEqual(self.parser._state[StateKey.POSITION], 128)
self.assertEqual(self.state_callback_value[StateKey.POSITION], 128)
self.assertEqual(self.publish_callback_value[0], self.particle_a_eng_telem)
self.assertEqual(self.publish_callback_value[1], self.particle_b_eng_telem)
self.assertEqual(self.publish_callback_value[2], self.particle_c_eng_telem)
self.assertEqual(self.publish_callback_value[3], self.particle_d_eng_telem)
self.assertEqual(self.file_ingested, True)
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:26,代码来源:test_wfp_eng__stc_imodem.py
示例5: test_simple_telemetered
def test_simple_telemetered(self):
"""
Read test data and pull out data particles one at a time.
Assert that the results are those we expected.
"""
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA_SHORT)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED), self.start_state, self.stream_handle,
self.state_callback, self.pub_callback)
# start with the start time record
result = self.parser.get_records(1)
self.assert_result(result, 24, self.particle_a_start_time_telem, False)
# next get engineering records
result = self.parser.get_records(1)
self.assert_result(result, 50, self.particle_a_eng_telem, False)
result = self.parser.get_records(1)
self.assert_result(result, 76, self.particle_b_eng_telem, False)
result = self.parser.get_records(1)
self.assert_result(result, 102, self.particle_c_eng_telem, False)
result = self.parser.get_records(1)
self.assert_result(result, 128, self.particle_d_eng_telem, True)
# no data left, dont move the position
result = self.parser.get_records(1)
self.assertEqual(result, [])
self.assertEqual(self.parser._state[StateKey.POSITION], 128)
self.assertEqual(self.state_callback_value[StateKey.POSITION], 128)
self.assert_(isinstance(self.publish_callback_value, list))
self.assertEqual(self.publish_callback_value[0], self.particle_d_eng_telem)
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:32,代码来源:test_wfp_eng__stc_imodem.py
示例6: test_bad_flags_telemetered
def test_bad_flags_telemetered(self):
"""
test that we don't parse any records when the flags are not what we expect
"""
with self.assertRaises(SampleException):
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA_BAD_FLAGS)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED), self.start_state, self.stream_handle,
self.state_callback, self.pub_callback)
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:9,代码来源:test_wfp_eng__stc_imodem.py
示例7: test_mid_state_start_telemetered
def test_mid_state_start_telemetered(self):
"""
Test starting the parser in a state in the middle of processing
"""
new_state = {StateKey.POSITION:76}
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA_SHORT)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED), new_state, self.stream_handle,
self.state_callback, self.pub_callback)
result = self.parser.get_records(1)
self.assert_result(result, 102, self.particle_c_eng_telem, False)
result = self.parser.get_records(1)
self.assert_result(result, 128, self.particle_d_eng_telem, True)
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:14,代码来源:test_wfp_eng__stc_imodem.py
示例8: test_bug_3241
def test_bug_3241(self):
"""
This test was created to validate fixes to bug #3241
It validates the parser can parse a file recovered from
Global platforms
"""
file_path = os.path.join(RESOURCE_PATH, 'E0000000.DAT')
with open(file_path, 'rb') as stream_handle:
parser = WfpEngStcImodemParser(
self._recov_config,
None,
stream_handle,
lambda state, ingested: None,
lambda data: None)
result = parser.get_records(100)
# make sure we get 100 particles back
self.assertEquals(len(result), 100)
# make sure there are no errors
self.assertEquals(len(self.exception_callback_value), 0)
开发者ID:GrimJ,项目名称:mi-dataset,代码行数:23,代码来源:test_wfp_eng__stc_imodem.py
示例9: test_bad_data_telemetered
def test_bad_data_telemetered(self):
"""
Ensure that missing data causes us to miss records
TODO: This test should be improved if we come up with a more accurate regex for the data sample
"""
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA_BAD_ENG)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED), self.start_state, self.stream_handle,
self.state_callback, self.pub_callback)
# start with the start time record
result = self.parser.get_records(1)
self.assert_result(result, 24, self.particle_a_start_time_telem, False)
# next get engineering records
result = self.parser.get_records(4)
if len(result) == 4:
self.fail("We got 4 records, the bad data should only make 3")
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:18,代码来源:test_wfp_eng__stc_imodem.py
示例10: test_after_header_recovered
def test_after_header_recovered(self):
"""
Test starting the parser in a state in the middle of processing
"""
new_state = {StateKey.POSITION: 24}
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA_SHORT)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_RECOVERED), new_state, self.stream_handle,
self.state_callback, self.pub_callback)
# get engineering records
result = self.parser.get_records(1)
self.assert_result(result, 50, self.particle_a_eng_recov, False)
result = self.parser.get_records(1)
self.assert_result(result, 76, self.particle_b_eng_recov, False)
result = self.parser.get_records(1)
self.assert_result(result, 102, self.particle_c_eng_recov, False)
result = self.parser.get_records(1)
self.assert_result(result, 128, self.particle_d_eng_recov, True)
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:19,代码来源:test_wfp_eng__stc_imodem.py
示例11: test_set_state_telemetered
def test_set_state_telemetered(self):
"""
Test changing to a new state after initializing the parser and
reading data, as if new data has been found and the state has
changed
"""
new_state = {StateKey.POSITION: 76}
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA_SHORT)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED), self.start_state, self.stream_handle,
self.state_callback, self.pub_callback)
# start with the start time record
result = self.parser.get_records(1)
self.assert_result(result, 24, self.particle_a_start_time_telem, False)
# set the new state, the essentially skips engineering a and b
self.parser.set_state(new_state)
result = self.parser.get_records(1)
self.assert_result(result, 102, self.particle_c_eng_telem, False)
result = self.parser.get_records(1)
self.assert_result(result, 128, self.particle_d_eng_telem, True)
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:22,代码来源:test_wfp_eng__stc_imodem.py
示例12: test_long_stream_telemetered
def test_long_stream_telemetered(self):
"""
Test a long stream of data
"""
self.stream_handle = StringIO(WfpEngStcImodemParserUnitTestCase.TEST_DATA)
self.parser = WfpEngStcImodemParser(
self.config.get(DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED), self.start_state, self.stream_handle,
self.state_callback, self.pub_callback)
# start with the start time record
result = self.parser.get_records(1)
self.assert_result(result, 24, self.particle_a_start_time_telem, False)
result = self.parser.get_records(32)
self.assertEqual(result[0], self.particle_a_eng_telem)
self.assertEqual(result[-1], self.particle_last_eng_telem)
self.assertEqual(self.parser._state[StateKey.POSITION], 856)
self.assertEqual(self.state_callback_value[StateKey.POSITION], 856)
self.assertEqual(self.publish_callback_value[-1], self.particle_last_eng_telem)
result = self.parser.get_records(1)
self.assert_result(result, 872, self.particle_a_status_telem, True)
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:24,代码来源:test_wfp_eng__stc_imodem.py
示例13: WfpEngStcImodemParserUnitTestCase
class WfpEngStcImodemParserUnitTestCase(ParserUnitTestCase):
"""
Wfp_eng__stc_imodem Parser unit test suite
"""
TEST_DATA_SHORT = "\x00\x01\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01\x00\x00\x00\x00R\x9d\xab\xa2R\x9d\xac\x19R\x9d\xac" \
"\x1d\x00\x00\x00\x00A:6\xe3\x00\x00\x00\x00\x00\x00\x00\x00\x01\x03\x00h\x00NR\x9d\xac!C\t\xf2\xf7A9A!\x00\x00\x00" \
"\x00\x00\x00\x00\x00\x00\xf2\x00c\x00OR\x9d\xac&C\xbc\x9f\xa7A7'\xbb\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc2\x00^" \
"\x00OR\x9d\xac*C\xc5\xad\x08A6\xd5\xd0\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb4\x00n\x00O"
TEST_DATA = "\x00\x01\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01\x00\x00\x00\x00R\x9d\xab\xa2R\x9d\xac\x19R\x9d\xac\x1d\x00" \
"\x00\x00\x00A:6\xe3\x00\x00\x00\x00\x00\x00\x00\x00\x01\x03\x00h\x00NR\x9d\xac!C\t\xf2\xf7A9A!\x00\x00\x00\x00" \
"\x00\x00\x00\x00\x00\xf2\x00c\x00OR\x9d\xac&C\xbc\x9f\xa7A7'\xbb\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc2\x00^" \
"\x00OR\x9d\xac*C\xc5\xad\x08A6\xd5\xd0\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb4\x00n\x00OR\x9d\xac/C\xb8COA6\xde" \
"\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x9d\x00p\x00QR\x9d\xac3C\x98\xe5TA733\x00\x00\x00\x00\x00\x00\x00\x00" \
"\x00\xa4\x00u\x00OR\x9d\xac8C\x9566A7!-\x00\x00\x00\x00\x00\x00\x00\x00\x00\x9a\x00o\x00OR\x9d\xac?C\xa1\xd7\xc3" \
"A6\xa6LB\x8bG\xae\x00\x00\x00\x00\x00\xb6\x00v\x00PR\x9d\xacECsS\xfeA7e\xfeB\x88\x00\x00\x00\x00\x00\x00\x00" \
"\x98\x00s\x00QR\x9d\xacKC\x89\x17\x8cA6\xe2\xecB\x84\x99\x9a\x00\x00\x00\x00\x00\xa4\x00\x81\x00PR\x9d\xacQC}\n" \
"\xbfA7\x00hB\x81G\xae\x00\x00\x00\x00\x00\xa2\x00|\x00NR\x9d\xacWCyW\xc7A6\x97\x8dB{\xe1H\x00\x00\x00\x00\x00\x9a" \
"\x00m\x00NR\x9d\xac]C\x8c!#A6\x9f\xbeBuQ\xec\x00\x00\x00\x00\x00\x97\x00s\x00QR\x9d\xaccC\x84!9A6h\nBn\x8f\\\x00" \
"\x00\x00\x00\x00\x9f\x00v\x00NR\x9d\xaciCE\xa5UA6a|Bh=q\x00\x00\x00\x00\x00\x97\x00l\x00PR\x9d\xacoC\xa5\xa5\xad" \
"A5\x94\xafBa\\)\x00\x00\x00\x00\x00\x9b\x00n\x00RR\x9d\xacuC\\\r\x08A6\x14{B[\n=\x00\x00\x00\x00\x00\x9a\x00s\x00" \
"OR\x9d\xac{C\xa3\x0b\xb8A5F\nBT33\x00\x00\x00\x00\x00\x98\x00q\x00NR\x9d\xac\x81CO\xc0+A5\xd7\xdcBM\xd7\n\x00\x00" \
"\x00\x00\x00\x97\x00n\x00PR\x9d\xac\x87Cxp\xd0A5#\xa3BGG\xae\x00\x00\x00\x00\x00\x9b\x00n\x00PR\x9d\xac\x8dC\x84" \
"\xdd\xd9A5X\[email protected]\xae\x14\x00\x00\x00\x00\x00\xa5\x00v\x00OR\x9d\xac\x93C\xa0\x85\x01A4j\x7fB:\x14{\x00\x00\x00\x00" \
"\x00\x9c\x00t\x00QR\x9d\xac\x99Cq\xa4\xdbA5:\x92B3\xc2\x8f\x00\x00\x00\x00\x00\x9c\x00x\x00PR\x9d\xac\x9fCg\x07#A5" \
"\x18+B-\x00\x00\x00\x00\x00\x00\x00\x9e\x00m\x00QR\x9d\xac\xa5C\x9bw\x96A4FtB&z\xe1\x00\x00\x00\x00\x00\xd7\x00s" \
"\x00OR\x9d\xac\xabCmP5A4\x9dJB\x1f\xd7\n\x00\x00\x00\x00\x00\x99\x00s\x00PR\x9d\xac\xb1C\xad\x960A3\x8a\tB\x19" \
"(\xf6\x00\x00\x00\x00\x00\x95\x00n\x00OR\x9d\xac\xb7C\x0c\xce]A5\x0f\xfaB\x12\xe1H\x00\x00\x00\x00\x00\x9c\x00u" \
"\x00PR\x9d\xac\xbdC\xa1\xeb\x02A3Z\x85B\x0c=q\x00\x00\x00\x00\x00\x95\x00u\x00OR\x9d\xac\xc3C$\xafOA4\xa23B\x05" \
"\xe1H\x00\x00\x00\x00\x00\x99\x00r\x00PR\x9d\xac\xc9C\xae\xddeA3\x0f(A\xfe(\xf6\x00\x00\x00\x00\x00\x9a\x00o\x00O" \
"R\x9d\xac\xcfA\xfa\xb2:A5\x0b\x0fA\xf2\x8f\\\x00\x00\x00\x00\x00\xaf\x00m\x00P\xff\xff\xff\xff\x00\x00\x00\rR\x9d" \
"\xac\xd4R\x9d\xadQ"
# all flags set to zero
TEST_DATA_BAD_FLAGS = "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00R\x9d\xab\xa2R\x9d\xac\x19R\x9d\xac\x1d" \
"\x00\x00\x00\x00A:6\xe3\x00\x00\x00\x00\x00\x00\x00\x00\x01\x03\x00h\x00NR\x9d\xac!C\t\xf2\xf7A9A!\x00\x00\x00\x00\x00" \
"\x00\x00\x00\x00\xf2\x00c\x00OR\x9d\xac&C\xbc\x9f\xa7A7'\xbb\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc2\x00^\x00OR\x9d\xac" \
"*C\xc5\xad\x08A6\xd5\xd0\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb4\x00n\x00O"
# took 5 bytes out of second engineering sample
TEST_DATA_BAD_ENG = "\x00\x01\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01\x00\x00\x00\x00R\x9d\xab\xa2R\x9d\xac\x19R\x9d\xac\x1d" \
"\x00\x00\x00\x00A:6\xe3\x00\x00\x00\x00\x00\x00\x00\x00\x01\x03\x00h\x00NR\x9d\xac!C\t!\x00\x00\x00\x00\x00" \
"\x00\x00\x00\x00\xf2\x00c\x00OR\x9d\xac&C\xbc\x9f\xa7A7'\xbb\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc2\x00^\x00OR\x9d\xac" \
"*C\xc5\xad\x08A6\xd5\xd0\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb4\x00n\x00O"
def state_callback(self, state, file_ingested):
""" Call back method to watch what comes in via the position callback """
self.file_ingested = file_ingested
self.state_callback_value = state
def pub_callback(self, pub):
""" Call back method to watch what comes in via the publish callback """
self.publish_callback_value = pub
def setUp(self):
ParserUnitTestCase.setUp(self)
self.config = {
DataTypeKey.WFP_ENG_STC_IMODEM_RECOVERED: {
DataSetDriverConfigKeys.PARTICLE_MODULE: 'mi.dataset.parser.wfp_eng__stc_imodem_particles',
DataSetDriverConfigKeys.PARTICLE_CLASS: None,
DataSetDriverConfigKeys.PARTICLE_CLASSES_DICT: {
'status_data_particle_class': WfpEngStcImodemStatusRecoveredDataParticle,
'start_data_particle_class': WfpEngStcImodemStartRecoveredDataParticle,
'engineering_data_particle_class': WfpEngStcImodemEngineeringRecoveredDataParticle
}
},
DataTypeKey.WFP_ENG_STC_IMODEM_TELEMETERED: {
DataSetDriverConfigKeys.PARTICLE_MODULE: 'mi.dataset.parser.wfp_eng__stc_imodem_particles',
DataSetDriverConfigKeys.PARTICLE_CLASS: None,
DataSetDriverConfigKeys.PARTICLE_CLASSES_DICT: {
'status_data_particle_class': WfpEngStcImodemStatusTelemeteredDataParticle,
'start_data_particle_class': WfpEngStcImodemStartTelemeteredDataParticle,
'engineering_data_particle_class': WfpEngStcImodemEngineeringTelemeteredDataParticle
}
},
}
self.start_state = {StateKey.POSITION: 0}
# Define test data particles and their associated timestamps which will be
# compared with returned results
timestamp1_time = self.timestamp_to_ntp('R\x9d\xac\x19')
self.particle_a_start_time_recov = WfpEngStcImodemStartRecoveredDataParticle(
b'\x00\x01\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01\x00\x00\x00\x00R\x9d\xab\xa2R\x9d\xac\x19',
internal_timestamp=timestamp1_time)
self.particle_a_start_time_telem = WfpEngStcImodemStartTelemeteredDataParticle(
b'\x00\x01\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01\x00\x00\x00\x00R\x9d\xab\xa2R\x9d\xac\x19',
internal_timestamp=timestamp1_time)
timestamp1_eng = self.timestamp_to_ntp('R\x9d\xac\x1d')
self.particle_a_eng_recov = WfpEngStcImodemEngineeringRecoveredDataParticle(
b'R\x9d\xac\x1d\x00\x00\x00\x00A:6\xe3\x00\x00\x00\x00\x00\x00\x00\x00\x01\x03\x00h\x00N',
internal_timestamp=timestamp1_eng)
self.particle_a_eng_telem = WfpEngStcImodemEngineeringTelemeteredDataParticle(
b'R\x9d\xac\x1d\x00\x00\x00\x00A:6\xe3\x00\x00\x00\x00\x00\x00\x00\x00\x01\x03\x00h\x00N',
internal_timestamp=timestamp1_eng)
timestamp2_eng = self.timestamp_to_ntp('R\x9d\xac!')
#.........这里部分代码省略.........
开发者ID:JeffRoy,项目名称:marine-integrations,代码行数:101,代码来源:test_wfp_eng__stc_imodem.py
注:本文中的mi.dataset.parser.wfp_eng__stc_imodem.WfpEngStcImodemParser类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论