How to use the fireworks.FiretaskBase function in FireWorks

parse_kwargs = self.get("parse_kwargs")
        pc_string = self.get("parse_class")
        parse_class = load_class(*pc_string.rsplit(".", 1))
        calc_dir = self.get("calc_dir", ".")
        with monty.os.cd(calc_dir):
            result = parse_class(**parse_kwargs)

        pass_dict = recursive_get_result(pass_dict, result)
        mod_spec_key = self.get("mod_spec_key", "prev_calc_result")
        mod_spec_cmd = self.get("mod_spec_cmd", "_set")
        return FWAction(mod_spec=[{mod_spec_cmd: {mod_spec_key: pass_dict}}])


#TODO: not sure this is the best to do this, will mull over it and do the recatoring later - matk
@explicit_serialize
class CopyFiles(FiretaskBase):
    """
    Task to copy the given list of files from the given directory to the destination directory.
    To customize override the setup_copy and copy_files methods.

    Optional params:
        from_dir (str): path to the directory containing the files to be copied. Supports env_chk.
        to_dir (str): path to the destination directory. Supports env_chk.
        filesystem (str)
        files_to_copy (list): list of file names. Defaults to copying everything in from_dir.
        exclude_files (list): list of file names to be excluded.
        suffix (str): suffix to append to each filename when copying 
            (e.g., rename 'INCAR' to 'INCAR.precondition')
        continue_on_missing(bool): Whether to continue copying when a file
            in filenames is missing. Defaults to False.
    """

Optional_params:
        radius (float): cluster radius in angstroms
        other_params (dict): **kwargs to pass into the desired InputSet if using str feff_input_set
    """
    required_params = ["absorbing_atom", "structure", "feff_input_set"]
    optional_params = ["radius", "other_params"]

    def run_task(self, fw_spec):
        feff_input_set = get_feff_input_set_obj(self["feff_input_set"], self["absorbing_atom"],
                                                self["structure"], self.get("radius", 10.0),
                                                **self.get("other_params", {}))
        feff_input_set.write_input(".")


@explicit_serialize
class WriteEXAFSPaths(FiretaskBase):
    """
    Write the scattering paths to paths.dat file.

    Required_params:
        feff_input_set: (FeffDictSet subclass)
        paths (list): list of paths. A path = list of site indices.

    Optional_params:
        degeneracies (list): list of path degeneracies.
    """
    required_params = ["feff_input_set", "paths"]
    optional_params = ["degeneracies"]

    def run_task(self, fw_spec):
        atoms = self['feff_input_set'].atoms
        paths = Paths(atoms, self["paths"], degeneracies=self.get("degeneracies", []))

with open("raman.json", "w") as f:
                f.write(json.dumps(d, default=DATETIME_HANDLER))
        else:
            db = VaspCalcDb.from_db_file(db_file, admin=True)
            db.collection = db.db["raman"]
            db.collection.insert_one(d)
            logger.info("Raman tensor calculation complete.")
        return FWAction()


# TODO: @computron: this requires a "tasks" collection to proceed. Merits of changing to FW passing
# method? -computron
# TODO: @computron: even if you use the db-centric method, embed information in tags rather than
# task_label? This workflow likely requires review with its authors. -computron
@explicit_serialize
class GibbsAnalysisToDb(FiretaskBase):
    """
    Compute the quasi-harmonic gibbs free energy. There are 2 options available for the
    quasi-harmonic approximation (set via 'qha_type' parameter):
    1. use the phonopy package quasi-harmonic approximation interface or
    2. use the debye model.
    Note: Instead of relying on fw_spec, this task gets the required data directly from the
    tasks collection for processing. The summary dict is written to 'gibbs.json' file.

    required_params:
        tag (str): unique tag appended to the task labels in other fireworks so that all the
            required data can be queried directly from the database.
        db_file (str): path to the db file

    optional_params:
        qha_type(str): quasi-harmonic approximation type: "debye_model" or "phonopy",
            default is "debye_model"

poscar_path,
                incar_path,
                None,
                option=option,
                dict_for_basis=user_supplied_basis,
            )
        additional_input = self.get("user_lobsterin_settings", None)
        if additional_input:
            for key, parameter in additional_input.items():
                lobsterinput[key] = parameter

        lobsterinput.write_lobsterin("lobsterin")


@explicit_serialize
class RunLobster(FiretaskBase):
    """
    Starts the Lobster Job
    Optional params:
        lobster_cmd (str): command to run lobster, supports env_chk
        gzip_output (bool): Default: True. If true, output (except WAVECAR) will be gzipped.
        gzip_WAVECAR (bool): Default: False. If true, WAVECAR will be gzipped
        handler_group (str or [ErrorHandler]): group of handlers to use. See handler_groups dict in the code for
            the groups and complete list of handlers in each group. Alternatively, you can
            specify a list of ErrorHandler objects. These handlers can be found in the lobster module of custodian.
        validator_group (str or [Validator]): group of validators to use. See validator_groups dict in the
            code for the groups and complete list of validators in each group. Alternatively, you can
            specify a list of Validator objects.
    """

    optional_params = [
        "lobster_cmd",

raise ValueError("Unsupported job type: {}".format(job_type))

        # construct handlers
        handlers = handler_groups[self.get("handler_group", "default")]

        c = Custodian(
            handlers,
            jobs,
            max_errors=max_errors,
            gzipped_output=gzipped_output)

        c.run()


@explicit_serialize
class RunNoQChem(FiretaskBase):
    """
    Do NOT run QChem. Do nothing.
    """

    def run_task(self, fw_spec):
        pass


@explicit_serialize
class RunQChemFake(FiretaskBase):
    """
     QChem Emulator

     Required params:
         ref_dir (string): Path to reference qchem run directory with input file in the folder
            named "mol.qin" and output file in the folder named "mol.qout".

if fw_spec.get("tags", None):
            summary["tags"] = fw_spec["tags"]

        # db_file itself is required but the user can choose to pass the results to db or not
        if to_db:
            mmdb.collection = mmdb.db["magnetic_deformation"]
            mmdb.collection.insert_one(summary)
        else:
            with open("magnetic_deformation.json", "w") as f:
                f.write(json.dumps(summary, default=DATETIME_HANDLER))

        logger.info("Magnetic deformation calculation complete.")


@explicit_serialize
class PolarizationToDb(FiretaskBase):
    """
    Recovers the same branch polarization and spontaneous polarization
    for a ferroelectric workflow.
    """

    optional_params = ["db_file"]

    def run_task(self, fw_spec):

        wfid = list(filter(lambda x: 'wfid' in x, fw_spec['tags'])).pop()
        db_file = env_chk(self.get("db_file"), fw_spec)
        vaspdb = VaspCalcDb.from_db_file(db_file, admin=True)

        # ferroelectric workflow groups calculations by generated wfid tag
        polarization_tasks = vaspdb.collection.find({"tags": wfid, "task_label": {"$regex": ".*polarization"}})

]
    optional_params = ["vasp_input_params", "autosort_tol", "potcar_spec"]

    def run_task(self, fw_spec):
        # Get interpolated structure.
        structure = GetInterpolatedPOSCAR.interpolate_poscar(self, fw_spec)

        # Assumes VaspInputSet String + parameters was provided
        vis_cls = load_class("pymatgen.io.vasp.sets", self["vasp_input_set"])
        vis = vis_cls(structure, **self.get("vasp_input_params", {}))
        potcar_spec = self.get("potcar_spec", False)
        vis.write_input(".", potcar_spec=potcar_spec)


@explicit_serialize
class WriteVaspFromPMGObjects(FiretaskBase):
    """
    Write VASP files using pymatgen objects.

    Note, that although this firetask has no required params, it is
    non-functional unless at least one optional param is set.

    Optional params:
        incar (Incar): pymatgen Incar object
        poscar (Poscar): pymatgen Poscar object
        kpoints (Kpoints): pymatgen Kpoints object
        potcar (Potcar): pymatgen Potcar object
    """

    optional_params = ["incar", "poscar", "kpoints", "potcar"]

    def run_task(self, fw_spec):

EnoughBandsValidator,
    LobsterFilesValidator,
)
from custodian.lobster.jobs import LobsterJob
from pymatgen.core.structure import Structure
from pymatgen.io.lobster import Lobsterout, Lobsterin

__author__ = "Janine George, Guido Petretto"
__email__ = "janine.george@uclouvain.be, guido.petretto@uclouvain.be"

MODULE_DIR = os.path.dirname(os.path.abspath(__file__))
logger = logging.getLogger(__name__)


@explicit_serialize
class WriteLobsterinputfromIO(FiretaskBase):
    """
    will write lobsterin from POSCAR, INCAR, POTCAR
    Optional Params:
        poscar_path (str): path of POSCAR (will use "POSCAR" if not specified)
        incar_path (str): path of INCAR (will use "INCAR" if not specified)
        potcar_path (str): address to POSCAR (will use "POTCAR" if not specified)
        option (str): options as in Lobsterin.standard_calculations_from_vasp_files (will use "standard" if not
        specified)
        user_supplied_basis (dict): dictionary including the basis for each atom type
        user_lobsterin_settings (dict): dictionary that will be used to overwrite settings in Lobsterin dict
    """

    optional_params = [
        "user_supplied_basis",
        "user_lobsterin_settings",
        "poscar_path",

converged_list.append(hmodel.as_dict())

        # If multiple Heisenberg Models converged, take the maximal cutoff
        if len(converged_list) > 0:
            hmodel = converged_list[-1]  # Largest cutoff
        else:
            hmodel = None

        # Update FW spec with converged hmodel or None
        update_spec = {"converged_heisenberg_model": hmodel}

        return FWAction(update_spec=update_spec)


@explicit_serialize
class VampireMC(FiretaskBase):
    """
    If there is a suitable Heisenberg model, run Monte Carlo to compute the critical temperature.

    Args:
        db_file (str): path to file containing the database credentials.
        exchange_wf_uuid (int): Unique id for record keeping.
        parent_structure (Structure): Structure object with magmoms.
        mc_settings (dict): A configuration dict for monte carlo.

    """

    required_params = ["db_file", "exchange_wf_uuid", "parent_structure", "mc_settings"]
    optional_params = []

    def run_task(self, fw_spec):

except FileNotFoundError as exc:
                if continue_on_missing:
                    continue
                else:
                    raise exc

    def run_task(self, fw_spec):
        self.setup_copy(self.get("from_dir", None), to_dir=self.get("to_dir", None),
                        filesystem=self.get("filesystem", None),
                        files_to_copy=self.get("files_to_copy", None),
                        exclude_files=self.get("exclude_files", []),
                        suffix=self.get("suffix",None),fw_spec=fw_spec)
        self.copy_files()

@explicit_serialize
class GzipDir(FiretaskBase):
    """
    Task to gzip the current directory.

    """

    required_params = []
    optional_params = []

    def run_task(self,fw_spec=None):
        cwd = os.getcwd()
        gzip_dir(cwd)