#! /usr/bin/env python
###############################################################################
import datetime
import tomli
import re
import urllib.request
###################################################################################
# Class inspired from https://gitlab.in2p3.fr/laurent-aphecetche/km3net-shifter-tools
# package
[docs]
class ShiftSummary:
def __init__(self, source):
if source == "sftp":
self.retrieve_schedule_from_sftp()
self.get_current_crew()
self.retrieve_links_from_sardine()
##############################
[docs]
def retrieve_schedule_from_sftp(self):
"""
Retrieve from the KM3NeT database the name/mails of shifters and
run coordinator
"""
self.shift_crew = {}
self.run_coordinator = {}
with urllib.request.urlopen("https://sftp.km3net.de/data/km3dq_lw_db/Common/shift_schedule.toml") as s_f:
tmp = s_f.read()
t = tomli.loads(tmp.decode("utf-8"))
self.update = t["last_update"]
re_week = re.compile(r"(\d+)/(\d+)/(\d+) - (\d+)/(\d+)/(\d+)")
for (_, i_value) in t["shift_crew"].items():
res = re_week.search(i_value["week"])
self.shift_crew[i_value["week"]] = {
"day_start": datetime.date(int(res.group(3)),
int(res.group(2)),
int(res.group(1))),
"shift_leader": i_value["shift_leader"],
"shifter": i_value["shifter"],
"name": i_value["name"],
"mail": i_value["mail"]
}
for (_, i_value) in t["run_coordinator"].items():
res = re_week.search(i_value["week"])
self.run_coordinator[i_value["week"]] = {
"day_start": datetime.date(int(res.group(3)),
int(res.group(2)),
int(res.group(1))),
"run_coordinator": i_value["run_coordinator"],
"name": i_value["name"],
"mail": i_value["mail"]
}
##############################
[docs]
def retrieve_links_from_sardine(self):
"""
"""
with urllib.request.urlopen("https://shifter-tools-km3net-operation-tools-prod.apps.wok2.in2p3.fr/sardine/shift_prop_crew.toml") as s_f:
tmp = s_f.read()
t = tomli.loads(tmp.decode("utf-8"))
self.weekly_report = t["weekly_report_archive"]
self.weekly_plots_arca = t["weekly_plots_arca_archive"]
self.weekly_plots_orca = t["weekly_plots_orca_archive"]
self.weekly_highlights_arca = t["weekly_highlights_arca_archive"]
self.weekly_highlights_orca = t["weekly_highlights_orca_archive"]
self.biweekly_aob = t["biweekly_aob"]
##############################
[docs]
def get_current_crew(self):
"""
Return the names of the two shifters as a single string
"""
tod = datetime.datetime.today()
today = datetime.date(tod.year, tod.month, tod.day)
current_crew = "Unknown"
for i_week in self.shift_crew.keys():
delta = (self.shift_crew[i_week]["day_start"]-today).days
if all((delta >= -7, delta < 0)):
self.current_week = i_week
self.current_week_short = re.sub(r'/20\d\d', "", i_week)
self.current_sc = self.shift_crew[i_week]["name"]
self.current_rc = self.run_coordinator[i_week]["name"]