Loading pysen/io/reader.py +10 −8 Original line number Diff line number Diff line Loading @@ -449,7 +449,8 @@ def read_echo(filename): Read echo file """ _log = logging.getLogger() lines = iter(open(filename).readlines()) with open(filename) as _fd: lines = iter(_fd.readlines()) result = AttrDict() result['scan'] = read_scan_info(lines) version = { 'type': result.scan['type'] } Loading Loading @@ -575,8 +576,8 @@ def read_datreat(filename): result = [] # result data = [] # data meta = OrderedDict() # meta data with open(filename, 'r') as fdesc: lines = _remove_empty_lines(fdesc.readlines()) with open(filename, 'r') as _fdesc: lines = _remove_empty_lines(_fdesc.readlines()) try: while True: # -------- meta section ------------ Loading Loading @@ -618,8 +619,8 @@ def list_detimage(inpfile): sample = None wavelength = None idx, k = 0,0 with open(inpfile, 'r') as fd: lines = fd.readlines() with open(inpfile, 'r') as _fdesc: lines = _fdesc.readlines() for k, line in enumerate(lines): line = line.strip() if not line: Loading Loading @@ -710,7 +711,8 @@ def read_magnetic(filename): "read magnetic up/down file" result = OrderedDict() lines = iter(open(filename).readlines()) with open(filename) as _fdesc: lines = iter(_fdesc.readlines()) _ = read_params_info(lines) tech = read_tech_info(lines) Loading @@ -726,8 +728,8 @@ def read_transmission(filename): result = OrderedDict() #transtab = { '"' : None, # '\n': None, } with open(filename) as fd: lines = fd.readlines() with open(filename) as _fdesc: lines = _fdesc.readlines() header = remove_chars(lines[1], delchars='"\n') result['header'] = header header = header.split(',') Loading test/test_echo_reduce.py +3 −7 Original line number Diff line number Diff line Loading @@ -23,17 +23,13 @@ class EchoReduceTestCase(unittest.TestCase): "Tear-down method" pass def xtest_reduce(): def test_reduce(): "reduction test" import h5py hdfile = h5py.File(os.path.join(TestDataDir,'s4863.h5'), 'a', driver='core', backing_store=False) process_data(hdfile, qbins=np.linspace(0.125, 0.250, 6), tbins=(2, 40)) display_echo(hdfile, iecho=0, show=True) #process_data(hdfile, qbins=np.linspace(0.125, 0.250, 6), tbins=(2, 40)) #display_echo(hdfile, iecho=0, show=True) if __name__ == "__main__": unittest.main(exit=False, verbosity=2) Loading test/test_io_reader.py +48 −47 Original line number Diff line number Diff line Loading @@ -90,36 +90,36 @@ class IOReaderTestCase(unittest.TestCase): for line in read_lines(BUFFER_MODE1): res.update(reader.parse_keywords(line, mode=1)) #print_dict(res) self.assertEquals(12, len(res)) self.assertEquals(res['scan_dimension'], 1) self.assertEquals(res['scan0_parameter'], 'tau') self.assertEquals(res['a'], 6.28) self.assertEquals(res['b'], 6.28e+10) self.assertEquals(res['c'], 6.28e-10) self.assertEquals(res['alpha'], 90.0) self.assertEquals(res['beta'], 60.0) self.assertEquals(res['gamma'], 120.0) self.assertEquals(res['c_0'], 1.0) self.assertEquals(res['extra'], []) self.assertEquals(res['l1'], [1.0, 2.0, 3e12, 4.0]) self.assertEquals(res['l2'], [2.0, 3.0]) self.assertEqual(12, len(res)) self.assertEqual(res['scan_dimension'], 1) self.assertEqual(res['scan0_parameter'], 'tau') self.assertEqual(res['a'], 6.28) self.assertEqual(res['b'], 6.28e+10) self.assertEqual(res['c'], 6.28e-10) self.assertEqual(res['alpha'], 90.0) self.assertEqual(res['beta'], 60.0) self.assertEqual(res['gamma'], 120.0) self.assertEqual(res['c_0'], 1.0) self.assertEqual(res['extra'], []) self.assertEqual(res['l1'], [1.0, 2.0, 3e12, 4.0]) self.assertEqual(res['l2'], [2.0, 3.0]) def test_parse_mode2(self): res = reader.AttrDict() for line in read_lines(BUFFER_MODE2): res.update(reader.parse_keywords(line, mode=2)) #print_dict(res) self.assertEquals(10, len(res)) self.assertEquals(res['chopper1.frequency'], [60.0, 60.0]) self.assertEquals(res['chopper1.phase'], [166.0, 166.0, 'off']) self.assertEquals(res['Temperature.sample1.lakeshore1.1'], self.assertEqual(10, len(res)) self.assertEqual(res['chopper1.frequency'], [60.0, 60.0]) self.assertEqual(res['chopper1.phase'], [166.0, 166.0, 'off']) self.assertEqual(res['Temperature.sample1.lakeshore1.1'], [121.0, 120.0]) self.assertEquals(res['Temperature.sample1.lakeshore1.2'], self.assertEqual(res['Temperature.sample1.lakeshore1.2'], [123.0, 122.0]) self.assertEquals(res['i5'], [3.18, 3.19, 0.39, 1.35, self.assertEqual(res['i5'], [3.18, 3.19, 0.39, 1.35, 'set', 'on', 'flow', 'solphase1inner']) self.assertEquals(res['i23'], [2.00, 1.99, 10.55, 9.89, self.assertEqual(res['i23'], [2.00, 1.99, 10.55, 9.89, 'fixed', 'on', 'flow', 'solanalyzer1', 'solanalyzer2']) Loading @@ -128,27 +128,27 @@ class IOReaderTestCase(unittest.TestCase): for line in read_lines(BUFFER_MODE3): res.update(reader.parse_keywords(line, mode=3)) #print_dict(res) self.assertEquals(11, len(res)) self.assertEquals(res['detector.xbins'], 32) self.assertEquals(res['detector.xbins.width'], 0.009375) self.assertEquals(res['detector.xbins.centerbinx'], 16.0) self.assertEquals(res['detector.ybins'], 32) self.assertEquals(res['detector.ybins.width'], 0.009375) self.assertEquals(res['detector.ybins.centerbiny'], 16.0) self.assertEquals(res['detector.tbins'], 42) self.assertEquals(res['detector.tbins.width'], 0.000396825) self.assertEquals(res['detector.storage'], 'yxt') self.assertEquals(res['detector.moderatordistance'], 21.3) self.assertEquals(res['detector.sampledistance'], 3.877) #self.assertEquals(res['chopper1.phase'], [166.0, 166.0, 'off']) #self.assertEquals(res['Temperature.sample1.lakeshore1.1'], self.assertEqual(11, len(res)) self.assertEqual(res['detector.xbins'], 32) self.assertEqual(res['detector.xbins.width'], 0.009375) self.assertEqual(res['detector.xbins.centerbinx'], 16.0) self.assertEqual(res['detector.ybins'], 32) self.assertEqual(res['detector.ybins.width'], 0.009375) self.assertEqual(res['detector.ybins.centerbiny'], 16.0) self.assertEqual(res['detector.tbins'], 42) self.assertEqual(res['detector.tbins.width'], 0.000396825) self.assertEqual(res['detector.storage'], 'yxt') self.assertEqual(res['detector.moderatordistance'], 21.3) self.assertEqual(res['detector.sampledistance'], 3.877) #self.assertEqual(res['chopper1.phase'], [166.0, 166.0, 'off']) #self.assertEqual(res['Temperature.sample1.lakeshore1.1'], # [121.0, 120.0]) #self.assertEquals(res['Temperature.sample1.lakeshore1.2'], #self.assertEqual(res['Temperature.sample1.lakeshore1.2'], # [123.0, 122.0]) #self.assertEquals(res['i5'], [3.18, 3.19, 0.39, 1.35, #self.assertEqual(res['i5'], [3.18, 3.19, 0.39, 1.35, # 'set', 'on', 'flow', 'solphase1inner']) #self.assertEquals(res['i23'], [2.00, 1.99, 10.55, 9.89, #self.assertEqual(res['i23'], [2.00, 1.99, 10.55, 9.89, # 'fixed', 'on', 'flow', # 'solanalyzer1', 'solanalyzer2']) Loading @@ -157,14 +157,14 @@ class IOReaderTestCase(unittest.TestCase): for line in read_lines(BUFFER_MODE4): res.update(reader.parse_keywords(line, mode=4)) #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) def test_parse_table_l(self): lines = read_lines(TABLE_L) res = reader.parse_tables(lines, delchars='()/') #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) nptest.assert_array_almost_equal( res['lambdaTable']*1e10, np.asarray([ 3.61359, 3.67109, 3.72860 ]), Loading @@ -174,7 +174,7 @@ class IOReaderTestCase(unittest.TestCase): lines = read_lines(TABLE_Q) res = reader.parse_tables(lines, delchars='()/', delwords=('m',)) #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) nptest.assert_array_equal( res['qTable'], np.asarray([[2657970000.0,203156000.0, -2650190000.0, 0], Loading @@ -185,7 +185,7 @@ class IOReaderTestCase(unittest.TestCase): lines = read_lines(TABLE_D) res = reader.parse_tables(lines, delchars='()/', delwords=('m',)) #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) nptest.assert_array_equal( res['dqxTable'], np.asarray([[ -6408490., 41554200., 0.], Loading @@ -197,11 +197,11 @@ class IOReaderTestCase(unittest.TestCase): 'dqxTable', 'dqyTable', 'hklTable', 'dhklyTable', 'dhklzTable', 'fouriertimeTable'] lines = open(os.path.join(TestDataDir,'param_section.dat')).readlines() lines = iter(lines) with open(os.path.join(TestDataDir,'param_section.dat')) as _fdesc: lines = iter(_fdesc.readlines()) res = reader.read_params_info(lines, delchars='()/', delwords=('m', 'sec')) ntable = 0 for k in res: if 'Table' in k: Loading Loading @@ -356,7 +356,8 @@ class IOReaderTestCase(unittest.TestCase): def test_detimage_parse(self): filename = os.path.join(TestDataDir,'detector.single') lines = iter(open(filename).readlines()) with open(filename) as _fd: lines = iter(_fd.readlines()) res = reader.parse_detimage(lines) #print res.keys() self.assertEqual(len(res), 48) Loading Loading
pysen/io/reader.py +10 −8 Original line number Diff line number Diff line Loading @@ -449,7 +449,8 @@ def read_echo(filename): Read echo file """ _log = logging.getLogger() lines = iter(open(filename).readlines()) with open(filename) as _fd: lines = iter(_fd.readlines()) result = AttrDict() result['scan'] = read_scan_info(lines) version = { 'type': result.scan['type'] } Loading Loading @@ -575,8 +576,8 @@ def read_datreat(filename): result = [] # result data = [] # data meta = OrderedDict() # meta data with open(filename, 'r') as fdesc: lines = _remove_empty_lines(fdesc.readlines()) with open(filename, 'r') as _fdesc: lines = _remove_empty_lines(_fdesc.readlines()) try: while True: # -------- meta section ------------ Loading Loading @@ -618,8 +619,8 @@ def list_detimage(inpfile): sample = None wavelength = None idx, k = 0,0 with open(inpfile, 'r') as fd: lines = fd.readlines() with open(inpfile, 'r') as _fdesc: lines = _fdesc.readlines() for k, line in enumerate(lines): line = line.strip() if not line: Loading Loading @@ -710,7 +711,8 @@ def read_magnetic(filename): "read magnetic up/down file" result = OrderedDict() lines = iter(open(filename).readlines()) with open(filename) as _fdesc: lines = iter(_fdesc.readlines()) _ = read_params_info(lines) tech = read_tech_info(lines) Loading @@ -726,8 +728,8 @@ def read_transmission(filename): result = OrderedDict() #transtab = { '"' : None, # '\n': None, } with open(filename) as fd: lines = fd.readlines() with open(filename) as _fdesc: lines = _fdesc.readlines() header = remove_chars(lines[1], delchars='"\n') result['header'] = header header = header.split(',') Loading
test/test_echo_reduce.py +3 −7 Original line number Diff line number Diff line Loading @@ -23,17 +23,13 @@ class EchoReduceTestCase(unittest.TestCase): "Tear-down method" pass def xtest_reduce(): def test_reduce(): "reduction test" import h5py hdfile = h5py.File(os.path.join(TestDataDir,'s4863.h5'), 'a', driver='core', backing_store=False) process_data(hdfile, qbins=np.linspace(0.125, 0.250, 6), tbins=(2, 40)) display_echo(hdfile, iecho=0, show=True) #process_data(hdfile, qbins=np.linspace(0.125, 0.250, 6), tbins=(2, 40)) #display_echo(hdfile, iecho=0, show=True) if __name__ == "__main__": unittest.main(exit=False, verbosity=2) Loading
test/test_io_reader.py +48 −47 Original line number Diff line number Diff line Loading @@ -90,36 +90,36 @@ class IOReaderTestCase(unittest.TestCase): for line in read_lines(BUFFER_MODE1): res.update(reader.parse_keywords(line, mode=1)) #print_dict(res) self.assertEquals(12, len(res)) self.assertEquals(res['scan_dimension'], 1) self.assertEquals(res['scan0_parameter'], 'tau') self.assertEquals(res['a'], 6.28) self.assertEquals(res['b'], 6.28e+10) self.assertEquals(res['c'], 6.28e-10) self.assertEquals(res['alpha'], 90.0) self.assertEquals(res['beta'], 60.0) self.assertEquals(res['gamma'], 120.0) self.assertEquals(res['c_0'], 1.0) self.assertEquals(res['extra'], []) self.assertEquals(res['l1'], [1.0, 2.0, 3e12, 4.0]) self.assertEquals(res['l2'], [2.0, 3.0]) self.assertEqual(12, len(res)) self.assertEqual(res['scan_dimension'], 1) self.assertEqual(res['scan0_parameter'], 'tau') self.assertEqual(res['a'], 6.28) self.assertEqual(res['b'], 6.28e+10) self.assertEqual(res['c'], 6.28e-10) self.assertEqual(res['alpha'], 90.0) self.assertEqual(res['beta'], 60.0) self.assertEqual(res['gamma'], 120.0) self.assertEqual(res['c_0'], 1.0) self.assertEqual(res['extra'], []) self.assertEqual(res['l1'], [1.0, 2.0, 3e12, 4.0]) self.assertEqual(res['l2'], [2.0, 3.0]) def test_parse_mode2(self): res = reader.AttrDict() for line in read_lines(BUFFER_MODE2): res.update(reader.parse_keywords(line, mode=2)) #print_dict(res) self.assertEquals(10, len(res)) self.assertEquals(res['chopper1.frequency'], [60.0, 60.0]) self.assertEquals(res['chopper1.phase'], [166.0, 166.0, 'off']) self.assertEquals(res['Temperature.sample1.lakeshore1.1'], self.assertEqual(10, len(res)) self.assertEqual(res['chopper1.frequency'], [60.0, 60.0]) self.assertEqual(res['chopper1.phase'], [166.0, 166.0, 'off']) self.assertEqual(res['Temperature.sample1.lakeshore1.1'], [121.0, 120.0]) self.assertEquals(res['Temperature.sample1.lakeshore1.2'], self.assertEqual(res['Temperature.sample1.lakeshore1.2'], [123.0, 122.0]) self.assertEquals(res['i5'], [3.18, 3.19, 0.39, 1.35, self.assertEqual(res['i5'], [3.18, 3.19, 0.39, 1.35, 'set', 'on', 'flow', 'solphase1inner']) self.assertEquals(res['i23'], [2.00, 1.99, 10.55, 9.89, self.assertEqual(res['i23'], [2.00, 1.99, 10.55, 9.89, 'fixed', 'on', 'flow', 'solanalyzer1', 'solanalyzer2']) Loading @@ -128,27 +128,27 @@ class IOReaderTestCase(unittest.TestCase): for line in read_lines(BUFFER_MODE3): res.update(reader.parse_keywords(line, mode=3)) #print_dict(res) self.assertEquals(11, len(res)) self.assertEquals(res['detector.xbins'], 32) self.assertEquals(res['detector.xbins.width'], 0.009375) self.assertEquals(res['detector.xbins.centerbinx'], 16.0) self.assertEquals(res['detector.ybins'], 32) self.assertEquals(res['detector.ybins.width'], 0.009375) self.assertEquals(res['detector.ybins.centerbiny'], 16.0) self.assertEquals(res['detector.tbins'], 42) self.assertEquals(res['detector.tbins.width'], 0.000396825) self.assertEquals(res['detector.storage'], 'yxt') self.assertEquals(res['detector.moderatordistance'], 21.3) self.assertEquals(res['detector.sampledistance'], 3.877) #self.assertEquals(res['chopper1.phase'], [166.0, 166.0, 'off']) #self.assertEquals(res['Temperature.sample1.lakeshore1.1'], self.assertEqual(11, len(res)) self.assertEqual(res['detector.xbins'], 32) self.assertEqual(res['detector.xbins.width'], 0.009375) self.assertEqual(res['detector.xbins.centerbinx'], 16.0) self.assertEqual(res['detector.ybins'], 32) self.assertEqual(res['detector.ybins.width'], 0.009375) self.assertEqual(res['detector.ybins.centerbiny'], 16.0) self.assertEqual(res['detector.tbins'], 42) self.assertEqual(res['detector.tbins.width'], 0.000396825) self.assertEqual(res['detector.storage'], 'yxt') self.assertEqual(res['detector.moderatordistance'], 21.3) self.assertEqual(res['detector.sampledistance'], 3.877) #self.assertEqual(res['chopper1.phase'], [166.0, 166.0, 'off']) #self.assertEqual(res['Temperature.sample1.lakeshore1.1'], # [121.0, 120.0]) #self.assertEquals(res['Temperature.sample1.lakeshore1.2'], #self.assertEqual(res['Temperature.sample1.lakeshore1.2'], # [123.0, 122.0]) #self.assertEquals(res['i5'], [3.18, 3.19, 0.39, 1.35, #self.assertEqual(res['i5'], [3.18, 3.19, 0.39, 1.35, # 'set', 'on', 'flow', 'solphase1inner']) #self.assertEquals(res['i23'], [2.00, 1.99, 10.55, 9.89, #self.assertEqual(res['i23'], [2.00, 1.99, 10.55, 9.89, # 'fixed', 'on', 'flow', # 'solanalyzer1', 'solanalyzer2']) Loading @@ -157,14 +157,14 @@ class IOReaderTestCase(unittest.TestCase): for line in read_lines(BUFFER_MODE4): res.update(reader.parse_keywords(line, mode=4)) #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) def test_parse_table_l(self): lines = read_lines(TABLE_L) res = reader.parse_tables(lines, delchars='()/') #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) nptest.assert_array_almost_equal( res['lambdaTable']*1e10, np.asarray([ 3.61359, 3.67109, 3.72860 ]), Loading @@ -174,7 +174,7 @@ class IOReaderTestCase(unittest.TestCase): lines = read_lines(TABLE_Q) res = reader.parse_tables(lines, delchars='()/', delwords=('m',)) #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) nptest.assert_array_equal( res['qTable'], np.asarray([[2657970000.0,203156000.0, -2650190000.0, 0], Loading @@ -185,7 +185,7 @@ class IOReaderTestCase(unittest.TestCase): lines = read_lines(TABLE_D) res = reader.parse_tables(lines, delchars='()/', delwords=('m',)) #print_dict(res) self.assertEquals(1, len(res)) self.assertEqual(1, len(res)) nptest.assert_array_equal( res['dqxTable'], np.asarray([[ -6408490., 41554200., 0.], Loading @@ -197,11 +197,11 @@ class IOReaderTestCase(unittest.TestCase): 'dqxTable', 'dqyTable', 'hklTable', 'dhklyTable', 'dhklzTable', 'fouriertimeTable'] lines = open(os.path.join(TestDataDir,'param_section.dat')).readlines() lines = iter(lines) with open(os.path.join(TestDataDir,'param_section.dat')) as _fdesc: lines = iter(_fdesc.readlines()) res = reader.read_params_info(lines, delchars='()/', delwords=('m', 'sec')) ntable = 0 for k in res: if 'Table' in k: Loading Loading @@ -356,7 +356,8 @@ class IOReaderTestCase(unittest.TestCase): def test_detimage_parse(self): filename = os.path.join(TestDataDir,'detector.single') lines = iter(open(filename).readlines()) with open(filename) as _fd: lines = iter(_fd.readlines()) res = reader.parse_detimage(lines) #print res.keys() self.assertEqual(len(res), 48) Loading