from __future__ import absolute_import, division, print_function
import NuRadioReco.framework.base_station
import NuRadioReco.framework.channel
import NuRadioReco.framework.sim_channel
import collections
try:
import cPickle as pickle
except ImportError:
import pickle
import logging
logger = logging.getLogger('NuRadioReco.SimStation')
[docs]class SimStation(NuRadioReco.framework.base_station.BaseStation):
def __init__(self, station_id):
NuRadioReco.framework.base_station.BaseStation.__init__(self, station_id)
self.__magnetic_field_vector = None
self.__simulation_weight = None
self.__channels = collections.OrderedDict()
self.__candidate = None
[docs] def set_candidate(self, candidate_status):
"""
Set the candidate for the simulation station. True means the station is a candidate for producing a trigger.
Parameters
----------
candidate_status : bool
If the station is a candidate for producing a trigger.
Returns
-------
None
"""
if not isinstance(candidate_status, bool) and candidate_status is not None:
raise ValueError("The candidate_status must be a bool or None.")
self.__candidate = candidate_status
[docs] def is_candidate(self):
"""
Returns whether the station is a candidate for producing a trigger.
Returns
-------
bool
True if the station is a candidate for producing a trigger, False otherwise.
"""
if self.__candidate is None:
raise ValueError("The candidate status has not been set.")
return self.__candidate
[docs] def get_magnetic_field_vector(self):
return self.__magnetic_field_vector
[docs] def set_magnetic_field_vector(self, magnetic_field_vector):
self.__magnetic_field_vector = magnetic_field_vector
[docs] def get_simulation_weight(self):
return self.__simulation_weight
[docs] def set_simulation_weight(self, simulation_weight):
self.__simulation_weight = simulation_weight
[docs] def iter_channels(self):
for channel in self.__channels.values():
yield channel
[docs] def add_channel(self, channel):
"""
adds a NuRadioReco.framework.sim_channel to the SimStation object
"""
if not isinstance(channel, NuRadioReco.framework.sim_channel.SimChannel):
raise AttributeError("channel needs to be of type NuRadioReco.framework.sim_channel")
if(channel.get_unique_identifier() in self.__channels):
raise AttributeError(f"channel with the unique identifier {channel.get_unique_identifier()} is already present in SimStation")
self.__channels[channel.get_unique_identifier()] = channel
[docs] def get_channel(self, unique_identifier):
"""
returns channel identified by the triple (channel_id, shower_id, ray_tracing_id)
"""
return self.__channels[unique_identifier]
[docs] def get_channel_ids(self):
"""
returns a list with the channel IDs of all simChannels of the simStation
"""
channel_ids = []
for unique_identifier in self.__channels.keys():
if unique_identifier[0] not in channel_ids:
channel_ids.append(unique_identifier[0])
channel_ids.sort()
return channel_ids
[docs] def get_shower_ids(self):
"""
returns a list with the shower IDs of all simChannels of the simStation
"""
shower_ids = []
for unique_identifier in self.__channels.keys():
if unique_identifier[1] not in shower_ids:
shower_ids.append(unique_identifier[1])
shower_ids.sort()
return shower_ids
[docs] def get_ray_tracing_ids(self):
"""
returns a list with the raytracing IDs of all simChannels of the simStation
"""
ray_tracing_ids = []
for unique_identifier in self.__channels.keys():
if unique_identifier[2] not in ray_tracing_ids:
ray_tracing_ids.append(unique_identifier[2])
ray_tracing_ids.sort()
return ray_tracing_ids
[docs] def get_channels_by_channel_id(self, channel_id):
"""
returns all simChannels that have the given channel_id
"""
for channel in self.__channels.values():
if channel.get_id() == channel_id:
yield channel
[docs] def get_channels_by_shower_id(self, shower_id):
"""
returns all simChannels that have the given shower_id
"""
for channel in self.__channels.values():
if channel.get_shower_id() == shower_id:
yield channel
[docs] def get_channels_by_ray_tracing_id(self, ray_tracing_id):
"""
returns all simChannels that have the given ray_tracing_id
"""
for channel in self.__channels.values():
if channel.get_ray_tracing_solution_id() == ray_tracing_id:
yield channel
[docs] def serialize(self, save_channel_traces, save_efield_traces):
base_station_pkl = NuRadioReco.framework.base_station.BaseStation.serialize(self, save_efield_traces=save_efield_traces)
channels_pkl = []
for channel in self.iter_channels():
channels_pkl.append(channel.serialize(save_trace=save_channel_traces))
data = {'__magnetic_field_vector': self.__magnetic_field_vector,
'__simulation_weight': self.__simulation_weight,
'channels': channels_pkl,
'base_station': base_station_pkl}
return pickle.dumps(data, protocol=4)
[docs] def deserialize(self, data_pkl):
data = pickle.loads(data_pkl)
NuRadioReco.framework.base_station.BaseStation.deserialize(self, data['base_station'])
self.__magnetic_field_vector = data['__magnetic_field_vector']
self.__simulation_weight = data['__simulation_weight']
if 'channels' in data.keys():
for channel_pkl in data['channels']:
channel = NuRadioReco.framework.sim_channel.SimChannel(0, 0, 0)
channel.deserialize(channel_pkl)
self.add_channel(channel)
def __add__(self, x):
"""
adds a SimStation object to another SimStation object
WARNING: Only channel and efield objects are added but no other meta information
"""
if not isinstance(x, SimStation):
raise AttributeError("Can only add SimStation to SimStation")
if self.get_id() != x.get_id():
raise AttributeError("Can only add SimStations with the same ID")
for channel in x.iter_channels():
if channel.get_unique_identifier() in self.__channels:
raise AttributeError(f"Channel with ID {channel.get_unique_identifier()} already present in SimStation")
self.add_channel(channel)
efield_ids = self.get_electric_field_ids()
for efield in x.get_electric_fields():
if efield.get_unique_identifier() in efield_ids:
raise AttributeError(f"Electric field with unique identifier {efield.get_unique_identifier()} already present in SimStation")
self.add_electric_field(efield)
return self