import argparse
import asyncio
import copy
import getpass
import glob
import json
import logging
import os
import sqlite3
import subprocess # FIXME: use asyncio.create_subprocess_*
import traceback
import urllib.parse
from asyncio import get_event_loop, shield
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List
import yaml
import zmq
import zmq.asyncio
import zmq.auth.thread
from git import InvalidGitRepositoryError, Repo
from metadata_client.metadata_client import MetadataClient
try:
from .messages import MDC, Errors, Success
except ImportError:
from messages import MDC, Errors, Success
async def init_job_db(config):
""" Initialize the sqlite job database
A new database is created if no pre-existing one is present. A single
table is created: jobs, which has columns:
rid, id, proposal, run, flg, status
:param config: the configuration parsed from the webservice YAML config
:return: a sqlite3 connection instance to the database
"""
# FIXME: sqlite3 is synchronous, it should be replaced with
# https://pypi.org/project/databases/
logging.info("Initializing database")
conn = sqlite3.connect(config['web-service']['job-db'])
c = conn.cursor()
try:
c.execute("SELECT * FROM jobs")
except Exception: # TODO: is it sqlite3.OperationalError?
logging.info("Creating initial job database")
c.execute("CREATE TABLE jobs(rid, jobid, proposal, run, status, time, det, act)")
return conn
async def init_md_client(config):
""" Initialize an MDC client connection
:param config: the configuration parsed from the webservice YAML config
:return: an MDC client connection
"""
# FIXME: this blocks the even loop, should use asyncio.Task
# FIXME: calls to this coro should be shielded
# TODO: could the client be a global? This would recuce passing it around
mdconf = config['metadata-client']
client_conn = MetadataClient(client_id=mdconf['user-id'],
client_secret=mdconf['user-secret'],
user_email=mdconf['user-email'],
token_url=mdconf['token-url'],
refresh_url=mdconf['refresh-url'],
auth_url=mdconf['auth-url'],
scope=mdconf['scope'],
base_api_url=mdconf['base-api-url'])
return client_conn
def init_config_repo(config):
""" Make sure the configuration repo is present and up-to-data
:param config: the configuration defined in the `config-repo` section
"""
# FIXME: keep this as func, but call this with asyncio.thread
os.makedirs(config['local-path'], exist_ok=True)
# check if it is a repo
try:
repo = Repo(config['local-path'])
except InvalidGitRepositoryError:
repo = Repo.clone_from(config['url'], config['local-path'])
try:
repo.remote().pull()
except Exception:
pass
logging.info("Config repo is initialized")
async def upload_config(socket, config, yaml, instrument, cycle, proposal):
""" Upload a new configuration YAML
:param socket: ZMQ socket to send reply on
:param config: the configuration defined in the `config-repo` section
of the webservice.yaml configuration.
:param yaml: the YAML contents to update
:param instrument: instrument for which the update is for
:param cycle: the facility cylce the update is for
:param proposal: the proposal the update is for
The YAML contents will be placed into a file at
{config.local-path}/{cycle}/{proposal}.yaml
If it exists it is overwritten and then the new version is pushed to
the configuration git repo.
"""
# FIXME: keep this as func, but call this with asyncio.thread
repo = Repo(config['local-path'])
# assure we are on most current version
repo.remote().pull()
prop_dir = os.path.join(repo.working_tree_dir, cycle)
os.makedirs(prop_dir, exist_ok=True)
with open("{}/{}.yaml".format(prop_dir, proposal), "w") as f:
f.write(yaml)
fpath = "{}/{}.yaml".format(prop_dir, proposal)
repo.index.add([fpath])
repo.index.commit(
"Update to proposal p{}: {}".format(proposal,
datetime.now().isoformat()))
repo.remote().push()
logging.info(Success.UPLOADED_CONFIG.format(cycle, proposal))
socket.send(Success.UPLOADED_CONFIG.format(cycle, proposal).encode())
def merge(source: Dict, destination: Dict) -> Dict:
"""Deep merge two dictionaries.
:param source: source dictionary to merge into destination
:param destination: destination dictionary which is being merged in
:return: the updated destination dictionary
Taken from:
https://stackoverflow.com/questions/20656135/
python-deep-merge-dictionary-data
"""
for key, value in source.items():
if isinstance(value, dict):
# check if destination (e.g. karabo-id) has none value
if key in destination.keys() and destination[key] is None:
destination[key] = {}
# get node or create one
node = destination.setdefault(key, {})
merge(value, node)
else:
destination[key] = value
return destination
async def change_config(socket, config, updated_config, karabo_id, instrument,
cycle, proposal, apply=False):
"""
Change the configuration of a proposal
If no proposal specific configuration yet exists, one is first created
based on the default configuration of the proposal
Changes are committed to git.
:param socket: ZMQ socket to send reply on
:param config: repo config as given in YAML config file
:param updated_config: a dictionary containing the updated config
:param instrument: the instrument to change config for
:param cycle: the cycle to change config for
:param proposal: the proposal to change config for
:param apply: set to True to actually commit a change, otherwise a dry-run
is performed
:return: The updated config to the requesting zmq socket
"""
# first check if a proposal specific config exists, if not create one
repo = Repo(config['local-path'])
repo.remote().pull()
prop_dir = os.path.join(repo.working_tree_dir, cycle)
os.makedirs(prop_dir, exist_ok=True)
fpath = "{}/{:06d}.yaml".format(prop_dir, int(proposal))
if not os.path.exists(fpath):
with open("{}/default.yaml".format(repo.working_tree_dir), "r") as f:
defconf = yaml.load(f.read(), Loader=yaml.FullLoader)
subconf = {}
for action, instruments in defconf.items():
subconf[action] = {}
if action != "data-mapping":
subconf[action][instrument] = instruments[instrument]
else:
subconf[action][karabo_id] = instruments[karabo_id]
with open(fpath, "w") as wf:
wf.write(yaml.dump(subconf, default_flow_style=False))
new_conf = None
with open(fpath, "r") as rf:
existing_conf = yaml.load(rf.read(), Loader=yaml.FullLoader)
new_conf = merge(updated_config, existing_conf)
if apply:
with open(fpath, "w") as wf:
wf.write(yaml.dump(new_conf, default_flow_style=False))
repo.index.add([fpath])
repo.index.commit(
"Update to proposal YAML: {}".format(datetime.now().isoformat()))
repo.remote().push()
logging.info(Success.UPLOADED_CONFIG.format(cycle, proposal))
socket.send(yaml.dump(new_conf, default_flow_style=False).encode())
async def slurm_status(filter_user=True):
""" Return the status of slurm jobs by calling squeue
:param filter_user: set to true to filter ony jobs from current user
:return: a dictionary indexed by slurm jobid and containing a tuple
of (status, run time) as values.
"""
cmd = ["squeue"]
if filter_user:
cmd += ["-u", getpass.getuser()]
ret = subprocess.run(cmd, stdout=subprocess.PIPE) # FIXME: asyncio
if ret.returncode == 0:
rlines = ret.stdout.decode().split("\n")
statii = {}
for r in rlines[1:]:
try:
jobid, _, _, _, status, runtime, _, _ = r.split()
jobid = jobid.strip()
statii[jobid] = status, runtime
except ValueError: # not enough values to unpack in split
pass
return statii
return None
async def slurm_job_status(jobid):
""" Return the status of slurm job
:param jobid: Slurm job Id
:return: Slurm state, Elapsed.
"""
cmd = ["sacct", "-j", str(jobid), "--format=JobID,Elapsed,state"]
ret = subprocess.run(cmd, stdout=subprocess.PIPE) # FIXME: asyncio
if ret.returncode == 0:
rlines = ret.stdout.decode().split("\n")
logging.debug("Job {} state {}".format(jobid, rlines[2].split()))
if len(rlines[2].split()) == 3:
return rlines[2].replace("+", "").split()
return "NA", "NA", "NA"
async def query_rid(conn, socket, rid):
c = conn.cursor()
c.execute("SELECT * FROM jobs WHERE rid LIKE ?", rid)
combined = {}
for r in c.fetchall():
rid, jobid, proposal, run, status, time_, _ = r
logging.debug(
"Job {}, proposal {}, run {} has status {}".format(jobid,
proposal,
run,
status))
cflg, cstatus = combined.get(rid, ([], []))
if status in ['R', 'PD']:
flg = 'R'
elif status == 'NA':
flg = 'NA'
else:
flg = 'A'
cflg.append(flg)
cstatus.append(status)
combined[rid] = cflg, cstatus
flg_order = {"R": 2, "A": 1, "NA": 0}
msg = ""
for rid, value in combined.items():
flgs, statii = value
flg = max(flgs, key=lambda i: flg_order[i])
msg += "\n".join(statii)
if msg == "":
msg = 'NA'
socket.send(msg.encode())
def parse_config(cmd: List[str], config: Dict[str, Any]) -> List[str]:
"""Convert a dictionary to a list of arguments.
Values that are not strings will be cast.
Lists will be converted to several strings following their `--key`
flag.
Booleans will be converted to a `--key` flag, where `key` is the
dictionary key.
"""
for key, value in config.items():
if ' ' in key or (isinstance(value, str) and ' ' in value):
raise ValueError('Spaces are not allowed', key, value)
if isinstance(value, list):
cmd.append(f"--{key}")
cmd += [str(v) for v in value]
elif isinstance(value, bool):
if value:
cmd += ["--{}".format(key)]
else:
if value in ['""', "''"]:
value = ""
cmd += ["--{}".format(key), str(value)]
return cmd
async def update_job_db(config):
""" Update the job database and send out updates to MDC
This coro runs as background task
:param config: configuration parsed from webservice YAML
"""
logging.info("Starting config db handling")
conn = await init_job_db(config)
mdc = await init_md_client(config)
time_interval = int(config['web-service']['job-update-interval'])
while True:
statii = await slurm_status()
# Check that slurm is giving proper feedback
if statii is None:
await asyncio.sleep(time_interval)
continue
try:
c = conn.cursor()
c.execute("SELECT * FROM jobs WHERE status IN ('R', 'PD', 'CG') ")
combined = {}
logging.debug("SLURM info {}".format(statii))
for r in c.fetchall():
rid, jobid, proposal, run, status, time, _, action = r
logging.debug("DB info {}".format(r))
cflg, cstatus = combined.get(rid, ([], []))
if jobid in statii:
slstatus, runtime = statii[jobid]
query = "UPDATE jobs SET status=?, time=? WHERE jobid LIKE ?"
c.execute(query, (slstatus, runtime, jobid))
cflg.append('R')
cstatus.append("{}-{}".format(slstatus, runtime))
else:
_, sltime, slstatus = await slurm_job_status(jobid)
query = "UPDATE jobs SET status=? WHERE jobid LIKE ?"
c.execute(query, (jobid, slstatus))
if slstatus == 'COMPLETED':
cflg.append("A")
else:
cflg.append("NA")
cstatus.append(slstatus)
combined[rid] = cflg, cstatus
conn.commit()
flg_order = {"R": 2, "A": 1, "NA": 0}
dark_flags = {'NA': 'E', 'R': 'IP', 'A': 'F'}
for rid, value in combined.items():
if int(rid) == 0: # this job was not submitted from MyMDC
continue
flgs, statii = value
# sort by least done status
flg = max(flgs, key=lambda i: flg_order[i])
msg = "\n".join(statii)
msg_debug = f"Update MDC {rid}, {msg}"
logging.debug(msg_debug.replace('\n', ', '))
if action == 'CORRECT':
response = mdc.update_run_api(rid,
{'flg_cal_data_status': flg,
'cal_pipeline_reply': msg})
else: # action == 'DARK' but it's dark_request
data = {'dark_run': {'flg_status': dark_flags[flg],
'calcat_feedback': msg}}
response = mdc.update_dark_run_api(rid, data)
if response.status_code != 200:
logging.error(Errors.MDC_RESPONSE.format(response))
except Exception as e:
e = str(e)
logging.error(f"Failure to update job DB: {e}")
await asyncio.sleep(time_interval)
async def copy_untouched_files(file_list, out_folder, run):
""" Copy those files whicha are not touched by calibration to outpot dir
:param file_list: The list of files to copy
:param out_folder: The output folder
:param run: The run which is being handled
Copying is done via an asyncio subprocess call
"""
os.makedirs("{}/r{:04d}".format(out_folder, int(run)), exist_ok=True)
for f in file_list:
of = f.replace("raw", "proc").replace("RAW", "CORR")
cmd = ["rsync", "-av", f, of]
await asyncio.subprocess.create_subprocess_shell(" ".join(cmd))
logging.info("Copying {} to {}".format(f, of))
async def run_action(job_db, cmd, mode, proposal, run, rid):
""" Run action command (CORRECT OR DARK)
:param job_db: jobs database
:param cmd: to run, should be a in list for as expected by subprocess.run
:param mode: "prod" or "sim", in the latter case nothing will be executed
but the command will be logged
:param proposal: proposal the command was issued for
:param run: run the command was issued for
:param: rid: run id in the MDC
Returns a formatted Success or Error message indicating outcome of the
execution.
"""
# FIXME: this coro has too many returns that can be simplified
if mode == "prod":
logging.info(" ".join(cmd))
ret = subprocess.run(cmd, stdout=subprocess.PIPE) # FIXME: asyncio
if ret.returncode == 0:
if "DARK" in cmd:
logging.info(Success.START_CHAR.format(proposal, run))
else:
logging.info(Success.START_CORRECTION.format(proposal, run))
# enter jobs in job db
c = job_db.cursor() # FIXME: asyncio
rstr = ret.stdout.decode()
query = "INSERT INTO jobs VALUES ('{rid}', '{jobid}', '{proposal}', '{run}', 'PD', '{now}', '{det}', '{act}')" # noqa
for r in rstr.split("\n"):
if "Submitted job:" in r:
_, jobid = r.split(":")
c.execute(
"INSERT INTO jobs VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(rid, jobid.strip(), proposal, run,
datetime.now().isoformat(), cmd[3], cmd[4])
)
job_db.commit()
logging.debug((" ".join(cmd)).replace(',', '').replace("'", ""))
if "DARK" in cmd:
return Success.START_CHAR.format(proposal, run)
else:
return Success.START_CORRECTION.format(proposal, run)
else:
logging.error(Errors.JOB_LAUNCH_FAILED.format(cmd, ret.returncode))
return Errors.JOB_LAUNCH_FAILED.format(cmd, ret.returncode)
else:
if "DARK" in cmd:
logging.debug(Success.START_CHAR_SIM.format(proposal, run))
else:
logging.debug(Success.START_CORRECTION_SIM.format(proposal, run))
logging.debug((" ".join(cmd)).replace(',', '').replace("'", ""))
if "DARK" in cmd:
return Success.START_CHAR_SIM.format(proposal, run)
else:
return Success.START_CORRECTION_SIM.format(proposal, run)
async def wait_on_transfer(rpath, max_tries=300):
"""
Wait on data files to be transferred to Maxwell
:param rpath: Folder, which contains data files migrated to Maxwell
:param max_tries: Maximum number of checks if files are transferred
:return: True if files are transferred
"""
# TODO: Make use of MyMDC to request whether the run has been copied.
# It is not sufficient to know that the files are on disk, but also to
# check the copy is finished (ie. that the files are complete).
if 'pnfs' in os.path.realpath(rpath):
return True
rstr = None
ret = None
tries = 0
# FIXME: if not kafka, then do event-driven, no sleep
# wait until folder gets created
while not os.path.exists(rpath):
if tries > max_tries:
return False
tries += 1
await asyncio.sleep(10)
# FIXME: if not kafka, then do event-driven, no sleep
# wait until files are migrated
while rstr is None or 'status="online"' in rstr or 'status="Online"' in rstr or ret.returncode != 0: # noqa
await asyncio.sleep(10)
# FIXME: make use of asyncio.subprocess.run
ret = subprocess.run(["getfattr", "-n", "user.status", rpath],
stdout=subprocess.PIPE)
rstr = ret.stdout.decode()
if tries > max_tries:
return False
tries += 1
return ret.returncode == 0
def check_files(in_folder: str,
runs: List[int],
karabo_das: List[str]) -> bool:
"""Check if files for given karabo-das exists for given runs
:param in_folder: Input folder
:param runs: List of runs
:param karabo_das: List of karabo-das
:return: True if files are there
"""
files_exists = True
for runnr in runs:
rpath = Path(in_folder, 'r{:04d}'.format(int(runnr)))
fl = rpath.glob('*.h5')
if not any(da in match.name for match in fl for da in karabo_das):
files_exists = False
return files_exists
async def update_darks_paths(mdc: MetadataClient, rid: int, in_path: str,
out_path: str, report_path: str):
"""Update data paths in MyMDC to provide Globus access
DarkRequests have the capability to provide a link to the report via
Globus. For this, the absolute path to the report needs to be provided.
Additionally, the input path to the raw data, as well as the output path to
the processed data can be specified.
:param mdc: an authenticated MyMDC client
:param rid: the DarkRequest id to update
:param in_path: the absolute path to the raw data being calibrated
:param out_path: the absolute path to the resulting processed data
:param report_path: the absolute path to the PDF report
"""
data = {'dark_run': {'output_path': out_path,
'input_path': in_path,
'globus_url': report_path}}
loop = get_event_loop()
response = await shield(loop.run_in_executor(None, mdc.update_dark_run_api,
rid, data))
if response.status_code != 200:
logging.error(Errors.MDC_RESPONSE.format(response))
async def update_mdc_status(mdc: MetadataClient, action: str,
rid: int, message: str):
"""Update MyMDC statuses
The message content and destination depend on the Action type.
A message to MyMDC expects a status flag, which differ between Correction
and DarkRequest.
Correction can have NA (Not Available), R(unning), A(vailable)
DarkRequest can have F(inished), E(rror), R(equested),
IP (In Progress), T(imeout)
The flag is extracted from the message.
Further informations are available at:
https://git.xfel.eu/gitlab/detectors/pycalibration/wikis/MyMDC-Communication
"""
if message.split(':')[0] in ('FAILED', 'WARN'): # Errors
flag = 'NA' if action == 'correct' else 'E'
elif message.split(':')[0] == 'SUCCESS': # Success
flag = 'R' if action == 'correct' else 'IP'
if 'Uploaded' in message or 'Finished' in message:
flag = 'A' if action == 'correct' else 'F'
else: # MDC Timeout
flag = 'NA' if action == 'correct' else 'T'
if action == 'correct':
func = mdc.update_run_api
data = {'flg_cal_data_status': flag, 'cal_pipeline_reply': message}
if action == 'dark_request':
func = mdc.update_dark_run_api
data = {'dark_run': {'flg_status': flag, 'calcat_feedback': message}}
loop = get_event_loop()
response = await shield(loop.run_in_executor(None, func, rid, data))
if response.status_code != 200:
logging.error(Errors.MDC_RESPONSE.format(response))
async def server_runner(config, mode):
""" The main server loop
The main server loop handles remote requests via a ZMQ interface.
Requests are the form of ZMQ.REQuest and have the format
command, *params
where *parms is a string-encoded python list as defined by the
commands. The following commands are currently understood:
- correct, with parmeters rid, sase, instrument, cycle, proposal, runnr
where
:param rid: is the runid within the MDC database
:param sase: is the sase beamline
:param instrument: is the instrument
:param cycle: is the facility cycle
:param proposal: is the proposal id
:param runnr: is the run number in integer form, e.g. without leading
"r"
This will trigger a correction process to be launched for that run in
the given cycle and proposal.
- dark_request, with parameters rid, sase, instrument, cycle, proposal,
did, operation_mode, pdu_names, karabo_das, runnr
where
:param rid: is the runid within the MDC database
:param sase: is the sase beamline
:param instrument: is the instrument
:param cycle: is the facility cycle
:param proposal: is the proposal id
:param did: is the detector karabo id
:param operation_mode: is the detector's operation mode, as defined in
CalCat
:param pdu_names: physical detector units for each modules
:param karabo_das: the Data Agreggators representing which detector
modules to calibrate
:param runnr: is the run number in integer form, i.e. without leading
"r"
- upload-yaml, with parameters sase, instrument, cycle, proposal, yaml
where
:param sase: is the sase beamline
:param instrument: is the instrument
:param cycle: is the facility cycle
:param proposal: is the proposal id
:param yaml: is url-encoded (quotes and spaces) representation of
new YAML file
This will create or replace the existing YAML configuration for the
proposal and cycle with the newly sent one, and then push it to the git
configuration repo.
"""
init_config_repo(config['config-repo'])
job_db = await init_job_db(config)
mdc = await init_md_client(config)
context = zmq.asyncio.Context()
auth = zmq.auth.thread.ThreadAuthenticator(context)
if mode == "prod-auth":
auth.start()
auth.allow(config['web-service']['allowed-ips'])
socket = context.socket(zmq.REP)
socket.zap_domain = b'global'
socket.bind("{}:{}".format(config['web-service']['bind-to'],
config['web-service']['port']))
while True:
response = await socket.recv_multipart()
if isinstance(response, list) and len(response) == 1:
try: # protect against unparseable requests
response = eval(response[0])
except SyntaxError as e:
logging.error(str(e))
socket.send(Errors.REQUEST_FAILED.encode())
continue
if len(response) < 2: # catch parseable but malformed requests
logging.error(Errors.REQUEST_MALFORMED.format(response))
socket.send(Errors.REQUEST_MALFORMED.format(response).encode())
continue
# FIXME: action should be an enum
action, payload = response[0], response[1:]
if action not in ['correct', 'dark', 'dark_request', 'query-rid',
'upload-yaml', 'update_conf']:
logging.warning(Errors.UNKNOWN_ACTION.format(action))
socket.send(Errors.UNKNOWN_ACTION.format(action).encode())
continue
logging.debug('{}, {}'.format(action, payload))
if action == "query-rid":
rid = payload[0]
await query_rid(job_db, socket, rid)
continue
async def do_action(action, payload): # FIXME: this needn't be nested
in_folder = None
run_mapping = {}
priority = None # TODO: Investigate argument
if action == 'update_conf':
updated_config = None
try:
sase, karabo_id, instrument, cycle, proposal, config_yaml, apply = payload # noqa
updated_config = json.loads(config_yaml)
await change_config(socket, config['config-repo'],
updated_config, karabo_id, instrument,
cycle, proposal,
apply.upper() == "TRUE")
except Exception as e:
e = str(e)
err_msg = (f"Failure applying config for {proposal}:"
f" {e}: {updated_config}")
logging.error(err_msg)
logging.error(f"Unexpected error: {traceback.format_exc()}") # noqa
socket.send(yaml.dump(err_msg,
default_flow_style=False).encode())
if action in ['dark', 'correct', 'dark_request']:
request_time = datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
try:
wait_runs: List[str] = []
rid, sase, instrument, cycle, proposal, *payload = payload
if action == 'correct':
runnr, priority = payload
runnr = runnr.strip('r')
wait_runs = [runnr]
if action == 'dark':
karabo_ids, karabo_das, *runs = payload
karabo_ids = karabo_ids.split(',')
karabo_das = karabo_das.split(',')
for i, run in enumerate(runs):
erun = eval(run)
if isinstance(erun, (list, tuple)):
typ, runnr = erun
if typ == "reservation":
continue
runnr = runnr.strip('r')
run_mapping[typ] = runnr
wait_runs.append(runnr)
else:
run_mapping['no_mapping_{}'.format(i)] = erun
wait_runs.append(erun)
if action == 'dark_request':
karabo_id, operation_mode, *payload = payload
payload = eval(','.join(payload))
pdus, karabo_das, wait_runs = payload
karabo_das = [val.strip() for val in karabo_das]
wait_runs = [str(val) for val in wait_runs]
proposal = proposal.strip('p')
proposal = "{:06d}".format(int(proposal))
logging.info(f'{action} of {proposal} run {wait_runs} at '
f'{instrument} is requested. Checking files.')
# Read calibration configuration from yaml
conf_file = Path(config['config-repo']['local-path'],
cycle, f'{proposal}.yaml')
if not conf_file.exists():
conf_file = Path(config['config-repo']['local-path'],
"default.yaml")
with open(conf_file, "r") as f:
pconf_full = yaml.load(f.read(),
Loader=yaml.FullLoader)
# FIXME: remove once MyMDC sends `dark` action
action_ = 'dark' if action == 'dark_request' else action
data_conf = pconf_full['data-mapping']
if instrument in pconf_full[action_]:
pconf = pconf_full[action_][instrument]
else:
socket.send(Errors.NOT_CONFIGURED.encode())
logging.info(f'Instrument {instrument} is unknown')
return
in_folder = config[action_]['in-folder'].format(
instrument=instrument, cycle=cycle, proposal=proposal)
msg = Success.QUEUED.format(proposal, wait_runs)
socket.send(msg.encode())
logging.debug(msg)
if action in ['correct', 'dark_request']:
await update_mdc_status(mdc, action, rid, msg)
except Exception as e:
e = str(e)
msg = Errors.JOB_LAUNCH_FAILED.format(action, e)
logging.error(msg)
socket.send(msg.encode())
if action in ['correct', 'dark_request']:
await update_mdc_status(mdc, action, rid, msg)
return
# Check if all files for given runs are transferred
all_transfers = []
# FIXME: this loop should be an asyncio.gather
for runnr in wait_runs:
rpath = "{}/r{:04d}/".format(in_folder, int(runnr))
transfer_complete = await wait_on_transfer(rpath)
all_transfers.append(transfer_complete)
if not transfer_complete:
logging.error(
Errors.TRANSFER_EVAL_FAILED.format(proposal,
runnr))
if action in ['correct', 'dark_request']:
await update_mdc_status(mdc, action, rid,
MDC.MIGRATION_TIMEOUT)
if not all(all_transfers):
logging.error(Errors.TRANSFER_EVAL_FAILED.format(proposal, ','.join(wait_runs))) # noqa
return
logging.debug(f"Now doing: {action}")
ts = datetime.now().strftime('%y%m%d_%H%M%S')
if action == 'dark':
detectors = {}
out_folder = config[action]['out-folder'].format(
instrument=instrument, cycle=cycle, proposal=proposal,
runs="_".join(wait_runs))
# Run over all available detectors
if karabo_ids[0] == 'all':
karabo_ids = list(pconf.keys())
# Prepare configs for all requested detectors
for karabo_id in karabo_ids:
# use selected karabo_das
if karabo_das[0] == 'all':
karabo_da = data_conf[karabo_id]["karabo-da"]
# Check if any files for given karabo-das exists
if check_files(in_folder, wait_runs, karabo_da):
thisconf = copy.copy(data_conf[karabo_id])
if (karabo_id in pconf and
isinstance(pconf[karabo_id], dict)):
thisconf.update(copy.copy(pconf[karabo_id]))
thisconf["in-folder"] = in_folder
thisconf["out-folder"] = '/'.join((out_folder,
karabo_id.replace('-', '_'))) # noqa FIXME Make use of pathlib
thisconf["karabo-id"] = karabo_id
thisconf["karabo-da"] = karabo_da
run_config = []
for typ, run in run_mapping.items():
if "no_mapping" in typ:
run_config.append(run)
else:
thisconf[typ] = run
if len(run_config):
thisconf["runs"] = ",".join(run_config)
detectors[karabo_id] = thisconf
else:
logging.warning("File list for {} at {} is empty"
.format(karabo_id,
"{}/*.h5".format(rpath)))
if len(detectors) == 0:
logging.warning(Errors.NOTHING_TO_DO.format(rpath))
if action == 'correct':
try:
runnr = wait_runs[0]
rpath = "{}/r{:04d}/".format(in_folder, int(runnr))
out_folder = config[action]['out-folder'].format(
instrument=instrument, cycle=cycle, proposal=proposal,
run='r{:04d}'.format(int(runnr)))
# Prepare configs for all detectors in run
fl = glob.glob(f"{rpath}/*.h5")
corr_file_list = set()
copy_file_list = set(fl)
detectors = {}
for karabo_id in pconf:
dconfig = data_conf[karabo_id]
# check for files according to mapping in raw run dir.
if any(y in x for x in fl
for y in dconfig['karabo-da']):
for karabo_da in dconfig['karabo-da']:
tfl = glob.glob(f"{rpath}/*{karabo_da}*.h5")
corr_file_list = corr_file_list.union(set(tfl))
thisconf = copy.copy(dconfig)
if isinstance(pconf[karabo_id], dict):
thisconf.update(copy.copy(pconf[karabo_id]))
thisconf["in-folder"] = in_folder
thisconf["out-folder"] = out_folder
thisconf["karabo-id"] = karabo_id
thisconf["run"] = runnr
if priority:
thisconf["priority"] = str(priority)
detectors[karabo_id] = thisconf
copy_file_list = copy_file_list.difference(corr_file_list)
asyncio.ensure_future(copy_untouched_files(copy_file_list,
out_folder,
runnr))
if len(detectors) == 0:
logging.warning(Errors.NOTHING_TO_DO.format(rpath))
await update_mdc_status(mdc, action, rid,
MDC.NOTHING_TO_DO)
return
except Exception as corr_e:
logging.error(f"Error during correction: {str(corr_e)}")
await update_mdc_status(mdc, action, rid,
Errors.REQUEST_FAILED)
if action == 'dark_request':
runs = [str(r) for r in wait_runs]
# Notebooks require one or three runs, depending on the
# detector type and operation mode.
triple = any(det in karabo_id for det in
["LPD", "AGIPD", "JUNGFRAU", "JF", "JNGFR"])
if triple and len(runs) == 1:
runs_dict = {'run-high': runs[0],
'run-med': '0',
'run-low': '0'}
elif triple and len(runs) == 3:
runs_dict = {'run-high': runs[0],
'run-med': runs[1],
'run-low': runs[2]}
else: # single
runs_dict = {'run': runs[0]}
out_folder = config['dark']['out-folder'].format(
instrument=instrument, cycle=cycle, proposal=proposal,
runs='_'.join(runs))
out_folder = str(Path(out_folder,
karabo_id.replace('-', '_')))
# We assume that MyMDC does not allow dark request if the data
# is not migrated, thus skipping some validation here.
thisconf = copy.copy(data_conf[karabo_id])
if (karabo_id in pconf
and isinstance(pconf[karabo_id], dict)):
thisconf.update(copy.copy(pconf[karabo_id]))
thisconf['in-folder'] = in_folder
thisconf['out-folder'] = out_folder
thisconf['karabo-id'] = karabo_id
thisconf['karabo-da'] = karabo_das
thisconf['operation-mode'] = operation_mode
thisconf.update(runs_dict)
detectors = {karabo_id: thisconf}
if action in ['correct', 'dark', 'dark_request']:
# run xfel_calibrate
action_ = 'dark' if action == 'dark_request' else action
for karabo_id, dconfig in detectors.items():
detector = dconfig['detector-type']
del dconfig['detector-type']
cmd = config[action_]['cmd'].format(
detector=detector,
sched_prio=str(config[action_]['sched-prio']),
action=action_, instrument=instrument,
cycle=cycle, proposal=proposal,
runs="_".join([f"r{r}" for r in wait_runs]),
time_stamp=ts,
det_instance=karabo_id,
request_time=request_time
).split()
cmd = parse_config(cmd, dconfig)
rid = rid if action in ['correct', 'dark_request'] else 0
ret = await run_action(job_db, cmd, mode,
proposal, runnr, rid)
if action == 'correct':
await update_mdc_status(mdc, action, rid, ret)
if action == 'dark_request':
await update_mdc_status(mdc, action, rid, ret)
report_idx = cmd.index('--report-to') + 1
report = cmd[report_idx] + '.pdf'
await update_darks_paths(mdc, rid, in_folder,
out_folder, report)
# TODO: moving this block further up reduces the need of so
# many nested ifs. Move up and return directly
if action == 'upload-yaml':
sase, instrument, cycle, proposal, this_yaml = payload
this_yaml = urllib.parse.unquote_plus(this_yaml)
await upload_config(socket, config['config-repo'], this_yaml,
instrument, cycle, proposal)
try:
asyncio.ensure_future(
do_action(copy.copy(action), copy.copy(payload)))
except Exception as e: # actions that fail are only error logged
logging.error(str(e))
parser = argparse.ArgumentParser(
description='Start the calibration webservice')
parser.add_argument('--config-file', type=str, default='./webservice.yaml')
parser.add_argument('--log-file', type=str, default='./web.log')
parser.add_argument('--mode', type=str, default="sim", choices=['sim', 'prod'])
parser.add_argument('--logging', type=str, default="INFO",
choices=['INFO', 'DEBUG', 'ERROR'])
if __name__ == "__main__":
args = vars(parser.parse_args())
conf_file = args["config_file"]
with open(conf_file, "r") as f:
config = yaml.load(f.read(), Loader=yaml.FullLoader)
mode = args["mode"]
logfile = args["log_file"]
fmt = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
logging.basicConfig(filename=logfile,
level=getattr(logging, args['logging']),
format=fmt)
loop = asyncio.get_event_loop()
loop.create_task(update_job_db(config))
loop.run_until_complete(server_runner(config, mode))
loop.close()