python.virtual_calculator Namespace Reference

Functions
Generator[Tuple[RunLumi, RunLumi, Mapping[str, DEFECT_IOV]], None, None]	generate_virtual_defects (Mapping[str, IOVSet] by_channel, Iterable[DefectLogic] logics, Union[int, Tuple[int, int], RunLumi] since_cr, Union[int, Tuple[int, int], RunLumi] until_cr, Optional[Container[str]] ignore)
bool	bad_iov (RunLumi since, RunLumi until)
IOVSet	calculate_virtual_defects (IOVSet primary_iovs, Iterable[DefectLogic] evaluation_order, Iterable[str] virtual_output_channels, Iterable[str] primary_output_channels, Optional[Union[int, Tuple[int, int], RunLumi]] since, Optional[Union[int, Tuple[int, int], RunLumi]] until, Optional[Container[str]] ignore)

Variables
	log

Function Documentation

bad_iov()

bool python.virtual_calculator.bad_iov	(	RunLumi	since,
		RunLumi	until
	)

Skip some commonly emitted nonsensical IoVs.

Definition at line 53 of file virtual_calculator.py.

def bad_iov(since: RunLumi, until: RunLumi) -> bool:
    """
    Skip some commonly emitted nonsensical IoVs.
    """
    return (
        # Skip 1-lb long iovs on run borders
        (until - since == 1 and since.run != until.run) or
        # Skip LB 0->1 IoVs
        (since.lumi == 0 and until.lumi == 1 and since.run == until.run)
    )
 

calculate_virtual_defects()

IOVSet python.virtual_calculator.calculate_virtual_defects	(	IOVSet	primary_iovs,
		Iterable[DefectLogic]	evaluation_order,
		Iterable[str]	virtual_output_channels,
		Iterable[str]	primary_output_channels,
		Optional[Union[int, Tuple[int, int], RunLumi]]	since,
		Optional[Union[int, Tuple[int, int], RunLumi]]	until,
		Optional[Container[str]]	ignore
	)

Returns a list of IoVs for a given query in the normal COOL order
(sorted by channelID, then by since)

Definition at line 64 of file virtual_calculator.py.

def calculate_virtual_defects(primary_iovs: IOVSet, evaluation_order: Iterable[DefectLogic],
                              virtual_output_channels: Iterable[str], primary_output_channels: Iterable[str], 
                              since: Optional[Union[int, Tuple[int, int], RunLumi]], 
                              until: Optional[Union[int, Tuple[int, int], RunLumi]], 
                              ignore: Optional[Container[str]]) -> IOVSet:
    """
    Returns a list of IoVs for a given query in the normal COOL order
    (sorted by channelID, then by since)
    """
    
    if since is None: since = 0
    if until is None: until = 2**63-1
    since, until = RunLumi(since), RunLumi(until)
    
    result = defaultdict(IOVSet)
    
    primary_by_channel = primary_iovs.by_channel
    
    # Copy desired primary channels to the result
    for primary_channel, primary_iovs in primary_by_channel.items():
        if primary_channel in primary_output_channels:
            if ignore is not None and primary_channel in ignore:
                continue
            result[primary_channel] = primary_iovs
    
    args = primary_by_channel, evaluation_order, since, until, ignore
 
    # Skip vfgen loop if there is no virtual logic to compute
    vfgen = [] if not evaluation_order else generate_virtual_defects(*args)
    
    for since, until, virtual_iovs in vfgen:
        if bad_iov(since, until):
            continue
            
        for output_name in virtual_output_channels:
            # Consider the state of each desired output for this 
            # (since, until) and write it to output_iovs
            iov = virtual_iovs.get(output_name)
            if iov and iov.present:
                result[output_name].add(since, until, *iov[2:])
    
    # Sort them by traditional COOL sort ordering (by channelId first, 
    # then by since. `iovs` are already ordered by since.)
    result_list = IOVSet()
    for _, iovs in sorted(result.items()):
        result_list.extend(iovs.solidify(DEFECT_IOV))
        
    return result_list

generate_virtual_defects()

Generator[Tuple[RunLumi, RunLumi, Mapping[str, DEFECT_IOV]], None, None] python.virtual_calculator.generate_virtual_defects	(	Mapping[str, IOVSet]	by_channel,
		Iterable[DefectLogic]	logics,
		Union[int, Tuple[int, int], RunLumi]	since_cr,
		Union[int, Tuple[int, int], RunLumi]	until_cr,
		Optional[Container[str]]	ignore
	)

An iterator which emits (since, until, {channel name : iov})

It also computes the state of the virtual flags for each (since, until)
region.

Definition at line 18 of file virtual_calculator.py.

def generate_virtual_defects(by_channel: Mapping[str, IOVSet], logics: Iterable[DefectLogic], 
                             since_cr: Union[int, Tuple[int, int], RunLumi], 
                             until_cr: Union[int, Tuple[int, int], RunLumi], 
                             ignore: Optional[Container[str]]
                            ) -> Generator[Tuple[RunLumi, RunLumi, Mapping[str, DEFECT_IOV]], None, None]:
    """
    An iterator which emits (since, until, {channel name : iov})
    
    It also computes the state of the virtual flags for each (since, until)
    region.
    """    
    all_channels = list(by_channel.keys()) + [l.name for l in logics]
    states = dict((channel, None) for channel in all_channels)
    
    channels, iovsets = zip(*sorted(by_channel.items()))
        
    for since, until, current_iovs, changes in process_iovs_changed(*iovsets):
        # Update things that changed since last iteration in the states dict
        for change in changes:
            if ignore is not None and channels[change] in ignore:
                continue
            states[channels[change]] = current_iovs[change]
        
        # If we're completely outside the calculation range, don't yield iovs
        if until <= since_cr or since >= until_cr:
            continue
            
        # Update flag calculation result
        for logic in logics:
            if ignore and logic.name in ignore:
                continue
            states[logic.name] = logic.evaluate(states)
        
        yield since, until, states
        

Variable Documentation

log

python.virtual_calculator.log

Definition at line 5 of file virtual_calculator.py.