Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/io/kan.py: 85%

1# http://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

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

5 

6''' 

7Reader for KAN files. 

8''' 

9 

10import os 

11import numpy as num 

12 

13from pyrocko import util, trace 

14from .io_common import FileLoadError 

15 

16 

17class KanError(Exception): 

18 pass 

19 

20 

21class KanFile: 

22 

23 nbytes_header = 512 

24 

25 def __init__(self, *args, **kwargs): 

26 if args: 

27 self.read(*args, **kwargs) 

28 else: 

29 self.clear() 

30 

31 def clear(self): 

32 ''' 

33 Empty file record. 

34 ''' 

35 

36 # set the required attributes 

37 self.delta = 1.0 

38 self.npts = 0 

39 self.b = 0.0 

40 self.data = [num.arange(0, dtype=num.int32)] 

41 

42 def read(self, filename, load_data=True): 

43 ''' 

44 Read SAC file. 

45 

46 filename -- Name of KAN file. 

47 load_data -- If True, the data is read, otherwise only read headers. 

48 ''' 

49 nbh = KanFile.nbytes_header 

50 

51 # read in all data 

52 with open(filename, 'rb') as f: 

53 if load_data: 

54 filedata = f.read() 

55 else: 

56 filedata = f.read(nbh) 

57 

58 if len(filedata) < nbh: 

59 raise KanError('File too short to be a KAN file.') 

60 

61 header_records = int(filedata[0:8]) 

62 npts = int(filedata[8:20]) 

63 leven = int(filedata[20:28]) 

64 datatype = int(filedata[28:36]) 

65 rate = float(filedata[36:50]) 

66 offset = float(filedata[50:64]) 

67 byte_sex = ('little', 'big')[int(filedata[382:390])] 

68 

69 assert header_records == 1 

70 assert leven == 1 

71 assert datatype == 2 

72 assert offset == 0.0 

73 

74 date = str(filedata[393:405].decode('ascii')).strip() 

75 tim = str(filedata[405:415].decode('ascii')).strip() 

76 microseconds = int(filedata[415:423]) 

77 ref_time = util.to_time_float(util.ctimegm('%s %s' % (date, tim))) \ 

78 + microseconds/1.0e6 

79 

80 stationname = os.path.basename(filename) 

81 stationname = stationname.replace('.kan', '') 

82 stationname = stationname.replace('.KAN', '') 

83 

84 self.npts = npts 

85 self.delta = 1.0/rate 

86 self.kstnm = stationname 

87 self.ref_time = ref_time 

88 

89 if load_data: 

90 if byte_sex == 'little': 

91 dtype = '<i4' 

92 else: 

93 dtype = '>i4' 

94 

95 self.data = num.frombuffer(filedata[nbh:], dtype=dtype) 

96 

97 assert self.data.size == self.npts 

98 else: 

99 self.data = None 

100 

101 def to_trace(self): 

102 return trace.Trace( 

103 '', self.kstnm, '', '', 

104 self.ref_time, 

105 self.ref_time+self.delta*(self.npts-1), 

106 self.delta, 

107 self.data) 

108 

109 

110def iload(filename, load_data): 

111 try: 

112 kanf = KanFile(filename, load_data=load_data) 

113 tr = kanf.to_trace() 

114 yield tr 

115 

116 except (OSError, KanError) as e: 

117 raise FileLoadError(e)