Source code for kuibit.simdir

#!/usr/bin/env python3

# Copyright (C) 2020-2023 Gabriele Bozzola
# Based on code originally developed by Wolfgang Kastaun. This file may contain
# algorithms and/or structures first implemented in
# GitHub:wokast/PyCactus/PostCactus/
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 3 of the License, or (at your option) any later
# version.
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
# You should have received a copy of the GNU General Public License along with
# this program; if not, see <>.

"""This module provides easy access to Cactus data files.

A simulation directory is represented by an instance of the :py:class:`~.SimDir`
class, which provides access to all supported data types.

This is the main entry point into ``kuibit``. When a :py:class:`~.SimDir` is
initialized, the simulation directory is scanned and all the data is organized.
:py:class:`~.SimDir` objects have attributes that are interfaces to the data:
each attribute is a dictionary-like object that indexes the relevant data in
some way. For example, :py:meth:`~.timeseries` contains all the time series in
the output, indexed by the type of reduction that produced them (for example,
``norm2``, ``max``, ...).

In case of uncertainty, it is always possible to print :py:class:`~.SimDir`,
or any of its attributes, to obtain a message with the available content of
such attribute.

The function :py:func:`~.load_SimDir` can be used to load a :py:class:`~.SimDir`
saved with the method :py:meth:``.

import os
import pickle

from kuibit import (

[docs]def load_SimDir(path): """Load file produced with :py:meth:``. Pickles have to be regenerated if the version of ``kuibit`` changes. :param path: Pickle file as produced by :py:meth:``. :type path: str :returns: SimDir :rtype: :py:class:`~.SimDir` """ with open(path, "rb") as file_: sim = pickle.load(file_) if not isinstance(sim, SimDir): raise RuntimeError(f"File {path} does not contain a SimDir") return sim
[docs]class SimDir: """This class represents a Cactus simulation directory. :py:class:`~.SimDir` can be used as a context manager. For instance: .. code-block with SimDir(sim_path) as sim: print(sim) By itself, this is not very useful. It becomes more useful in conjunction with using pickles. Pickles are used to save the work done and resume it later. Since ``kuibit`` does a lot of lazy-loading, it can be useful to save the operations performed to disk and restart from them. For example, when initializing a :py:class:`~.SimDir`, the files have to be scanned and organized. It is pointless to this all the times if the simulation has not changed. For this, we use pickles. .. code-block with SimDir(sim_path, pickle_file="sim.pickle") as sim: print(sim) What happens here is that if ``pickle_file`` exists, it will be loaded. If it does not exist, it will be created. Using context managers here is useful because it automatically saves all the progress done. Alternatively, one has to call the :py:meth:`` method manually. .. warning:: When using pickles, it is important to make sure that the underlying data does not change, as the new/changed data will be not considered. To refresh :py:class:`~.SimDir`, you can always use the :py:meth:`~.rescan` method. Pickles have to be regenerated from scratch if the version of ``kuibit`` changes. Data is searched recursively in all subfolders. No particular folder structure (e.g. ``simfactory`` style) is assumed. The following attributes allow access to the supported data types: :ivar path: Top level path of simulation directory. :ivar dirs: All directories in which data is searched. :ivar logfiles: The locations of all log files (.out). :ivar errfiles: The location of all error log files (.err). :ivar ts: Scalar data of various type, see :py:class:`~.ScalarsDir` :ivar gf: Access to grid function data, see :py:class:`~.GridFunctionsDir`. :ivar gws: GW signal from the Weyl scalar multipole decomposition, see :py:class:`~.GravitationalWavesDir`. :ivar emws: EM signal from the Weyl scalar multipole decomposition, see :py:class:`~.ElectromagneticWavesDir`. :ivar horizons: Apparent horizon information, see :py:class:`~.HorizonsDir`. :ivar timers: Timer information, see :py:class:`~.TimersDir`. :ivar twopunctures: Metadata information from TwoPunctures. :py:class:`~.TwoPuncturesDir`. :ivar multipoles: Multipole components, see :py:class:`~.CactusMultipoleDir`. """ @staticmethod def _sanitize_path(path): # Make sure to have complete paths with respect to the current folder abs_path = os.path.abspath(os.path.expanduser(path)) if not os.path.isdir(abs_path): raise RuntimeError(f"Folder does not exist: {path}") return abs_path def _scan_folders(self, max_depth): """Scan all the folders in self.path up to depth ``max_depth`` and categorize all the files. :param max_depth: Maximum recursion depth to scan. :type max_depth: int """ def listdir_process_symlinks(path): """Return a list of files in path. If self.ignore_symlinks, exclude the symlinks, otherwise keep them around. """ dir_content = [os.path.join(path, p) for p in os.listdir(path)] if not self.ignore_symlinks: return dir_content return [p for p in dir_content if not os.path.islink(p)] def filter_ext(files, ext): """Return a list from the input list of file that has file extension ext.""" return [f for f in files if os.path.splitext(f)[1] == ext] def walk_rec(path, level=0): """Walk_rec is a recursive function that steps down all the subdirectories (except the ones with name defined in self.ignored_dirs) up to max_depth and add to self.allfiles the files found in the directories. """ if level >= max_depth: return self.dirs.append(path) all_files_in_path = listdir_process_symlinks(path) files_in_path = list(filter(os.path.isfile, all_files_in_path)) self.allfiles += files_in_path directories_in_path = list( filter(os.path.isdir, all_files_in_path) ) # We ignore the ones in self.ignored_dirs directories_to_scan = [ p for p in directories_in_path if (os.path.basename(p) not in self.ignored_dirs) ] # Apply walk_rec to all the subdirectory, but with level increased for p in directories_to_scan: walk_rec(p, level + 1) walk_rec(self.path) self.logfiles = filter_ext(self.allfiles, ".out") self.errfiles = filter_ext(self.allfiles, ".err") self.parfiles = filter_ext(self.allfiles, ".par") # Sort by time self.parfiles.sort(key=os.path.getmtime) self.logfiles.sort(key=os.path.getmtime) self.errfiles.sort(key=os.path.getmtime) simfac = os.path.join(self.path, "SIMFACTORY", "par") # Simfactory has a folder SIMFACTORY with a subdirectory for par files # Even if SIMFACTORY is excluded, we should include that par file if os.path.isdir(simfac): mainpar = filter_ext(listdir_process_symlinks(simfac), ".par") self.parfiles = mainpar + self.parfiles self.has_parfile = bool(self.parfiles) def __init__( self, path, max_depth=8, ignored_dirs=None, ignore_symlinks=True, pickle_file=None, ): """Constructor. :param path: Path to output of the simulation. :type path: str :param max_depth: Maximum recursion depth for subfolders. :type max_depth: int :param ignored_dirs: Names of folders to ignore (e.g. SIMFACTORY). :type ignored_dirs: set :param ignore_symlink: If True, do not consider symlinks. :type ignore_symlink: bool :param pickle_file: If ``pickle_file`` is not None, do not scan the folders and load the pickle file. All the other parameters are ignored. :type pickle_file: bool Parfiles (``*.par``) will be searched in all data directories and the top-level SIMFACTORY/par folder, if it exists. The parfile in the latter folder, if available, or else the oldest parfile in any of the data directories, will be used to extract the simulation parameters. Logfiles (``*.out``) and errorfiles (``*.err``) will be searched for in all data directories. """ if ignored_dirs is None: ignored_dirs = {"SIMFACTORY", "report", "movies", "tmp", "temp"} # We update self.path in _sanitize_path to make sure it is an absolute # path self.path = self._sanitize_path(str(path)) self.max_depth = int(max_depth) self.ignored_dirs = ignored_dirs self.ignore_symlinks = ignore_symlinks self.dirs = [] self.parfiles = [] self.logfiles = [] self.errfiles = [] self.allfiles = [] self.has_parfile = False self.__timeseries = None self.__multipoles = None self.__gravitationalwaves = None self.__electromagneticwaves = None self.__gridfunctions = None self.__horizons = None self.__timers = None self.__twopunctures = None if (pickle_file is None) or (not os.path.exists(pickle_file)): self._populate() else: sim = load_SimDir(pickle_file) # Overwrite all the local variables self.__dict__ = sim.__dict__ # We set this later, so that if it was read from the pickle, we override # it in such a way that we have consistency. self.pickle_file = pickle_file def _populate(self): """Scan the folders and populate basic attributes.""" self._scan_folders(self.max_depth) self.__timeseries = None self.__multipoles = None self.__gravitationalwaves = None self.__electromagneticwaves = None self.__gridfunctions = None self.__horizons = None self.__timers = None self.__twopunctures = None
[docs] def rescan(self): """Reset the SimDir and rescan all the files.""" self._populate()
@property def timeseries(self): """Return all the available timeseries in the data. :returns: Interface to all the timeseries in the directory. :rtype: :py:class:`~.ScalarsDir` """ if self.__timeseries is None: self.__timeseries = cactus_scalars.ScalarsDir(self) return self.__timeseries ts = timeseries @property def multipoles(self): """Return all the available multipole data. :returns: Interface to all the multipole data in the directory. :rtype: :py:class:`~.MultipolesDir` """ if self.__multipoles is None: self.__multipoles = cactus_multipoles.MultipolesDir(self) return self.__multipoles @property def gravitationalwaves(self): """Return all the available ``Psi4`` data. :returns: Interface to all the ``Psi4`` data in the directory. :rtype: :py:class:`~.GravitationalWavesDir` """ if self.__gravitationalwaves is None: self.__gravitationalwaves = cactus_waves.GravitationalWavesDir( self ) return self.__gravitationalwaves gws = gravitationalwaves @property def electromagneticwaves(self): """Return all the available ``Phi2`` data. :returns: Interface to all the ``Phi2`` data in the directory. :rtype: :py:class:`~.ElectromagneticWavesDir` """ if self.__electromagneticwaves is None: self.__electromagneticwaves = cactus_waves.ElectromagneticWavesDir( self ) return self.__electromagneticwaves emws = electromagneticwaves @property def gridfunctions(self): """Return all the available grid data. :returns: Interface to all the grid data in the directory. :rtype: :py:class:`~.GridFunctionsDir` """ if self.__gridfunctions is None: self.__gridfunctions = cactus_grid_functions.GridFunctionsDir(self) return self.__gridfunctions gf = gridfunctions @property def horizons(self): """Return all the available horizon data. :returns: Interface to all the horizon data in the directory. :rtype: :py:class:`~.HorizonsDir` """ if self.__horizons is None: self.__horizons = cactus_horizons.HorizonsDir(self) return self.__horizons @property def timers(self) -> cactus_timers.TimersDir: """Return all the available timertree data. :returns: Interface to all the timertree data in the directory. :rtype: :py:class:`~.TimertreeDir` """ if self.__timers is None: self.__timers = cactus_timers.TimersDir(self) return self.__timers @property def twopunctures(self): """Return the metadata for TwoPunctures. :returns: Interface to the metadata in TwoPunctures. :rtype: :py:class:`~.TwoPuncturesDir` """ if self.__twopunctures is None: self.__twopunctures = cactus_twopunctures.TwoPuncturesDir(self) return self.__twopunctures def __str__(self): header = f"Indexed {len(self.allfiles)} files" header += f" and {len(self.dirs)} subdirectories\n" ts_ret = f"{self.ts}" mp_ret = f"{self.multipoles}" gf_ret = f"{}" if len(self.gravitationalwaves) > 0: gw_ret = "Available gravitational wave data" else: gw_ret = "" if len(self.electromagneticwaves) > 0: em_ret = "Available electromagnetic wave data" else: em_ret = "" hor_ret = f"{self.horizons}" tim_ret = f"\n{self.timers}" return ( header + ts_ret + mp_ret + gw_ret + em_ret + gf_ret + hor_ret + tim_ret ) def __enter__(self): """This is classed when the object is used as a context manager.""" # All the work in done in __init__ return self def __exit__(self, _1, _2, _3): """Save the SimDir to disk as pickle. This is called when the object is used as a context manager. """ if self.pickle_file is not None:
[docs] def save(self, path): """Save this object as a pickle. The object can be loaded with the function :py:func:`~.load_SimDir`. :param path: Path where to save the file. :type path: str """ with open(path, "wb") as file_: pickle.dump(self, file_, protocol=pickle.HIGHEST_PROTOCOL)