kiwi-booster 0.4.6

Kiwi Booster

Python utils and classes for KiwiBot AI&Robotics team
Make a Pull Request · Report Bug · Request Feature

Table of contents

• About The Project
  • MCAP ROSBag Decoder
• Getting started
  • Installation
  • Usage
• Contributing
  • License
  • Contact

About The Project

This library contains utility functions and classes from Python that are commonly used in the AI&Robotics team. It is divided into the following sections:

• common_utils: Some common utils that are normally used in most of the projects.

  • kiwi_booster.common_utils.loggers This module contains local/cloud loggers with a predefined format: get_logger (a simple, idempotent stdlib logger with no external config file or environment variable required) and get_loggers (a structlog-based dual logger for structured/cloud logging, environment-aware via the ENV variable).

  • kiwi_booster.mixed This module contains miscellaneous utils from multiple objectives.

  • kiwi_booster.requests This module contains utils for working with HTTP requests.

  • kiwi_booster.video This module contains utils for working with videos. This includes the VideoWriter class, which is used to write videos in a predefined format. IMPORTANT: This class needs CV2 to be installed in the environment.

• gcp_utils: Utils that are related to the Google Cloud Platform.

  • kiwi_booster.gcp_utils.bigquery This module contains utils for working with BigQuery: table_exists, and the BQClient wrapper class (query/insert/load/create/partition/parameterised-query helpers), plus ingest_dataframe/upsert_rows_dedup for dedup-aware ingestion and TableSchema/ schema_from_dicts for declaring table schemas as plain Python data. BQClient's DataFrame-returning methods require the bigquery extra: pip install kiwi-booster[bigquery].

  • kiwi_booster.gcp_utils.kfp This module contains utils for working with Vertex (Kubeflow) Pipelines.

  • kiwi_booster.gcp_utils.vertex This module contains utils for submitting Vertex AI training jobs: VertexConfig, submit_container_training_job, and build_wandb_env (fetches WANDB_API_KEY from Secret Manager and merges it into the training container's environment variables). Requires the vertex extra: pip install kiwi-booster[vertex].

  • kiwi_booster.gcp_utils.secrets This module contains utils for working with Google Cloud Secrets Manager: access_secret_version(project_id, secret_id, version_id="latest") (returns the raw secret payload as bytes, defaults to the latest version) and get_token (impersonation-based Google identity token for authenticating API calls).

  • kiwi_booster.gcp_utils.storage This module contains utils for working with Google Cloud Storage: parse_gcs_path and the GCSClient wrapper class (upload/download/list/exists), plus signed-URL generation, console deep links, and auth-identity lookup (generate_download_signed_url_v4, get_signing_credentials, get_google_auth_user, gcs_to_http, gcs_to_storage_search). Image/YAML helpers (upload_image, download_image, upload_yaml, download_yaml) and threaded batch download (parallel_download) require the gcs extra: pip install kiwi-booster[gcs].

• cvat_utils: REST client for a self-hosted CVAT annotation server (v1/v2 auto-detect).

  • kiwi_booster.cvat_utils CVATAnnotationAPI (project/task/user CRUD, pagination, annotation upload; usable as a context manager) and CVATConfig (server URL, username/password, optional organization and extra headers).

• wandb_utils: Weights & Biases run lifecycle and artifact management.

  • kiwi_booster.wandb_utils WandbConfig, WandbRunManager (context manager wrapping wandb.init/wandb.finish, plus get_pl_logger() for a pytorch_lightning.loggers.WandbLogger - requires the wandb-pl extra), log_artifact, link_to_registry, get_latest_artifact, fetch_run_metrics. Requires the wandb extra: pip install kiwi-booster[wandb].

• deployment_utils: Client for the internal "Ouroboros" model deployment server.

  • kiwi_booster.deployment_utils Deployer (orchestrates GCS upload + server registration for a trained model - project- specific steps like ONNX export are the caller's responsibility), DeploymentServerConfig, deploy_model_version, get_latest_model_info, get_model_version_info, get_token (impersonation-based auth for the server), generate_model_codename (requires the deployment extra: pip install kiwi-booster[deployment]), upload_model_to_gcs, download_model_from_gcs.

• dataset_utils: Dataset download and train/val/test split helpers.

  • kiwi_booster.dataset_utils DownloadSpec, parallel_download_specs, download_from_df, write_split_txt_files, DataSplit, random_split, cluster_split (geo-temporal DBSCAN clustering to avoid data leakage between splits). Requires the dataset extra: pip install kiwi-booster[dataset].
  • kiwi_booster.dataset_utils.lightly open_dataset/attach_metadata - wraps the lightly_studio local dataset viewer (not re-exported at the package level, since it pulls in a much heavier dependency than the rest of dataset_utils). Requires the lightly extra (Python >=3.9 only): pip install kiwi-booster[lightly].

• viz_utils: Streamlit cross-model evaluation comparison dashboard.

  • kiwi_booster.viz_utils DashboardConfig, MetricsConfig, render (renders the full dashboard app - heatmap, bar chart, metrics table, and an optional prediction-image viewer driven by caller-supplied predict_fn/label_map_fn callbacks), build_colormap, make_overlay, colormap_legend_html, sample_image_paths. Requires the viz extra (streamlit needs Python >=3.10): pip install kiwi-booster[viz].

• decorators: Decorators that are used to improve the codebase.

• slack_utils: Utils that are related to Slack.

• mcap_utils: Utils related to read and decode ROSbags messages in mcap format without the need of a ROS environment. More information on the next section.

MCAP ROSBag Decoder

The mcap utils are designed as tools to decode MCAP ROSBags without needing a ROS environment. It supports various message types and provides a custom decoder factory for handling ROS2IDL-encoded messages.

IDLDecoderFactory Class The IDLDecoderFactory class extends the DecoderFactory class. It is specifically designed to work with ROS2IDL encoded messages. The class provides a method decoder_for that returns a decoder for a given message encoding and schema. Currently, it uses a dummy decoder that does nothing but can be extended to use more complex decoders.

Supported Messages The module supports decoding of the following message types:

• CompressedImage
• Image
• NavSatFix
• TFMessage
• Odometry
• PointCloud2

Each message type has a corresponding decode function that takes in the message and returns the decoded data along with some metadata.

CompressedImage and Image The decode_compressed_image and decode_image functions decode sensor_msgs/msg/CompressedImage and sensor_msgs/msg/Image messages respectively. They return a tuple containing the image as a Numpy array and a metadata dictionary. The metadata includes the image format, the image's timestamp as a datetime object, the image's frame id, and the image's step (only for decode_image).

NavSatFix The decode_navsatfix function decodes sensor_msgs/msg/NavSatFix messages. It returns a dictionary of the message's fields, the measurement's accuracy, and some metadata. The metadata includes the message's timestamp as a datetime object, the message's frame ID, the measurement's accuracy, the altitude, latitude, longitude, position covariance, and position covariance type of the measurement.

TFMessage The decode_tfmessage function decodes tf2_msgs/msg/TFMessage messages. It returns a list of dictionaries of the transforms. Each dictionary includes the child frame id, frame id, timestamp as a datetime object, translation as a 3D vector, and rotation as a Quaternion.

Odometry The decode_odometry function decodes nav_msgs/msg/Odometry messages. It returns a dictionary with the timestamp as a datetime object, frame id, child frame id, pose (including covariance, position as a 3D vector, and orientation as a Quaternion), twist (including covariance, linear as a 3D vector, and angular as a 3D vector), and covariance of the message.

PointCloud2 The decode_pointcloud function decodes sensor_msgs/msg/PointCloud2 messages. It returns a tuple containing the point cloud data as a Numpy array and a metadata dictionary. The metadata includes the message's timestamp as a datetime object, the message's frame ID, the point cloud's height, width, if the point cloud is dense, the point cloud's is_bigendian, the point cloud's point step, and the point cloud's row step.

Reading Messages To read a message, call the corresponding decode function with the message as the argument. For example, to decode a CompressedImage message, you would do:

image, metadata = decode_compressed_image(msg)

Where msg is the CompressedImage message to decode, the function returns the image as a Numpy array and the metadata as a dictionary.

Here is an example of how to use read and mcap file:

from kiwi_booster.mcap_utils.decode import (
        IDLDecoderFactory,
        decode_compressed_image,
        decode_image,
        decode_navsatfix,
        decode_odometry,
        decode_image_marker,
        decode_pointcloud
    )
from mcap.reader import make_reader
from mcap_ros2.decoder import DecoderFactory

# Initialize the list for all the topics
topics = {topic: [] for topic in topics_to_read}

# Read the rosbag file
with open(rosbag_path, "rb") as f:
    reader = make_reader(
        f, decoder_factories=[DecoderFactory(), IDLDecoderFactory()]
    )
    for schema, channel, _, decoded_msg in reader.iter_decoded_messages(
        log_time_order=True # This is used to return the messages in the real time order 
    ):
        if channel.topic in topics_to_read:
            if schema.name == "sensor_msgs/msg/CompressedImage":
                image, image_metadata = decode_compressed_image(decoded_msg)
                topics[channel.topic].append([image, image_metadata])
            elif schema.name == "sensor_msgs/msg/NavSatFix":
                latlon = decode_navsatfix(decoded_msg)
                topics[channel.topic].append(latlon)
            elif schema.name == "nav_msgs/msg/Odometry":
                odometry = decode_odometry(decoded_msg)
                topics[channel.topic].append(odometry)
            elif schema.name == "sensor_msgs/msg/Image":
                image, image_metadata = decode_image(decoded_msg)
                topics[channel.topic].append([image, image_metadata])
            elif schema.name == "visualization_msgs/msg/ImageMarker":
                marker_array = decode_image_marker(decoded_msg)
                topics[channel.topic].append(marker_array)
            elif schema.name == "sensor_msgs/msg/PointCloud2":
                cloud, metadata = decode_pointcloud(decoded_msg)
                topics[channel.topic].append([cloud, metadata])
            else:
                raise ValueError(f"Unknown schema {schema.name}")

(back to top)

Getting started

Installation

To install the package, simply run the following command:

pip install kiwi-booster

Usage

To use the package, we recommend using relative imports for each function or class you want to import to improve readability. For example, if you want to use the SlackBot class, you can import it as follows:

from kiwi_booster.slack_utils import SlackBot

slack_bot = SlackBot(
        SLACK_TOKEN,
        SLACK_CHANNEL_ID,
        SLACK_BOT_IMAGE_URL,
        image_alt_text="Bot description",
)

Or any decorator as follows:

from kiwi_booster.decorators import try_catch_log

@try_catch_log
def my_function():
    # Do something

As well, we recommend importing them in a separate section from the rest of the imports.

(back to top)

Release lines

The fleet runs two ROS2 distros that can't take the same dependency versions, so the package is maintained on two branches that both publish to the same PyPI project:

• main — ROS2 Jazzy line, versions 0.4.x and up (e.g. ipregistry ^4.0.0).
• iron — ROS2 Iron maintenance line, versions 0.3.x (e.g. ipregistry ^3.2.0). Only backport fixes here; new features go to main.

Note: 0.3.18 was published with the Jazzy-only ipregistry ^4.0.0 dependency by mistake and is not Iron-compatible — Iron consumers should pin >=0.3.17,<0.4.0,!=0.3.18 (or exactly 0.3.19 and up) until it's yanked on PyPI.

Consumers should pin to the range matching their distro, e.g. kiwi-booster = ">=0.3.19,<0.4.0" for Iron or kiwi-booster = ">=0.4.0,<0.5.0" for Jazzy.

Publishing to pypi

Publishing is automated: bump the version in pyproject.toml (poetry version <patch|minor|major>) and merge to main (Jazzy line) or iron (Iron line). The publish workflow builds and publishes to PyPI using the PYPI_API_TOKEN repo secret whenever the version on that branch hasn't already been published to PyPI.

To publish manually instead, authenticate first with pypi. Generate a token in pypi account settings and run the following command:

poetry config pypi-token.pypi <your-token>

Then, run the following commands:

poetry build
poetry publish

Contributing

If you'd like to contribute to Kiwi Booster, please feel free to submit a pull request! We're always looking for ways to improve our codebase and make it more useful to a wider range of use cases. You can also request a new feature by submitting an issue.

License

Kiwi Booster is licensed under the GNU license. See the LICENSE file for more information.

Contact

Sebastian Hernández Reyes - Machine Learning Engineer - Mail contact

Carlos Alvarez - Machine Learning Engineer Lead - Mail contact

(back to top)