Coverage for /usr/local/lib/python3.13/dist-packages/pyrocko/squirrel/io/backends/hdf5_optodas.py: 54%

1# http://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

4# ---|P------/S----------~Lg---------- 

5 

6''' 

7Squirrel IO adaptor to :py:mod:`pyrocko.io.tdms_idas`. 

8''' 

9from __future__ import annotations 

10 

11from typing import Any, Generator 

12 

13from ... import model 

14 

15 

16SEGMENT_SIZE = 1024*1024 

17 

18try: 

19 import simpledas 

20except ImportError: 

21 simpledas = None 

22 

23 

24def provided_formats() -> list[str]: 

25 return ["hdf5_optodas"] 

26 

27 

28def detect(first512: bytes) -> bool: 

29 from pyrocko.io import hdf5_optodas 

30 if hdf5_optodas.detect(first512): 

31 return 'hdf5_optodas' 

32 else: 

33 return None 

34 

35 

36def iload( 

37 format: str, 

38 file_path: str, 

39 segment: int, 

40 content: tuple[str, ...]) -> Generator[model.Nut, Any, None]: 

41 

42 assert format == 'hdf5_optodas' 

43 from pyrocko.io import hdf5_optodas 

44 

45 load_data = 'waveform' in content 

46 

47 file_segment = None 

48 itr = 0 

49 for tr in hdf5_optodas.iload( 

50 file_path, load_data=load_data): 

51 

52 # if file_segment != tr.meta['offset_start']: 

53 # itr = 0 

54 # file_segment = tr.meta['offset_start'] 

55 

56 nsamples = int(round((tr.tmax - tr.tmin) / tr.deltat)) + 1 

57 nut = model.make_waveform_nut( 

58 file_segment=file_segment, 

59 file_element=itr, 

60 codes=tr.codes, 

61 tmin=tr.tmin, 

62 tmax=tr.tmin + tr.deltat * nsamples, 

63 deltat=tr.deltat) 

64 

65 if 'waveform' in content: 

66 nut.content = tr 

67 

68 yield nut 

69 itr += 1