Source Code for Module EDStatus

  1  #!/usr/bin/env python 
  2  # -*- coding: utf8 -*- 
  3  # 
  4  #    Project: EDNA-Kernel 
  5  #             http://www.edna-site.org 
  6  # 
  7  #    File: "$Id: EDParallelExecute.py 1990 2010-08-26 09:10:15Z svensson $" 
  8  # 
  9  #    Copyright (C) European Synchrotron Radiation Facility, Grenoble, France 
 10  # 
 11  #    Principal author:       Jérôme Kieffer (Jerome.Kieffer@ESRF.eu) 
 12  # 
 13  #    This program is free software: you can redistribute it and/or modify 
 14  #    it under the terms of the GNU Lesser General Public License as published 
 15  #    by the Free Software Foundation, either version 3 of the License, or 
 16  #    (at your option) any later version. 
 17  # 
 18  #    This program is distributed in the hope that it will be useful, 
 19  #    but WITHOUT ANY WARRANTY; without even the implied warranty of 
 20  #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 21  #    GNU Lesser General Public License for more details. 
 22  # 
 23  #    You should have received a copy of the GNU General Public License 
 24  #    and the GNU Lesser General Public License  along with this program.   
 25  #    If not, see <http://www.gnu.org/licenses/>. 
 26  # 
 27  from __future__ import with_statement 
 28   
 29  __authors__ = ["Jérôme Kieffer"] 
 30  __contact__ = "Jerome.Kieffer@esrf.eu" 
 31  __license__ = "LGPLv3+" 
 32  __copyright__ = "European Synchrotron Radiation Facility, Grenoble, France" 
 33  __date__ = "20110919" 
 34  __status__ = "Stable" 
 35   
 36  from EDVerbose import EDVerbose 
 37  from threading import Semaphore 
 38   
 39 -class EDStatus(object): 
 40      """ 
 41      This static class keeps track of plugins under execution and which have succeeded or failed 
 42      """ 
 43      __listRunning = [] 
 44      __listSuccess = [] 
 45      __listFailure = [] 
 46      __sem = Semaphore() 
 47   
 48   
 49      @classmethod 
 50 -    def getRunning(cls): 
 51          return cls.__listRunning 
 52   
 53   
 54      @classmethod 
 55 -    def getSuccess(cls): 
 56          return cls.__listSuccess 
 57   
 58   
 59      @classmethod 
 60 -    def getFailure(cls): 
 61          return cls.__listFailure 
 62   
 63   
 64      @classmethod 
 65 -    def tellRunning(cls, strPluginId): 
 66          """ 
 67          Tell the strPluginId has started. 
 68          @param strPluginId:  concatenation of the plugin-name and plugin-Id 
 69          """ 
 70          with cls.__sem: 
 71              cls.__listRunning.append(strPluginId) 
 72   
 73   
 74      @classmethod 
 75 -    def tellSuccess(cls, strPluginId): 
 76          """ 
 77          Tell the strPluginId has finished with success. 
 78          @param strPluginId: concatenation of the plugin-name and plugin-Id 
 79          """ 
 80          with cls.__sem: 
 81              try: 
 82                  cls.__listRunning.remove(strPluginId) 
 83              except ValueError: 
 84                  EDVerbose.ERROR("Unable to remove %s from running plugins' list" % strPluginId) 
 85              else: 
 86                  cls.__listSuccess.append(strPluginId) 
 87   
 88   
 89      @classmethod 
 90 -    def tellFailure(cls, strPluginId): 
 91          """ 
 92          Tell the strPluginId has finished with failure. 
 93          @param strPluginId: concatenation of the plugin-name and plugin-Id 
 94          """ 
 95          with cls.__sem: 
 96              try: 
 97                  cls.__listRunning.remove(strPluginId) 
 98              except ValueError: 
 99                  EDVerbose.ERROR("Unable to remove %s from running plugins' list" % strPluginId) 
100              else: 
101                  cls.__listFailure.append(strPluginId) 
102