FWCore/ParameterSet/python/Modules.py - master - githubmirror/cmssw

FWCore/ParameterSet/python/Modules.py
318 строк13 KB
Matti Kortelainen
Remove SwitchProducer
16 сен 2025, 19:13
16 сен 2025, 19:13c2c7612
from .Mixins import _ConfigureComponent, saveOrigin
from .Mixins import _Unlabelable, _Labelable
from .Mixins import _TypedParameterizable, _Parameterizable, PrintOptions, specialImportRegistry
from .SequenceTypes import _SequenceLeaf
from .Types import vstring, EDAlias


import copy
from .ExceptionHandling import *
class Service(_ConfigureComponent,_TypedParameterizable,_Unlabelable):
    def __init__(self,type_,*arg,**kargs):
        super(Service,self).__init__(type_,*arg,**kargs)
        self._inProcess = False
    def _placeImpl(self,name:str,proc):
        self._inProcess = True
        proc._placeService(self.type_(),self)
    def insertInto(self, processDesc):
        newpset = processDesc.newPSet()
        newpset.addString(True, "@service_type", self.type_())
        self.insertContentsInto(newpset)
        processDesc.addService(newpset)
    def dumpSequencePython(self, options:PrintOptions=PrintOptions()) -> str:
        return "process." + self.type_()
    def _isTaskComponent(self) -> bool:
        return True
    def isLeaf(self) -> bool:
        return True
    def __str__(self):
        return str(self.type_())

class ESSource(_ConfigureComponent,_TypedParameterizable,_Unlabelable,_Labelable):
    def __init__(self,type_,*arg,**kargs):
        super(ESSource,self).__init__(type_,*arg,**kargs)
        saveOrigin(self, 1)
    def _placeImpl(self,name:str,proc):
        if name == '':
            name=self.type_()
        proc._placeESSource(name,self)
    def moduleLabel_(self,myname:str) -> str:
       result = myname
       if self.type_() == myname:
           result = ""
       return result
    def nameInProcessDesc_(self, myname:str) -> str:
       result = self.type_() + "@" + self.moduleLabel_(myname)
       return result
    def _isTaskComponent(self) -> bool:
        return True
    def isLeaf(self) -> bool:
        return True

class ESProducer(_ConfigureComponent,_TypedParameterizable,_Unlabelable,_Labelable):
    def __init__(self,type_,*arg,**kargs):
        super(ESProducer,self).__init__(type_,*arg,**kargs)
    def _placeImpl(self,name:str,proc):
        if name == '':
            name=self.type_()
        proc._placeESProducer(name,self)
    def moduleLabel_(self,myname:str) -> str:
       result = myname
       if self.type_() == myname:
           result = ''
       return result
    def nameInProcessDesc_(self, myname:str) -> str:
       result = self.type_() + "@" + self.moduleLabel_(myname)
       return result
    def _isTaskComponent(self) -> bool:
        return True
    def isLeaf(self) -> bool:
        return True

class ESPrefer(_ConfigureComponent,_TypedParameterizable,_Unlabelable,_Labelable):
    """Used to set which EventSetup provider should provide a particular data item
    in the case where multiple providers are capable of delivering the data.
    The first argument specifies the C++ class type of the prodiver.
    If the provider has been given a label, you must specify that label as the second argument.
    Additional 'vstring' arguments maybe used to specify exactly which EventSetup Records
    are being preferred and optionally which data items within that Record.
    E.g.,
        #prefer all data in record 'OrangeRecord' from 'juicer'
        ESPrefer("ESJuicerProd", OrangeRecord=cms.vstring())
    or
        #prefer only "Orange" data in "OrangeRecord" from "juicer" 
        ESPrefer("ESJuicerProd", OrangeRecord=cms.vstring("ExtraPulp"))
    or 
        #prefer only "Orange" data with label "ExtraPulp" in "OrangeRecord" from "juicer" 
        ESPrefer("ESJuicerProd", OrangeRecord=cms.vstring("Orange/ExtraPulp"))
    """
    def __init__(self,type_,targetLabel:str='',*arg,**kargs):
        super(ESPrefer,self).__init__(type_,*arg,**kargs)
        self._targetLabel = targetLabel
        if targetLabel is None:
            self._targetLabel = str('')
        if kargs:
            for k,v in kargs.items():
                if not isinstance(v,vstring):
                    raise RuntimeError('ESPrefer only allows vstring attributes. "'+k+'" is a '+str(type(v)))
    def _placeImpl(self,name:str,proc):
        proc._placeESPrefer(name,self)
    def nameInProcessDesc_(self, myname:str) -> str:
        # the C++ parser can give it a name like "label@prefer".  Get rid of that.
        return "esprefer_" + self.type_() + "@" + self._targetLabel
    def copy(self):
        returnValue = ESPrefer.__new__(type(self))
        returnValue.__init__(self.type_(), self._targetLabel)
        return returnValue
    def moduleLabel_(self, myname:str) -> str:
        return self._targetLabel
    def targetLabel_(self) -> str:
        return self._targetLabel
    def dumpPythonAs(self, label, options:PrintOptions=PrintOptions()) -> str:
       result = options.indentation()
       basename = self._targetLabel
       if basename == '':
           basename = self.type_()
       if options.isCfg:
           # do either type or label
           result += 'process.prefer("'+basename+'"'
           if self.parameterNames_():
               result += ",\n"+_Parameterizable.dumpPython(self,options)+options.indentation()
           result +=')\n'
       else:
           # use the base class Module
           result += 'es_prefer_'+basename+' = cms.ESPrefer("'+self.type_()+'"'
           if self._targetLabel != '':
              result += ',"'+self._targetLabel+'"'
           if self.parameterNames_():
               result += ",\n"+_Parameterizable.dumpPython(self,options)+options.indentation()
           result += ')\n'
       return result

class _Module(_ConfigureComponent,_TypedParameterizable,_Labelable,_SequenceLeaf):
    """base class for classes which denote framework event based 'modules'"""
    __isStrict__ = False  
    def __init__(self,type_,*arg,**kargs):
        super(_Module,self).__init__(type_,*arg,**kargs)
        if _Module.__isStrict__:
            self.setIsFrozen()
        saveOrigin(self, 2)    
    def _clonesequence(self, lookuptable):
        try:
            return lookuptable[id(self)]
        except:
            raise ModuleCloneError(self._errorstr())
    def _errorstr(self):
         # return something like "EDAnalyzer("foo", ...)"
        typename = format_typename(self)
        return "%s('%s', ...)" %(typename, self.type_())
    
    def setPrerequisites(self, *libs):
        self.__dict__["libraries_"] = libs

    def insertInto(self, parameterSet, myname:str):
        if "libraries_" in self.__dict__:
            from ctypes import LibraryLoader, CDLL
            import platform
            loader = LibraryLoader(CDLL)
            ext = platform.uname()[0] == "Darwin" and "dylib" or "so"
            [loader.LoadLibrary("lib%s.%s" % (l, ext)) for l in self.libraries_]
        super(_Module,self).insertInto(parameterSet,myname)

class EDProducer(_Module):
    def __init__(self,type_,*arg,**kargs):
        super(EDProducer,self).__init__(type_,*arg,**kargs)
    def _placeImpl(self,name:str,proc):
        proc._placeProducer(name,self)
    def _isTaskComponent(self):
        return True

class EDFilter(_Module):
    def __init__(self,type_,*arg,**kargs):
        super(EDFilter,self).__init__(type_,*arg,**kargs)
    def _placeImpl(self,name:str,proc):
        proc._placeFilter(name,self)
    def _isTaskComponent(self):
        return True

class EDAnalyzer(_Module):
    def __init__(self,type_,*arg,**kargs):
        super(EDAnalyzer,self).__init__(type_,*arg,**kargs)
    def _placeImpl(self,name:str,proc):
        proc._placeAnalyzer(name,self)


class OutputModule(_Module):
    def __init__(self,type_,*arg,**kargs):
        super(OutputModule,self).__init__(type_,*arg,**kargs)
    def _placeImpl(self,name:str,proc):
        proc._placeOutputModule(name,self)


class Source(_ConfigureComponent,_TypedParameterizable):
    def __init__(self,type_,*arg,**kargs):
        super(Source,self).__init__(type_,*arg,**kargs)
    def _placeImpl(self,name:str,proc):
        proc._placeSource(name,self)
    def moduleLabel_(self,myname:str):
        return "@main_input"
    def nameInProcessDesc_(self,myname:str):
        return "@main_input"


class Looper(_ConfigureComponent,_TypedParameterizable):
    def __init__(self,type_,*arg,**kargs):
        super(Looper,self).__init__(type_,*arg,**kargs)
    def _placeImpl(self,name:str,proc):
        proc._placeLooper(name,self)
    def moduleLabel_(self,myname:str):
        return "@main_looper"
    def nameInProcessDesc_(self, myname:str):
        return "@main_looper"


if __name__ == "__main__":
    import unittest
    from .Types import *
    from .SequenceTypes import *

    class TestModules(unittest.TestCase):
        def testEDAnalyzer(self):
            empty = EDAnalyzer("Empty")
            withParam = EDAnalyzer("Parameterized",foo=untracked(int32(1)), bar = untracked(string("it")))
            self.assertEqual(withParam.foo.value(), 1)
            self.assertEqual(withParam.bar.value(), "it")
            aCopy = withParam.copy()
            self.assertEqual(aCopy.foo.value(), 1)
            self.assertEqual(aCopy.bar.value(), "it")
            withType = EDAnalyzer("Test",type = int32(1))
            self.assertEqual(withType.type.value(),1)
            block = PSet(i = int32(9))
            m = EDProducer("DumbProducer", block, j = int32(10))
            self.assertEqual(9, m.i.value())
            self.assertEqual(10, m.j.value())
        def testESPrefer(self):
            juicer = ESPrefer("JuiceProducer")
            options = PrintOptions()
            options.isCfg = True
            self.assertEqual(juicer.dumpPythonAs("juicer", options), "process.prefer(\"JuiceProducer\")\n")
            options.isCfg = False
            self.assertEqual(juicer.dumpPythonAs("juicer", options), "es_prefer_JuiceProducer = cms.ESPrefer(\"JuiceProducer\")\n")

            juicer = ESPrefer("JuiceProducer","juicer")
            options = PrintOptions()
            options.isCfg = True
            self.assertEqual(juicer.dumpPythonAs("juicer", options), 'process.prefer("juicer")\n')
            options.isCfg = False
            self.assertEqual(juicer.dumpPythonAs("juicer", options), 'es_prefer_juicer = cms.ESPrefer("JuiceProducer","juicer")\n')
            juicer = ESPrefer("JuiceProducer",fooRcd=vstring())
            self.assertEqual(juicer.dumpConfig(options),
"""JuiceProducer { 
    vstring fooRcd = {
    }

}
""")
            options = PrintOptions()
            options.isCfg = True
            self.assertEqual(juicer.dumpPythonAs("juicer"),
"""process.prefer("JuiceProducer",
    fooRcd = cms.vstring()
)
""")
            options.isCfg = False
            self.assertEqual(juicer.dumpPythonAs("juicer", options),
"""es_prefer_JuiceProducer = cms.ESPrefer("JuiceProducer",
    fooRcd = cms.vstring()
)
""")
        
        def testService(self):
            empty = Service("Empty")
            withParam = Service("Parameterized",foo=untracked(int32(1)), bar = untracked(string("it")))
            self.assertEqual(withParam.foo.value(), 1)
            self.assertEqual(withParam.bar.value(), "it")
            self.assertEqual(empty.dumpPython(), "cms.Service(\"Empty\")\n")
            self.assertEqual(withParam.dumpPython(), "cms.Service(\"Parameterized\",\n    bar = cms.untracked.string(\'it\'),\n    foo = cms.untracked.int32(1)\n)\n")
        def testSequences(self):
            m = EDProducer("MProducer")
            n = EDProducer("NProducer")
            m.setLabel("m")
            n.setLabel("n")
            s1 = Sequence(m*n)
            options = PrintOptions()

        def testIsTaskComponent(self):
            m = EDProducer("x")
            self.assertTrue(m._isTaskComponent())
            self.assertTrue(m.isLeaf())
            m = EDFilter("x")
            self.assertTrue(m._isTaskComponent())
            self.assertTrue(m.isLeaf())
            m = OutputModule("x")
            self.assertFalse(m._isTaskComponent())
            self.assertTrue(m.isLeaf())
            m = EDAnalyzer("x")
            self.assertFalse(m._isTaskComponent())
            self.assertTrue(m.isLeaf())
            m = Service("x")
            self.assertTrue(m._isTaskComponent())
            self.assertTrue(m.isLeaf())
            m = ESProducer("x")
            self.assertTrue(m._isTaskComponent())
            self.assertTrue(m.isLeaf())
            m = ESSource("x")
            self.assertTrue(m._isTaskComponent())
            self.assertTrue(m.isLeaf())
            m = Sequence()
            self.assertFalse(m._isTaskComponent())
            self.assertFalse(m.isLeaf())
            m = Path()
            self.assertFalse(m._isTaskComponent())
            m = EndPath()
            self.assertFalse(m._isTaskComponent())
            m = Task()
            self.assertTrue(m._isTaskComponent())
            self.assertFalse(m.isLeaf())

    unittest.main()