Coverage for larch/io/xas_data_source/read.py: 92%

37 statements  

« prev     ^ index     » next       coverage.py v7.6.0, created at 2024-10-16 21:04 +0000

1from typing import Optional 

2from . import sources 

3from . import hdf5_utils 

4from larch import Group 

5 

6 

7def open_xas_source(filename, **kw): 

8 with open(filename, "rb") as fh: 

9 topbytes = fh.read(10) 

10 

11 if topbytes.startswith(b"\x89HDF\r"): 

12 creator = hdf5_utils.nexus_creator(filename).lower() 

13 class_name = None 

14 if creator == "bliss": 

15 class_name = "esrf" 

16 if not class_name: 

17 source = hdf5_utils.nexus_source(filename).lower() 

18 if "soleil" in source: 

19 class_name = "soleil" 

20 if not class_name: 

21 instrument = hdf5_utils.nexus_instrument(filename).lower() 

22 if "soleil" in instrument: 

23 class_name = "soleil" 

24 if not class_name: 

25 class_name = "nexus" 

26 elif topbytes.startswith(b"#S ") or topbytes.startswith(b"#F "): 

27 class_name = "spec" 

28 else: 

29 raise ValueError(f"Unknown file format: {filename}") 

30 return sources.get_source_type(class_name)(filename, **kw) 

31 

32 

33def read_xas_source(filename: str, scan: Optional[str] = None) -> Optional[Group]: 

34 if scan is None: 

35 return None 

36 source = open_xas_source(filename) 

37 scan = source.get_scan(scan) 

38 

39 lgroup = Group( 

40 __name__=f"{source.TYPE} file: {filename}, scan: {scan.name}", 

41 filename=filename, 

42 source_info=source.get_source_info(), 

43 datatype="xas", 

44 ) 

45 for name, value in scan._asdict().items(): 

46 setattr(lgroup, name, value) 

47 for name, value in zip(scan.labels, scan.data): 

48 setattr(lgroup, name, value) 

49 return lgroup