personalPlayback.Playback Class Reference

Inheritance diagram for personalPlayback.Playback:

Public Member Functions
def	discover_files (self)
def	do_create_lumi (self)
def	do_exec (self)
def	do_init (self)
Public Member Functions inherited from personalPlayback.Applet
def	__init__ (self, name, opts, kwargs)
def	do_exec (self)
def	do_init (self)
def	log (self)
def	write (self, fp)

Public Attributes
	global_file
	input
	lumi_backlog
	lumi_backlog_size
	lumi_found
	lumi_order
	next_lumi_index
	output
	ramdisk
	run
Public Attributes inherited from personalPlayback.Applet
	control_fp
	kwargs
	name
	opts

Static Public Attributes
	re_pattern

Additional Inherited Members
Static Public Member Functions inherited from personalPlayback.Applet
def	read (fp)

Detailed Description

Definition at line 64 of file personalPlayback.py.

Member Function Documentation

discover_files()

def personalPlayback.Playback.discover_files

(

self

)

Definition at line 67 of file personalPlayback.py.

    def discover_files(self):
        self.lumi_found = {}

        files_found = set()
        streams_found = set()
        run_found = None
        for f in os.listdir(self.input):
            r = self.re_pattern.match(f)
            if r:
                run, lumi, stream, stream_source = r.groups()
                run, lumi = int(run), int(lumi)

                if run_found is None:
                    run_found = run
                elif run_found != run:
                    raise Exception("Files from multiple runs are not (yet) supported for as playback input.")

                lumi_dct = self.lumi_found.setdefault(lumi, { 'streams': {} })
                lumi_dct["streams"][stream] = (f, stream_source)
                files_found.add(f)
                streams_found.add(stream)

        if run_found is None:
            raise Exception("Playback files not found.")

        if self.run < 0:
            self.run = run_found

        self.log.info("Found run %s, will map output to run %s", run_found, self.run)
        self.log.info("Found %d lumisections with %d files", len(self.lumi_found), len(files_found))
        self.log.info("Found %d streams: %s", len(streams_found), list(streams_found))

        self.lumi_order = list(self.lumi_found.keys())
        self.lumi_order.sort()
        self.log.info("Lumi order: %s", str(self.lumi_order))

do_create_lumi()

def personalPlayback.Playback.do_create_lumi

(

self

)

Definition at line 126 of file personalPlayback.py.

References mps_setup.append, info(), WorkFlow.WorkFlow.input, personalPlayback.Playback.input, personalPlayback.Applet.log(), conddbCopyTest.CopyTest.log(), conditionUploadTest.UploadTest.log(), cond::persistency::Logger.log(), cond::CredentialStore.log(), crabFunctions.CrabTask.log, personalPlayback.Playback.lumi_backlog, personalPlayback.Playback.lumi_backlog_size, personalPlayback.Playback.lumi_found, personalPlayback.Playback.lumi_order, personalPlayback.Playback.next_lumi_index, GeometryComparisonPlotter.output, personalPlayback.Playback.output, PhysicsTools::Calibration::MVAComputer.output, RPCRunIOV::RunIOV_Item.run, TB06Reco.run, TB06RecoH2.run, RPCDQMObject.run, EfficiencyPlotter.run, WZInterestingEventSelector::event.run, MuonRecoTest.run, Point.run, DTResolutionAnalysisTest.run, BeamProfile2DBReader::TheBSfromDB.run, BeamSpotRcdReader::theBSfromDB.run, DTDataIntegrityTest.run, BeamSpotOnlineRecordsReader::theBSOfromDB.run, DTBlockedROChannelsTest.run, DTResolutionTest.run, DTChamberEfficiencyTest.run, DTEfficiencyTest.run, ME::Header.run, DQMNet::CoreObject.run, personalPlayback.Playback.run, DTLocalTriggerBaseTest.run, MatrixUtil.InputInfo.run, HitEff.run, cond::RunInfo_t.run, SiPixelErrorEstimation.run, and l1ct::Event.run.

Referenced by personalPlayback.Playback.do_exec().

    def do_create_lumi(self):
        orig_lumi = self.lumi_order[(self.next_lumi_index - 1) % len(self.lumi_order)]
        play_lumi = self.next_lumi_index;
        self.next_lumi_index += 1

        self.log.info("Start copying lumi (original) %06d -> %06d (playback)", orig_lumi, play_lumi)

        lumi_dct = self.lumi_found[orig_lumi]
        streams = lumi_dct["streams"]

        def ijoin(f):
            return os.path.join(self.input, f)

        def ojoin(f):
            return os.path.join(self.output, f)

        written_files = set()
        for stream, v  in streams.items():
            jsn_orig_fn, stream_source = v
            jsn_play_fn = "run%06d_ls%04d_stream%s_%s.jsn" % (self.run, play_lumi, stream, stream_source)

            # define dat filename
            ext = "dat"
            if stream.startswith("streamDQMHistograms"):
                ext = "pb"
            dat_play_fn = "run%06d_ls%04d_stream%s_%s.%s" % (self.run, play_lumi, stream, stream_source, ext)

            # read the original file name, for copying
            with open(ijoin(jsn_orig_fn), 'r') as f:
                jsn_data = json.load(f)
                dat_orig_fn = jsn_data["data"][3]

            # copy the data file
            if os.path.exists(ijoin(dat_orig_fn)):
                self.log.info("C: %s -> %s", dat_orig_fn, dat_play_fn)
                shutil.copyfile(ijoin(dat_orig_fn), ojoin(dat_play_fn))

                written_files.add(dat_play_fn)
            else:
                log.warning("Dat file is missing: %s", dat_orig_fn)

            # write a new json file point to a different data file
            # this has to be atomic!
            jsn_data["data"][3] = dat_play_fn

            f = tempfile.NamedTemporaryFile(prefix=jsn_play_fn+ ".", suffix=".tmp", dir = self.output, delete=False)
            tmp_fp = f.name
            json.dump(jsn_data, f)
            f.close()

            os.rename(tmp_fp, ojoin(jsn_play_fn))
            written_files.add(jsn_play_fn)

        self.log.info("Copied %d files for lumi %06d", len(written_files), play_lumi)

        self.lumi_backlog.append((play_lumi, written_files))
        while len(self.lumi_backlog) > self.lumi_backlog_size:
            old_lumi, files_to_delete = self.lumi_backlog.popleft()

            self.log.info("Deleting %d files for old lumi %06d", len(files_to_delete), old_lumi)
            for f in files_to_delete:
                os.unlink(ojoin(f))

do_exec()

def personalPlayback.Playback.do_exec

(

self

)

Definition at line 189 of file personalPlayback.py.

References personalPlayback.Playback.do_create_lumi(), info(), personalPlayback.Applet.log(), conddbCopyTest.CopyTest.log(), conditionUploadTest.UploadTest.log(), cond::persistency::Logger.log(), cond::CredentialStore.log(), crabFunctions.CrabTask.log, personalPlayback.Applet.opts, GeometryComparisonPlotter.output, personalPlayback.Playback.output, PhysicsTools::Calibration::MVAComputer.output, RPCRunIOV::RunIOV_Item.run, TB06RecoH2.run, RPCDQMObject.run, TB06Reco.run, EfficiencyPlotter.run, WZInterestingEventSelector::event.run, MuonRecoTest.run, Point.run, BeamProfile2DBReader::TheBSfromDB.run, DTResolutionAnalysisTest.run, BeamSpotRcdReader::theBSfromDB.run, DTDataIntegrityTest.run, BeamSpotOnlineRecordsReader::theBSOfromDB.run, DTBlockedROChannelsTest.run, DTResolutionTest.run, DTChamberEfficiencyTest.run, DTEfficiencyTest.run, ME::Header.run, personalPlayback.Playback.run, DQMNet::CoreObject.run, DTLocalTriggerBaseTest.run, MatrixUtil.InputInfo.run, HitEff.run, cond::RunInfo_t.run, SiPixelErrorEstimation.run, and l1ct::Event.run.

    def do_exec(self):
        last_write = 0
        lumi_produced = 0

        while True:
            time.sleep(1)

            now = time.time()
            if (now - last_write) > self.opts.playback_time_lumi:
                last_write = now

                if self.opts.playback_nlumi > -1 and lumi_produced >= self.opts.playback_nlumi:
                    break

                self.do_create_lumi()
                lumi_produced += 1

        # write eor
        eor_fn = "run%06d_ls0000_EoR.jsn" % (self.run, )
        eor_fp = os.path.join(self.output, eor_fn)
        with open(eor_fp, "w"):
            pass

        self.log.info("Wrote EoR: %s", eor_fp)

do_init()

def personalPlayback.Playback.do_init

(

self

)

Definition at line 103 of file personalPlayback.py.

    def do_init(self):
        # check if our input directory is okay
        self.input = self.opts.playback
        self.ramdisk = self.opts.work_ramdisk
        self.run = self.opts.run
        self.log.info("Using input directory: %s", self.input)

        self.discover_files()

        self.output = os.path.join(self.ramdisk, "run%06d" % self.run)
        if not os.path.isdir(self.output):
            os.makedirs(self.output)
        self.log.info("Using output directory: %s", self.output)

        self.global_file = os.path.join(self.ramdisk, ".run%06d.global" % self.run)
        self.log.info("Writing: %s", self.global_file)
        with open(self.global_file, "w") as f:
            f.write("run_key = pp_run")

        self.lumi_backlog = collections.deque()
        self.lumi_backlog_size = 10
        self.next_lumi_index = 1

Member Data Documentation

global_file

personalPlayback.Playback.global_file

Definition at line 117 of file personalPlayback.py.

input

personalPlayback.Playback.input

Definition at line 105 of file personalPlayback.py.

Referenced by collectionMerger.collectionMerger.analyze(), and personalPlayback.Playback.do_create_lumi().

lumi_backlog

personalPlayback.Playback.lumi_backlog

Definition at line 122 of file personalPlayback.py.

Referenced by personalPlayback.Playback.do_create_lumi().

lumi_backlog_size

personalPlayback.Playback.lumi_backlog_size

Definition at line 123 of file personalPlayback.py.

Referenced by personalPlayback.Playback.do_create_lumi().

lumi_found

personalPlayback.Playback.lumi_found

Definition at line 68 of file personalPlayback.py.

Referenced by personalPlayback.Playback.do_create_lumi().

lumi_order

personalPlayback.Playback.lumi_order

Definition at line 99 of file personalPlayback.py.

Referenced by personalPlayback.Playback.do_create_lumi().

next_lumi_index

personalPlayback.Playback.next_lumi_index

Definition at line 124 of file personalPlayback.py.

Referenced by personalPlayback.Playback.do_create_lumi().

output

personalPlayback.Playback.output

Definition at line 112 of file personalPlayback.py.

Referenced by collectionMerger.collectionMerger.analyze(), personalPlayback.Playback.do_create_lumi(), and personalPlayback.Playback.do_exec().

ramdisk

personalPlayback.Playback.ramdisk

Definition at line 106 of file personalPlayback.py.

Referenced by personalPlayback.FrameworkJob.discover_latest().

re_pattern

personalPlayback.Playback.re_pattern

static

Definition at line 65 of file personalPlayback.py.

run

personalPlayback.Playback.run

Definition at line 93 of file personalPlayback.py.

Referenced by generateEDF.LumiInfo.__str__(), DTWorkflow.DTWorkflow.all(), Types.EventID.cppID(), Types.LuminosityBlockID.cppID(), personalPlayback.Playback.do_create_lumi(), personalPlayback.Playback.do_exec(), o2olib.O2OTool.execute(), generateEDF.LumiInfo.fixXingInfo(), and dqm_interfaces.DirWalkerFile.walk().