本文整理汇总了Python中mi.dataset.parser.ctdmo.CtdmoParser类的典型用法代码示例。如果您正苦于以下问题：Python CtdmoParser类的具体用法？Python CtdmoParser怎么用？Python CtdmoParser使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了CtdmoParser类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_update
def test_update(self):
"""
Test a file which has had a section of data replaced by 0s, as if a block of data has not been received yet,
then using the returned state make a new parser with the test data that has the 0s filled in
"""
self.state = {StateKey.UNPROCESSED_DATA:[[0, 7160]],
StateKey.IN_PROCESS_DATA:[], StateKey.TIMESTAMP:0.0}
# this file has a block of CT data replaced by 0s
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_replace.dat'))
self.parser = CtdmoParser(self.config, self.state, self.stream_handle,
self.state_callback, self.pub_callback) # last one is the link to the data source
result = self.parser.get_records(1)
self.assert_result(result, [[394, 467, 3, 1, 1], [6970, 7160, 12, 0, 1]],
[[0, 12], [336, 2010], [5354, 7160]],
self.timestamp3, self.particle_a)
result = self.parser.get_records(1)
self.assert_result(result, [[394, 467, 3, 2, 1], [6970, 7160, 12, 0, 1]],
[[0, 12], [336, 2010], [5354, 7160]],
self.timestamp3, self.particle_b)
result = self.parser.get_records(1)
self.assert_result(result, [[6970, 7160, 12, 0, 1]],
[[0, 12], [336,394], [467, 2010], [5354, 7160]],
self.timestamp3, self.particle_c)
result = self.parser.get_records(1)
self.assert_result(result, [[6970, 7160, 12, 1, 1]],
[[0, 12], [336,394], [467, 2010], [5354, 7160]],
self.timestamp3, self.particle_p)
self.stream_handle.close()
next_state = self.parser._state
# this file has the block of CT data that was missing in the previous file
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, next_state, self.stream_handle,
self.state_callback, self.pub_callback) # last one is the link to the data source
# first get the old 'in process' records from [6970-7160]
# Once those are done, the un processed data will be checked
result = self.parser.get_records(11)
self.assertEqual(result[0], self.particle_q)
self.assert_state([], [[0, 12], [336,394], [467, 2010], [5354, 6970]],
self.timestamp3)
# this should be the first of the newly filled in particles from [5354-5544]
result = self.parser.get_records(1)
self.assert_result(result, [[5354, 5544, 12, 1, 1]],
[[0, 12], [336,394], [467, 2010], [5354, 6970]],
self.timestamp2, self.particle_d)
self.stream_handle.close()
开发者ID:gxtchen，项目名称:marine-integrations，代码行数:51，代码来源:test_ctdmo.py
示例2: test_long_stream
def test_long_stream(self):
self.state = {
StateKey.UNPROCESSED_DATA: [[0, len(CtdmoParserUnitTestCase.TEST_DATA)]],
StateKey.IN_PROCESS_DATA: [],
StateKey.TIMESTAMP: 0.0,
}
self.stream_handle = StringIO(CtdmoParserUnitTestCase.TEST_DATA)
self.parser = CtdmoParser(
self.config, self.state, self.stream_handle, self.state_callback, self.pub_callback
) # last one is the link to the data source
result = self.parser.get_records(29)
self.stream_handle.close()
self.assertEqual(result[0], self.particle_a)
self.assertEqual(result[1], self.particle_b)
self.assertEqual(result[2], self.particle_c)
self.assertEqual(result[3], self.particle_d)
self.assertEqual(result[-16], self.particle_x)
self.assertEqual(result[-15], self.particle_z)
self.assertEqual(result[-14], self.particle_xx)
self.assertEqual(result[-13], self.particle_zz)
self.assertEqual(result[-2], self.particle_aa)
self.assertEqual(result[-1], self.particle_bb)
self.assert_state([[2295, 2485, 12, 2]], [[0, 50], [374, 432], [1197, 1470], [2295, 2485]], self.timestamp4)
self.assertEqual(self.publish_callback_value[-1], self.particle_bb)
开发者ID:sfoley，项目名称:marine-integrations，代码行数:25，代码来源:test_ctdmo.py
示例3: test_in_process_start
def test_in_process_start(self):
"""
test starting a parser with a state in the middle of processing
"""
new_state = {
StateKey.IN_PROCESS_DATA: [[1470, 1661, 12, 1], [2295, 2485, 12, 0]],
StateKey.UNPROCESSED_DATA: [[0, 50], [374, 432], [1197, 2485]],
StateKey.TIMESTAMP: self.timestamp1,
}
self.stream_handle = StringIO(CtdmoParserUnitTestCase.TEST_DATA)
self.parser = CtdmoParser(
self.config, new_state, self.stream_handle, self.state_callback, self.pub_callback
) # last one is the link to the data source
result = self.parser.get_records(1)
# note about timestamp3: this is because when in process state is filled in,
# only enough data is processed to return the requested number of samples
# since only 1 is requested, it only reads [1470-1661], which has timestamp3
self.assert_result(
result,
[[1470, 1661, 12, 2], [2295, 2485, 12, 0]],
[[0, 50], [374, 432], [1197, 2485]],
self.timestamp3,
self.particle_zz,
)
result = self.parser.get_records(11)
self.assertEqual(result[-1], self.particle_aa)
self.assert_state([[2295, 2485, 12, 1]], [[0, 50], [374, 432], [1197, 1470], [1661, 2485]], self.timestamp4)
self.assertEqual(self.publish_callback_value[-1], self.particle_aa)
开发者ID:sfoley，项目名称:marine-integrations，代码行数:30，代码来源:test_ctdmo.py
示例4: test_simple
def test_simple(self):
"""
Read test data from the file and pull out data particles one at a time.
Assert that the results are those we expected.
"""
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, None, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
result = self.parser.get_records(1)
self.assert_result(result, [[853,1043,1,0], [1429,1619,1,0], [5349,5539,1,0], [6313,6503,1,0], [6958,7148,1,0], [7534,7724,1,0]],
[[0, 12], [336, 394], [853,1043], [1429,1619], [5349,5539], [5924,5927], [6313,6503], [6889,7148], [7534,7985]],
self.particle_a)
result = self.parser.get_records(1)
self.assert_result(result, [[1429,1619,1,0], [5349,5539,1,0], [6313,6503,1,0], [6958,7148,1,0], [7534,7724,1,0]],
[[0, 12], [336, 394], [1429,1619], [5349,5539], [5924,5927], [6313,6503], [6889,7148], [7534,7985]],
self.particle_b)
result = self.parser.get_records(1)
self.assert_result(result, [[5349,5539,1,0], [6313,6503,1,0], [6958,7148,1,0], [7534,7724,1,0]],
[[0, 12], [336, 394], [5349,5539], [5924,5927], [6313,6503], [6889,7148], [7534,7985]],
self.particle_c)
result = self.parser.get_records(1)
self.assert_result(result, [[6313,6503,1,0], [6958,7148,1,0], [7534,7724,1,0]],
[[0, 12], [336, 394], [5924,5927], [6313,6503], [6889,7148], [7534,7985]],
self.particle_d)
self.stream_handle.close()
self.assertEqual(self.exception_callback_value, None)
开发者ID:StevenMyerson，项目名称:marine-integrations，代码行数:28，代码来源:test_ctdmo.py
示例5: test_set_state
def test_set_state(self):
"""
test changing the state after initializing
"""
self.state = {StateKey.UNPROCESSED_DATA:[[0, 500]], StateKey.IN_PROCESS_DATA:[]}
new_state = {StateKey.UNPROCESSED_DATA:[[0, 12], [336, 394], [1429,7500]],
StateKey.IN_PROCESS_DATA:[]}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, self.state, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
# there should only be 1 records, make sure we stop there
result = self.parser.get_records(1)
self.assertEqual(result[0], self.particle_a)
result = self.parser.get_records(1)
self.assertEqual(result, [])
self.parser.set_state(new_state)
result = self.parser.get_records(1)
self.stream_handle.close()
self.assert_result(result, [[5349,5539,1,0], [6313,6503,1,0], [6958,7148,1,0]],
[[0, 12], [336, 394], [5349,5539], [5924,5927], [6313,6503], [6889,7500]],
self.particle_c)
self.assertEqual(self.exception_callback_value, None)
开发者ID:StevenMyerson，项目名称:marine-integrations，代码行数:25，代码来源:test_ctdmo.py
示例6: test_set_state
def test_set_state(self):
"""
test changing the state after initializing
"""
self.state = {StateKey.UNPROCESSED_DATA: [[0, 1197]], StateKey.IN_PROCESS_DATA: [], StateKey.TIMESTAMP: 0.0}
new_state = {
StateKey.UNPROCESSED_DATA: [[0, 50], [374, 432], [1197, 2485]],
StateKey.IN_PROCESS_DATA: [],
StateKey.TIMESTAMP: self.timestamp2,
}
self.stream_handle = StringIO(CtdmoParserUnitTestCase.TEST_DATA)
self.parser = CtdmoParser(
self.config, self.state, self.stream_handle, self.state_callback, self.pub_callback
) # last one is the link to the data source
# there should only be 16 records, make sure we stop there
result = self.parser.get_records(16)
result = self.parser.get_records(1)
self.assertEqual(result, [])
self.parser.set_state(new_state)
result = self.parser.get_records(1)
self.stream_handle.close()
self.assert_result(
result,
[[1470, 1661, 12, 1], [2295, 2485, 12, 0]],
[[0, 50], [374, 432], [1197, 1661], [2295, 2485]],
self.timestamp4,
self.particle_xx,
)
开发者ID:sfoley，项目名称:marine-integrations，代码行数:30，代码来源:test_ctdmo.py
示例7: test_set_state
def test_set_state(self):
"""
test changing the state after initializing
"""
self.state = {StateKey.UNPROCESSED_DATA:[[0, 500]], StateKey.IN_PROCESS_DATA:[],
StateKey.TIMESTAMP:0.0}
new_state = {StateKey.UNPROCESSED_DATA:[[0, 12], [336, 394], [467, 2010], [5354, 6000]],
StateKey.IN_PROCESS_DATA:[],
StateKey.TIMESTAMP:self.timestamp1}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, self.state, self.stream_handle,
self.state_callback, self.pub_callback) # last one is the link to the data source
# there should only be 3 records, make sure we stop there
result = self.parser.get_records(3)
result = self.parser.get_records(1)
self.assertEqual(result, [])
self.parser.set_state(new_state)
result = self.parser.get_records(1)
self.stream_handle.close()
self.assert_result(result, [[5354, 5544, 12, 1, 1]],
[[0, 12], [336, 394], [467, 2010], [5354, 6000]],
self.timestamp2, self.particle_d)
开发者ID:gxtchen，项目名称:marine-integrations，代码行数:25，代码来源:test_ctdmo.py
示例8: test_longest_for_co
def test_longest_for_co(self):
"""
Test an even longer file which contains more of the CO samples
"""
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_longest.dat'))
self.parser = CtdmoParser(self.config, None, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
result = self.parser.get_records(36)
self.assertEqual(result[0], self.particle_a)
self.assertEqual(result[1], self.particle_b)
self.assertEqual(result[2], self.particle_c)
self.assertEqual(result[3], self.particle_d)
self.assertEqual(result[9], self.particle_a_offset)
self.assertEqual(result[12], self.particle_z)
self.assertEqual(result[22], self.particle_b_offset)
self.assertEqual(result[-1], self.particle_c_offset)
self.assert_state([],
[[0, 12], [336, 394], [5924,5927], [6889, 6958], [8687,8756],
[8946,9522], [14576,14647], [16375,16444], [18173,18240],
[20130,20199], [21927,21996], [29707,29776], [30648,30746]])
self.stream_handle.close()
self.assertEqual(self.exception_callback_value, None)
开发者ID:StevenMyerson，项目名称:marine-integrations，代码行数:26，代码来源:test_ctdmo.py
示例9: test_get_many
def test_get_many(self):
"""
Read test data from the file and pull out multiple data particles at one time.
Assert that the results are those we expected.
"""
self.state = {
StateKey.UNPROCESSED_DATA: [[0, len(CtdmoParserUnitTestCase.TEST_DATA)]],
StateKey.IN_PROCESS_DATA: [],
StateKey.TIMESTAMP: 0.0,
}
self.stream_handle = StringIO(CtdmoParserUnitTestCase.TEST_DATA)
self.parser = CtdmoParser(
self.config, self.state, self.stream_handle, self.state_callback, self.pub_callback
) # last one is the link to the data source
result = self.parser.get_records(3)
self.stream_handle.close()
self.assertEqual(result, [self.particle_a, self.particle_b, self.particle_c])
self.assert_state(
[[892, 1083, 12, 0], [1470, 1661, 12, 0], [2295, 2485, 12, 0]],
[[0, 50], [374, 432], [892, 1083], [1197, 1661], [2295, 2485]],
self.timestamp4,
)
self.assertEqual(self.publish_callback_value[0], self.particle_a)
self.assertEqual(self.publish_callback_value[1], self.particle_b)
self.assertEqual(self.publish_callback_value[2], self.particle_c)
开发者ID:sfoley，项目名称:marine-integrations，代码行数:26，代码来源:test_ctdmo.py
示例10: test_simple
def test_simple(self):
"""
Read test data from the file and pull out data particles one at a time.
Assert that the results are those we expected.
"""
self.state = {StateKey.UNPROCESSED_DATA:[[0, 6000]],
StateKey.IN_PROCESS_DATA:[], StateKey.TIMESTAMP:0.0}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, self.state, self.stream_handle,
self.state_callback, self.pub_callback)
result = self.parser.get_records(1)
self.stream_handle.close()
self.assert_result(result, [[394, 467, 3, 1, 1], [5354, 5544, 12, 0, 1]],
[[0, 12], [336, 2010], [5354, 6000]],
self.timestamp2, self.particle_a)
result = self.parser.get_records(1)
self.assert_result(result, [[394, 467, 3, 2, 1], [5354, 5544, 12, 0, 1]],
[[0, 12], [336, 2010], [5354, 6000]],
self.timestamp2, self.particle_b)
result = self.parser.get_records(1)
self.assert_result(result, [[5354, 5544, 12, 0, 1]],
[[0, 12], [336, 394], [467, 2010], [5354, 6000]],
self.timestamp2, self.particle_c)
result = self.parser.get_records(1)
self.assert_result(result, [[5354, 5544, 12, 1, 1]],
[[0, 12], [336, 394], [467, 2010], [5354, 6000]],
self.timestamp2, self.particle_d)
result = self.parser.get_records(1)
self.assert_result(result, [[5354, 5544, 12, 2, 1]],
[[0, 12], [336, 394], [467, 2010], [5354, 6000]],
self.timestamp2, self.particle_e)
开发者ID:gxtchen，项目名称:marine-integrations，代码行数:33，代码来源:test_ctdmo.py
示例11: test_update
def test_update(self):
"""
Test a file which has had a section of data replaced by 0s, as if a block of data has not been received yet,
then using the returned state make a new parser with the test data that has the 0s filled in
"""
# this file has a block of CT data replaced by 0s
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_replace.dat'))
self.parser = CtdmoParser(self.config, None, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
result = self.parser.get_records(4)
# particle d has been replaced in this file with zeros
self.assertEqual(result, [self.particle_a, self.particle_b, self.particle_c, self.particle_e])
self.assert_state([[6958,7148,1,0], [7534,7724,1,0]],
[[0, 12], [336, 394], [5349,5539], [5924,5927], [6889,7148], [7534,7985]])
self.assertEqual(self.publish_callback_value[0], self.particle_a)
self.assertEqual(self.publish_callback_value[1], self.particle_b)
self.assertEqual(self.publish_callback_value[2], self.particle_c)
self.assertEqual(self.publish_callback_value[3], self.particle_e)
self.stream_handle.close()
next_state = self.parser._state
# this file has the block of CT data that was missing in the previous file
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, next_state, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
# first get the old 'in process' records from [6970-7160]
# Once those are done, the un processed data will be checked
result = self.parser.get_records(2)
self.assertEqual(result, [self.particle_f, self.particle_g])
self.assert_state([], [[0, 12], [336, 394], [5349,5539], [5924,5927], [6889,6958], [7724,7985]])
self.assertEqual(self.publish_callback_value[0], self.particle_f)
self.assertEqual(self.publish_callback_value[1], self.particle_g)
# this should be the first of the newly filled in particles from [5354-5544]
result = self.parser.get_records(1)
self.assert_result(result, [], [[0, 12], [336, 394], [5924,5927], [6889,6958], [7724,7985]],
self.particle_d)
self.stream_handle.close()
self.assertEqual(self.exception_callback_value, None)
开发者ID:StevenMyerson，项目名称:marine-integrations，代码行数:44，代码来源:test_ctdmo.py
示例12: test_simple_section
def test_simple_section(self):
"""
Read test data from the file and pull out data particles one at a time.
Assert that the results are those we expected.
"""
self.state = {
StateKey.UNPROCESSED_DATA: [[0, len(CtdmoParserUnitTestCase.REPLACE_TEST_DATA)]],
StateKey.IN_PROCESS_DATA: [],
StateKey.TIMESTAMP: 0.0,
}
self.stream_handle = StringIO(CtdmoParserUnitTestCase.REPLACE_TEST_DATA)
self.parser = CtdmoParser(self.config, self.state, self.stream_handle, self.state_callback, self.pub_callback)
result = self.parser.get_records(1)
self.assert_result(
result,
[[432, 505, 3, 1], [1470, 1661, 12, 0], [2295, 2485, 12, 0]],
[[0, 50], [374, 505], [892, 1083], [1197, 1661], [2295, 2485]],
self.timestamp4,
self.particle_a,
)
result = self.parser.get_records(1)
self.assert_result(
result,
[[432, 505, 3, 2], [1470, 1661, 12, 0], [2295, 2485, 12, 0]],
[[0, 50], [374, 505], [892, 1083], [1197, 1661], [2295, 2485]],
self.timestamp4,
self.particle_b,
)
result = self.parser.get_records(1)
self.assert_result(
result,
[[1470, 1661, 12, 0], [2295, 2485, 12, 0]],
[[0, 50], [374, 432], [892, 1083], [1197, 1661], [2295, 2485]],
self.timestamp4,
self.particle_c,
)
result = self.parser.get_records(1)
self.assert_result(
result,
[[1470, 1661, 12, 1], [2295, 2485, 12, 0]],
[[0, 50], [374, 432], [892, 1083], [1197, 1661], [2295, 2485]],
self.timestamp4,
self.particle_xx,
)
result = self.parser.get_records(1)
self.assert_result(
result,
[[1470, 1661, 12, 2], [2295, 2485, 12, 0]],
[[0, 50], [374, 432], [892, 1083], [1197, 1661], [2295, 2485]],
self.timestamp4,
self.particle_zz,
)
self.stream_handle.close()
开发者ID:sfoley，项目名称:marine-integrations，代码行数:54，代码来源:test_ctdmo.py
示例13: test_missing_inductive_id_config
def test_missing_inductive_id_config(self):
"""
Make sure that the driver complains about a missing inductive ID in the config
"""
self.state = {StateKey.UNPROCESSED_DATA:[[0, 8000]],
StateKey.IN_PROCESS_DATA:[], StateKey.TIMESTAMP:0.0}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
bad_config = {
DataSetDriverConfigKeys.PARTICLE_MODULE: 'mi.dataset.parser.ctdmo',
DataSetDriverConfigKeys.PARTICLE_CLASS: 'CtdmoParserDataParticle',
}
with self.assertRaises(DatasetParserException):
self.parser = CtdmoParser(bad_config, self.state, self.stream_handle,
self.state_callback, self.pub_callback)
开发者ID:cwingard，项目名称:marine-integrations，代码行数:15，代码来源:test_ctdmo.py
示例14: test_mid_state_start
def test_mid_state_start(self):
"""
test starting a parser with a state in the middle of processing
"""
new_state = {StateKey.IN_PROCESS_DATA:[],
StateKey.UNPROCESSED_DATA:[[0, 12], [336, 394], [467, 2010], [5354, 6000]],
StateKey.TIMESTAMP:self.timestamp1}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, new_state, self.stream_handle,
self.state_callback, self.pub_callback) # last one is the link to the data source
result = self.parser.get_records(1)
self.stream_handle.close()
self.assert_result(result, [[5354, 5544, 12, 1, 1]],
[[0, 12], [336, 394], [467, 2010], [5354, 6000]],
self.timestamp2, self.particle_d)
开发者ID:gxtchen，项目名称:marine-integrations，代码行数:16，代码来源:test_ctdmo.py
示例15: test_mid_state_start
def test_mid_state_start(self):
"""
test starting a parser with a state in the middle of processing
"""
new_state = {StateKey.IN_PROCESS_DATA:[],
StateKey.UNPROCESSED_DATA:[[0, 12], [336, 394], [1429,7500]]}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, new_state, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
result = self.parser.get_records(1)
self.stream_handle.close()
self.assert_result(result, [[5349,5539,1,0], [6313,6503,1,0], [6958,7148,1,0]],
[[0, 12], [336, 394], [5349,5539], [5924,5927], [6313,6503], [6889,7500]],
self.particle_c)
self.assertEqual(self.exception_callback_value, None)
开发者ID:StevenMyerson，项目名称:marine-integrations，代码行数:16，代码来源:test_ctdmo.py
示例16: test_in_process_start
def test_in_process_start(self):
"""
test starting a parser with a state in the middle of processing
"""
new_state = {StateKey.IN_PROCESS_DATA:[[6970,7160,1,0,1], [7547,7737,1,0,1]],
StateKey.UNPROCESSED_DATA:[[0, 12], [336, 394], [467, 2010], [5544, 8000]],
StateKey.TIMESTAMP:self.timestamp2}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, new_state, self.stream_handle,
self.state_callback, self.pub_callback) # last one is the link to the data source
result = self.parser.get_records(2)
self.assertEqual(result[0], self.particle_c)
self.assertEqual(result[-1], self.particle_d)
self.assert_state([],
[[0, 12], [336, 394], [467, 2010], [5544, 6970], [7160, 7547],[7737, 8000]],
self.timestamp4)
self.assertEqual(self.publish_callback_value[-1], self.particle_d)
开发者ID:cwingard，项目名称:marine-integrations，代码行数:19，代码来源:test_ctdmo.py
示例17: test_in_process_start
def test_in_process_start(self):
"""
test starting a parser with a state in the middle of processing
"""
new_state = {StateKey.IN_PROCESS_DATA:[[5349,5539,1,0], [6313,6503,1,0], [6958,7148,1,0], [7534,7724,1,0]],
StateKey.UNPROCESSED_DATA:[[0, 12], [336, 394], [5349,5539], [5924,5927], [6313,6503], [6889,7148], [7534,7985]],
StateKey.FILE_SIZE: 8000}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, new_state, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
result = self.parser.get_records(2)
self.assertEqual(result[0], self.particle_d)
self.assertEqual(result[-1], self.particle_e)
self.assert_state([[6958,7148,1,0], [7534,7724,1,0]],
[[0, 12], [336, 394], [5924,5927], [6889,7148], [7534,7985]])
self.assertEqual(self.publish_callback_value[-1], self.particle_e)
self.assertEqual(self.exception_callback_value, None)
开发者ID:danmergens，项目名称:marine-integrations，代码行数:19，代码来源:test_ctdmo.py
示例18: test_long_stream
def test_long_stream(self):
self.state = {StateKey.UNPROCESSED_DATA:[[0, 14000]],
StateKey.IN_PROCESS_DATA:[], StateKey.TIMESTAMP:0.0}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_longer.dat'))
self.parser = CtdmoParser(self.config, self.state, self.stream_handle,
self.state_callback, self.pub_callback) # last one is the link to the data source
result = self.parser.get_records(6)
self.assertEqual(result[0], self.particle_a)
self.assertEqual(result[1], self.particle_b)
self.assertEqual(result[2], self.particle_c)
self.assertEqual(result[3], self.particle_d)
self.assertEqual(result[-1], self.particle_z)
self.assert_state([],
[[0, 12], [336, 394], [467, 2010], [5544, 6970], [7160, 7547],
[7737, 8773], [8963, 10037], [10283, 10672], [12873, 14000]],
self.timestamp_last)
self.assertEqual(self.publish_callback_value[-1], self.particle_z)
self.stream_handle.close()
开发者ID:cwingard，项目名称:marine-integrations，代码行数:20，代码来源:test_ctdmo.py
示例19: test_get_many
def test_get_many(self):
"""
Read test data from the file and pull out multiple data particles at one time.
Assert that the results are those we expected.
"""
self.state = {StateKey.UNPROCESSED_DATA:[[0, 6000]],
StateKey.IN_PROCESS_DATA:[], StateKey.TIMESTAMP:0.0}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_shorter.dat'))
self.parser = CtdmoParser(self.config, self.state, self.stream_handle,
self.state_callback, self.pub_callback) # last one is the link to the data source
result = self.parser.get_records(3)
self.stream_handle.close()
self.assertEqual(result, [self.particle_a, self.particle_b, self.particle_c])
self.assert_state([[5354, 5544, 12, 0, 1]],
[[0, 12], [336, 394], [467, 2010], [5354, 6000]], self.timestamp2)
self.assertEqual(self.publish_callback_value[0], self.particle_a)
self.assertEqual(self.publish_callback_value[1], self.particle_b)
self.assertEqual(self.publish_callback_value[2], self.particle_c)
开发者ID:gxtchen，项目名称:marine-integrations，代码行数:20，代码来源:test_ctdmo.py
示例20: test_long_stream
def test_long_stream(self):
self.state = {StateKey.UNPROCESSED_DATA:[[0, 14000]],
StateKey.IN_PROCESS_DATA:[]}
self.stream_handle = open(os.path.join(RESOURCE_PATH,
'node59p1_longer.dat'))
self.parser = CtdmoParser(self.config, self.state, self.stream_handle,
self.state_callback, self.pub_callback, self.exception_callback)
result = self.parser.get_records(13)
self.assertEqual(result[0], self.particle_a)
self.assertEqual(result[1], self.particle_b)
self.assertEqual(result[2], self.particle_c)
self.assertEqual(result[3], self.particle_d)
self.assertEqual(result[-1], self.particle_z)
self.assert_state([],
[[0, 12], [336, 394], [5924,5927], [6889, 6958], [8687,8756],
[8946,9522], [13615, 14000]])
self.assertEqual(self.publish_callback_value[-1], self.particle_z)
self.stream_handle.close()
self.assertEqual(self.exception_callback_value, None)
开发者ID:manalang，项目名称:marine-integrations，代码行数:20，代码来源:test_ctdmo.py
注：本文中的mi.dataset.parser.ctdmo.CtdmoParser类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论