Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CHAP/common/map_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -443,7 +443,7 @@ def get_independent_dimensions(_scanparser):
{
'label': 'Start_Epoch',
'units': 'seconds',
'data_type': 'scan_start_time',
'data_type': 'spec_epoch',
'name': 'Start_Epoch',
},
]
Expand Down
30 changes: 29 additions & 1 deletion CHAP/common/models/map.py
Original file line number Diff line number Diff line change
Expand Up @@ -433,7 +433,7 @@ class PointByPointScanData(CHAPBaseModel):
data_type: Literal[
'detector_log_timestamps', 'expression', 'scan_column',
'scan_number', 'scan_start_time', 'scan_step_index',
'smb_par', 'spec_motor', 'spec_motor_absolute',
'smb_par', 'spec_epoch', 'spec_motor', 'spec_motor_absolute',
'spec_motor_static'
]
name: constr(strip_whitespace=True, min_length=1)
Expand Down Expand Up @@ -621,6 +621,12 @@ def get_value(
scanparser = get_scanparser(spec_scans.spec_file, scan_number)
return np.array([start_time] * scanparser.spec_scan_npts)
return start_time
if self.data_type == 'spec_epoch':
spec_epoch = get_spec_epoch(spec_scans.spec_file, scan_number)
if scan_step_index < 0:
scanparser = get_scanparser(spec_scans.spec_file, scan_number)
return np.array([spec_epoch] * scanparser.spec_scan_npts)
return spec_epoch
if self.data_type == 'smb_par':
return get_smb_par_value(
spec_scans.spec_file, scan_number, self.name)
Expand Down Expand Up @@ -776,6 +782,28 @@ def get_scan_start_time(spec_file, scan_number):
start_time = start_time.astimezone(tz=datetime.timezone.utc)
return start_time.timestamp()

@cache
def get_spec_epoch(spec_file, scan_number):
"""Return the previous most recent epoch (``#E`` line) recorded in
the spec file."""
_scanparser = get_scanparser(spec_file, scan_number)
scan_firstline = _scanparser.spec_scan.firstline
headers = _scanparser.spec_file._headers
useheader_i = -1
spec_epoch = None
while useheader_i < len(headers) - 1:
if headers[useheader_i + 1].firstline < scan_firstline:
useheader_i += 1
else:
break
spec_epoch = headers[useheader_i]._epoch
if spec_epoch is None:
raise RuntimeError(
'No headers containing epoch preceding scan '
f'{scan_number} in {spec_file}'
)
return spec_epoch

def get_expression_value(
spec_scans, scan_number, scan_step_index, expression,
scalar_data):
Expand Down
Loading