Source code for pwem.protocols.protocol_movie_eraser

import os
import time

from pwem.objects import SetOfMovies
from pwem.protocols import EMProtocol
from pyworkflow import BETA
from pyworkflow.protocol import params
import pyworkflow.utils as pwutils

[docs]class ProtMovieEraser(EMProtocol): """ It will REMOVE movies that already have been aligned into micrographs. WARNING: There is no way back. Be sure you understand the consequences.""" _label = "movie eraser" _devStatus = BETA def __init__(self, **kwargs): super().__init__(**kwargs) self.deletedMovies = [] def _defineParams(self, form): form.addSection(label='Input') form.addParam('inputMicrographs', params.PointerParam, pointerClass='SetOfMicrographs', label=pwutils.Message.LABEL_INPUT_MIC, important=True, help='Select the SetOfMicrographs to drive the movie DELETION process.') form.addParam('movieSet', params.PointerParam, pointerClass='SetOfMovies', label="Movies source", important=True, help='Select the SetOfMovies to locate the movie files. Match will be done by ids!.') form.addParam("dryMode", params.BooleanParam, label="Dry mode", default=True, expertLevel=params.LEVEL_ADVANCED) # -------------------------- STEP methods --------------------------------- def _insertAllSteps(self): self._insertFunctionStep(self.deleteMoviesStep)
[docs] def deleteMoviesStep(self): """ Delete all movies origin of the micset until set is closed""" lastMicId = 0 while not self._isAllDone(): # For each mic not done yet for mic in self._getMicSet().iterItems(where="id > %s" % lastMicId): self._deleteMovieFromMic(mic) lastMicId = mic.getObjId() # wait some time time.sleep(30)
def _deleteMovieFromMic(self, mic): """ Deletes the movie source of the mic object""" movieFile = self._getMovieFileFromMic(mic) # Check existence if not os.path.exists(movieFile): print("%s is gone already!." % movieFile) # if dry mode elif self.dryMode.get(): print("%s would have been deleted" % movieFile, flush=True) # Delete!!! else: # Delete the move os.remove(movieFile) print("%s deleted." % movieFile, flush=True) # Annotate it self.deletedMovies.append(movieFile) def _getMovieFileFromMic(self, mic): """ Returns the movie path that was used in the micrograph movie alignment""" movieSet = self._getMovieSet() # To match the movie use the ID: # movie alignment protocols are copying the movie id to the mic --> mic.copyObjId(movie) movie = movieSet[mic.getObjId()] movieFile = movie.getFileName() if os.path.islink(movieFile): movieFile = os.path.realpath(movieFile) return movieFile def _getMovieSet(self): """ Returns the SetOFMovies related to the input set of micrographs""" # This is not possible since the relations are stored in the run.db of the input protocol # and there is no easy way to get them from this run # parents= self.getProject().getSourceParents(self._getMicSet()) # # # Get the parent protocols # for parent in parents: # # We need to update them taking the most recent data form its own run.db # self.getProject()._updateProtocol(parent) # for attr, output in parent.iterOutputAttributes(): # if isinstance(output, SetOfMovies): # return output # # return None return self.movieSet.get() def _getMicSet(self): """ Returns te input set of micrographs""" mics = self.inputMicrographs.get() mics.loadAllProperties() return mics def _isAllDone(self): """ Returns True if all work is done. All movies deleted """ mics = self._getMicSet() return mics.isStreamClosed() and mics.getSize() == len(self.deletedMovies)
[docs] @classmethod def worksInStreaming(cls): return True
# -------------------------- INFO functions ------------------------------- def _warnings(self): if not self.dryMode.get(): return ["If you continue, all movies related to the set of micrographs will be DELETED!. Try to run it in \"dry mode\" to double check first."] def _summary(self): summary = [] summary.append("Dry mode is active. Harmless" if self.dryMode.get() else "Dry mode deactivated, harm can be done!") return summary