Source code for NuRadioReco.detector.RNO_G.rnog_detector_mod

from NuRadioReco.utilities import fft
from NuRadioReco.detector.response import Response

from NuRadioReco.detector.RNO_G.rnog_detector import \
    Detector, _keys_not_in_dict, _check_detector_time

import copy
import logging
import numpy as np


[docs] def replace_value_in_dict(d, keys, value): """ Replaces the value of a nested dict entry. Examples -------- .. code-block:: d = {1: {2: {3: 1, 4: 2}}} replace_value_in_dict(d, [1, 2, 4], 14) print(d) # {1: {2: {3: 1, 4: 14}}} """ if isinstance(keys, str): keys = [keys] d_tmp = d while True: key = keys.pop(0) # get the first key if not len(keys): d_tmp[key] = value break else: d_tmp = d_tmp[key]
[docs] class ModDetector(Detector): def __init__(self, *args, **kwargs): super(ModDetector, self).__init__(*args, **kwargs) self.logger = logging.getLogger("NuRadioReco.RNOGDetector.Mod")
[docs] def modify_channel_description(self, station_id, channel_id, keys, value): """ This function allows you to replace/modifty the description of a channel. Parameters ---------- station_id: int The station id channel_id: int The channel id keys: list of str The list of keys of the corresponding part of the description to be changed value: various types The value of the description to be changed """ if not self.has_station(station_id): err = f"Station id {station_id} not commission at {self.get_detector_time()}" self.logger.error(err) raise ValueError(err) channel_dict = self._Detector__get_channel(station_id, channel_id, with_position=True, with_signal_chain=True) if _keys_not_in_dict(channel_dict, keys): # to simplify the code here all keys have to exist already raise KeyError( f"Could not find {keys} for station.channel {station_id}.{channel_id}.") replace_value_in_dict(channel_dict, keys, value)
[docs] @_check_detector_time def get_channel(self, station_id, channel_id): """ Returns a dictionary of all channel parameters Parameters ---------- station_id: int The station id channel_id: int The channel id Returns ------- channel_info: dict Dictionary of channel parameters """ self.get_signal_chain_response(station_id, channel_id) # this adds `total_response` to dict channel_data = copy.deepcopy(self._Detector__get_channel(station_id, channel_id, with_position=True, with_signal_chain=True)) for key in self._Detector__default_values: # In this class overwritting is valid if isinstance(self._Detector__default_values[key], dict): channel_data[key] = self._Detector__default_values[key][channel_id] else: channel_data[key] = self._Detector__default_values[key] return channel_data
[docs] def modify_station_description(self, station_id, keys, value): """ This function allows you to replace/modifty the description of a channel. Parameters ---------- station_id: int The station id keys: list of str The list of keys of the corresponding part of the description to be changed value: various types The value of the description to be changed """ station_data = self.get_station(station_id) if _keys_not_in_dict(station_data, keys): # to simplify the code here all keys have to exist already raise KeyError( f"Could not find {keys} for station {station_id}.") replace_value_in_dict(station_data, keys, value)
[docs] def add_response(self, station_id, channel_id, response): """ Add an additional response to the `total_response` Parameters ---------- station_id: int The station id channel_id: int The channel id response: response.Response A response object to be added to the `total_response` """ orig_response = self.get_signal_chain_response(station_id, channel_id) signal_chain_dict = self.get_channel_signal_chain(station_id, channel_id) # modify the total response signal_chain_dict["total_response"] = orig_response * response # write the modified signal chain back to the buffered station self._Detector__buffered_stations[station_id]["channels"][channel_id]['signal_chain'] = signal_chain_dict
[docs] def add_component(self, station_id, channel_id, component): """ Add an additional component to the `response_chain` and the corresponding response to the `total_response` Parameters ---------- station_id: int The station id channel_id: int The channel id componennt: dict A dictionary with the properties of the component to be added """ # generate a response object from the component dict component_response = Response( **component, station_id=station_id, channel_id=channel_id ) orig_response = self.get_signal_chain_response(station_id, channel_id) signal_chain_dict = self.get_channel_signal_chain(station_id, channel_id) # modify the signal chain signal_chain_dict["total_response"] = orig_response * component_response signal_chain_dict["response_chain"][component['name']] = component # write the modified signal chain back to the buffered station self._Detector__buffered_stations[station_id]["channels"][channel_id]['signal_chain'] = signal_chain_dict
[docs] def add_manual_time_delay(self, station_id, channel_id, time_delay, weight=1, name="MOD_manual_time_delay"): """ Add an additional time delay to the signal chain and total response Parameters ---------- station_id: int The station id channel_id: int The channel id time_delay: float The manual time delay to be added weight: float (default: 1) The weight of the time delay in the total response (either 1 or -1) name: str (default: "MOD_manual_time_delay") The name of the component to be added """ if time_delay < 0.0: raise ValueError("Expect positive additional delay; use 'weight = -1' to implement a negative delay.") sampling_rate = self.get_sampling_frequency(station_id, channel_id) # number of samples a trace would have with a length at least that of the time delay n_samples = int(time_delay * 2.0 * sampling_rate) + 1 freqs = fft.freqs(n_samples, sampling_rate) # pseudo data phase = -2 * np.pi * time_delay * freqs mag = np.ones_like(phase) component = { 'weight': weight, 'y_unit': ['mag', 'rad'], 'y': [list(mag), list(phase)], 'frequency': list(freqs), 'name': name, 'time_delay': time_delay } # add the component to the response chain self.add_component(station_id, channel_id, component)
[docs] def set_channel_position(self, station_id, channel_id, value): """ Set the relative position of a channel. Parameters ---------- station_id: int The station id channel_id: int The channel id value: array/list of float The relative position of the channel """ channel_info = self._Detector__get_channel( station_id, channel_id, with_position=True) channel_info["channel_position"]['position'] = list(value)
[docs] def set_device_position(self, station_id, device_id, value): """ Set the relative position of a device. Parameters ---------- station_id: int The station id device_id: int The device id value: array/list of float The relative position of the channel """ self.modify_station_description(station_id, ["devices", device_id, "device_position", "position"], list(value))
if __name__ == "__main__": det = ModDetector(select_stations=11) import datetime det.update(datetime.datetime(2023, 1, 1, 0, 0, 0)) resp = det.get_signal_chain_response(11, 0) print(resp.get_time_delay(), resp._calculate_time_delay()) det.add_manual_time_delay(11, 0, 250) resp = det.get_signal_chain_response(11, 0) print(resp.get_time_delay(), resp._calculate_time_delay())