"""CALDB interface
"""
import os
import re
import logging
import subprocess
import requests
logger = logging.getLogger(__name__)
ext_cfg = {
"host": "svom-caldb.lal.in2p3.fr",
"path": "/api/",
"port": ""
}
[docs]def send_file_to_caldb(file):
"""Send whatever file to CALDB.
Usefull for tests. Not to be used in production.
Following keywords MUST appear in primary hdu file:
"TELESCOP" = "SVOM"
"INSTRUME" = "ECL"
"CCLS0001" = "BCF"
"CDTP0001" = "DATA"
"CCNM0001" = "ARF"
"CVSD0001" = "1999-01-01"
"CVST0001" = "00:00:00"
"CDES0001" = "test dataset"
The KW CCNM0001 specifies the type in your request.
:param file: path to the file to be sent to CALDB
:type file: str
"""
with open(file, 'rb') as fp:
data = fp.read()
url = 'https://' + ext_cfg['host'] + ext_cfg['path']
response = requests.post(url + "/admin/", data=data)
logger.debug(f"CALDB response: {response}")
[docs]def download_file_from_caldb(dest_dir, file_calib,
path_caldb="user/data/svom/ecl/cpf/arf_effarea"):
"""Simple download with API REST interface / JM Colley
.. note: for the moment the download does not account for the port 5000.
Otherwise it does not work.
:param dest_dir: destination directory
:type dest_dir: str
:param file_calib: name of the calibration file to be downloaded
:type file_calib: str
:param path_caldb: root path in caldb
:type path_caldb: str
"""
if not os.path.exists(dest_dir):
logger.info(f"Destination directory {dest_dir} does not exist !")
return False
if ext_cfg['port'] is "":
raw_url = 'https://' + ext_cfg['host'] + ext_cfg['path']
else:
raw_url = 'https://' + ext_cfg['host'] + ':' + \
str(ext_cfg['port']) + ext_cfg['path']
# https://svom-caldb.lal.in2p3.fr/api/user/data/svom/ecl/bcf/aux-ecl-arf-rsp/aux-ecl-arf-rsp_2017-01-01.fits
url = os.path.join(raw_url, path_caldb, file_calib)
p_file = os.path.join(dest_dir, file_calib)
cmd = f'curl {url} > {p_file}'
logger.debug(f"cmd : {cmd}")
os.system(cmd)
ret = os.stat(p_file).st_size > 0
rename_arf_dc2(p_file)
return ret
[docs]def receive_arf_file_from_caldb(dest_dir):
"""Receive ARf file from the CALDB.
:param dest_dir: path of the destination directory
:type dest_dir: str
:return: if the operation succeeded
:rtype: bool
"""
if not os.path.exists(dest_dir):
logger.info(f"Destination directory {dest_dir} does not exist !")
return False
filetype = 'ARF'
headers = {
"Cache-Control": "no-cache",
"Pragma": "no-cache"
}
try:
url = call_quzcif(filetype)
response = requests.get(url, headers=headers)
except requests.exceptions.ConnectionError:
logger.info("Failed to establish a new connection with CALDB")
return False
response_header = response.headers['content-disposition']
caldb_filename = re.findall("filename=(.+)", response_header)[0]
abspath_filename = os.path.join(dest_dir, caldb_filename)
with open(abspath_filename, 'wb') as f:
f.write(response.content)
# rename from ecl_rsp_ene to ECL-RSP-ENE.
rename_arf(abspath_filename)
return True
[docs]def execute(cmd):
"""Simple subprocess run
"""
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
return proc.communicate()
[docs]def call_quzcif(filetype='ECL-RSP-ENE'):
""" CALDB request quzcif
"""
if "CALDB" not in os.environ:
logger.error("CALDB should be in environment")
if "CALDBCONFIG" not in os.environ:
logger.error("CALDBCONFIG should be in environment")
if "CALDBALIAS" not in os.environ:
logger.error("CALDBALIAS should be in environment")
# filetype = 'ECL-RSP-ENE'
cmd = ["quzcif svom ecl - - %s now now -" % (filetype)]
out, _ = execute(cmd)
return out.decode('utf-8').split()[0]
# TODO: Check the os independence
[docs]def rename_arf(arf_path):
"""Rename arf with correct syntaxe
"""
root = '/'.join(arf_path.split('/')[:-1]) + '/'
f = arf_path.split('/')[-1]
f = f.replace(f[:11], f[:11].upper())
os.rename(arf_path, root + f)
# TODO: Check the os independence
[docs]def rename_arf_dc2(arf_path):
"""Rename arf with correct syntaxe
"""
root = '/'.join(arf_path.split('/')[:-1]) + '/'
f_arf = arf_path.split('/')[-1]
f_arf = 'ECL-RSP-ENE_' + f_arf
os.rename(arf_path, root + f_arf)