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

# rb_utils

`rb_utils`는 백엔드 전반에서 재사용하는 유틸리티 모음 패키지입니다.

## 모듈/함수 상세

## 1) `rb_utils.parser`

- `snake_to_camel(s: str) -> str`: snake_case 문자열을 CamelCase 형태로 변환
- `camel_to_snake(s: str) -> str`: CamelCase 문자열을 snake_case로 변환
- `t_to_dict(obj: Any) -> Any`: 중첩 객체(FlatBuffer T, dict/list/tuple/set, bytes, numpy 배열)를 Python 기본 타입으로 재귀 변환
- `_normalize_filter(filter_: Any) -> dict`: `None/dict/JSON 문자열`을 공통 dict로 정규화
- `to_json(obj: Any) -> str`: `t_to_dict` 결과를 JSON 문자열로 직렬화
- `to_iso(val: str | datetime) -> str`: datetime이면 ISO 문자열, 문자열이면 그대로 반환

예시:

```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`: 제한된 AST 규칙으로 안전 평가
- `_pick_target_loop(func, provided) -> asyncio.AbstractEventLoop | None`: callable 실행 대상 loop 선택
- `call_with_matching_args(func, **provided) -> Any`: 함수 시그니처에 맞춰 인자만 골라 호출 (sync/async 공통)
- `eval_value(value, variables=None, get_global_variable=None) -> Any`: 문자열/리스트/튜플/딕셔너리를 재귀 평가
- `make_builtins_allow_most() -> dict`: 위험 builtins 제외한 builtins dict 생성

예시:

```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)`: 현재 loop에 task 생성 후 완료/예외 로깅
- `_ensure_private_loop() -> asyncio.AbstractEventLoop`: 전용 백그라운드 loop 확보
- `fire_and_log(coro, *, name=None, loop=None)`: 현재 loop/외부 loop/전용 loop에 안전하게 실행
- `_attach_done_logging(fut_or_task, *, name=None)`: Future/Task 완료 시 예외 로깅 callback 연결

예시:

```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)`:
  - 필드: `message: str`, `status_code: int | None`
  - `__str__()`: `message (status=...)` 형태 출력
  - `to_dict()`: `{message, status_code}` 반환

예시:

```python
from rb_utils.service_exception import ServiceException

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

## 5) `rb_utils.date`

- `is_valid_date(d)`: `date/str/int(timestamp)` 유효성 검사
- `timestamp_ms_to_seconds(timestamp, tz=UTC)`: ms/seconds timestamp를 seconds로 정규화
- `is_has_time(s: str)`: 문자열에 시간 정보 포함 여부 판단
- `parse_date_with_time(mode, d, tz=UTC)`: 날짜/문자열/타임스탬프를 timezone-aware datetime으로 변환
- `ensure_datetime(dt, tz)`: `date | datetime`을 지정 tz의 datetime으로 보정
- `_tz(tz_name)`: timezone 캐시 조회
- `convert_dt(local, in_tz, out_tz)`: 입력 tz 기준 datetime/date를 출력 tz로 변환
- `get_current_dt_yyyymmddhhmmss(tz)`: 현재 시각을 `YYYYMMDDHHMMSS` 문자열 반환

예시:

```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`: 다운로드용 `Content-Disposition` 헤더 생성
- `sanitize_filename(name, default="export.csv", max_name_len=100) -> str`: OS 금지문자 제거/길이 제한
- `get_env_path() -> Path | None`: 실행 위치 기준으로 `.env` 파일 탐색

예시:

```python
from rb_utils.file import sanitize_filename

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

## 7) `rb_utils.pagination`

- `class PageInfo(TypedDict)`: 페이지 응답 메타 타입 (`mode/page/pages/limit/total/sort/order`)
- `class LogsResponse(TypedDict)`: 로그 목록 응답 타입 (`items`, `pageInfo`)

## 8) `rb_utils.helper`

- `ManipulateZenohResHelper(obj)`: Zenoh 응답 객체 보정 유틸 의도 함수

주의:
- 현재 구현은 dict 처리가 아닌 `.hasAttr/.set` 접근과 즉시 `ValueError`를 포함하므로, 사용 전 동작 검증이 필요합니다.

## 실무 사용 패턴

- FlatBuffer/SDK 응답 평탄화: `t_to_dict`
- Flow step 동적 인자 평가: `eval_value`, `safe_eval_expr`
- 비동기 fire-and-forget + 예외 로깅: `fire_and_log`
- 서비스 레이어 공통 예외: `ServiceException`
- 파일 다운로드/이름 정리: `content_disposition_header`, `sanitize_filename`

## 공통 주의사항

- `flow_manager` 평가기는 허용된 연산/함수만 실행합니다.
- timezone 변환 함수는 `Asia/Seoul` 같은 IANA tz 문자열을 명시해야 합니다.
