#!/usr/bin/env python3
import sys
import re
import datetime
import pandas
import tomli
import urllib.request
from km3dq_common.common_library import get_run_properties_from_db
###############################################################################
[docs]
def date_to_datetime(d):
"""
Convert a date-string "yyyy-mm-dd hh:mm" into a datetime
"""
re_time = re.compile(r"(\d\d\d\d)-(\d\d)-(\d\d)\s(\d\d):(\d\d)")
spl_time = re_time.match(d)
if spl_time:
output = datetime.datetime(int(spl_time.group(1)),
int(spl_time.group(2)),
int(spl_time.group(3)),
int(spl_time.group(4)),
int(spl_time.group(5)))
else:
print(f"Wrong datetime format {d}")
sys.exit()
return output
###############################################################################
[docs]
def get_det_run_timerange(site):
"""
Return all the start/end of time/run for all detectors of a site
Data read from the lw_db/Common sftp area
This script is meanly to bypass the request to KM3DB that can be problematic
for nicegui website
"""
tr = {}
# Soon obsolete
# if site == "ARCA":
# start_date_time = "2019-01-16 06:00"
# else:
# start_date_time = "2020-01-26 15:38"
#
# tr["Any"] = {"start": start_date_time,
# "start_run": 0}
tr["Any"] = {"det_start": "",
"first_run": 0}
with urllib.request.urlopen(f"https://sftp.km3net.de/data/km3dq_lw_db/{site}/detector_status_config.toml") as s_f:
toml = tomli.loads(s_f.read().decode("utf-8"))
for i_det in toml.keys():
# det_start: detector start - A priori equal to the previous-detector end
# This is the only mandatory characteristics
tr[i_det] = {"det_start": toml[i_det]["det_start"]}
# det_end: detector end - Undefined only for the current detector
try:
tr[i_det]["det_end"] = toml[i_det]["det_end"]
except KeyError:
tr[i_det]["det_end"] = "2035-01-01 00:00"
# det_id: detector id
try:
tr[i_det]["det_id"] = toml[i_det]["det_id"]
except KeyError:
tr[i_det]["det_id"] = "Unknown"
# first_comm_run / last_comm_run: First/last commissioning run
try:
tr[i_det]["first_comm_run"] = toml[i_det]["first_comm_run"]
try:
tr[i_det]["last_comm_run"] = toml[i_det]["last_comm_run"]
except KeyError:
tr[i_det]["last_comm_run"] = 1e6
except KeyError:
tr[i_det]["first_comm_run"] = 1e6
tr[i_det]["last_comm_run"] = 0
# first_phys_run / last_phys_run: First/last commissioning run
try:
tr[i_det]["first_phys_run"] = toml[i_det]["first_phys_run"]
try:
tr[i_det]["last_phys_run"] = toml[i_det]["last_phys_run"]
except KeyError:
tr[i_det]["last_phys_run"] = 1e6
except KeyError:
tr[i_det]["first_phys_run"] = 1e6
tr[i_det]["last_phys_run"] = 0
tr[i_det]["first_run"] = min(tr[i_det]["first_comm_run"], tr[i_det]["first_phys_run"])
tr[i_det]["last_run"] = max(tr[i_det]["last_comm_run"], tr[i_det]["last_phys_run"])
# Reminder: the detector facts are chronologically ordered
if tr["Any"]["first_run"] == 0:
tr["Any"]["first_run"] = tr[i_det]["first_run"]
if tr["Any"]["det_start"] == "":
tr["Any"]["det_start"] = tr[i_det]["det_start"]
tr["Any"]["last_run"] = tr[i_det]["last_run"]
tr["Any"]["det_end"] = tr[i_det]["det_end"]
return tr
###############################################################################
[docs]
def get_det_run_from_date(site, date_str):
"""
Input:
- site: ARCA/ORCA
- date/time with the format "yyyy-mm-dd hh:mm"
Output:
- Detector name / run number corresponding to this date/time.
"""
out = {"detector": "",
"run": 1e5}
tr = get_det_run_timerange(site)
date_time = date_to_datetime(date_str)
date_time_epoch = date_time.replace(tzinfo=datetime.timezone.utc).timestamp()
for i_det in tr.keys():
d_t_0 = date_to_datetime(tr[i_det]["start"])
d_t_1 = date_to_datetime(tr[i_det]["end"])
if d_t_0 <= date_time < d_t_1:
out["detector"] = i_det
continue
run_prop = get_run_properties_from_db(out["detector"], "PHYS")
for i_run in run_prop.keys():
if all((date_time_epoch < float(run_prop[i_run]['UNIXSTARTTIME'])/1000.,
i_run < out["run"])):
out["run"] = i_run
return out