Modules

Command-line interface (CLI) for Fleetmaster.

This module provides the main entrypoint for the Fleetmaster CLI, a tool for running hydrodynamic simulations with Capytaine. It uses the click library to define the main command group and handles global options like version and verbosity.

Functions:

Name	Description
cli	The main CLI entrypoint that registers subcommands and sets up logging.

Commands

run: Runs a batch of Capytaine simulations based on a settings file or CLI options. gui: Launches the Fleetmaster graphical user interface (GUI).

cli(verbose)

The main entrypoint for the Fleetmaster CLI.

This function configures the application's logging level based on the --verbose flag and registers all available subcommands.

Source code in src\fleetmaster\cli.py

@click.group(
    context_settings={"ignore_unknown_options": False},
    help="A CLI for running hydrodynamic simulations with Fleetmaster.",
    invoke_without_command=True,
)
@click.version_option(
    __version__,
    "--version",
    message="Version: %(version)s",
    help="Show the version and exit.",
)
@click.option(
    "-v",
    "--verbose",
    count=True,
    help="Increase verbosity level. Use -v for info, -vv for debug.",
)
def cli(verbose: int) -> None:
    """
    The main entrypoint for the Fleetmaster CLI.

    This function configures the application's logging level based on the
    --verbose flag and registers all available subcommands.
    """
    # Get the root logger of the package and set its level
    package_logger = logging.getLogger("fleetmaster")

    # Clear existing handlers to avoid duplicates
    if package_logger.hasHandlers():
        package_logger.handlers.clear()

    # Add RichHandler for beautiful console output
    handler = RichHandler(rich_tracebacks=True)
    package_logger.addHandler(handler)
    package_logger.propagate = False  # Prevent duplicate logging to the root logger

    # Define verbosity levels
    VERBOSITY_DEBUG = 2
    VERBOSITY_INFO = 1

    if verbose >= VERBOSITY_DEBUG:
        log_level = logging.DEBUG
    elif verbose == VERBOSITY_INFO:
        log_level = logging.INFO
    else:
        log_level = logging.WARNING

    package_logger.setLevel(log_level)
    for h in package_logger.handlers:
        h.setLevel(log_level)

    # If no subcommand is invoked, show the help message.
    ctx = click.get_current_context()
    if ctx.invoked_subcommand is None:
        click.echo(ctx.get_help())
        return

    if log_level <= logging.INFO:
        logger.info(
            "🚀 Fleetmaster CLI — ready to start your capytaine simulations.",
        )

EngineMesh dataclass

Represents a mesh object with its configuration.

Source code in src\fleetmaster\core\engine.py

@dataclass
class EngineMesh:
    """Represents a mesh object with its configuration."""

    name: str
    mesh: trimesh.Trimesh
    config: MeshConfig

    def copy(self) -> EngineMesh:
        """Creates a deep copy of the EngineMesh instance."""
        return EngineMesh(
            name=self.name,
            mesh=self.mesh.copy(),
            config=self.config.model_copy(deep=True),
        )

copy()

Creates a deep copy of the EngineMesh instance.

Source code in src\fleetmaster\core\engine.py

def copy(self) -> EngineMesh:
    """Creates a deep copy of the EngineMesh instance."""
    return EngineMesh(
        name=self.name,
        mesh=self.mesh.copy(),
        config=self.config.model_copy(deep=True),
    )

add_mesh_to_database(output_file, mesh_to_add, mesh_name, overwrite=False, mesh_config=None)

Adds a mesh and its geometric properties to the HDF5 database under the MESH_GROUP_NAME.

Checks if the mesh already exists by comparing SHA256 hashes. If the data is different, it will either raise a warning or overwrite if overwrite is True.

Parameters:

Name	Type	Description	Default
mesh_to_add	Trimesh	The trimesh object of the mesh to be added.	required

Source code in src\fleetmaster\core\engine.py

def add_mesh_to_database(
    output_file: Path,
    mesh_to_add: trimesh.Trimesh,
    mesh_name: str,
    overwrite: bool = False,
    mesh_config: MeshConfig | None = None,
) -> None:
    """
    Adds a mesh and its geometric properties to the HDF5 database under the MESH_GROUP_NAME.

    Checks if the mesh already exists by comparing SHA256 hashes.
    If the data is different, it will either raise a warning or overwrite if `overwrite` is True.

    Args:
        mesh_to_add: The trimesh object of the mesh to be added.
    """
    if not isinstance(mesh_to_add, trimesh.Trimesh) or mesh_to_add.is_empty:
        logger.warning(f"Attempted to add an empty or invalid mesh named '{mesh_name}' to the database. Skipping.")
        return

    mesh_group_path = f"{MESH_GROUP_NAME}/{mesh_name}"
    new_stl_content, new_hash = _get_mesh_hash(mesh_to_add)

    with h5py.File(output_file, "a") as f:
        if _handle_existing_mesh(f, mesh_group_path, new_hash, overwrite, mesh_name):
            return

        logger.debug(f"Adding mesh '{mesh_name}' to group '{MESH_GROUP_NAME}'...")
        group = f.create_group(mesh_group_path)
        _write_mesh_to_group(group, mesh_to_add, mesh_config, new_hash, new_stl_content)

create_hyddb_from_capytaine_file(filename)

Loads hydrodynamic data from a dataset produced with capytaine.

• Wave forces,
• radiation_damping and
• added_mass are read.

See Also

Rao.wave_force_from_capytaine

Source code in src\fleetmaster\core\engine.py

def create_hyddb_from_capytaine_file(filename: str | Path) -> Any:
    """Loads hydrodynamic data from a  dataset produced with capytaine.

    - Wave forces,
    - radiation_damping and
    - added_mass are read.

    See Also:
        Rao.wave_force_from_capytaine
    """

    from capytaine.io.xarray import merge_complex_values

    dataset = merge_complex_values(xr.open_dataset(filename))

    hyddb = create_hyd_from_capytaine_data(dataset)

    return hyddb

create_rao_from_capytaine_wave_force(dataset, mode)

Reads hydrodynamic data from a netCFD file created with capytaine and copies the data for the requested mode into the object.

Parameters:

Name	Type	Description	Default
dataset	Dataset	the xarray.Dataset read from the capytaine netCDF file	required
mode	Any	Name of the mode to read MotionMode	required

Returns:

Type	Description
Any	None

Examples:

_test = Rao() _test.wave_force_from_capytaine(r"capytaine.nc", MotionMode.HEAVE)

Source code in src\fleetmaster\core\engine.py

def create_rao_from_capytaine_wave_force(dataset: xr.Dataset, mode: Any) -> Any:
    """
    Reads hydrodynamic data from a netCFD file created with capytaine and copies the
    data for the requested mode into the object.

    Args:
        dataset: the xarray.Dataset read from the capytaine netCDF file
        mode: Name of the mode to read MotionMode

    Returns:
        None

    Examples:
        _test = Rao()
        _test.wave_force_from_capytaine(r"capytaine.nc", MotionMode.HEAVE)

    """
    rao = Rao()

    wave_direction = dataset["wave_direction"] * (180 / np.pi)  # convert rad to deg
    dataset = dataset.assign_coords(wave_direction=wave_direction)

    if "excitation_force" not in dataset:
        dataset["excitation_force"] = dataset["Froude_Krylov_force"] + dataset["diffraction_force"]

    cmode = MotionModeToStr(mode)

    da = dataset["excitation_force"].sel(influenced_dof=cmode)

    rao._data = xr.Dataset()

    rao._data["amplitude"] = np.abs(da)

    rao._data["phase"] = rao._data["amplitude"]  # To avoid shape mismatch,
    rao._data["phase"].values = np.angle(da)  # first copy with dummy data - then fill

    rao.mode = mode
    return rao

export_hdf5_case_to_dhyd(hdf5_file, case_group, output_dhyd_file, *, overwrite=False)

Exports one simulation case group from the Fleetmaster HDF5 database to a standalone .dhyd file.

Source code in src\fleetmaster\core\engine.py

def export_hdf5_case_to_dhyd(
    hdf5_file: str | Path,
    case_group: str,
    output_dhyd_file: str | Path,
    *,
    overwrite: bool = False,
) -> Path:
    """Exports one simulation case group from the Fleetmaster HDF5 database to a standalone .dhyd file."""
    hdf5_path = Path(hdf5_file)
    output_path = Path(output_dhyd_file)

    if not hdf5_path.exists():
        msg = f"HDF5 database not found: {hdf5_path}"
        raise FileNotFoundError(msg)

    if output_path.exists() and not overwrite:
        msg = f"Output .dhyd file already exists: {output_path}. Use --overwrite to replace it."
        raise FileExistsError(msg)

    with h5py.File(hdf5_path, "r") as stream:
        if case_group not in stream:
            available = sorted(str(name) for name in stream if name != MESH_GROUP_NAME)
            msg = (
                f"Case group '{case_group}' not found in {hdf5_path}. "
                f"Available cases: {available if available else 'none'}"
            )
            raise ValueError(msg)

    dataset = xr.open_dataset(hdf5_path, group=case_group, engine="h5netcdf")
    try:
        _write_case_to_dhyd(dataset.load(), output_path)
    finally:
        dataset.close()

    return output_path

export_hdf5_cases_to_dhyd(hdf5_file, case_groups=None, output_dhyd_file=None, *, overwrite=False)

Exports one or more simulation case groups from the Fleetmaster HDF5 database to standalone .dhyd files.

Source code in src\fleetmaster\core\engine.py

def export_hdf5_cases_to_dhyd(
    hdf5_file: str | Path,
    case_groups: list[str] | None = None,
    output_dhyd_file: str | Path | None = None,
    *,
    overwrite: bool = False,
) -> list[Path]:
    """Exports one or more simulation case groups from the Fleetmaster HDF5 database to standalone .dhyd files."""
    hdf5_path = Path(hdf5_file)

    if not hdf5_path.exists():
        msg = f"HDF5 database not found: {hdf5_path}"
        raise FileNotFoundError(msg)

    available_cases = list_case_groups_in_hdf5(hdf5_path)
    selected_cases = available_cases if case_groups is None else case_groups

    missing_cases = [case for case in selected_cases if case not in available_cases]
    if missing_cases:
        msg = f"Case group(s) not found in {hdf5_path}: {missing_cases}. Available cases: {available_cases}"
        raise ValueError(msg)

    output_template = Path(output_dhyd_file) if output_dhyd_file is not None else None
    exporting_multiple_cases = len(selected_cases) != 1
    exported_paths: list[Path] = []

    for case_group in selected_cases:
        output_path = _resolve_bulk_output_dhyd_path(hdf5_path, case_group, output_template, exporting_multiple_cases)
        exported_paths.append(
            export_hdf5_case_to_dhyd(
                hdf5_file=hdf5_path,
                case_group=case_group,
                output_dhyd_file=output_path,
                overwrite=overwrite,
            )
        )

    return exported_paths

list_case_groups_in_hdf5(hdf5_file)

Lists all simulation case groups in an HDF5 database (excluding the meshes group).

Source code in src\fleetmaster\core\engine.py

def list_case_groups_in_hdf5(hdf5_file: str | Path) -> list[str]:
    """Lists all simulation case groups in an HDF5 database (excluding the meshes group)."""
    hdf5_path = Path(hdf5_file)
    if not hdf5_path.exists():
        msg = f"HDF5 database not found: {hdf5_path}"
        raise FileNotFoundError(msg)

    with h5py.File(hdf5_path, "r") as stream:
        return sorted(str(name) for name in stream if name is not None and name != MESH_GROUP_NAME)

make_database(body, omegas, wave_directions, water_depth, water_level, forward_speed, case_label=None)

Create a dataset of BEM results for a given body and conditions.

Source code in src\fleetmaster\core\engine.py

def make_database(
    body: Any,
    omegas: list | npt.NDArray[np.float64],
    wave_directions: list | npt.NDArray[np.float64],
    water_depth: float,
    water_level: float,
    forward_speed: float,
    case_label: str | None = None,
) -> Any:
    """Create a dataset of BEM results for a given body and conditions."""
    bem_solver = _create_bem_solver(water_depth)
    problems = _build_bem_problems(body, omegas, wave_directions, water_depth, water_level, forward_speed)

    total_problems = len(problems)
    label_prefix = f"[{case_label}] " if case_label else ""
    logger.info(f"{label_prefix}Starting BEM solve for {total_problems} problems.")

    results = []
    start_time = time.perf_counter()
    progress_step = max(1, total_problems // 20)  # Roughly every 5%
    for idx, problem in enumerate(problems, start=1):
        problem_name = type(problem).__name__
        omega_val = getattr(problem, "omega", None)
        wave_direction = getattr(problem, "wave_direction", None)
        radiating_dof = getattr(problem, "radiating_dof", None)
        detail_bits = [problem_name]
        if omega_val is not None:
            detail_bits.append(f"omega={omega_val:.4f}")
        if wave_direction is not None:
            detail_bits.append(f"beta={wave_direction:.4f}")
        if radiating_dof is not None:
            detail_bits.append(f"dof={radiating_dof}")
        logger.info(
            "%sSolving %d/%d: %s",
            label_prefix,
            idx,
            total_problems,
            ", ".join(detail_bits),
        )
        results.append(bem_solver.solve(problem))

        if idx == 1 or idx == total_problems or idx % progress_step == 0:
            elapsed = max(time.perf_counter() - start_time, 1e-9)
            rate = idx / elapsed
            remaining = (total_problems - idx) / rate if rate > 0 else 0.0
            progress_pct = 100.0 * idx / total_problems
            logger.info(
                "%sBEM progress: %d/%d (%.1f%%), elapsed %.1fs, ETA %.1fs",
                label_prefix,
                idx,
                total_problems,
                progress_pct,
                elapsed,
                remaining,
            )

    database = cpt.assemble_dataset(results)

    # Rename phony dimensions that might be created by capytaine.
    # Based on user feedback, we expect phony_dim_0, 1, and 2.
    rename_map = {
        "phony_dim_0": "i",  # Likely a 3x3 matrix row
        "phony_dim_1": "j",  # Likely a 3x3 matrix column
        "phony_dim_2": "mesh_nodes",  # Likely a mesh-related dimension
    }
    # Filter for dims that actually exist in the dataset to avoid errors
    dims_to_rename = {k: v for k, v in rename_map.items() if k in database.dims}
    if dims_to_rename:
        logger.info(f"Renaming phony dimensions: {dims_to_rename}")
        database = database.rename_dims(dims_to_rename)

    for coord_name, coord_data in database.coords.items():
        if hasattr(coord_data.dtype, "categories"):  # Check for categorical dtype without pandas
            logger.debug(f"Converting coordinate '{coord_name}' from Categorical to unicode dtype.")
            database[coord_name] = database[coord_name].astype("U")

    return database

run_simulation_batch(settings)

Runs a batch of Capytaine simulations and saves all results to a single HDF5 file.

If settings.drafts is provided, it generates new meshes by translating a single base STL file for each draft. Otherwise, it processes the provided list of STL files.

Parameters:

Name	Type	Description	Default
settings	SimulationSettings	A SimulationSettings object with all necessary parameters.	required

Source code in src\fleetmaster\core\engine.py

def run_simulation_batch(settings: SimulationSettings) -> None:
    """
    Runs a batch of Capytaine simulations and saves all results to a single HDF5 file.

    If `settings.drafts` is provided, it generates new meshes by translating a single
    base STL file for each draft. Otherwise, it processes the provided list of STL files.

    Args:
        settings: A SimulationSettings object with all necessary parameters.
    """
    logger.info("Starting simulation batch...")
    try:
        output_file = _setup_output_file(settings)
    except ValueError as e:
        logger.warning(e)
        return

    # Determine the base mesh and the origin translation
    all_mesh_configs = [MeshConfig.model_validate(mc) for mc in settings.stl_files]
    all_files = [mc.file for mc in all_mesh_configs]

    origin_translation = np.array([0.0, 0.0, 0.0])
    base_mesh_path: str | None = settings.base_mesh
    if not base_mesh_path and all_files:
        base_mesh_path = all_files[0]

    if base_mesh_path:
        # Load the base mesh geometry once, as it might be needed for origin calculation or saving.
        base_mesh_trimesh = _prepare_trimesh_geometry(base_mesh_path)
        base_mesh_name = Path(base_mesh_path).stem

        if settings.base_origin:
            # If base_origin is specified, it's a point in the local coordinates of the base_mesh.
            # This point becomes the origin of our world coordinate system.
            origin_translation = np.array(settings.base_origin)
            logger.info(f"Using local point {origin_translation} from '{base_mesh_path}' as the world origin.")
        else:
            origin_translation = base_mesh_trimesh.center_mass
            logger.info(f"Database origin (center of mass of base mesh) set to: {origin_translation}")

        # Add the base mesh to the HDF5 database under the 'meshes' group.
        add_mesh_to_database(output_file, base_mesh_trimesh, base_mesh_name, overwrite=settings.overwrite_meshes)

        # Store the base reference information in the root of the HDF5 file
        with h5py.File(output_file, "a") as f:
            f.attrs["base_mesh"] = base_mesh_name
            if settings.base_origin:
                f.attrs["base_origin"] = settings.base_origin
            else:
                f.attrs["base_origin"] = origin_translation  # Store the calculated CoM as origin
    else:
        logger.warning("No base mesh provided.")

    if settings.drafts and base_mesh_path:
        if len(all_files) != 1:
            msg = f"When using --drafts, exactly one base STL file must be provided, but {len(all_files)} were given."
            logger.error(msg)
            raise ValueError(msg)

        base_mesh_name = Path(base_mesh_path).stem
        for draft in settings.drafts:
            logger.info(f"Processing for draft: {draft}")

            # Create a copy of the settings to modify for this specific draft
            draft_settings = settings.model_copy(deep=True)

            # Create a MeshConfig for this specific draft
            base_mesh_config = next((mc for mc in all_mesh_configs if mc.file == base_mesh_path), None)
            draft_translation = base_mesh_config.translation.copy() if base_mesh_config else [0.0, 0.0, 0.0]
            draft_translation[2] -= draft  # Positive draft means sinking, so subtract from Z

            # Create a unique name for this draft-specific mesh configuration
            draft_str = _format_value_for_name(draft)
            mesh_name_for_draft = f"{base_mesh_name}_draft_{draft_str}"

            draft_mesh_config = MeshConfig(file=base_mesh_path, translation=draft_translation)

            # Process this specific configuration
            _process_single_stl(
                draft_mesh_config,
                draft_settings,
                output_file,
                mesh_name_override=mesh_name_for_draft,
                origin_translation=origin_translation,
            )

    else:
        # Standard mode: process files as they are
        logger.info("Starting standard processing for provided STL files.")
        for mesh_config in all_mesh_configs:
            _process_single_stl(
                mesh_config, settings, output_file, mesh_name_override=None, origin_translation=origin_translation
            )

    logger.info(f"✅ Simulation batch finished. Results saved to {output_file}")

Main window of the Fleetmaster GUI.

MainWindow

Bases: QMainWindow

Main window of the Fleetmaster GUI.

Source code in src\fleetmaster\gui\main_window.py

class MainWindow(QMainWindow):
    """Main window of the Fleetmaster GUI."""

    def __init__(self) -> None:
        """Initialize the main window."""
        super().__init__()
        self.setWindowTitle("Fleetmaster")
        self.setCentralWidget(QLabel("Hello, World!"))

__init__()

Initialize the main window.

Source code in src\fleetmaster\gui\main_window.py

def __init__(self) -> None:
    """Initialize the main window."""
    super().__init__()
    self.setWindowTitle("Fleetmaster")
    self.setCentralWidget(QLabel("Hello, World!"))

main()

Create and show the main window.

Source code in src\fleetmaster\gui\main_window.py

def main() -> None:
    """Create and show the main window."""
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec())