sf-daq/sf_daq_broker
0
0
Fork 0
mirror of https://github.com/paulscherrerinstitute/sf_daq_broker.git synced 2026-08-10 18:20:30 +02:00
Code Issues Packages Projects Releases Wiki Activity

proper timestamp conversion for data-api calls

Browse Source
This commit is contained in:
Dmitry Ozerov
2022-01-20 16:26:35 +01:00
committed by Data Backend account
parent 492501f100
commit 3a45bbc3d1
3 changed files with 51 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sf_daq_broker/broker_manager.py
+9 -8
View File
@@ -324,21 +324,22 @@ class BrokerManager(object):
if not os.path.exists(path_to_pgroup):
return {"status" : "failed", "message" : f'pgroup directory {path_to_pgroup} not reachable'}
if "run_number" not in request:
request["run_number"] = get_current_run_number(daq_directory, file_run="LAST_RUN")
run_number = request.get("run_number", 0)
output_run_directory = f'run{run_number:04}'
full_path = f'{path_to_pgroup}{output_run_directory}'
daq_directory = f'{path_to_pgroup}{DIR_NAME_RUN_INFO}'
if not os.path.exists(daq_directory):
try:
os.mkdir(daq_directory)
except:
return {"status" : "failed", "message" : "no permission or possibility to make run_info directory in pgroup space"}
if "run_number" not in request:
request["run_number"] = get_current_run_number(daq_directory, file_run="LAST_RUN")
run_number = request.get("run_number", 0)
output_run_directory = f'run{run_number:04}'
full_path = f'{path_to_pgroup}{output_run_directory}'
if os.path.exists(f'{daq_directory}/CLOSED'):
return {"status" : "failed", "message" : f'{path_to_pgroup} is closed for writing'}
sf_daq_broker/utils.py
+20 -6
View File
@@ -82,21 +82,35 @@ def pulse_id_to_seconds(pulse_id):
raise RuntimeError("Cannot convert pulse_id to time")
return sec
def pulse_id_to_timestamp(pulse_id):
tm = 0
try:
request = requests.get(f'{config.DATA_API3_QUERY_ADDRESS}/map/pulse/{pulse_id}')
if request.status_code == 200:
ts = request.json()
else:
_logger.error(f'Problem to convert {pulse_id} to timestamp. return code {request.status_code}')
except Exception as e:
_logger.error(e)
raise RuntimeError("Cannot convert pulse_id to time")
return ts
def transform_range_from_pulse_id_to_timestamp_new(data_api_request):
new_data_api_request = deepcopy(data_api_request)
try:
start_seconds = pulse_id_to_seconds(data_api_request["range"]["startPulseId"]-1)
stop_seconds = pulse_id_to_seconds(data_api_request["range"]["endPulseId"]+1)
start_ts = pulse_id_to_timestamp(data_api_request["range"]["startPulseId"])
stop_ts = pulse_id_to_timestamp(data_api_request["range"]["endPulseId"]+1)
if start_seconds != 0 and stop_seconds != 0 and start_seconds < stop_seconds:
if start_ts!= 0 and stop_ts != 0 and start_ts < stop_ts:
del new_data_api_request["range"]["startPulseId"]
new_data_api_request["range"]["startSeconds"] = start_seconds
new_data_api_request["range"]["startTS"] = start_ts
del new_data_api_request["range"]["endPulseId"]
new_data_api_request["range"]["endSeconds"] = stop_seconds
new_data_api_request["range"]["endTS"] = stop_ts
else:
_logger.error(f'Convertion pulse_id to time failed {start_seconds} {stop_seconds}')
_logger.error(f'Convertion pulse_id to time failed {start_ts} {stop_ts}')
except Exception as e:
_logger.error(e)
sf_daq_broker/writer/bsread_writer.py
+22 -10
View File
@@ -2,6 +2,7 @@ import logging
import os
from datetime import datetime
from time import time
import pytz
import h5py
import numpy
@@ -20,6 +21,13 @@ except:
_logger.warning("There is no ujson in this environment. Performance will suffer.")
import json
def tsfmt(ts):
ts = ts // 1000
n = ts // 1000000
m = ts % 1000000
s = datetime.fromtimestamp(n).astimezone(pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S")
s = f"{s}.{m:06d}Z"
return s
def check_data_consistency(start_pulse_id, stop_pulse_id, rate_multiplicator, channels, output_file):
@@ -115,7 +123,6 @@ def write_from_databuffer(data_api_request, output_file, metadata):
def write_from_imagebuffer(data_api_request, output_file, parameters):
import data_api3.h5 as h5
import pytz
start_pulse_id = data_api_request["range"]["startPulseId"]
stop_pulse_id = data_api_request["range"]["endPulseId"]
@@ -127,10 +134,13 @@ def write_from_imagebuffer(data_api_request, output_file, parameters):
channels = [channel["name"] for channel in data_api_request_timestamp["channels"]]
start = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["startSeconds"])).astimezone(
pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
end = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["endSeconds"])).astimezone(
pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
#start = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["startSeconds"])).astimezone(
# pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
#end = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["endSeconds"])).astimezone(
# pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
start = tsfmt(data_api_request_timestamp["range"]["startTS"])
end = tsfmt(data_api_request_timestamp["range"]["endTS"])
query = {
"channels": channels,
@@ -159,7 +169,6 @@ def write_from_imagebuffer(data_api_request, output_file, parameters):
def write_from_databuffer_api3(data_api_request, output_file, parameters):
import data_api3.h5 as h5
import pytz
_logger.debug("Data3 API request: %s", data_api_request)
@@ -171,10 +180,13 @@ def write_from_databuffer_api3(data_api_request, output_file, parameters):
channels = [channel["name"] for channel in data_api_request_timestamp["channels"]]
start = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["startSeconds"])).astimezone(
pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
end = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["endSeconds"])).astimezone(
pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
#start = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["startSeconds"])).astimezone(
# pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
#end = datetime.fromtimestamp(float(data_api_request_timestamp["range"]["endSeconds"])).astimezone(
# pytz.timezone('UTC')).strftime("%Y-%m-%dT%H:%M:%S.%fZ") # isoformat() # "2019-12-13T09:00:00.00
start = tsfmt(data_api_request_timestamp["range"]["startTS"])
end = tsfmt(data_api_request_timestamp["range"]["endTS"])
query = {
"channels": channels,