Calculation Class Reference

Inheritance diagram for Calculation:

Public Member Functions
def	__init__ (self, process_list=None)
def	__iter__ (self)
def	__len__ (self)
def	stop (self, index=None)
def	start_batched_and_wait_for_end (self, max_processes=None)
def	start (self, index=None)
def	ensure_running (self, max_processes)
def	wait_for_end (self, display_bar=True, send_notification=False, max_processes=None)
def	show_end_result (self, process, process_bars)
def	map_on_processes (self, map_function, index)
def	is_running (self, index=None)
def	is_finished (self, index=None)
def	has_failed (self, index=None)
def	get (self, name, index=None)
def	get_keys (self, index=None)
def	get_log (self, index=None)
def	get_status (self, index=None)
def	get_parameters (self, index=None)
def	create_widgets_for_all_processes (self, widget_function, index=None)
def	show_log (self, index=None)
def	show_statistics (self, index=None)
def	show_collections (self, index=None)
def	get_statistics (self, index=None)
def	append (self, result_queue, log_file_name, parameters, **kwargs)

Public Attributes
	process_list
	The process list (possibly empty, even later)

Protected Attributes
	_calculation_process_type
	The calculation process type to use.

Detailed Description

Create a Calculation from the given Process that handles
the status of the process and the actions like start, stop or wait_for_end
Do not create instances of this class by yourself but rather use the IPythonHandler for this.

Definition at line 19 of file calculation.py.

Constructor & Destructor Documentation

__init__()

def __init__	(		self,
			process_list = None
	)

 Init with an empty list of processes 

Reimplemented in Basf2Calculation.

Definition at line 27 of file calculation.py.

    def __init__(self, process_list=None):
        """ Init with an empty list of processes """
        
        if process_list:
            self.process_list = process_list
        else:
            self.process_list = []
 
        
        self._calculation_process_type = CalculationProcess
 

Member Function Documentation

__iter__()

def __iter__

(

self

)

Make the class iterable over all single processes

Definition at line 38 of file calculation.py.

    def __iter__(self):
        """
        Make the class iterable over all single processes
        """
        for process in self.process_list:
            yield Calculation([process])
 

__len__()

def __len__

(

self

)

 Return the number of handled processes 

Definition at line 45 of file calculation.py.

    def __len__(self):
        """ Return the number of handled processes """
        return len(self.process_list)
 

append()

def append	(		self,
			result_queue,
			log_file_name,
			parameters,
		**	kwargs
	)

Construct a new process with the given parameters and add it to the process_list.

Definition at line 368 of file calculation.py.

    def append(self, result_queue, log_file_name, parameters, **kwargs):
        """
        Construct a new process with the given parameters and add it to the process_list.
        """
        self.process_list.append(self._calculation_process_type(result_queue=result_queue, log_file_name=log_file_name,
                                                                parameters=parameters, **kwargs))

create_widgets_for_all_processes()

def create_widgets_for_all_processes	(		self,
			widget_function,
			index = None
	)

Create a overview widget for all processes or only one for the given process.

Definition at line 298 of file calculation.py.

    def create_widgets_for_all_processes(self, widget_function, index=None):
        """
        Create a overview widget for all processes or only one for the given process.
        """
 
        if len(self.process_list) == 1:
            widget = widget_function(self.process_list[0])
        else:
            if index is None:
                widget = viewer.ProcessViewer(list(map(widget_function, self.process_list)))
            else:
                if isinstance(index, int):
                    widget = widget_function(self.process_list[index])
                else:
                    widget = widget_function(index)
 
        widget.show()
 

ensure_running()

def ensure_running	(		self,
			max_processes
	)

Ensure that the max_processes number of processes is running and will start
processes which are waiting and have not been started yet.

Definition at line 90 of file calculation.py.

    def ensure_running(self, max_processes):
        """
        Ensure that the max_processes number of processes is running and will start
        processes which are waiting and have not been started yet.
        """
 
        running_processes = [p for p in self.process_list if self.is_running(p)]
 
        processes_to_start = max(0, max_processes - len(running_processes))
 
        for process in self.process_list:
            # start processes as long as processes_to_start is larger than zero
            if not process.already_run and processes_to_start > 0:
                process.start()
                process.already_run = True
                processes_to_start = processes_to_start - 1
 

get()

def get	(		self,
			name,
			index = None
	)

Return the saved queue item with the given name

Definition at line 247 of file calculation.py.

    def get(self, name, index=None):
        """
        Return the saved queue item with the given name
        """
        def f(process, name):
            try:
                return process.get(name)
            except KeyError:
                return None
        return self.map_on_processes(lambda process: f(process, name), index)
 

get_keys()

def get_keys	(		self,
			index = None
	)

Return the names of the items that were saved in the queue

Definition at line 258 of file calculation.py.

    def get_keys(self, index=None):
        """
        Return the names of the items that were saved in the queue
        """
        return self.map_on_processes(lambda process: process.get_keys(), index)
 

get_log()

def get_log	(		self,
			index = None
	)

Return the log of the process if finished

Definition at line 264 of file calculation.py.

    def get_log(self, index=None):
        """
        Return the log of the process if finished
        """
        def f(process):
            if self.is_running(process) or self.is_finished(process):
                return process.get_log()
            else:
                raise AssertionError("Calculation has not been started.")
 
        return self.map_on_processes(f, index)
 

get_parameters()

def get_parameters	(		self,
			index = None
	)

Return the parameters used to start this calculation

Definition at line 292 of file calculation.py.

    def get_parameters(self, index=None):
        """
        Return the parameters used to start this calculation
        """
        return self.map_on_processes(lambda process: process.parameters, index)
 

get_statistics()

def get_statistics	(		self,
			index = None
	)

Return the statistics of the process if finished

Definition at line 356 of file calculation.py.

    def get_statistics(self, index=None):
        """
        Return the statistics of the process if finished
        """
        def f(process):
            if self.is_finished(process) and not self.has_failed(process):
                return self.get("ipython.statistics", process)
            else:
                raise AssertionError("Calculation has not finished.")
 
        return self.map_on_processes(f, index)
 

get_status()

def get_status	(		self,
			index = None
	)

Return a string describing the current status if the calculation

Definition at line 276 of file calculation.py.

    def get_status(self, index=None):
        """
        Return a string describing the current status if the calculation
        """
        def f(process):
            if not process.already_run:
                return "not started"
            elif self.is_running(process):
                return "running"
            elif self.has_failed(process):
                return "failed"
            elif self.is_finished(process):
                return "finished"
 
        return self.map_on_processes(f, index)
 

has_failed()

def has_failed	(		self,
			index = None
	)

Test if the process has failed.

Definition at line 235 of file calculation.py.

    def has_failed(self, index=None):
        """
        Test if the process has failed.
        """
        def f(process):
            if self.is_finished(process):
                return process.exitcode != 0
            else:
                raise AssertionError("Calculation has not finished.")
 
        return self.map_on_processes(f, index)
 

is_finished()

def is_finished	(		self,
			index = None
	)

Test if the process has finished

Definition at line 229 of file calculation.py.

    def is_finished(self, index=None):
        """
        Test if the process has finished
        """
        return self.map_on_processes(lambda process: process.already_run and not self.is_running(process), index)
 

is_running()

def is_running	(		self,
			index = None
	)

Test if the process is still running

Definition at line 223 of file calculation.py.

    def is_running(self, index=None):
        """
        Test if the process is still running
        """
        return self.map_on_processes(lambda process: process.is_alive(), index)
 

map_on_processes()

def map_on_processes	(		self,
			map_function,
			index
	)

Calculate a function on all processes and collect the results if index is None.
Else calculate the function only one the given process or the process number.

Definition at line 207 of file calculation.py.

    def map_on_processes(self, map_function, index):
        """
        Calculate a function on all processes and collect the results if index is None.
        Else calculate the function only one the given process or the process number.
        """
        if len(self.process_list) == 1:
            return map_function(self.process_list[0])
        else:
            if index is None:
                return list(map(map_function, self.process_list))
            else:
                if isinstance(index, int):
                    return map_function(self.process_list[index])
                else:
                    return map_function(index)
 

show_collections()

def show_collections	(		self,
			index = None
	)

Show some snapshots on the collections.
Remember to add the PrintCollectionsPython Module for that!

Definition at line 342 of file calculation.py.

    def show_collections(self, index=None):
        """
        Show some snapshots on the collections.
        Remember to add the PrintCollectionsPython Module for that!
        """
 
        def f(process):
            if "ipython.store_content" in self.get_keys(process):
                return viewer.CollectionsViewer(self.get("ipython.store_content", process))
            else:
                return None
 
        self.create_widgets_for_all_processes(f, index)
 

show_end_result()

def show_end_result	(		self,
			process,
			process_bars
	)

Shows the end result (finished or failed) for all processes in the process_bars list

Definition at line 196 of file calculation.py.

    def show_end_result(self, process, process_bars):
        """
        Shows the end result (finished or failed) for all processes in the process_bars list
        """
        if process in process_bars:
            process_bar = process_bars[process]
            if self.has_failed(process):
                process_bar.update("failed!")
            else:
                process_bar.update("finished")
 

show_log()

def show_log	(		self,
			index = None
	)

Show the log of the underlying process(es).

Definition at line 316 of file calculation.py.

    def show_log(self, index=None):
        """
        Show the log of the underlying process(es).
        """
 
        def f(process):
            if self.is_running(process) or self.is_finished(process):
                return viewer.LogViewer(process.get_log())
            else:
                raise AssertionError("Calculation has not been started.")
 
        self.create_widgets_for_all_processes(f, index)
 

show_statistics()

def show_statistics	(		self,
			index = None
	)

Show the statistics in a smart manner

Definition at line 329 of file calculation.py.

    def show_statistics(self, index=None):
        """
        Show the statistics in a smart manner
        """
 
        def f(process):
            if "ipython.statistics" in self.get_keys(process):
                return viewer.StatisticsViewer(self.get("ipython.statistics", process))
            else:
                return None
 
        self.create_widgets_for_all_processes(f, index)
 

start()

def start	(		self,
			index = None
	)

Start the processes in the background.
Raises an error of the process has been started already.
You can not restart a process. If you want to do so, create another Calculation instance.

Definition at line 75 of file calculation.py.

    def start(self, index=None):
        """
        Start the processes in the background.
        Raises an error of the process has been started already.
        You can not restart a process. If you want to do so, create another Calculation instance.
        """
        def f(process):
            if not process.already_run:
                process.start()
                process.already_run = True
            else:
                raise AssertionError("Calculation can not be started twice.")
 
        self.map_on_processes(f, index)
 

start_batched_and_wait_for_end()

def start_batched_and_wait_for_end	(		self,
			max_processes = None
	)

Start part of the processes and wait for all to finish. If max_processes is None,
only n processes will be started where n is the number of cores on the current machine.
As soon as one process is finished, a waiting one will be started in order to fill all cores
of the machine.

Parameters:
  max_processes: The number of processes which can be run at the same time

Definition at line 59 of file calculation.py.

    def start_batched_and_wait_for_end(self, max_processes=None):
        """
        Start part of the processes and wait for all to finish. If max_processes is None,
        only n processes will be started where n is the number of cores on the current machine.
        As soon as one process is finished, a waiting one will be started in order to fill all cores
        of the machine.
 
        Parameters:
          max_processes: The number of processes which can be run at the same time
        """
        if not max_processes:
            max_processes = multiprocessing.cpu_count()
 
        self.ensure_running(max_processes)
        self.wait_for_end(max_processes=max_processes)
 

stop()

def stop	(		self,
			index = None
	)

Kill the processes. Please keep in mind that killing a process is normaly accompanied with data loss.

Definition at line 49 of file calculation.py.

    def stop(self, index=None):
        """
        Kill the processes. Please keep in mind that killing a process is normaly accompanied with data loss.
        """
        def f(process):
            if self.is_running(process):
                process.terminate()
 
        self.map_on_processes(f, index)
 

wait_for_end()

def wait_for_end	(		self,
			display_bar = True,
			send_notification = False,
			max_processes = None
	)

Send the calculation into the foreground by halting the notebook as
long as the process is running.  Shows a progress bar with the number
of processed events.  Please keep in mind that you can not execute
cells in the notebook when having called wait_for_end (but before -
although a calculation is running.)


Parameters:
  display_bar: If true, the display bar is used to show in the notebook
        that the computation is complete.
  send_notification: If true, the notify2 library will be used to
        notify the user if the computation is complete. This will only
        work if the jupyter notebook is hosted on the local desktop
        machine.
  max_processes: The maximum number of processes which will be run on
        the machine. This has no effect when start() has been called
        before.  This parameter can not be used directly, but
        start_batched_and_wait_for_end() should be used.

Definition at line 107 of file calculation.py.

    def wait_for_end(self, display_bar=True, send_notification=False, max_processes=None):
        """
        Send the calculation into the foreground by halting the notebook as
        long as the process is running.  Shows a progress bar with the number
        of processed events.  Please keep in mind that you can not execute
        cells in the notebook when having called wait_for_end (but before -
        although a calculation is running.)
 
 
        Parameters:
          display_bar: If true, the display bar is used to show in the notebook
                that the computation is complete.
          send_notification: If true, the notify2 library will be used to
                notify the user if the computation is complete. This will only
                work if the jupyter notebook is hosted on the local desktop
                machine.
          max_processes: The maximum number of processes which will be run on
                the machine. This has no effect when start() has been called
                before.  This parameter can not be used directly, but
                start_batched_and_wait_for_end() should be used.
        """
 
        if display_bar:
            # Initialize all process bars
            process_bars = {process: viewer.ProgressBarViewer()
                            for process in self.process_list if process.is_valid}
 
        started_processes = [p for p in self.process_list if p.is_valid and p.already_run]
        running_processes = started_processes
        # Update all process bars as long as minimum one process is running
        while len(running_processes) > 0:
 
            if max_processes:
                self.ensure_running(max_processes)
 
            started_processes = [p for p in self.process_list if p.is_valid and p.already_run]
            running_processes = [p for p in started_processes if self.is_running(p)]
            ended_processes = [p for p in started_processes if not self.is_running(p)]
 
            for process in ended_processes:
                if display_bar:
                    self.show_end_result(process, process_bars)
                    # Only update the process bar once
                    if process in process_bars:
                        del process_bars[process]
 
            for process in running_processes:
                # Check if the process is valid
                if not process.is_valid:
                    if display_bar:
                        self.show_end_result(process, process_bars)
 
                # Check if the process is still running. If not set the end result correctly.
                elif not self.is_running(process):
                    if display_bar:
                        self.show_end_result(process, process_bars)
 
                # Check if there are news from the process python module (a new percentage)
                elif process.progress_queue_local.poll():
                    result = process.progress_queue_local.recv()
                    if result != "end" and display_bar:
                        process_bar = process_bars[process]
                        process_bar.update(result)
 
                else:
                    process.result_queue.fill_results()
                    process.join(timeout=0.01)
 
            time.sleep(0.01)
 
        if display_bar:
            for process in self.process_list:
                self.show_end_result(process, process_bars)
 
        # send notification if requested and notify2 library is available
        if send_notification:
            try:
                import notify2
 
                notify2.init("basf2")
                n = notify2.Notification("basf2",  # head line
                                         "Calculation finished",  # Description text
                                         "notification-message-im"   # Icon name
                                         )
                n.show()
            except ImportError:
                # re-throw with more useful message
                raise ImportError("notify2 library must be installed to show Desktop notifications.")
 

Member Data Documentation

_calculation_process_type

_calculation_process_type

protected

The calculation process type to use.

Definition at line 36 of file calculation.py.

process_list

process_list

The process list (possibly empty, even later)

Definition at line 31 of file calculation.py.

---

The documentation for this class was generated from the following file:

• framework/scripts/hep_ipython_tools/calculation.py