Source code for km3dq_common.lw_db_fact_library

#! /usr/bin/env python
###############################################################################
import re
import sys
import calendar
import time
import urllib.request


###############################################################################
[docs] def read_basic_fact_file(det, fact_type): """ Read defect files stored on SFTP. Simply returns lines === Arguments === - det : detector name - [string] - Ex: "D0ARCA021", "D0ORCA018"... - fact_type === Output === - Array of lines """ if "ORCA" in det: site = "ORCA" else: site = "ARCA" lines = [] fact_url = ( "https://sftp.km3net.de/data/km3dq_lw_db/" f"{site}/{det}/Facts/" f"{fact_type}.txt" ) try: with urllib.request.urlopen(fact_url) as def_file: tmp = (def_file.read()).split(b"\n") for i_line in tmp: if i_line != "": lines.append(i_line.decode("utf-8")) except urllib.error.URLError: # No fact file found lines = None return lines
###############################################################################
[docs] def read_fact_file(det, fact_type): """ Read defect files stored on SFTP. === Arguments === - det : detector name - [string] - Ex: "D0ARCA021", "D0ORCA018"... - fact_type === Output === - Array of lines """ if "ORCA" in det: site = "ORCA" else: site = "ARCA" re_line = re.compile( r"\s*(\S*)\s*\|" r"\s*(\S*)\s*\|" r"\s*(\S.*\d)\s\-\s(\S.*\d)\s*\|" r"(.*)\|" r"(.*)\|" r"(.*)" ) re_line_no_end = re.compile( r"\s*(\S*)\s*\|" r".*\|" r"\s*(\S.*\d)\s\-\s*\|" r"(.*)\|" r"(.*)\|" r"(.*)" ) fact_list = [] fact_url = ( "https://sftp.km3net.de/data/km3dq_lw_db/" f"{site}/{det}/Facts/" f"{fact_type}.txt" ) try: with urllib.request.urlopen(fact_url) as def_file: tmp = (def_file.read()).split(b"\n") for i_line in tmp: r_m = re_line.match(i_line.decode("utf-8")) if r_m: # Intermediate step to handle facts with and without second accuracy # Accuracy must be the same for both end/start by construction if r_m.group(3).count(":") == 1: t_s = int(calendar.timegm(time.strptime(r_m.group(3), "%a, %d %b %Y %H:%M"))) t_e = int(calendar.timegm(time.strptime(r_m.group(4), "%a, %d %b %Y %H:%M"))) else: t_s = int(calendar.timegm(time.strptime(r_m.group(3), "%a, %d %b %Y %H:%M:%S"))) t_e = int(calendar.timegm(time.strptime(r_m.group(4), "%a, %d %b %Y %H:%M:%S"))) fact_list.append({ "run_start": r_m.group(1), "run_end": r_m.group(2), "time_start": t_s, "time_end": t_e, "comment": r_m.group(5), "documentation": r_m.group(6), "author": r_m.group(7), }) continue r_m = re_line_no_end.match(i_line.decode("utf-8")) if r_m: # Intermediate step to handle facts with and without second accuracy if r_m.group(2).count(":") == 1: t_s = int(calendar.timegm(time.strptime(r_m.group(2), "%a, %d %b %Y %H:%M"))) else: t_s = int(calendar.timegm(time.strptime(r_m.group(2), "%a, %d %b %Y %H:%M:%S"))) fact_list.append({ "run_start": r_m.group(1), "run_end": 1e6, "time_start": t_s, "time_end": 0, "comment": r_m.group(3), "documentation": r_m.group(4), "author": r_m.group(5), }) except urllib.error.URLError: # No fact file found lines = None return fact_list