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

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)

make_database(body, omegas, wave_directions, water_depth, water_level, forward_speed)

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,
) -> Any:
    """Create a dataset of BEM results for a given body and conditions."""
    bem_solver = cpt.BEMSolver()
    problems: list[Any] = []
    logger.debug(f"Solving for water_depth={water_depth} water_level={water_level} forward_speed={forward_speed}")
    for omega in omegas:
        logger.debug(f"RadiationProblem and DiffractionProblem for omega {omega}")
        problems.extend(
            cpt.RadiationProblem(
                omega=omega,
                body=body,
                radiating_dof=dof,
                water_depth=water_depth,
                free_surface=water_level,
                forward_speed=forward_speed,
            )
            for dof in body.dofs
        )
        for wave_direction in wave_directions:
            logger.debug(f"DiffractionProblem for wave_direction {wave_direction} ")
            problems.append(
                cpt.DiffractionProblem(
                    omega=omega,
                    body=body,
                    wave_direction=wave_direction,
                    water_depth=water_depth,
                    free_surface=water_level,
                    forward_speed=forward_speed,
                )
            )

    results = [bem_solver.solve(problem) for problem in problems]

    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 string dtype.")
            database[coord_name] = database[coord_name].astype(str)

    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())