Monitors / Recordings

Create Monitors

CompNeuroPy provides a CompNeuroMonitors class that can be used to easily create and control multiple ANNarchy monitors at once. To create a CompNeuroMonitors object, all that is needed is a monitors_dictionary that defines which variables should be recorded for each model component. All populations and projections have to have unique names to work with CompNeuroMonitors. The keys of the monitor_dictionary are the names of the model components (in example below "my_pop1" and "my_pop2"). The key can also include a recording period (the time between two recordings, given after a ";"), e.g. record the variables of my_pop1 only every 10 ms would look like this: 'pop;my_pop1;10':['v', 'spike']. The default period is the time step of the simulation for populations and 1000 times the timestep for projections. The values of the monitor_dictionary are lists of all the variables that should be recorded from the corresponding components. The names of components (populations, projections) could be provided by a CompNeuroModel.

Example:

Here the variables v and spike should be recorded of the population with the name "my_pop1" and the variable v should be recorded from the population with the name "my_pop2":

from CompNeuroPy import CompNeuroMonitors
monitor_dictionary = {'my_pop1':['v', 'spike'], 'my_pop2':['v']}
mon = CompNeuroMonitors(monitor_dictionary)

A full example is available in the Examples.

Chunks and periods

In CompNeuroPy, recordings are divided into so-called chunks and periods. Chunks are simulation sections that are separated by monitor resets (optionally also reset the model). A chunk can consist of several periods. A period represents the time span between the start and pause of a monitor recording. To divide a simulation into chunks and periods, a CompNeuroMonitors object provides the three functions start(), pause() and reset().

At the beginning of a simulation, the monitors do not start automatically which is why the start() function must be called at least once. The start() function can also be used to resume paused recordings. With the function pause() recordings are paused. The function reset() starts a new chunk for the recordings (the end of a chunk is also always the end of a period, i.e. the last period of the corresponding chunk). After calling reset() the monitors remain in their current mode (active or paused). By default reset() also resets the model to the compile status (time = 0) by calling the ANNarchy reset() function and has the same arguments. If the argument model is set to False, the ANNarchy reset() function is not called and only a new chunk is created.

Example:

### first chunk, one period
simulate(100) # 100 ms not recorded
mon.start()   # start all monitors
simulate(100) # 100 ms recorded

### second chunk, two periods
mon.reset()   # model reset, beginning of new chunk
simulate(100) # 100 ms recorded (monitors were active before reset --> still active)
mon.pause()   # pause all monitors
simulate(100) # 100 ms not recorded
mon.start()   # start all monitors
simulate(100) # 100 ms recorded

Get recordings

The recordings can be obtained from the CompNeuroMonitors object using the get_recordings() function. This returns a list of dictionaries (one for each chunk). The dictionaries contain the recorded data defined with the monitor_dictionary at the CompNeuroMonitors initialization. In the recordings dictionaries the keys have the following structure: "<component_name>;variable"; the corresponding dictionary values are the recordings of the respective variable. The dictionaries always contain the time step of the simulation (key = "dt"), the periods (time between recorded values) for each component (key = "<component_name>;period") and the attributes of each component (key = "<component_name>;parameter_dict").

Example:

recordings = mon.get_recordings()
y1 = recordings[0]['my_pop1;v'] ### variable v of my_pop1 from 1st chunk
y2 = recordings[1]['my_pop1;v'] ### variable v of my_pop1 from 2nd chunk

Get recording times

In addition to the recordings themselves, recording times can also be obtained from the CompNeuroMonitors object, which is very useful for later analyses. With the function get_recording_times() of the CompNeuroMonitors object a RecordingTimes object can be obtained. From the RecordingTimes object one can get time limits (in ms) and coresponding indizes for the recordings.

Example:

recording_times = mon.get_recording_times()
start_time = recording_times.time_lims(chunk=1, period=1)[0] ### 200 ms
start_idx  = recording_times.idx_lims(chunk=1, period=1)[0]  ### 1000, if dt == 0.1
end_time   = recording_times.time_lims(chunk=1, period=1)[1] ### 300 ms
end_idx    = recording_times.idx_lims(chunk=1, period=1)[1]  ### 2000

You can combine the recordings of both chunks of the example simulation shown above into a single time array and a single value array using the RecordingTimes object's combine_chunks function:

time_arr, value_arr = recording_times.combine_chunks(recordings, 'my_pop1;v', 'consecutive')

Plot recordings

To get a quick overview of the recordings, CompNeuroPy provides the PlotRecordings class.

CompNeuroPy.monitors.CompNeuroMonitors

Class to bring together ANNarchy monitors into one object.

Source code in src/CompNeuroPy/monitors.py

class CompNeuroMonitors:
    """
    Class to bring together ANNarchy monitors into one object.
    """

    def __init__(self, mon_dict={}):
        """
        Initialize CompNeuroMonitors object by creating ANNarchy monitors.

        Args:
            mon_dict (dict):
                dict with key="compartment_name;period" where period is optional and
                val=list with variables to record.
        """
        self.mon = self._add_monitors(mon_dict)
        self.mon_dict = mon_dict
        self._init_internals(init_call=True)

    def _init_internals(self, init_call=False):
        """
        Initialize the following internal variables:
            - timings (dict):
                dict with key="pop_name" for populations and "proj_name" for projections
                for each recorded population and projection and
                val={"currently_paused": True, "start": [], "stop": []}
            - recordings (list):
                list with recordings of all chunks. Set to empty list.
            - recording_times (list):
                list with recording times of all chunks. Set to empty list.
            - already_got_recordings (bool):
                True if recordings were already requested, False otherwise. Set to
                False.
            - already_got_recording_times (bool):
                True if recording_times were already requested, False otherwise. Set to
                False.
            - get_recordings_reset_call (bool):
                True if get_recordings() and get_recording_times() are called within
                reset(), False otherwise. Set to False.

        Args:
            init_call (bool, optional):
                True if called from __init__(), False otherwise. Default: False.
        """
        if init_call is False:
            #### pause all ANNarchy monitors because currently paused will be set to False
            self.pause()

        ### initialize timings
        timings = {}
        for key, val in self.mon_dict.items():
            _, compartment, _ = self._unpack_mon_dict_keys(key)
            timings[compartment] = {"currently_paused": True, "start": [], "stop": []}
        self.timings = timings

        ### initialize recordings and recording_times etc.
        self.recordings = []
        self.recording_times = []
        self.already_got_recordings = False
        self.already_got_recording_times = False
        self.get_recordings_reset_call = False

    @check_types()
    def start(self, compartment_list: list | None = None):
        """
        Start or resume recording of all recorded compartments in compartment_list.

        Args:
            compartment_list (list, optional):
                List with compartment names to start or resume recording. Default: None,
                i.e., all compartments of initialized mon_dict are started or resumed.
        """
        if compartment_list == None:
            mon_dict_key_list = list(self.mon_dict.keys())
            compartment_list = [
                self._unpack_mon_dict_keys(key)[1] for key in mon_dict_key_list
            ]

        self.timings = self._start_monitors(compartment_list, self.mon, self.timings)

    @check_types()
    def pause(self, compartment_list: list | None = None):
        """
        Pause recording of all recorded compartments in compartment_list.

        Args:
            compartment_list (list, optional):
                List with compartment names to pause recording. Default: None,
                i.e., all compartments of initialized mon_dict are paused.
        """
        if compartment_list == None:
            mon_dict_key_list = list(self.mon_dict.keys())
            compartment_list = [
                self._unpack_mon_dict_keys(key)[1] for key in mon_dict_key_list
            ]

        self.timings = self._pause_monitors(compartment_list, self.mon, self.timings)

    def reset(
        self,
        populations=True,
        projections=False,
        synapses=False,
        monitors=True,
        model=True,
        parameters=True,
        net_id=0,
    ):
        """
        Create a new recording chunk by getting recordings and recording times of the
        current chunk and optionally resetting the model. Recordings are automatically
        resumed in the new chunk if they are not paused.

        Args:
            populations (bool, optional):
                If True, reset populations. Default: True.
            projections (bool, optional):
                If True, reset projections. Default: False.
            synapses (bool, optional):
                If True, reset synapses. Default: False.
            monitors (bool, optional):
                If True, reset ANNarchy monitors. Default: True.
            model (bool, optional):
                If True, reset model. Default: True.
            parameters (bool, optional):
                If True, reset the parameters of popilations and projections. Default:
                True.
            net_id (int, optional):
                Id of the network to reset. Default: 0.
        """
        ### TODO rename this function to new_chunk() or something like that and let
        ### recordings and recording times be returned
        self.get_recordings_reset_call = True
        self.get_recordings()
        self.get_recording_times()
        self.get_recordings_reset_call = False
        self.already_got_recordings = (
            False  # after reset one can still update recordings
        )
        self.already_got_recording_times = (
            False  # after reset one can still update recording_times
        )

        ### reset timings, after reset, add a zero to start if the monitor is still
        ### running (this is not resetted by reset())
        ### if the model was not resetted --> do add current time instead of zero
        for key in self.timings.keys():
            self.timings[key]["start"] = []
            self.timings[key]["stop"] = []
            if self.timings[key]["currently_paused"] == False:
                if model:
                    self.timings[key]["start"].append(0)
                else:
                    self.timings[key]["start"].append(
                        np.round(get_time(), af.get_number_of_decimals(dt()))
                    )

        ### reset model
        if model:
            if parameters is False:
                ### if parameters=False, get parameters before reset and set them after
                ### reset
                parameters_dict = mf._get_all_parameters()
            reset(populations, projections, synapses, monitors, net_id=net_id)
            if parameters is False:
                ### if parameters=False, set parameters after reset
                mf._set_all_parameters(parameters_dict)

    def current_chunk(self):
        """
        Get the index of the current chunk.

        Returns:
            current_chunk_idx (int):
                Index of the current chunk. If no recordings are currently active,
                returns None.
        """
        ### if recordings are currently active --> return chunk in which these recordings will be saved
        ### check if there are currently active recordings
        active_recordings = False
        for key, val in self.mon_dict.items():
            _, compartment, _ = self._unpack_mon_dict_keys(key)
            if not (self.timings[compartment]["currently_paused"]):
                ### tere are currently active recordings
                active_recordings = True

        if active_recordings:
            current_chunk_idx = len(self.recordings)
            return current_chunk_idx
        else:
            ### if currently no recordings are active return None
            return None

    def get_recordings(self) -> list[dict]:
        """
        Get recordings of all recorded compartments.

        Returns:
            recordings (list):
                List with recordings of all chunks.
        """
        ### only if recordings in current chunk and get_recodings was not already called add current chunk to recordings
        if (
            self._any_recordings_in_current_chunk()
            and self.already_got_recordings is False
        ):
            ### update recordings
            self.recordings.append(self._get_monitors(self.mon_dict, self.mon))
            ### upade already_got_recordings --> it will not update recordings again
            self.already_got_recordings = True

            if not (self.get_recordings_reset_call):
                if len(self.recordings) == 0:
                    print(
                        "WARNING get_recordings: no recordings available, empty list returned. Maybe forgot start()?"
                    )
            return self.recordings
        else:
            if not (self.get_recordings_reset_call):
                if len(self.recordings) == 0:
                    print(
                        "WARNING get_recordings: no recordings available, empty list returned. Maybe forgot start()?"
                    )
            return self.recordings

    def get_recording_times(self):
        """
        Get recording times of all recorded compartments.

        Returns:
            recording_times (recording_times_cl):
                Object with recording times of all chunks.
        """

        temp_timings = self._get_temp_timings()

        ### only append temp_timings of current chunk if there are recordings in current chunk at all and if get_recordings was not already called (double call would add the same chunk again)
        if (
            self._any_recordings_in_current_chunk()
            and self.already_got_recording_times is False
        ):
            self.recording_times.append(temp_timings)

        ### upade already_got_recording_times --> it will not update recording_times again
        self.already_got_recording_times = True

        ### generate a object from recording_times and return this instead of the dict
        recording_times_ob = RecordingTimes(self.recording_times)

        if not (self.get_recordings_reset_call):
            if len(self.recording_times) == 0:
                print(
                    "WARNING get_recording_times: no recordings available, empty list returned. Maybe forgot start()?"
                )
        return recording_times_ob

    def get_recordings_and_clear(self):
        """
        The default get_recordings method should be called at the end of the simulation.
        The get_recordings_and_clear method allows to get several times recordings with
        the same monitor object and to simulate between the calls. Sets the internal
        variables back to their initial state. Usefull if you repeat a simulation +
        recording several times and you do not want to always create new chunks.

        !!! warning
            If you want to continue recording after calling this method, you have to
            call start() again.

        Returns:
            recordings (list):
                List with recordings of all chunks.
            recording_times (recording_times_cl):
                Object with recording times of all chunks.
        """
        ret0 = self.get_recordings()
        ret1 = self.get_recording_times()
        self._init_internals()
        ret = (ret0, ret1)
        return ret

    def _correct_start_stop(self, start_time_arr, stop_time_arr, period):
        """
        Corrects the start and stop times of recordings to the actual start and stop
        times of recorded values.

        Args:
            start_time_arr (np.array):
                Array with start times of recordings, obtained with get_time() function
                of ANNarchy.
            stop_time_arr (np.array):
                Array with stop times of recordings, obtained with get_time() function
                of ANNarchy.
            period (float):
                Time difference between recording values specified by the user.

        Returns:
            actual_start_time (np.array):
                Array with actual start times of recorded values.
            actual_stop_time (np.array):
                Array with actual stop times of recorded values.
            nr_rec_vals (np.array):
                Array with number of recorded values between start and stop.
        """
        # actual_period = int(period / dt()) * dt()
        actual_start_time = np.ceil(start_time_arr / period) * period

        actual_stop_time = np.ceil(stop_time_arr / period - 1) * period

        nr_rec_vals = 1 + (actual_stop_time - actual_start_time) / period

        return (actual_start_time, actual_stop_time, nr_rec_vals)

    def _get_temp_timings(self):
        """
        Generates a timings dictionary with time lims and idx lims for each compartment.
        Calculates the idx lims of the recordings based on the time lims.

        Returns:
            temp_timings (dict):
                Dict with time lims and idx lims for each compartment.
        """
        temp_timings = {}
        for key in self.mon_dict.keys():
            _, compartment, period = self._unpack_mon_dict_keys(key)
            if len(self.timings[compartment]["start"]) > len(
                self.timings[compartment]["stop"]
            ):
                ### was started/resumed but never stoped after --> use current time for stop time
                self.timings[compartment]["stop"].append(get_time())
            ### calculate the idx of the recorded arrays which correspond to the timings and remove 'currently_paused'
            ### get for each start-stop pair the corrected start stop timings (when teh values were actually recorded, depends on period and timestep)
            ### and also get the number of recorded values for start-stop pair
            start_time_arr = np.array(self.timings[compartment]["start"])
            stop_time_arr = np.array(self.timings[compartment]["stop"])
            (
                start_time_arr,
                stop_time_arr,
                nr_rec_vals_arr,
            ) = self._correct_start_stop(start_time_arr, stop_time_arr, period)

            ### with the number of recorded values -> get start and end idx for each start-stop pair
            start_idx = [
                np.sum(nr_rec_vals_arr[0:i]).astype(int)
                for i in range(nr_rec_vals_arr.size)
            ]
            stop_idx = [
                np.sum(nr_rec_vals_arr[0 : i + 1]).astype(int) - 1
                for i in range(nr_rec_vals_arr.size)
            ]

            ### return start-stop pair info in timings format
            temp_timings[compartment] = {
                "start": {
                    "ms": np.round(
                        start_time_arr, af.get_number_of_decimals(dt())
                    ).tolist(),
                    "idx": start_idx,
                },
                "stop": {
                    "ms": np.round(
                        stop_time_arr, af.get_number_of_decimals(dt())
                    ).tolist(),
                    "idx": stop_idx,
                },
            }
        return temp_timings

    def _any_recordings_in_current_chunk(self):
        """
        Check if there are any recordings in the current chunk.

        Returns:
            any_recordings (bool):
                True if there are any recordings in the current chunk, False otherwise.
        """
        temp_timings = self._get_temp_timings()

        ### generate a temp object of temp timings to check if there were recordings at all
        recording_times_ob_temp = RecordingTimes([temp_timings])
        return recording_times_ob_temp._any_recordings(chunk=0)

    def _add_monitors(self, mon_dict: dict):
        """
        Generate monitors defined by mon_dict.

        Args:
            mon_dict (dict):
                dict with key="compartment_name;period" where period is optional and
                val=list with variables to record.

        Returns:
            mon (dict):
                dict with key="pop_name" for populations and key="proj_name" for
                projections and val=ANNarchy monitor object.
        """
        mon = {}
        for key, val in mon_dict.items():
            compartmentType, compartment, period = self._unpack_mon_dict_keys(
                key, warning=True
            )
            ### check if compartment is pop
            if compartmentType == "pop":
                mon[compartment] = Monitor(
                    get_population(compartment), val, start=False, period=period
                )
            ### check if compartment is proj
            if compartmentType == "proj":
                mon[compartment] = Monitor(
                    get_projection(compartment), val, start=False, period=period
                )
        return mon

    def _start_monitors(self, compartment_list, mon, timings=None):
        """
        Starts or resumes monitores defined by compartment_list.

        Args:
            compartment_list (list):
                List with compartment names to start or resume recording.
            mon (dict):
                Dict with key="pop_name" for populations and key="proj_name" for
                projections and val=ANNarchy monitor object.
            timings (dict, optional):
                timings variable of the CompNeuroMonitors object. Default: None.

        Returns:
            timings (dict):
                timings variable of the CompNeuroMonitors object.
        """
        ### for each compartment generate started variable (because compartments can ocure multiple times if multiple variables of them are recorded --> do not start same monitor multiple times)
        started = {}
        for compartment_name in compartment_list:
            started[compartment_name] = False

        if timings == None:
            ### information about pauses not available, just start
            for compartment_name in compartment_list:
                if started[compartment_name] == False:
                    mon[compartment_name].start()
                    print("start", compartment_name)
                    started[compartment_name] = True
            return None
        else:
            ### information about pauses available, start if not paused, resume if paused
            for compartment_name in compartment_list:
                if started[compartment_name] == False:
                    if timings[compartment_name]["currently_paused"]:
                        if len(timings[compartment_name]["start"]) > 0:
                            ### resume
                            mon[compartment_name].resume()
                        else:
                            ### initial start
                            mon[compartment_name].start()
                    started[compartment_name] = True
                    ### update currently_paused
                    timings[compartment_name]["currently_paused"] = False
                    ### never make start longer than stop+1!... this can be caused if start is called multiple times without pause in between
                    if len(timings[compartment_name]["start"]) <= len(
                        timings[compartment_name]["stop"]
                    ):
                        timings[compartment_name]["start"].append(get_time())
            return timings

    def _pause_monitors(self, compartment_list, mon, timings=None):
        """
        Pause monitores defined by compartment_list.

        Args:
            compartment_list (list):
                List with compartment names to pause recording.
            mon (dict):
                Dict with key="pop_name" for populations and key="proj_name" for
                projections and val=ANNarchy monitor object.
            timings (dict, optional):
                timings variable of the CompNeuroMonitors object. Default: None.

        Returns:
            timings (dict):
                timings variable of the CompNeuroMonitors object.
        """
        ### for each compartment generate paused variable (because compartments can ocure multiple times if multiple variables of them are recorded --> do not pause same monitor multiple times)
        paused = {}
        for compartment_name in compartment_list:
            paused[compartment_name] = False

        for compartment_name in compartment_list:
            if paused[compartment_name] == False:
                mon[compartment_name].pause()
                paused[compartment_name] = True

        if timings != None:
            ### information about pauses is available, update it
            for key, val in paused.items():
                timings[key]["currently_paused"] = True
                ### never make pause longer than start, this can be caused if pause is called multiple times without start in between
                if len(timings[key]["stop"]) < len(timings[key]["start"]):
                    timings[key]["stop"].append(get_time())
                ### if pause is directly called after start --> start == stop --> remove these entries, this is no actual period
                if (
                    len(timings[key]["stop"]) == len(timings[key]["start"])
                    and timings[key]["stop"][-1] == timings[key]["start"][-1]
                ):
                    timings[key]["stop"] = timings[key]["stop"][:-1]
                    timings[key]["start"] = timings[key]["start"][:-1]
            return timings
        else:
            return None

    def _get_monitors(self, mon_dict, mon):
        """
        Get recorded values from ANNarchy monitors defined by mon_dict.

        Args:
            mon_dict (dict):
                dict with key="compartment_name;period" where period is optional and
                val=list with variables to record.
            mon (dict):
                Dict with key="pop_name" for populations and key="proj_name" for
                projections and val=ANNarchy monitor object.

        Returns:
            recordings (dict):
                Dict with key="compartment_name;variable" and val=list with recorded
                values.
        """
        recordings = {}
        for key, val in mon_dict.items():
            compartment_type, compartment, period = self._unpack_mon_dict_keys(key)
            recordings[f"{compartment};period"] = period
            if compartment_type == "pop":
                pop = get_population(compartment)
                parameter_dict = {
                    param_name: getattr(pop, param_name)
                    for param_name in pop.parameters
                }
                recordings[f"{compartment};parameter_dict"] = parameter_dict
            if compartment_type == "proj":
                proj = get_projection(compartment)
                parameter_dict = {
                    param_name: getattr(proj, param_name)
                    for param_name in proj.parameters
                }
                recordings[f"{compartment};parameters"] = parameter_dict
            for val_val in val:
                temp = mon[compartment].get(val_val)
                recordings[f"{compartment};{val_val}"] = temp
        recordings["dt"] = dt()
        return recordings

    def _unpack_mon_dict_keys(self, s: str, warning: bool = False):
        """
        Unpacks a string of the form "compartment_name;period" or
        "compartment_name" into its components. If period is not provided
        it is set to dt() for populations and dt()*1000 for projections.

        Args:
            s (str):
                String to be unpacked
            warning (bool, optional):
                If True, print warning if period is not provided for projections.

        Returns:
            compartment_type (str):
                Compartment type
            compartment_name (str):
                Compartment name
            period (float):
                Period of the compartment
        """
        ### split string
        splitted_s = s.split(";")

        ### get name
        compartment_name = splitted_s[0]

        ### get type
        pop_list = [pop.name for pop in populations()]
        proj_list = [proj.name for proj in projections()]
        if compartment_name in pop_list and compartment_name in proj_list:
            ### raise error because name is in both lists
            print(
                "ERROR CompNeuroMonitors._unpack_mon_dict_keys(): compartment_name is both populaiton and projection"
            )
            quit()
        elif compartment_name in pop_list:
            compartment_type = "pop"
        elif compartment_name in proj_list:
            compartment_type = "proj"

        ### get period
        if len(splitted_s) == 2:
            period = float(splitted_s[1])
        else:
            period = {"pop": dt(), "proj": dt() * 1000}[compartment_type]
            ### print warning for compartment_type proj
            if compartment_type == "proj" and warning:
                print(
                    f"WARNING CompNeuroMonitors: no period provided for projection {compartment_name}, period set to {period} ms"
                )
        period = round(period / dt()) * dt()

        return compartment_type, compartment_name, period

__init__(mon_dict={})

Initialize CompNeuroMonitors object by creating ANNarchy monitors.

Parameters:

Name	Type	Description	Default
mon_dict	dict	dict with key="compartment_name;period" where period is optional and val=list with variables to record.	{}

Source code in src/CompNeuroPy/monitors.py

def __init__(self, mon_dict={}):
    """
    Initialize CompNeuroMonitors object by creating ANNarchy monitors.

    Args:
        mon_dict (dict):
            dict with key="compartment_name;period" where period is optional and
            val=list with variables to record.
    """
    self.mon = self._add_monitors(mon_dict)
    self.mon_dict = mon_dict
    self._init_internals(init_call=True)

start(compartment_list=None)

Start or resume recording of all recorded compartments in compartment_list.

Parameters:

Name	Type	Description	Default
compartment_list	list	List with compartment names to start or resume recording. Default: None, i.e., all compartments of initialized mon_dict are started or resumed.	None

Source code in src/CompNeuroPy/monitors.py

@check_types()
def start(self, compartment_list: list | None = None):
    """
    Start or resume recording of all recorded compartments in compartment_list.

    Args:
        compartment_list (list, optional):
            List with compartment names to start or resume recording. Default: None,
            i.e., all compartments of initialized mon_dict are started or resumed.
    """
    if compartment_list == None:
        mon_dict_key_list = list(self.mon_dict.keys())
        compartment_list = [
            self._unpack_mon_dict_keys(key)[1] for key in mon_dict_key_list
        ]

    self.timings = self._start_monitors(compartment_list, self.mon, self.timings)

pause(compartment_list=None)

Pause recording of all recorded compartments in compartment_list.

Parameters:

Name	Type	Description	Default
compartment_list	list	List with compartment names to pause recording. Default: None, i.e., all compartments of initialized mon_dict are paused.	None

Source code in src/CompNeuroPy/monitors.py

@check_types()
def pause(self, compartment_list: list | None = None):
    """
    Pause recording of all recorded compartments in compartment_list.

    Args:
        compartment_list (list, optional):
            List with compartment names to pause recording. Default: None,
            i.e., all compartments of initialized mon_dict are paused.
    """
    if compartment_list == None:
        mon_dict_key_list = list(self.mon_dict.keys())
        compartment_list = [
            self._unpack_mon_dict_keys(key)[1] for key in mon_dict_key_list
        ]

    self.timings = self._pause_monitors(compartment_list, self.mon, self.timings)

reset(populations=True, projections=False, synapses=False, monitors=True, model=True, parameters=True, net_id=0)

Create a new recording chunk by getting recordings and recording times of the current chunk and optionally resetting the model. Recordings are automatically resumed in the new chunk if they are not paused.

Parameters:

Name	Type	Description	Default
populations	bool	If True, reset populations. Default: True.	True
projections	bool	If True, reset projections. Default: False.	False
synapses	bool	If True, reset synapses. Default: False.	False
monitors	bool	If True, reset ANNarchy monitors. Default: True.	True
model	bool	If True, reset model. Default: True.	True
parameters	bool	If True, reset the parameters of popilations and projections. Default: True.	True
net_id	int	Id of the network to reset. Default: 0.	0

Source code in src/CompNeuroPy/monitors.py

def reset(
    self,
    populations=True,
    projections=False,
    synapses=False,
    monitors=True,
    model=True,
    parameters=True,
    net_id=0,
):
    """
    Create a new recording chunk by getting recordings and recording times of the
    current chunk and optionally resetting the model. Recordings are automatically
    resumed in the new chunk if they are not paused.

    Args:
        populations (bool, optional):
            If True, reset populations. Default: True.
        projections (bool, optional):
            If True, reset projections. Default: False.
        synapses (bool, optional):
            If True, reset synapses. Default: False.
        monitors (bool, optional):
            If True, reset ANNarchy monitors. Default: True.
        model (bool, optional):
            If True, reset model. Default: True.
        parameters (bool, optional):
            If True, reset the parameters of popilations and projections. Default:
            True.
        net_id (int, optional):
            Id of the network to reset. Default: 0.
    """
    ### TODO rename this function to new_chunk() or something like that and let
    ### recordings and recording times be returned
    self.get_recordings_reset_call = True
    self.get_recordings()
    self.get_recording_times()
    self.get_recordings_reset_call = False
    self.already_got_recordings = (
        False  # after reset one can still update recordings
    )
    self.already_got_recording_times = (
        False  # after reset one can still update recording_times
    )

    ### reset timings, after reset, add a zero to start if the monitor is still
    ### running (this is not resetted by reset())
    ### if the model was not resetted --> do add current time instead of zero
    for key in self.timings.keys():
        self.timings[key]["start"] = []
        self.timings[key]["stop"] = []
        if self.timings[key]["currently_paused"] == False:
            if model:
                self.timings[key]["start"].append(0)
            else:
                self.timings[key]["start"].append(
                    np.round(get_time(), af.get_number_of_decimals(dt()))
                )

    ### reset model
    if model:
        if parameters is False:
            ### if parameters=False, get parameters before reset and set them after
            ### reset
            parameters_dict = mf._get_all_parameters()
        reset(populations, projections, synapses, monitors, net_id=net_id)
        if parameters is False:
            ### if parameters=False, set parameters after reset
            mf._set_all_parameters(parameters_dict)

current_chunk()

Get the index of the current chunk.

Returns:

Name	Type	Description
current_chunk_idx	int	Index of the current chunk. If no recordings are currently active, returns None.

Source code in src/CompNeuroPy/monitors.py

def current_chunk(self):
    """
    Get the index of the current chunk.

    Returns:
        current_chunk_idx (int):
            Index of the current chunk. If no recordings are currently active,
            returns None.
    """
    ### if recordings are currently active --> return chunk in which these recordings will be saved
    ### check if there are currently active recordings
    active_recordings = False
    for key, val in self.mon_dict.items():
        _, compartment, _ = self._unpack_mon_dict_keys(key)
        if not (self.timings[compartment]["currently_paused"]):
            ### tere are currently active recordings
            active_recordings = True

    if active_recordings:
        current_chunk_idx = len(self.recordings)
        return current_chunk_idx
    else:
        ### if currently no recordings are active return None
        return None

get_recordings()

Get recordings of all recorded compartments.

Returns:

Name	Type	Description
recordings	list	List with recordings of all chunks.

Source code in src/CompNeuroPy/monitors.py

def get_recordings(self) -> list[dict]:
    """
    Get recordings of all recorded compartments.

    Returns:
        recordings (list):
            List with recordings of all chunks.
    """
    ### only if recordings in current chunk and get_recodings was not already called add current chunk to recordings
    if (
        self._any_recordings_in_current_chunk()
        and self.already_got_recordings is False
    ):
        ### update recordings
        self.recordings.append(self._get_monitors(self.mon_dict, self.mon))
        ### upade already_got_recordings --> it will not update recordings again
        self.already_got_recordings = True

        if not (self.get_recordings_reset_call):
            if len(self.recordings) == 0:
                print(
                    "WARNING get_recordings: no recordings available, empty list returned. Maybe forgot start()?"
                )
        return self.recordings
    else:
        if not (self.get_recordings_reset_call):
            if len(self.recordings) == 0:
                print(
                    "WARNING get_recordings: no recordings available, empty list returned. Maybe forgot start()?"
                )
        return self.recordings

get_recording_times()

Get recording times of all recorded compartments.

Returns:

Name	Type	Description
recording_times	recording_times_cl	Object with recording times of all chunks.

Source code in src/CompNeuroPy/monitors.py

def get_recording_times(self):
    """
    Get recording times of all recorded compartments.

    Returns:
        recording_times (recording_times_cl):
            Object with recording times of all chunks.
    """

    temp_timings = self._get_temp_timings()

    ### only append temp_timings of current chunk if there are recordings in current chunk at all and if get_recordings was not already called (double call would add the same chunk again)
    if (
        self._any_recordings_in_current_chunk()
        and self.already_got_recording_times is False
    ):
        self.recording_times.append(temp_timings)

    ### upade already_got_recording_times --> it will not update recording_times again
    self.already_got_recording_times = True

    ### generate a object from recording_times and return this instead of the dict
    recording_times_ob = RecordingTimes(self.recording_times)

    if not (self.get_recordings_reset_call):
        if len(self.recording_times) == 0:
            print(
                "WARNING get_recording_times: no recordings available, empty list returned. Maybe forgot start()?"
            )
    return recording_times_ob

get_recordings_and_clear()

The default get_recordings method should be called at the end of the simulation. The get_recordings_and_clear method allows to get several times recordings with the same monitor object and to simulate between the calls. Sets the internal variables back to their initial state. Usefull if you repeat a simulation + recording several times and you do not want to always create new chunks.

Warning

If you want to continue recording after calling this method, you have to call start() again.

Returns:

Name	Type	Description
recordings	list	List with recordings of all chunks.
recording_times	recording_times_cl	Object with recording times of all chunks.

Source code in src/CompNeuroPy/monitors.py

def get_recordings_and_clear(self):
    """
    The default get_recordings method should be called at the end of the simulation.
    The get_recordings_and_clear method allows to get several times recordings with
    the same monitor object and to simulate between the calls. Sets the internal
    variables back to their initial state. Usefull if you repeat a simulation +
    recording several times and you do not want to always create new chunks.

    !!! warning
        If you want to continue recording after calling this method, you have to
        call start() again.

    Returns:
        recordings (list):
            List with recordings of all chunks.
        recording_times (recording_times_cl):
            Object with recording times of all chunks.
    """
    ret0 = self.get_recordings()
    ret1 = self.get_recording_times()
    self._init_internals()
    ret = (ret0, ret1)
    return ret

CompNeuroPy.monitors.RecordingTimes

Source code in src/CompNeuroPy/monitors.py

class RecordingTimes:
    def __init__(self, recording_times_list):
        """
        Initialize RecordingTimes object.

        Args:
            recording_times_list (list):
                List with recording times of all chunks.
        """
        self.recording_times_list = recording_times_list

    def time_lims(
        self,
        chunk: int | None = None,
        compartment: str | None = None,
        period: int | None = None,
    ):
        """
        Get the time limits recordings of of a specified chunk/model compartment in ms.

        chunk (int, optional):
            Index of the chunk. Default: None, i.e., first chunk.
        compartment (str, optional):
            Name of the compartment. Default: None, i.e., first model compartment from
            monitor.
        period (int, optional):
            Index of the period. Default: None, i.e., all periods.

        Returns:
            lims (tuple):
                Tuple with start and stop time of the specified chunk/model compartment.
        """
        assert (
            len(self.recording_times_list) > 0
        ), "ERROR time_lims(): No recordings/recording_times available."
        return self._lims("ms", chunk, compartment, period)

    def idx_lims(
        self,
        chunk: int | None = None,
        compartment: str | None = None,
        period: int | None = None,
    ):
        """
        Get the index limits of recordings of a specified chunk/model compartment.

        chunk (int, optional):
            Index of the chunk. Default: None, i.e., first chunk.
        compartment (str, optional):
            Name of the compartment. Default: None, i.e., first model compartment from
            monitor.
        period (int, optional):
            Index of the period. Default: None, i.e., all periods.

        Returns:
            lims (tuple):
                Tuple with start and stop index of the specified chunk/model
                compartment.
        """
        assert (
            len(self.recording_times_list) > 0
        ), "ERROR idx_lims(): No recordings/recording_times available."
        return self._lims("idx", chunk, compartment, period)

    def all(self):
        """
        Get the recording times of all chunks, compartments, periods in ms and index.

        Returns:
            recording_times_list (list):
                List with recording times of all chunks.
        """
        return self.recording_times_list

    def nr_periods(self, chunk=None, compartment=None):
        """
        Get the number of recording periods (start-pause) of a specified chunk/model
        compartment.

        Args:
            chunk (int, optional):
                Index of the chunk. Default: None, i.e., first chunk.
            compartment (str, optional):
                Name of the compartment. Default: None, i.e., first model compartment
                from monitor.

        Returns:
            nr_periods (int):
                Number of recording periods (start-pause) of a specified chunk/model
                compartment.
        """
        chunk = self._check_chunk(chunk)
        compartment = self.__check_compartment__(compartment, chunk)
        return self._get_nr_periods(chunk, compartment)

    def combine_periods(
        self,
        recordings: list,
        recording_data_str: str,
        chunk: int = 0,
        fill: str | float = "none",
    ):
        """
        Combines the data of all periods of a specified chunk/model compartment.

        Args:
            recordings (list):
                List with recordings of all chunks.
            recording_data_str (str):
                String specifying the compartment name and the variable to combine.
                Format: "compartment_name;variable_name"
            chunk (int, optional):
                Index of the chunk. Default: 0.
            fill (str|float, optional):
                How recording pauses should be filled. Can be "none" (no filling),
                "nan" (fill with nan), "interpolate" (interpolate with nearest) or a
                float (fill with this value). Default: "none".

        Returns:
            time_arr (np.array):
                Array with time values in ms.
            data_arr (np.array):
                Array with the recorded variable.
        """
        compartment = recording_data_str.split(";")[0]
        period_time = recordings[0][f"{compartment};period"]
        nr_periods = self._get_nr_periods(chunk, compartment)
        time_step = recordings[0]["dt"]
        time_list = []

        ### get data arr
        data_arr = recordings[chunk][recording_data_str]

        ### get time arr
        for period in range(nr_periods):
            start_time, end_time = self.time_lims(
                chunk=chunk, compartment=compartment, period=period
            )
            start_time = round(start_time, af.get_number_of_decimals(time_step))
            end_time = round(end_time, af.get_number_of_decimals(time_step))
            times = np.arange(start_time, end_time + period_time / 2, period_time)
            time_list.append(times)
        time_arr = np.concatenate(time_list, 0)

        ### fill gaps with nan or interpolate
        if fill == "nan":
            time_arr, data_arr = af.time_data_add_nan(
                time_arr,
                data_arr,
                fill_time_step=period_time,
            )
        elif fill == "interpolate":
            full_time_arr = np.arange(
                time_arr[0], time_arr[-1] + period_time / 2, period_time
            )
            f = interp1d(time_arr, data_arr, kind="nearest", axis=0)
            data_arr = f(full_time_arr)
            time_arr = full_time_arr
        elif isinstance(fill, (int, float)):
            time_arr, data_arr = af.time_data_fill_gaps(
                time_arr,
                data_arr,
                fill_time_step=period_time,
                fill=fill,
            )

        return time_arr, data_arr

    def combine_chunks(
        self, recordings: list, recording_data_str: str, mode="sequential"
    ):
        """
        Combines the data of all chunks of recordings. Time periods without recordings
        AT THE BEGINNING OF chunks or WITHIN chunks (i.e. recording pauses) are
        considered and filled with nan values.

        !!! warning
            If you use mode="consecutive": Missing recordings at the end of chunks
            (simulated but not recorded) are not considered, this leads to times which
            differ from the original simulation times (these time periods without
            recording are simply ignored)!

        Args:
            recordings (list):
                List with recordings of all chunks.
            recording_data_str (str):
                String specifying the compartment name and the variable to combine.
                Format: "compartment_name;variable_name"
            mode (str, optional):
                How should the time array be generated. Can be "sequential" or
                "consecutive". Default: "sequential".
                - "sequential": each chunk starts at zero e.g.: [0,100] + [0,250] -->
                    [0, 1, ..., 100, 0, 1, ..., 250]
                - "consecutive": each chunk starts at the last stop time of the previous
                    chunk e.g.: [0,100] + [0,250] --> [0, 1, ..., 100, 101, 102, ..., 350]

        Returns:
            time_arr (np.array):
                Array with time values in ms.
            data_arr (np.array):
                Array with the recorded variable.
        """
        assert (
            len(self.recording_times_list) > 0
        ), "ERROR combine_chunks(): No recordings/recording_times available."

        compartment = recording_data_str.split(";")[0]
        period_time = recordings[0][f"{compartment};period"]
        time_step = recordings[0]["dt"]
        nr_chunks = self._get_nr_chunks()
        data_list = []
        time_list = []
        pre_chunk_start_time = 0

        for chunk in range(nr_chunks):
            ### append data list with data of all periods of this chunk
            data_list.append(recordings[chunk][recording_data_str])

            ### nr of periods in this chunk
            nr_periods = self._get_nr_periods(chunk, compartment)

            ### start time of chunk depends on mode
            if mode == "sequential":
                chunk_start_time = 0
            elif mode == "consecutive":
                if chunk == 0:
                    chunk_start_time = 0
                else:
                    ### TODO this uses the wrong end time if there is a pause at the end
                    ### of the last chunk but it is not possible to get the info how long
                    ### nothing was recorded at the end of a chunk (all I save is from
                    ### when to when something is recorded)
                    last_stop_time = self.recording_times_list[chunk - 1][compartment][
                        "stop"
                    ]["ms"][-1]
                    chunk_start_time = (
                        pre_chunk_start_time + last_stop_time + period_time
                    )
                    pre_chunk_start_time = chunk_start_time
            else:
                print("ERROR recording_times.combine_data, Wrong mode.")
                quit()

            ### append the time list with all times of the periods
            for period in range(nr_periods):
                start_time = (
                    self.time_lims(chunk=chunk, compartment=compartment, period=period)[
                        0
                    ]
                    + chunk_start_time
                )
                end_time = (
                    self.time_lims(chunk=chunk, compartment=compartment, period=period)[
                        1
                    ]
                    + chunk_start_time
                )
                start_time = round(start_time, af.get_number_of_decimals(time_step))
                end_time = round(end_time, af.get_number_of_decimals(time_step))
                times = np.arange(start_time, end_time + period_time / 2, period_time)
                time_list.append(times)

        ### flatten the two lists
        data_arr = np.concatenate(data_list, 0)
        time_arr = np.concatenate(time_list, 0)

        ### check if there are gaps in the time array
        ### fill them with the corersponding times and
        ### the data array with nan values
        time_arr, data_arr = af.time_data_add_nan(
            time_arr,
            data_arr,
            fill_time_step=period_time,
        )

        return time_arr, data_arr

    def _lims(self, string, chunk=None, compartment=None, period=None):
        """
        Get the limits of recordings of a specified chunk/model compartment.

        Args:
            string (str):
                String specifying the type of limits to return. Can be "ms" for time
                limits in ms or "idx" for index limits.
            chunk (int, optional):
                Index of the chunk. Default: None, i.e., first chunk.
            compartment (str, optional):
                Name of the compartment. Default: None, i.e., first model compartment
                from monitor.
            period (int, optional):
                Index of the period. Default: None, i.e., all periods.

        Returns:
            lims (tuple):
                Tuple with start and stop time/index of the specified chunk/model
                compartment.
        """

        chunk = self._check_chunk(chunk)
        compartment = self.__check_compartment__(compartment, chunk)
        period_0, period_1 = self._check_period(period, chunk, compartment)
        lims = (
            self.recording_times_list[chunk][compartment]["start"][string][period_0],
            self.recording_times_list[chunk][compartment]["stop"][string][period_1],
        )
        return lims

    def __check_compartment__(self, compartment, chunk):
        if compartment == None:
            ### by default just use the first compartment
            compartment = list(self.recording_times_list[chunk].keys())[0]
        elif compartment in list(self.recording_times_list[chunk].keys()):
            compartment = compartment
        else:
            print(
                'ERROR recording_times, given compartment "'
                + str(compartment)
                + '" not available'
            )
            quit()

        return compartment

    def _check_period(self, period, chunk, compartment):
        """
        Check if period is given.

        Args:
            period (int, optional):
                Index of the period. Default: None, i.e., all periods.
            chunk (int):
                Index of the chunk.
            compartment (str):
                Name of the compartment.

        Returns:
            period_0 (int):
                Index of the first period.
            period_1 (int):
                Index of the last period. If perios is given, period_0 == period_1.
        """
        if period == None:
            ### by default use all periods
            period_0 = 0
            period_1 = (
                len(self.recording_times_list[chunk][compartment]["start"]["idx"]) - 1
            )
        elif period < len(
            self.recording_times_list[chunk][compartment]["start"]["idx"]
        ):
            period_0 = period
            period_1 = period
        else:
            print("ERROR recording_times, given period not available")
            quit()

        return period_0, period_1

    def _check_chunk(self, chunk):
        """
        Check if chunk is given.

        Args:
            chunk (int, optional):
                Index of the chunk. Default: None, i.e., first chunk.

        Returns:
            chunk (int):
                Index of the chunk.
        """
        if chunk is None:
            ### by default use first chunk
            chunk = 0
        elif chunk < self._get_nr_chunks():
            chunk = chunk
        else:
            print("ERROR recording_times, given chunk not available")
            quit()

        return chunk

    def _get_nr_chunks(self):
        """
        Get the number of chunks of the recordings.

        Returns:
            nr_chunks (int):
                Number of chunks.
        """
        return len(self.recording_times_list)

    def _get_nr_periods(self, chunk, compartment):
        """
        Get the number of recording periods (start-pause) of a specified chunk/model
        compartment.

        Args:
            chunk (int):
                Index of the chunk.
            compartment (str):
                Name of the compartment.

        Returns:
            nr_periods (int):
                Number of recording periods (start-pause) of a specified chunk/model
                compartment.
        """
        return len(self.recording_times_list[chunk][compartment]["start"]["idx"])

    def _any_recordings(self, chunk):
        """
        Check all periods and compartments if there are any recordings.

        Args:
            chunk (int):
                Index of the chunk.

        Returns:
            found_recordings (bool):
                True if there are any recordings in the chunk, False otherwise.
        """
        compartment_list = list(self.recording_times_list[chunk].keys())
        found_recordings = False
        for compartment in compartment_list:
            nr_periods_of_compartment = len(
                self.recording_times_list[chunk][compartment]["start"]["idx"]
            )

            for period_idx in range(nr_periods_of_compartment):
                idx_lims = self.idx_lims(
                    chunk=chunk, compartment=compartment, period=period_idx
                )
                if np.diff(idx_lims)[0] > 0:
                    found_recordings = True

        return found_recordings

__init__(recording_times_list)

Initialize RecordingTimes object.

Parameters:

Name	Type	Description	Default
recording_times_list	list	List with recording times of all chunks.	required

Source code in src/CompNeuroPy/monitors.py

def __init__(self, recording_times_list):
    """
    Initialize RecordingTimes object.

    Args:
        recording_times_list (list):
            List with recording times of all chunks.
    """
    self.recording_times_list = recording_times_list

time_lims(chunk=None, compartment=None, period=None)

Get the time limits recordings of of a specified chunk/model compartment in ms.

chunk (int, optional): Index of the chunk. Default: None, i.e., first chunk. compartment (str, optional): Name of the compartment. Default: None, i.e., first model compartment from monitor. period (int, optional): Index of the period. Default: None, i.e., all periods.

Returns:

Name	Type	Description
lims	tuple	Tuple with start and stop time of the specified chunk/model compartment.

Source code in src/CompNeuroPy/monitors.py

def time_lims(
    self,
    chunk: int | None = None,
    compartment: str | None = None,
    period: int | None = None,
):
    """
    Get the time limits recordings of of a specified chunk/model compartment in ms.

    chunk (int, optional):
        Index of the chunk. Default: None, i.e., first chunk.
    compartment (str, optional):
        Name of the compartment. Default: None, i.e., first model compartment from
        monitor.
    period (int, optional):
        Index of the period. Default: None, i.e., all periods.

    Returns:
        lims (tuple):
            Tuple with start and stop time of the specified chunk/model compartment.
    """
    assert (
        len(self.recording_times_list) > 0
    ), "ERROR time_lims(): No recordings/recording_times available."
    return self._lims("ms", chunk, compartment, period)

idx_lims(chunk=None, compartment=None, period=None)

Get the index limits of recordings of a specified chunk/model compartment.

chunk (int, optional): Index of the chunk. Default: None, i.e., first chunk. compartment (str, optional): Name of the compartment. Default: None, i.e., first model compartment from monitor. period (int, optional): Index of the period. Default: None, i.e., all periods.

Returns:

Name	Type	Description
lims	tuple	Tuple with start and stop index of the specified chunk/model compartment.

Source code in src/CompNeuroPy/monitors.py

def idx_lims(
    self,
    chunk: int | None = None,
    compartment: str | None = None,
    period: int | None = None,
):
    """
    Get the index limits of recordings of a specified chunk/model compartment.

    chunk (int, optional):
        Index of the chunk. Default: None, i.e., first chunk.
    compartment (str, optional):
        Name of the compartment. Default: None, i.e., first model compartment from
        monitor.
    period (int, optional):
        Index of the period. Default: None, i.e., all periods.

    Returns:
        lims (tuple):
            Tuple with start and stop index of the specified chunk/model
            compartment.
    """
    assert (
        len(self.recording_times_list) > 0
    ), "ERROR idx_lims(): No recordings/recording_times available."
    return self._lims("idx", chunk, compartment, period)

all()

Get the recording times of all chunks, compartments, periods in ms and index.

Returns:

Name	Type	Description
recording_times_list	list	List with recording times of all chunks.

Source code in src/CompNeuroPy/monitors.py

def all(self):
    """
    Get the recording times of all chunks, compartments, periods in ms and index.

    Returns:
        recording_times_list (list):
            List with recording times of all chunks.
    """
    return self.recording_times_list

nr_periods(chunk=None, compartment=None)

Get the number of recording periods (start-pause) of a specified chunk/model compartment.

Parameters:

Name	Type	Description	Default
chunk	int	Index of the chunk. Default: None, i.e., first chunk.	None
compartment	str	Name of the compartment. Default: None, i.e., first model compartment from monitor.	None

Returns:

Name	Type	Description
nr_periods	int	Number of recording periods (start-pause) of a specified chunk/model compartment.

Source code in src/CompNeuroPy/monitors.py

def nr_periods(self, chunk=None, compartment=None):
    """
    Get the number of recording periods (start-pause) of a specified chunk/model
    compartment.

    Args:
        chunk (int, optional):
            Index of the chunk. Default: None, i.e., first chunk.
        compartment (str, optional):
            Name of the compartment. Default: None, i.e., first model compartment
            from monitor.

    Returns:
        nr_periods (int):
            Number of recording periods (start-pause) of a specified chunk/model
            compartment.
    """
    chunk = self._check_chunk(chunk)
    compartment = self.__check_compartment__(compartment, chunk)
    return self._get_nr_periods(chunk, compartment)

combine_periods(recordings, recording_data_str, chunk=0, fill='none')

Combines the data of all periods of a specified chunk/model compartment.

Parameters:

Name	Type	Description	Default
recordings	list	List with recordings of all chunks.	required
recording_data_str	str	String specifying the compartment name and the variable to combine. Format: "compartment_name;variable_name"	required
chunk	int	Index of the chunk. Default: 0.	0
fill	str | float	How recording pauses should be filled. Can be "none" (no filling), "nan" (fill with nan), "interpolate" (interpolate with nearest) or a float (fill with this value). Default: "none".	'none'

Returns:

Name	Type	Description
time_arr	array	Array with time values in ms.
data_arr	array	Array with the recorded variable.

Source code in src/CompNeuroPy/monitors.py

def combine_periods(
    self,
    recordings: list,
    recording_data_str: str,
    chunk: int = 0,
    fill: str | float = "none",
):
    """
    Combines the data of all periods of a specified chunk/model compartment.

    Args:
        recordings (list):
            List with recordings of all chunks.
        recording_data_str (str):
            String specifying the compartment name and the variable to combine.
            Format: "compartment_name;variable_name"
        chunk (int, optional):
            Index of the chunk. Default: 0.
        fill (str|float, optional):
            How recording pauses should be filled. Can be "none" (no filling),
            "nan" (fill with nan), "interpolate" (interpolate with nearest) or a
            float (fill with this value). Default: "none".

    Returns:
        time_arr (np.array):
            Array with time values in ms.
        data_arr (np.array):
            Array with the recorded variable.
    """
    compartment = recording_data_str.split(";")[0]
    period_time = recordings[0][f"{compartment};period"]
    nr_periods = self._get_nr_periods(chunk, compartment)
    time_step = recordings[0]["dt"]
    time_list = []

    ### get data arr
    data_arr = recordings[chunk][recording_data_str]

    ### get time arr
    for period in range(nr_periods):
        start_time, end_time = self.time_lims(
            chunk=chunk, compartment=compartment, period=period
        )
        start_time = round(start_time, af.get_number_of_decimals(time_step))
        end_time = round(end_time, af.get_number_of_decimals(time_step))
        times = np.arange(start_time, end_time + period_time / 2, period_time)
        time_list.append(times)
    time_arr = np.concatenate(time_list, 0)

    ### fill gaps with nan or interpolate
    if fill == "nan":
        time_arr, data_arr = af.time_data_add_nan(
            time_arr,
            data_arr,
            fill_time_step=period_time,
        )
    elif fill == "interpolate":
        full_time_arr = np.arange(
            time_arr[0], time_arr[-1] + period_time / 2, period_time
        )
        f = interp1d(time_arr, data_arr, kind="nearest", axis=0)
        data_arr = f(full_time_arr)
        time_arr = full_time_arr
    elif isinstance(fill, (int, float)):
        time_arr, data_arr = af.time_data_fill_gaps(
            time_arr,
            data_arr,
            fill_time_step=period_time,
            fill=fill,
        )

    return time_arr, data_arr

combine_chunks(recordings, recording_data_str, mode='sequential')

Combines the data of all chunks of recordings. Time periods without recordings AT THE BEGINNING OF chunks or WITHIN chunks (i.e. recording pauses) are considered and filled with nan values.

Warning

If you use mode="consecutive": Missing recordings at the end of chunks (simulated but not recorded) are not considered, this leads to times which differ from the original simulation times (these time periods without recording are simply ignored)!

Parameters:

Name	Type	Description	Default
recordings	list	List with recordings of all chunks.	required
recording_data_str	str	String specifying the compartment name and the variable to combine. Format: "compartment_name;variable_name"	required
mode	str	How should the time array be generated. Can be "sequential" or "consecutive". Default: "sequential". - "sequential": each chunk starts at zero e.g.: [0,100] + [0,250] --> [0, 1, ..., 100, 0, 1, ..., 250] - "consecutive": each chunk starts at the last stop time of the previous chunk e.g.: [0,100] + [0,250] --> [0, 1, ..., 100, 101, 102, ..., 350]	'sequential'

Returns:

Name	Type	Description
time_arr	array	Array with time values in ms.
data_arr	array	Array with the recorded variable.

Source code in src/CompNeuroPy/monitors.py

def combine_chunks(
    self, recordings: list, recording_data_str: str, mode="sequential"
):
    """
    Combines the data of all chunks of recordings. Time periods without recordings
    AT THE BEGINNING OF chunks or WITHIN chunks (i.e. recording pauses) are
    considered and filled with nan values.

    !!! warning
        If you use mode="consecutive": Missing recordings at the end of chunks
        (simulated but not recorded) are not considered, this leads to times which
        differ from the original simulation times (these time periods without
        recording are simply ignored)!

    Args:
        recordings (list):
            List with recordings of all chunks.
        recording_data_str (str):
            String specifying the compartment name and the variable to combine.
            Format: "compartment_name;variable_name"
        mode (str, optional):
            How should the time array be generated. Can be "sequential" or
            "consecutive". Default: "sequential".
            - "sequential": each chunk starts at zero e.g.: [0,100] + [0,250] -->
                [0, 1, ..., 100, 0, 1, ..., 250]
            - "consecutive": each chunk starts at the last stop time of the previous
                chunk e.g.: [0,100] + [0,250] --> [0, 1, ..., 100, 101, 102, ..., 350]

    Returns:
        time_arr (np.array):
            Array with time values in ms.
        data_arr (np.array):
            Array with the recorded variable.
    """
    assert (
        len(self.recording_times_list) > 0
    ), "ERROR combine_chunks(): No recordings/recording_times available."

    compartment = recording_data_str.split(";")[0]
    period_time = recordings[0][f"{compartment};period"]
    time_step = recordings[0]["dt"]
    nr_chunks = self._get_nr_chunks()
    data_list = []
    time_list = []
    pre_chunk_start_time = 0

    for chunk in range(nr_chunks):
        ### append data list with data of all periods of this chunk
        data_list.append(recordings[chunk][recording_data_str])

        ### nr of periods in this chunk
        nr_periods = self._get_nr_periods(chunk, compartment)

        ### start time of chunk depends on mode
        if mode == "sequential":
            chunk_start_time = 0
        elif mode == "consecutive":
            if chunk == 0:
                chunk_start_time = 0
            else:
                ### TODO this uses the wrong end time if there is a pause at the end
                ### of the last chunk but it is not possible to get the info how long
                ### nothing was recorded at the end of a chunk (all I save is from
                ### when to when something is recorded)
                last_stop_time = self.recording_times_list[chunk - 1][compartment][
                    "stop"
                ]["ms"][-1]
                chunk_start_time = (
                    pre_chunk_start_time + last_stop_time + period_time
                )
                pre_chunk_start_time = chunk_start_time
        else:
            print("ERROR recording_times.combine_data, Wrong mode.")
            quit()

        ### append the time list with all times of the periods
        for period in range(nr_periods):
            start_time = (
                self.time_lims(chunk=chunk, compartment=compartment, period=period)[
                    0
                ]
                + chunk_start_time
            )
            end_time = (
                self.time_lims(chunk=chunk, compartment=compartment, period=period)[
                    1
                ]
                + chunk_start_time
            )
            start_time = round(start_time, af.get_number_of_decimals(time_step))
            end_time = round(end_time, af.get_number_of_decimals(time_step))
            times = np.arange(start_time, end_time + period_time / 2, period_time)
            time_list.append(times)

    ### flatten the two lists
    data_arr = np.concatenate(data_list, 0)
    time_arr = np.concatenate(time_list, 0)

    ### check if there are gaps in the time array
    ### fill them with the corersponding times and
    ### the data array with nan values
    time_arr, data_arr = af.time_data_add_nan(
        time_arr,
        data_arr,
        fill_time_step=period_time,
    )

    return time_arr, data_arr