pandaserver/dataservice/event_lookup_client_ei.py

0001 """
0002 EventLookupClientEI is a class for looking up events in the EventIndex.
0003 """
0004 import os
0005 import subprocess
0006 import tempfile
0007
0008 from typing import List, Tuple
0009
0010
0011 class EventLookupClientEI:
0012     """
0013     EventLookupClientEI is a class for looking up events in the EventIndex.
0014     """
0015     def do_lookup(self, event_run_list: List[Tuple[int, int]], stream: str = None, tokens: str = None, ami_tag: str = None) -> Tuple[List[str], str, str, str]:
0016         """
0017         Performs a lookup in the EventIndex for the given parameters.
0018
0019         Parameters:
0020             event_run_list (List[Tuple[int, int]]): The list of run events.
0021             stream (str): The name of the stream.
0022             tokens (str): The tokens.
0023             ami_tag (str): The AMI tag.
0024
0025         Returns:
0026             Tuple[List[str], str, str, str]: A tuple containing the list of GUIDs, the command, the output, and the error.
0027         """
0028         command = os.path.join(
0029             os.getenv(
0030                 "EIDIR",
0031                 "/cvmfs/atlas.cern.ch/repo/ATLASLocalRootBase/x86_64/EIClient/current",
0032             ),
0033             "bin",
0034             "event-lookup",
0035         )
0036         with tempfile.NamedTemporaryFile(mode="w+t") as tmp_event_file:
0037             command += f" -F {tmp_event_file.name} "
0038             for run_event in event_run_list:
0039                 # creating a preformatted string. The :08d and :09d parts are formatting these integers to be eight and nine digits long
0040                 tmp_string = f"{int(run_event[0]):08d} {int(run_event[1]):09d}\n"
0041                 tmp_event_file.write(tmp_string)
0042             tmp_event_file.flush()
0043             if stream not in [None, ""]:
0044                 command += f"-s {stream} "
0045             if ami_tag not in [None, ""]:
0046                 command += f"-a {ami_tag} "
0047             command += "-c plain "
0048             with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True,
0049                                   universal_newlines=True) as execute_process:
0050                 tmp_output, tmp_error = execute_process.communicate()
0051             guids = {}
0052             if tokens == "":
0053                 tokens = None
0054             try:
0055                 # The output is expected to be a string with multiple lines, each line containing information about a single event.
0056                 # Each line is split into items, the first two items are interpreted as integers representing a run number and an event number.
0057                 # These two numbers are combined into a tuple, run_event, which is used as a key in the guids dictionary.
0058                 # The fourth item in the line is appended to the string "Stream" to form tmp_token, and the third item is assigned to tmp_guid.
0059                 # An example input (tmp_output) is:
0060                 # 123 456 789abc StreamXYZ
0061                 # An example output (guids dictionary) is:
0062                 # (123, 456): {"789abc"}
0063                 for tmp_line in tmp_output.split("\n"):
0064                     tmp_items = tmp_line.split()
0065                     run_event = (int(tmp_items[0]), int(tmp_items[1]))
0066                     guids.setdefault(run_event, set())
0067                     # check type
0068                     tmp_token = "Stream" + tmp_items[3]
0069                     tmp_guid = tmp_items[2]
0070                     if not tokens or tokens == tmp_token:
0071                         guids[run_event].add(tmp_guid)
0072                 if not guids:
0073                     # add dummy
0074                     guids[None] = None
0075             except Exception:
0076                 pass
0077             return guids, command, tmp_output, tmp_error