diff --git a/sf_daq_broker/broker_manager.py b/sf_daq_broker/broker_manager.py index 9a1f51e..a8c63da 100644 --- a/sf_daq_broker/broker_manager.py +++ b/sf_daq_broker/broker_manager.py @@ -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'} diff --git a/sf_daq_broker/utils.py b/sf_daq_broker/utils.py index 63e26f5..8bfaf52 100644 --- a/sf_daq_broker/utils.py +++ b/sf_daq_broker/utils.py @@ -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) diff --git a/sf_daq_broker/writer/bsread_writer.py b/sf_daq_broker/writer/bsread_writer.py index 79997cd..39aaab0 100644 --- a/sf_daq_broker/writer/bsread_writer.py +++ b/sf_daq_broker/writer/bsread_writer.py @@ -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,