Metadata-Version: 2.4
Name: rainbow-rb-utils
Version: 0.10.4
Summary: Rainbow Python Utilities
Author-email: Derek <dfd1123@rainbow-robotics.com>
Requires-Python: <3.14,>=3.11
Description-Content-Type: text/markdown
Requires-Dist: psutil<8.0.0,>=7.0.0
Requires-Dist: rainbow-rb-log==0.10.4

# rb_utils

`rb_utils` is a collection of utilities reused across the backend.

## Modules / Functions

## 1) `rb_utils.parser`

- `snake_to_camel(s: str) -> str`: Converts a snake_case string to CamelCase
- `camel_to_snake(s: str) -> str`: Converts a CamelCase string to snake_case
- `t_to_dict(obj: Any) -> Any`: Recursively converts nested objects (FlatBuffer T, dict/list/tuple/set, bytes, numpy arrays) into Python built-in types
- `_normalize_filter(filter_: Any) -> dict`: Normalizes `None/dict/JSON string` into a common dict
- `to_json(obj: Any) -> str`: Serializes the result of `t_to_dict` to a JSON string
- `to_iso(val: str | datetime) -> str`: Returns an ISO string for a datetime, or the string unchanged

Example:

```python
from rb_utils.parser import t_to_dict, to_json

payload_dict = t_to_dict(obj_payload)
payload_json = to_json(obj_payload)
```

## 2) `rb_utils.flow_manager`

- `safe_eval_expr(expr, variables=None, get_global_variable=None) -> Any`: Safely evaluates an expression under restricted AST rules
- `_pick_target_loop(func, provided) -> asyncio.AbstractEventLoop | None`: Selects the loop on which to run the callable
- `call_with_matching_args(func, **provided) -> Any`: Calls a function with only the arguments matching its signature (works for both sync and async)
- `eval_value(value, variables=None, get_global_variable=None) -> Any`: Recursively evaluates strings/lists/tuples/dicts
- `make_builtins_allow_most() -> dict`: Builds a builtins dict excluding dangerous builtins

Example:

```python
from rb_utils.flow_manager import eval_value

value = eval_value("a + 10", variables={"local": {"a": 5}, "global": {}})
# value == 15
```

## 3) `rb_utils.asyncio_helper`

- `spawn(coro, *, name=None, log_ex=True)`: Creates a task on the current loop and logs its completion/exceptions
- `_ensure_private_loop() -> asyncio.AbstractEventLoop`: Ensures a dedicated background loop exists
- `fire_and_log(coro, *, name=None, loop=None)`: Safely runs on the current loop, an external loop, or the dedicated loop
- `_attach_done_logging(fut_or_task, *, name=None)`: Attaches a callback that logs exceptions when a Future/Task completes

Example:

```python
from rb_utils.asyncio_helper import fire_and_log

fire_and_log(socket_client.emit("muscat/program/log", {"msg": "hello"}), name="program-log")
```

## 4) `rb_utils.service_exception`

- `class ServiceException(Exception)`:
  - Fields: `message: str`, `status_code: int | None`
  - `__str__()`: Outputs in the form `message (status=...)`
  - `to_dict()`: Returns `{message, status_code}`

Example:

```python
from rb_utils.service_exception import ServiceException

raise ServiceException("invalid request", status_code=400)
```

## 5) `rb_utils.date`

- `is_valid_date(d)`: Validates a `date/str/int(timestamp)`
- `timestamp_ms_to_seconds(timestamp, tz=UTC)`: Normalizes a ms/seconds timestamp to seconds
- `is_has_time(s: str)`: Determines whether a string contains time information
- `parse_date_with_time(mode, d, tz=UTC)`: Converts a date/string/timestamp to a timezone-aware datetime
- `ensure_datetime(dt, tz)`: Coerces a `date | datetime` into a datetime in the given tz
- `_tz(tz_name)`: Looks up a timezone from the cache
- `convert_dt(local, in_tz, out_tz)`: Converts a datetime/date from the input tz to the output tz
- `get_current_dt_yyyymmddhhmmss(tz)`: Returns the current time as a `YYYYMMDDHHMMSS` string

Example:

```python
from rb_utils.date import parse_date_with_time
from datetime import UTC

dt = parse_date_with_time("start", "2026-02-25", tz=UTC)
```

## 6) `rb_utils.file`

- `content_disposition_header(filename) -> str`: Builds a `Content-Disposition` header for downloads
- `sanitize_filename(name, default="export.csv", max_name_len=100) -> str`: Removes OS-forbidden characters and limits length
- `get_env_path() -> Path | None`: Searches for a `.env` file relative to the working location

Example:

```python
from rb_utils.file import sanitize_filename

safe_name = sanitize_filename("../../report?.csv")
```

## 7) `rb_utils.env`

- `env_get(key, default=None, start=None)`: Looks up a value from `os.environ` first, then the nearest `.env.development` → `.env`, in that order
- `load_env(start=None, override=False)`: Loads the nearest env files into `os.environ`
- `iter_env_paths(start=None)`: Returns the env files that will actually be read, in priority order
- `parse_env_file(path)`: Parses an env file into a dict

Example:

In new code, you don't need `load_env()` to read a single value directly.

```python
from rb_utils.env import env_get

username = env_get("MONGO_USERNAME")
password = env_get("MONGO_PASSWORD")
host = env_get("MONGO_HOST", "127.0.0.1")
```

If existing code uses `os.getenv(...)` heavily, load once at process startup.

```python
import os

from rb_utils.env import load_env

load_env()

username = os.getenv("MONGO_USERNAME")
password = os.getenv("MONGO_PASSWORD")
```

Default priority:

1. The `.env.development` closest to the current location
2. `.env` in the same directory
3. `.env.development` in the parent directory
4. `.env` in the parent directory

In other words, in a development environment `.env.development` overrides values in `.env`.
Production builds do not copy `.env.development`, so only `.env` applies.

If you need a different suffix, specify it explicitly:

```python
env_get("KEY", env_files=(".env.local", ".env.development", ".env"))
```

## 8) `rb_utils.pagination`

- `class PageInfo(TypedDict)`: Page response metadata type (`mode/page/pages/limit/total/sort/order`)
- `class LogsResponse(TypedDict)`: Log list response type (`items`, `pageInfo`)

## 9) `rb_utils.helper`

- `ManipulateZenohResHelper(obj)`: Function intended as a utility for fixing up Zenoh response objects

Caution:
- The current implementation uses `.hasAttr/.set` access instead of dict handling and raises `ValueError` immediately, so verify its behavior before use.

## Common Usage Patterns

- Flattening FlatBuffer/SDK responses: `t_to_dict`
- Evaluating dynamic Flow step arguments: `eval_value`, `safe_eval_expr`
- Async fire-and-forget + exception logging: `fire_and_log`
- Common service-layer exception: `ServiceException`
- File downloads / filename sanitization: `content_disposition_header`, `sanitize_filename`

## General Notes

- The `flow_manager` evaluator only executes allowed operations/functions.
- Timezone conversion functions require an explicit IANA tz string such as `Asia/Seoul`.
