Data Formats

@trait_docs
class SaveHDF5(Operator):
    """Operator which saves observations to HDF5.

    This creates a file for each observation.  Detector data compression can be enabled
    by specifying a tuple for each item in the detdata list.  The first item in the
    tuple is the field name.  The second item is either None, or a dictionary of FLAC
    comppression properties.  Allowed compression parameters are:

        "level": (int) the compression level
        "quanta": (float) the quantization value, only for floating point data
        "precision": (float) the fixed precision, only for floating point data

    For integer data, an empty dictionary may be passed, and FLAC compression
    will use the default level (5).  Floating point data *must* specify either the
    quanta or precision parameters.

    """

    # Class traits

    API = Int(0, help="Internal interface version for this operator")

    volume = Unicode(
        "toast_out_hdf5",
        allow_none=True,
        help="Top-level directory for the data volume",
    )

    volume_index = Unicode(
        "DEFAULT",
        allow_none=True,
        help=(
            "Path to index file.  None disables use of index.  "
            "'DEFAULT' uses the default VolumeIndex name at the top of the volume."
        ),
    )

    volume_index_fields = Dict(
        {}, help="Extra observation metadata to index.  See `VolumeIndex.append()`."
    )

    session_dirs = Bool(
        False, help="Organize observation files in session sub-directories"
    )

    unix_time_dirs = Bool(
        False, help="If True, organize files in 5 digit, time-prefix sub-directories"
    )

    # FIXME:  We should add a filtering mechanism here to dump a subset of
    # observations and / or detectors, as well as figure out subdirectory organization.

    meta = List([], allow_none=True, help="Only save this list of meta objects")

    attrs = List([], allow_none=True, help="Only save this list of obs attributes")

    detdata = List(
        [defaults.det_data, defaults.det_flags],
        help="Only save this list of detdata objects",
    )

    shared = List([], help="Only save this list of shared objects")

    intervals = List([], help="Only save this list of intervals objects")

    times = Unicode(defaults.times, help="Observation shared key for timestamps")

    config = Dict({}, help="Write this job config to the file")

    force_serial = Bool(
        False, help="Use serial HDF5 operations, even if parallel support available"
    )

    detdata_float32 = Bool(
        False, help="(Deprecated) Specify the per-field compression parameters."
    )

    detdata_in_place = Bool(
        False,
        help="If True, all compressed detector data will be decompressed and written "
        "over the input data.",
    )

    compress_detdata = Bool(
        False, help="(Deprecated) Specify the per-field compression parameters"
    )

    compress_precision = Int(
        None,
        allow_none=True,
        help="(Deprecated) Specify the per-field compression parameters",
    )

    verify = Bool(False, help="If True, immediately load data back in and verify")

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    @function_timer
    def _exec(self, data, detectors=None, **kwargs):
        log = Logger.get()

        if self.volume is None:
            msg = "You must set the volume trait prior to calling exec()"
            log.error(msg)
            raise RuntimeError(msg)

        # Warn for deprecated traits that will be removed eventually.

        if self.detdata_float32:
            msg = "The detdata_float32 option is deprecated.  Instead, specify"
            msg = " a compression quanta / precision that is appropriate for"
            msg = " each detdata field."
            warnings.warn(msg, DeprecationWarning)

        if self.compress_detdata:
            msg = "The compress_detdata option is deprecated.  Instead, specify"
            msg = " a compression quanta / precision that is appropriate for"
            msg = " each detdata field."
            warnings.warn(msg, DeprecationWarning)

        if self.compress_precision is not None:
            msg = "The compress_precision option is deprecated.  Instead, specify"
            msg = " a compression quanta / precision that is appropriate for"
            msg = " each detdata field."
            warnings.warn(msg, DeprecationWarning)

        # One process creates the top directory
        if data.comm.world_rank == 0:
            os.makedirs(self.volume, exist_ok=True)
        if data.comm.comm_world is not None:
            data.comm.comm_world.barrier()

        # Index for the volume, if we are using it.
        if self.volume_index is None:
            vindx = None
        elif self.volume_index == "DEFAULT":
            vindx = VolumeIndex(os.path.join(self.volume, VolumeIndex.default_name))
        else:
            vindx = VolumeIndex(self.volume_index)

        meta_fields = None
        if len(self.meta) > 0:
            meta_fields = list(self.meta)

        attrs_fields = None
        if len(self.attrs) > 0:
            attrs_fields = list(self.attrs)

        shared_fields = None
        if len(self.shared) > 0:
            shared_fields = list(self.shared)

        intervals_fields = None
        if len(self.intervals) > 0:
            intervals_fields = list(self.intervals)

        if len(self.detdata) > 0:
            detdata_fields = list(self.detdata)
        else:
            detdata_fields = list()

        # Handle parsing of deprecated global compression options.  All
        # new code should specify the FLAC compression parameters per
        # field.
        for ifield, field in enumerate(detdata_fields):
            if not isinstance(field, str):
                # User already specified compression parameters
                continue
            cprops = {"level": 5}
            if self.compress_detdata:
                # Try to guess what to do.
                if "flag" not in field:
                    # Might be float data
                    if self.compress_precision is None:
                        # Compress to 32bit floats
                        cprops["quanta"] = np.finfo(np.float32).eps
                    else:
                        cprops["precision"] = self.compress_precision
                detdata_fields[ifield] = (field, cprops)
            elif self.detdata_float32:
                # Implement this truncation as just compression to 32bit float
                # precision
                cprops["quanta"] = np.finfo(np.float32).eps
                detdata_fields[ifield] = (field, cprops)
            else:
                # No compression
                detdata_fields[ifield] = (field, None)

        for ob in data.obs:
            # Observations must have a name for this to work
            if ob.name is None:
                raise RuntimeError(
                    "Observations must have a name in order to save to HDF5 format"
                )

            # Check to see if any detector data objects are temporary and have just
            # a partial list of detectors.  Delete these.
            for dd in list(ob.detdata.keys()):
                if ob.detdata[dd].detectors != ob.local_detectors:
                    del ob.detdata[dd]

            time_prefix = None
            if self.unix_time_dirs:
                time_prefix = f"{ob.session.start.timestamp()}"[:5]

            if self.session_dirs:
                if time_prefix is None:
                    outdir = os.path.join(self.volume, ob.session.name)
                else:
                    outdir = os.path.join(self.volume, time_prefix, ob.session.name)
                # One process creates the top directory
                if data.comm.group_rank == 0:
                    os.makedirs(outdir, exist_ok=True)
                if data.comm.comm_group is not None:
                    data.comm.comm_group.barrier()
            else:
                if time_prefix is None:
                    outdir = self.volume
                else:
                    outdir = os.path.join(self.volume, time_prefix)

            outpath = save_hdf5(
                ob,
                outdir,
                meta=meta_fields,
                attrs=attrs_fields,
                detdata=detdata_fields,
                shared=shared_fields,
                intervals=intervals_fields,
                config=self.config,
                times=str(self.times),
                force_serial=self.force_serial,
                detdata_in_place=self.detdata_in_place,
            )

            log.info_rank(f"Wrote {outpath}", comm=data.comm.comm_group)

            # Add this observation to the volume index
            if vindx is not None:
                rel_path = os.path.relpath(outpath, start=self.volume)
                if len(self.volume_index_fields) == 0:
                    index_fields = None
                else:
                    index_fields = self.volume_index_fields
                msg = f"Add {ob.name} to index with extra fields {index_fields}"
                log.verbose_rank(msg, comm=data.comm.comm_group)
                vindx.append(ob, rel_path, indexfields=index_fields)

            if self.verify:
                # We are going to load the data back in, but first we need to make
                # a modified copy of the input observation for comparison.  This is
                # because we may have only a portion of the data on disk and we
                # might have also converted data to 32bit floats.

                loadpath = outpath

                if len(self.detdata) > 0:
                    # There might be compression info
                    if isinstance(detdata_fields[0], (tuple, list)):
                        verify_fields = [x[0] for x in detdata_fields]
                    else:
                        verify_fields = list(detdata_fields)
                else:
                    # We saved nothing
                    verify_fields = list()

                original = ob.duplicate(
                    times=str(self.times),
                    meta=meta_fields,
                    shared=shared_fields,
                    detdata=verify_fields,
                    intervals=intervals_fields,
                )

                compare = load_hdf5(
                    loadpath,
                    data.comm,
                    process_rows=ob.comm_col_size,
                    meta=meta_fields,
                    detdata=verify_fields,
                    shared=shared_fields,
                    intervals=intervals_fields,
                    force_serial=self.force_serial,
                )

                failed = 0
                fail_msg = None
                if not original.__eq__(compare, approx=True):
                    fail_msg = "Observation HDF5 verify failed:\n"
                    fail_msg += f"Input = {original}\n"
                    fail_msg += f"Loaded = {compare}\n"
                    fail_msg += f"Input signal[0] = {original.detdata['signal'][0]}\n"
                    fail_msg += f"Loaded signal[0] = {compare.detdata['signal'][0]}"
                    log.error(fail_msg)
                    failed = 1
                if data.comm.comm_group is not None:
                    failed = data.comm.comm_group.allreduce(failed, op=MPI.SUM)
                if failed:
                    raise RuntimeError(fail_msg)

        return

    def _finalize(self, data, **kwargs):
        return

    def _requires(self):
        req = {
            "meta": list(),
            "detdata": list(),
            "intervals": list(),
            "shared": [
                self.times,
            ],
        }
        if self.meta is not None:
            req["meta"].extend(self.meta)
        if self.detdata is not None:
            req["detdata"].extend(self.detdata)
        if self.intervals is not None:
            req["intervals"].extend(self.intervals)
        if self.shared is not None:
            req["shared"].extend(self.shared)
        return req

    def _provides(self):
        return dict()

@trait_docs
class LoadHDF5(Operator):
    """Operator which loads HDF5 data files into observations.

    Selected observation files are distributed across process groups in a load-balanced
    way, using the number of samples and (if using an index), the number of valid
    detectors in each observation.

    The list of observation files can be built in multiple ways:

    - If `files` is specified, that is used and `volume` is ignored.

    - If `volume_select` is specified, it should contain the conditional portion of the
      SQL select command (e.g. "where X < Y").  This will be used to extract the
      observation metadata needed to load files.

    - If `volume_select` is not specified then all observations are chosen (either
      from the volume index if specified, or using the filesystem).  This full list
      of observations then has the regex `pattern` applied to the basename of each
      observation file.

    """

    # Class traits

    API = Int(0, help="Internal interface version for this operator")

    volume = Unicode(
        None, allow_none=True, help="Top-level directory containing the data volume"
    )

    volume_index = Unicode(
        "DEFAULT",
        allow_none=True,
        help=(
            "Path to index file.  None disables use of index.  "
            "'DEFAULT' uses the default VolumeIndex name at the top of the volume."
        ),
    )

    volume_select = Unicode(
        None,
        allow_none=True,
        help="SQL selection string applied to the observation table of the index.",
    )

    pattern = Unicode(
        r".*\.h5",
        help="Regexp pattern to match files against (if not using volume_select).",
    )

    files = List(
        [],
        help="Override `volume` and load a list of files",
    )

    meta = List([], help="Only load this list of meta objects")

    attrs = List([], help="Only load this list of observation attributes")

    detdata = List(
        [defaults.det_data, defaults.det_flags],
        help="Only load this list of detdata objects",
    )

    det_select = Dict(
        {}, help="Keep a subset of detectors whose focalplane columns match."
    )

    shared = List([], help="Only load this list of shared objects")

    intervals = List([], help="Only load this list of intervals objects")

    sort_by_size = Bool(
        False, help="If True, sort observations by size before load balancing"
    )

    process_rows = Int(
        None,
        allow_none=True,
        help="The size of the rectangular process grid in the detector direction.",
    )

    force_serial = Bool(
        False, help="Use serial HDF5 operations, even if parallel support available"
    )

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    @function_timer
    def _exec(self, data, detectors=None, **kwargs):
        meta_fields = None
        if len(self.meta) > 0:
            meta_fields = list(self.meta)

        attrs_fields = None
        if len(self.attrs) > 0:
            attrs_fields = list(self.attrs)

        shared_fields = None
        if len(self.shared) > 0:
            shared_fields = list(self.shared)

        if len(self.detdata) > 0:
            detdata_fields = list(self.detdata)
        else:
            detdata_fields = list()

        intervals_fields = None
        if len(self.intervals) > 0:
            intervals_fields = list(self.intervals)

        # Get our list of observation files and relative sizes for load-balancing
        obs_props = self._get_obs_props(data.comm.comm_world)
        if len(obs_props) == 0:
            msg = f"Did not find any observations to load"
            raise RuntimeError(msg)

        # Distribute observations among groups
        obs_sizes = [x[0] for x in obs_props]
        groupdist = distribute_discrete(obs_sizes, data.comm.ngroups)
        group_firstobs = groupdist[data.comm.group].offset
        group_numobs = groupdist[data.comm.group].n_elem

        # Every group loads its observations
        for obindx in range(group_firstobs, group_firstobs + group_numobs):
            obfile = obs_props[obindx][1]

            ob = load_hdf5(
                obfile,
                data.comm,
                process_rows=self.process_rows,
                meta=meta_fields,
                attrs=attrs_fields,
                detdata=detdata_fields,
                shared=shared_fields,
                intervals=intervals_fields,
                force_serial=self.force_serial,
                det_select=self.det_select,
            )

            data.obs.append(ob)

        return

    def _get_obs_samples(self, path):
        """Helper function to extract the number of samples in each observation."""
        n_samp = 0
        with h5py.File(path, "r") as hf:
            n_samp = int(hf.attrs["observation_samples"])
        return n_samp

    def _get_obs_props(self, comm):
        """Query the index or filesystem for observations and their properties."""
        log = Logger.get()
        if comm is None:
            rank = 0
        else:
            rank = comm.rank

        # Index for the volume, if we are using it.
        if self.volume_index is None or len(self.files) > 0:
            # Either we are loading a list of files or disabling use of the index
            vindx = None
        else:
            if self.volume_index == "DEFAULT":
                index_path = os.path.join(self.volume, VolumeIndex.default_name)
            else:
                index_path = self.volume_index
            index_exists = False
            if rank == 0:
                if os.path.isfile(index_path):
                    index_exists = True
            if comm is not None:
                index_exists = comm.bcast(index_exists, root=0)
            if index_exists:
                vindx = VolumeIndex(index_path)
            else:
                msg = f"Volume index '{index_path}' does not exist, "
                msg += "scanning filesystem for observations."
                log.warning_rank(msg, comm=comm)
                vindx = None

        # If using the index, the fields we are querying
        select_prefix = "select name, path, samples, valid_dets from "
        select_prefix += VolumeIndex.obs_table
        select_prefix += " "

        obs_props = None
        if rank == 0:
            # Perform all filesystem and index queries on one process.
            obs_props = list()
            if len(self.files) > 0:
                # The user gave an explicit list of files.  We use the number of
                # samples for load-balancing.
                if self.volume is not None:
                    log.warning(
                        f'LoadHDF5: volume="{self.volume}" trait overridden by '
                        f"files={self.files}"
                    )
                for ofile in self.files:
                    fsize = self._get_obs_samples(ofile)
                    obs_props.append((fsize, ofile))
                msg = "LoadHDF5 using specified file list with sizes: "
                msg += f"{obs_props}"
                log.verbose(msg)
            else:
                # We are using the volume trait.  Get the list of relative file paths
                # for each observation.
                if self.volume is None:
                    msg = "Either the volume or a list of files must be specified"
                    log.error(msg)
                    raise RuntimeError(msg)
                if self.volume_select is None:
                    # We are selecting all observations, and matching a regex.
                    if vindx is None:
                        # We have no index, just check the filesystem.  This is slow
                        # for many observations.  Use the number of samples for load
                        # balancing.
                        if not os.path.isdir(self.volume):
                            msg = f"{self.volume} is not a valid directory."
                            raise RuntimeError(msg)
                        rel_files = VolumeIndex.find_observations(
                            self.volume, pattern_str=self.pattern
                        )
                        for rfile in rel_files:
                            full_path = os.path.join(self.volume, rfile)
                            fsize = self._get_obs_samples(full_path)
                            obs_props.append((fsize, full_path))
                        msg = "LoadHDF5 using volume with NO index and matching"
                        msg += f" filename pattern '{self.pattern}' found sizes: "
                        msg += f"{obs_props}"
                        log.verbose(msg)
                    else:
                        # We are using the index.  Get the full list of obs and then
                        # apply the regex.  We use the number of valid detector-samples
                        # for load balancing.
                        pattern = re.compile(self.pattern)
                        all_obs = vindx.select(select_prefix)
                        for oname, opath, osamples, odets in all_obs:
                            if pattern.search(opath) is not None:
                                sz = osamples * odets
                                full_path = os.path.join(self.volume, opath)
                                obs_props.append((sz, full_path))
                        msg = "LoadHDF5 using volume WITH index and matching"
                        msg += f" filename pattern '{self.pattern}' found sizes: "
                        msg += f"{obs_props}"
                        log.verbose(msg)
                else:
                    # Using a custom selection with the index.  Use the number of
                    # valid detector-samples for load balancing.
                    sel_str = f"{select_prefix}{self.volume_select}"
                    sel_obs = vindx.select(sel_str)
                    for oname, opath, osamples, odets in sel_obs:
                        sz = osamples * odets
                        full_path = os.path.join(self.volume, opath)
                        obs_props.append((sz, full_path))
                    msg = f"LoadHDF5 using volume with query '{self.volume_select}'"
                    msg += f" found sizes: {obs_props}"
                    log.verbose(msg)
            if self.sort_by_size:
                obs_props.sort(key=lambda x: x[0])
            else:
                obs_props.sort(key=lambda x: x[1])
        if comm is not None:
            obs_props = comm.bcast(obs_props, root=0)
        return obs_props

    def _finalize(self, data, **kwargs):
        return

    def _requires(self):
        return dict()

    def _provides(self):
        return dict()

@trait_docs
class SaveSpt3g(Operator):
    """Operator which saves observations to SPT3G frame files.

    This creates a directory for each observation, and writes framefiles with
    a target size.

    """

    # Class traits

    API = Int(0, help="Internal interface version for this operator")

    directory = Unicode("spt3g_data", help="Top-level export directory")

    framefile_mb = Float(100.0, help="Target frame file size in MB")

    gzip = Bool(False, help="If True, gzip compress the frame files")

    # FIXME:  We should add a filtering mechanism here to dump a subset of
    # observations and / or detectors.

    obs_export = Instance(
        klass=object,
        allow_none=True,
        help="Export class to create frames from an observation",
    )

    purge = Bool(False, help="If True, delete observation data as it is saved")

    @traitlets.validate("obs_export")
    def _check_export(self, proposal):
        ex = proposal["value"]
        if ex is not None:
            # Check that this class has an "export_rank" attribute and is callable.
            if not callable(ex):
                raise traitlets.TraitError("obs_export class must be callable")
            if not hasattr(ex, "export_rank"):
                raise traitlets.TraitError(
                    "obs_export class must have an export_rank attribute"
                )
        return ex

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        if not available:
            raise RuntimeError("spt3g is not available")

    @function_timer
    def _exec(self, data, detectors=None, **kwargs):
        log = Logger.get()

        # Check that the export class is set
        if self.obs_export is None:
            raise RuntimeError(
                "You must set the obs_export trait before calling exec()"
            )

        # One process creates the top directory
        if data.comm.world_rank == 0:
            os.makedirs(self.directory, exist_ok=True)
        if data.comm.comm_world is not None:
            data.comm.comm_world.barrier()

        for ob in data.obs:
            # Observations must have a name for this to work
            if ob.name is None:
                raise RuntimeError(
                    "Observations must have a name in order to save to SPT3G format"
                )
            ob_dir = os.path.join(self.directory, ob.name)
            if ob.comm.group_rank == 0:
                # Make observation directory.  This should NOT already exist.
                os.makedirs(ob_dir)
            if ob.comm.comm_group is not None:
                ob.comm.comm_group.barrier()

            # Export observation to frames
            frames = self.obs_export(ob)

            # If the export rank is set, then frames will be gathered to one
            # process and written.  Otherwise each process will write to
            # sequential, independent frame files.
            ex_rank = self.obs_export.export_rank
            if ex_rank is None:
                # All processes write independently
                emitter = frame_emitter(frames=frames)
                save_pipe = c3g.G3Pipeline()
                save_pipe.Add(emitter)
                fname = f"frames-{ob.comm.group_rank:04d}.g3"
                if self.gzip:
                    fname += ".gz"
                save_pipe.Add(
                    c3g.G3Writer,
                    filename=os.path.join(ob_dir, fname),
                )
                save_pipe.Run()
                del save_pipe
                del emitter
            else:
                # Gather frames to one process and write
                if ob.comm.group_rank == ex_rank:
                    emitter = frame_emitter(frames=frames)
                    save_pipe = c3g.G3Pipeline()
                    save_pipe.Add(emitter)
                    fpattern = "frames-%04u.g3"
                    if self.gzip:
                        fpattern += ".gz"
                    save_pipe.Add(
                        c3g.G3MultiFileWriter,
                        filename=os.path.join(ob_dir, fpattern),
                        size_limit=int(self.framefile_mb * 1024**2),
                    )
                    save_pipe.Run()
                    del save_pipe
                    del emitter

            if ob.comm.comm_group is not None:
                ob.comm.comm_group.barrier()

            if self.purge:
                ob.clear()

        if self.purge:
            data.obs.clear()
        return

    def _finalize(self, data, **kwargs):
        return

    def _requires(self):
        return dict()

    def _provides(self):
        return dict()

@trait_docs
class LoadSpt3g(Operator):
    """Operator which loads SPT3G data files into observations.

    This operator expects a top-level data directory.

    """

    # Class traits

    API = Int(0, help="Internal interface version for this operator")

    directory = Unicode(
        None, allow_none=True, help="Top-level directory of observations"
    )

    # FIXME:  We should add a filtering mechanism here to load a subset of
    # observations and / or detectors.

    obs_import = Instance(
        klass=object,
        allow_none=True,
        help="Import class to create observations from frame files",
    )

    @traitlets.validate("obs_import")
    def _check_import(self, proposal):
        im = proposal["value"]
        if im is not None:
            # Check that this class has an "import_rank" attribute and is callable.
            if not callable(im):
                raise traitlets.TraitError("obs_import class must be callable")
            if not hasattr(im, "import_rank"):
                raise traitlets.TraitError(
                    "obs_import class must have an import_rank attribute"
                )
        return im

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        if not available:
            raise RuntimeError("spt3g is not available")

    @function_timer
    def _exec(self, data, detectors=None, **kwargs):
        log = Logger.get()

        # Check that the import class is set
        if self.obs_import is None:
            raise RuntimeError(
                "You must set the obs_import trait before calling exec()"
            )

        # Find the process rank which will load all frames for a given group.
        im_rank = self.obs_import.import_rank
        if im_rank is None:
            raise RuntimeError(
                "The obs_import class must be configured to import frames on one process"
            )

        # FIXME:  we should add a check that self.obs_import.comm is the same as
        # the data.comm.comm_group communicator.

        # One global process computes the list of observations and their approximate
        # relative size (based on the total framefile sizes for each).

        obs_props = list()
        if data.comm.world_rank == 0:
            for root, dirs, files in os.walk(self.directory):
                for d in dirs:
                    framefiles = glob.glob(os.path.join(root, d, "*.g3"))
                    if len(framefiles) > 0:
                        # This is an observation directory
                        flist = list()
                        obtotal = 0
                        for ffile in sorted(framefiles):
                            fsize = os.path.getsize(ffile)
                            obtotal += fsize
                            flist.append(ffile)
                        obs_props.append((d, obtotal, flist))
                break
        obs_props.sort(key=lambda x: x[0])
        if data.comm.comm_world is not None:
            obs_props = data.comm.comm_world.bcast(obs_props, root=0)

        # Distribute observations among groups

        obs_sizes = [x[1] for x in obs_props]
        groupdist = distribute_discrete(obs_sizes, data.comm.ngroups)
        group_firstobs = groupdist[data.comm.group][0]
        group_numobs = groupdist[data.comm.group][1]

        # Every group creates its observations
        for obindx in range(group_firstobs, group_firstobs + group_numobs):
            obname = obs_props[obindx][0]
            obfiles = obs_props[obindx][2]

            collect = frame_collector()

            if data.comm.group_rank == im_rank:
                all_frames = list()
                # We are loading the frames
                for ffile in obfiles:
                    load_pipe = c3g.G3Pipeline()
                    load_pipe.Add(c3g.G3Reader(ffile))
                    load_pipe.Add(collect)
                    load_pipe.Run()

            # Convert collected frames to an observation
            ob = self.obs_import(collect.frames)
            data.obs.append(ob)
            del collect

        return

    def _finalize(self, data, **kwargs):
        return

    def _requires(self):
        return dict()

    def _provides(self):
        return dict()