Source code for src.experiment_prototype.interface_classes.interface_class_base


This is the base module for all InterfaceClassBase types (iterable for an experiment given certain
parameters). These types include the Scan class, the AveragingPeriod class, and the Sequence

:copyright: 2018 SuperDARN Canada
:author: Marci Detwiller

# built-in
import inspect
import itertools
from pathlib import Path

# third-party
import structlog

# local
from experiment_prototype.experiment_exception import ExperimentException

# Obtain the module name that imported this log_config
caller = Path(inspect.stack()[-1].filename)
module_name =".")[0]
log = structlog.getLogger(module_name)

[docs]class InterfaceClassBase(object): """ The base class for the classes Scan, AveragingPeriod, and Sequence. Scans are made up of AveragingPeriods, these are typically a 3sec time of the same pulse sequence pointing in one direction. AveragingPeriods are made up of Sequences, typically the same sequence run ave. 20-30 times after a clear frequency search. Sequences are made up of pulses, which is a list of dictionaries where each dictionary describes a pulse. :param object_keys: slice_ids that need to be included in this interface_class_base type. :type object_keys: list :param object_slice_dict: the slice dictionary that explains the parameters of each slice that is included in this interface_class_base type. Keys are the slice_ids included and values are dictionaries including all necessary slice parameters as keys. :type object_slice_dict: dict :param object_interface: the interfacing dictionary that describes how to interface the slices that are included in this interface_class_base type. Keys are tuples of format (slice_id_1, slice_id_2) and values are of interface_types set up in experiment_prototype. :type object_interface: dict :param transmit_metadata: a dictionary of the experiment-wide transmit metadata for building pulse sequences. The keys of the transmit_metadata are: - 'output_rx_rate' [Hz], - 'main_antenna_count', - 'intf_antenna_count', - 'tr_window_time' [s], - 'main_antenna_spacing' [m], - 'intf_antenna_spacing' [m], - 'pulse_ramp_time' [s], - 'max_usrp_dac_amplitude' [V peak], - 'rx_sample_rate' [Hz], - 'min_pulse_separation' [us], - 'txrate' [Hz], - 'intf_offset' [m,m,m], - 'dm_rate' :type transmit_metadata: dict """ def __init__( self, object_keys, object_slice_dict, object_interface, transmit_metadata ): # list of slice_ids included in this interface_class_base self.slice_ids = object_keys # dictionary (key = slice_id) of dictionaries (value = slice parameters) self.slice_dict = object_slice_dict # interfacing dictionary (key = (slice_id_1, slice_id_2), value = one of # interface_types) self.interface = object_interface # The nested slice list is filled in a child class before the prep_for_nested_interface_class # function is run. This list is of format [[], [], ...] where the length of the outer list # is equal to the number of the lower interface_class_base instance within the instance of the # higher interface_class_base ( ex. number of sequences within averagingperiods) self.nested_slice_list = [] # all necessary experiment-wide transmit metadata self.transmit_metadata = transmit_metadata # List of lists, each inner list is all slice ids that share a scan self.nested_slice_list = self.get_nested_slice_ids()
[docs] def prep_for_nested_interface_class(self): """ Retrieve the params needed for the nested class (also with base InterfaceClassBase). This class reduces duplicate code by breaking down the InterfaceClassBase class into the separate portions for the nested instances. For Scan class, the nested class is AveragingPeriod, and we will need to break down the parameters given to the Scan instance because there may be multiple AveragingPeriods within. For AveragingPeriod, the nested class is Sequence. :returns: params for the nested class's instantiation. :rtype: list """ # TODO documentation make a detailed example of this and diagram nested_class_param_lists = [] log.debug(self.nested_slice_list) for slice_list in self.nested_slice_list: slices_for_nested_class = {} for slice_id in slice_list: try: slices_for_nested_class[slice_id] = self.slice_dict[slice_id] except KeyError: errmsg = ( f"Error with slice list - slice id {slice_id} cannot be found." ) raise ExperimentException(errmsg) # now take a subset of the interface dictionary that applies to this nested object # of interface_class_base type. nested_class_interface = {} for i in itertools.combinations(slice_list, 2): # slice_list is sorted so we should have the following effect: # combinations([1, 3, 5], 2) --> [1,3], [1,5], [3,5] nested_class_interface[tuple(i)] = self.interface[tuple(i)] nested_class_param_lists.append( [ slice_list, slices_for_nested_class, nested_class_interface, self.transmit_metadata, ] ) return nested_class_param_lists
[docs] @staticmethod def slice_combos_sorter(list_of_combos, all_keys): """ Sort keys of a list of combinations so that keys only appear once in the list. This function modifies the input list_of_combos so that all associated slices are in the same list. For example, if input is list_of_combos = [[0,1], [0,2], [0,4], [1,4], [2,4]] and all_keys = [0,1,2,4,5] then the output should be [[0,1,2,4], [5]]. This is used to get the slice dictionary for nested class instances. In the above example, we would then have two instances of the nested class to create: one with slices 0,1,2,4 and another with slice 5. :param list_of_combos: list of lists of length two associating two slices together. :type list_of_combos: list :param all_keys: list of all keys included in this object (scan, ave_period, or sequence). :type all_keys: list :returns: list of combos that is sorted so that each key only appears once and the lists within the list are of however long necessary :rtype: list """ disjoint_sets = [] # Go through all interfaces and create sets of mutually-interfaced slices for slice_interfacing in list_of_combos: slice_already_interfaced_with = False for disjoint_set in disjoint_sets: # Add both slice ids to this set if either one is already a member if ( slice_interfacing[0] in disjoint_set or slice_interfacing[1] in disjoint_set ): disjoint_set.update({slice_interfacing[0], slice_interfacing[1]}) slice_already_interfaced_with = True if not slice_already_interfaced_with: # Create a new set with just these two disjoint_sets.append({slice_interfacing[0], slice_interfacing[1]}) # Check all slice ids and add any missing ones to their own set. for key in all_keys: slice_in_set = False for disjoint_set in disjoint_sets: if key in disjoint_set: # This slice interfaces with others slice_in_set = True break if ( not slice_in_set ): # This slice doesn't interface with the others, make a new set disjoint_sets.append({key}) # Go through all sets and make sure they are disjoint (don't share any slice ids) for i in range(len(disjoint_sets)): for j in range(i + 1, len(disjoint_sets)): bad_slices = disjoint_sets[i].intersection(disjoint_sets[j]) if len(bad_slices) != 0: raise ExperimentException( f"The following slices do not interface well with other slices: " f"{bad_slices}" ) disjoint_sets[i] = sorted(list(disjoint_sets[i])) # Convert to a list disjoint_sets.sort(key=lambda x: x[0]) # Sort by the first key in each list return disjoint_sets
[docs] def get_nested_slice_ids(self): """ Organize the slice_ids by interface. This method is inherited by child classes and organizes all slices in each child class which should be combined by the class. For example, all slices in a Scan should be combined if they share an AveragingPeriod or Sequence or are concurrent. Returns a list of lists where each inner list contains the slices that are combined inside this object. e.g. for InterfaceClassBase: len(nested_slice_list) = # of scans in this experiment, len(nested_slice_list[0]) = # of slices in the first scan :returns: A list that has one element per scan. Each element is a list of slice_ids signifying which slices are combined inside that scan. The list returned could be of length 1, meaning only one scan is present in the experiment. :rtype: list of lists """ nested_combos = [] combine_below_dict = { "InterfaceClassBase": [ "AVEPERIOD", "SEQUENCE", "CONCURRENT", ], # Combine everything except SCAN interfaced "Scan": [ "SEQUENCE", "CONCURRENT", ], # Combine everything SEQUENCE or CONCURRENT interfaced "AveragingPeriod": [ "CONCURRENT" ], # Combine everything CONCURRENT interfaced "Sequence": [], # All slices in a Sequence are already CONCURRENT and should be combined already } combine_list = combine_below_dict[ type(self).__name__ ] # Returns the class name of the calling instance for slice_ids_combo, interface_value in self.interface.items(): if interface_value in combine_list: nested_combos.append(list(slice_ids_combo)) combos = self.slice_combos_sorter(nested_combos, self.slice_ids) return combos