DEV Community: Minwook Je

LeRobot with Custom Robot Arm

Minwook Je — Thu, 07 May 2026 02:30:38 +0000

Bring Your Own Hardware

This tutorial will explain how to integrate your own robot design into the LeRobot ecosystem and have it access all of our tools.

data collection
control pipelines
policy training
policy inference
robot base class

# Copyright 2024 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import abc
import builtins
from pathlib import Path

import draccus

from lerobot.motors import MotorCalibration
from lerobot.types import RobotAction, RobotObservation
from lerobot.utils.constants import HF_LEROBOT_CALIBRATION, ROBOTS

from .config import RobotConfig


# TODO(aliberts): action/obs typing such as Generic[ObsType, ActType] similar to gym.Env ?
# https://github.com/Farama-Foundation/Gymnasium/blob/3287c869f9a48d99454306b0d4b4ec537f0f35e3/gymnasium/core.py#L23
class Robot(abc.ABC):
    """
    The base abstract class for all LeRobot-compatible robots.

    This class provides a standardized interface for interacting with physical robots.
    Subclasses must implement all abstract methods and properties to be usable.

    Attributes:
        config_class (RobotConfig): The expected configuration class for this robot.
        name (str): The unique robot name used to identify this robot type.
    """

    # Set these in ALL subclasses
    config_class: builtins.type[RobotConfig]
    name: str

    def __init__(self, config: RobotConfig):
        self.robot_type = self.name
        self.id = config.id
        self.calibration_dir = (
            config.calibration_dir if config.calibration_dir else HF_LEROBOT_CALIBRATION / ROBOTS / self.name
        )
        self.calibration_dir.mkdir(parents=True, exist_ok=True)
        self.calibration_fpath = self.calibration_dir / f"{self.id}.json"
        self.calibration: dict[str, MotorCalibration] = {}
        if self.calibration_fpath.is_file():
            self._load_calibration()

    def __str__(self) -> str:
        return f"{self.id} {self.__class__.__name__}"

    def __enter__(self):
        """
        Context manager entry.
        Automatically connects to the camera.
        """
        self.connect()
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        """
        Context manager exit.
        Automatically disconnects, ensuring resources are released even on error.
        """
        self.disconnect()

    def __del__(self) -> None:
        """
        Destructor safety net.
        Attempts to disconnect if the object is garbage collected without cleanup.
        """
        try:
            if self.is_connected:
                self.disconnect()
        except Exception:  # nosec B110
            pass

    # TODO(aliberts): create a proper Feature class for this that links with datasets
    @property
    @abc.abstractmethod
    def observation_features(self) -> dict:
        """
        A dictionary describing the structure and types of the observations produced by the robot.
        Its structure (keys) should match the structure of what is returned by :pymeth:`get_observation`.
        Values for the dict should either be:
            - The type of the value if it's a simple value, e.g. `float` for single proprioceptive value (a joint's position/velocity)
            - A tuple representing the shape if it's an array-type value, e.g. `(height, width, channel)` for images

        Note: this property should be able to be called regardless of whether the robot is connected or not.
        """
        pass

    @property
    @abc.abstractmethod
    def action_features(self) -> dict:
        """
        A dictionary describing the structure and types of the actions expected by the robot. Its structure
        (keys) should match the structure of what is passed to :pymeth:`send_action`. Values for the dict
        should be the type of the value if it's a simple value, e.g. `float` for single proprioceptive value
        (a joint's goal position/velocity)

        Note: this property should be able to be called regardless of whether the robot is connected or not.
        """
        pass

    @property
    @abc.abstractmethod
    def is_connected(self) -> bool:
        """
        Whether the robot is currently connected or not. If `False`, calling :pymeth:`get_observation` or
        :pymeth:`send_action` should raise an error.
        """
        pass

    @abc.abstractmethod
    def connect(self, calibrate: bool = True) -> None:
        """
        Establish communication with the robot.

        Args:
            calibrate (bool): If True, automatically calibrate the robot after connecting if it's not
                calibrated or needs calibration (this is hardware-dependant).
        """
        pass

    @property
    @abc.abstractmethod
    def is_calibrated(self) -> bool:
        """Whether the robot is currently calibrated or not. Should be always `True` if not applicable"""
        pass

    @abc.abstractmethod
    def calibrate(self) -> None:
        """
        Calibrate the robot if applicable. If not, this should be a no-op.

        This method should collect any necessary data (e.g., motor offsets) and update the
        :pyattr:`calibration` dictionary accordingly.
        """
        pass

    def _load_calibration(self, fpath: Path | None = None) -> None:
        """
        Helper to load calibration data from the specified file.

        Args:
            fpath (Path | None): Optional path to the calibration file. Defaults to `self.calibration_fpath`.
        """
        fpath = self.calibration_fpath if fpath is None else fpath
        with open(fpath) as f, draccus.config_type("json"):
            self.calibration = draccus.load(dict[str, MotorCalibration], f)

    def _save_calibration(self, fpath: Path | None = None) -> None:
        """
        Helper to save calibration data to the specified file.

        Args:
            fpath (Path | None): Optional path to save the calibration file. Defaults to `self.calibration_fpath`.
        """
        fpath = self.calibration_fpath if fpath is None else fpath
        with open(fpath, "w") as f, draccus.config_type("json"):
            draccus.dump(self.calibration, f, indent=4)

    @abc.abstractmethod
    def configure(self) -> None:
        """
        Apply any one-time or runtime configuration to the robot.
        This may include setting motor parameters, control modes, or initial state.
        """
        pass

    @abc.abstractmethod
    def get_observation(self) -> RobotObservation:
        """
        Retrieve the current observation from the robot.

        Returns:
            RobotObservation: A flat dictionary representing the robot's current sensory state. Its structure
                should match :pymeth:`observation_features`.
        """

        pass

    @abc.abstractmethod
    def send_action(self, action: RobotAction) -> RobotAction:
        """
        Send an action command to the robot.

        Args:
            action (RobotAction): Dictionary representing the desired action. Its structure should match
                :pymeth:`action_features`.

        Returns:
            RobotAction: The action actually sent to the motors potentially clipped or modified, e.g. by
                safety limits on velocity.
        """
        pass

    @abc.abstractmethod
    def disconnect(self) -> None:
        """Disconnect from the robot and perform any necessary cleanup."""
        pass

[C++] Get Unique serial ID

Minwook Je — Wed, 21 Jan 2026 02:26:22 +0000

Python get device id is easy.

import uuid

def get_device_id():
    return f"BH_{uuid.getnode():012X}"

In C++

DeviceProvider.cpp

#include "DeviceIdProvider.h"
#include <iomanip> // std::setw, std::setfill
#include <sstream> // for std::stringstream
#include <vector>
#include <cstring>
#include <random>

// OS-specific headers
#include <sys/types.h>
#include <sys/socket.h> // 네트워크 인터페이스 접근용
#include <ifaddrs.h>    // getifaddrs(), 네트워크 인터페이스 리스트 조회
#include <net/if.h>     // IFF_LOOPBACK 플래그 정의

#if defined(__linux__)
#include <netpacket/packet.h> // sockaddr_ll 구조체 (ll = link layer)
#elif defined(__APPLE__)
#include <net/if_dl.h> // sockaddr_dl 구조체 (dl = data link, L2)
#endif

// 6-byte MAC address를 정수형 uint64_t 하나로 변환
// 이더넷, WiFi NIC의 MAC 주소는 6바이트(48비트)
uint64_t DeviceIdProvider::mac_to_uint64(const unsigned char *mac_ptr)
{
    uint64_t mac_val = 0;
    for (int i = 0; i < 6; i++)
    {
        // 1바이트(8비트) 왼쪽으로 밀고, 새 바이트를 Append
        mac_val = (mac_val << 8) | mac_ptr[i];
    }
    return mac_val;
}

// 시스템 MAC address 검색 (Python의 uuid.getnode 로직 참조)
uint64_t DeviceIdProvider::get_mac_address_uint64()
{
    // 네트워크 인터페이스if (eth0, wlan0, enp45s0 등) 리스트 조회용
    struct ifaddrs *ifaddr = nullptr, *ifa = nullptr;
    uint64_t candidate_uaa = 0; // 1순위: Universally Administered Address (제조사가 부여한 고유 MAC)
    uint64_t candidate_laa = 0; // 2순위: Locally Administered Address (사용자/OS가 임의로 설정한 로컬 MAC)

    // 시스템의 모든 네트워크 인터페이스 정보를 ll(링크드리스트) 또는 dl(더블링크드리스트)로 가져옴
    if (getifaddrs(&ifaddr) == -1)
    {
        return 0;
    }

    // 링크드 리스트 검색
    for (ifa = ifaddr; ifa != nullptr; ifa = ifa->ifa_next)
    {
        if (ifa->ifa_addr == nullptr)
            continue;

        // Loopback(127.0.0.1, localhost)은 제외
        if (ifa->ifa_flags & IFF_LOOPBACK)
            continue;

        const unsigned char *mac_ptr = nullptr;

        // Memory Overlay로 MAC 주소 추출
#if defined(__linux__)
        // AF_PACKET: Linux계열의 L2 링크레이어
        if (ifa->ifa_addr->sa_family == AF_PACKET)
        {
            struct sockaddr_ll *s = (struct sockaddr_ll *)ifa->ifa_addr;
            // 6바이트 MAC 주소인지 확인
            // halen(hardware address length)
            if (s->sll_halen == 6)
            {
                // sll_addr 배열로 MAC 주소 위치 얻기
                mac_ptr = (const unsigned char *)s->sll_addr;
            }
        }
#elif defined(__APPLE__)
        // AF_LINK: BSD계열의 L2 링크레이어
        if (ifa->ifa_addr->sa_family == AF_LINK)
        {
            struct sockaddr_dl *sdl = (struct sockaddr_dl *)ifa->ifa_addr;
            // 6바이트 MAC 주소인지 확인
            // alen(address length)
            if (sdl->sdl_alen == 6)
            {
                // LLADDR 매크로로 MAC 주소 위치 얻기
                mac_ptr = (const unsigned char *)LLADDR(sdl);
            }
        }
#endif

        // Validate MAC address
        if (mac_ptr)
        {
            // 1. Check for 00:00:00:00:00:00 (Invalid)
            bool is_zero = true;
            for (int i = 0; i < 6; i++)
            {
                if (mac_ptr[i] != 0)
                {
                    is_zero = false;
                    break;
                }
            }
            if (is_zero)
                continue;

            // 2. Check Multicast bit (Odd first byte means multicast, not physical MAC)
            // Example: 01:00:5e...
            // 유니캐스트가 아닌 멀티캐스트의 MAC은 한 장치의 고유 MAC이 아님
            if (mac_ptr[0] & 0x01)
                continue;

            uint64_t current_mac = mac_to_uint64(mac_ptr);

            // 3. Distinguish UAA(Universal) vs LAA(Local)
            // 첫 바이트의 두 번째 최하위 비트가 0이면 UAA, 1이면 LAA
            bool is_local = (mac_ptr[0] & 0x02);

            if (!is_local)
            {
                // Highest priority인 UAA를 발견 했으니 break;
                if (candidate_uaa == 0)
                {
                    candidate_uaa = current_mac;
                }
                break;
            }
            else
            {
                // LAA found, UAA를 찾을 때까지 loop 지속
                if (candidate_laa == 0)
                {
                    candidate_laa = current_mac;
                }
            }
        }
    }

    // getifaddrs로 할당받은 메모리 해제
    freeifaddrs(ifaddr);

    // Priority: UAA > LAA > 0
    if (candidate_uaa != 0)
        return candidate_uaa;
    if (candidate_laa != 0)
        return candidate_laa;

    return 0; // MAC not found
}

// RFC 4122(UUID 프로토콜) compliant random node ID generation
uint64_t DeviceIdProvider::generate_random_node_id()
{
    std::random_device rd;
    std::mt19937_64 gen(rd());
    std::uniform_int_distribution<uint64_t> dis(0, 0xFFFFFFFFFFFF); // 6바이트(48비트) 범위

    uint64_t random_node = dis(gen);

    // RFC 4122
    // 41번째(0...47)비트인 멀티캐스트 비트를 1로 설정하여 Locally Administered Address임을 표시
    // UUID의 경우, MAC 주소가 아님을 멀티캐스트 비트에 명시해야함.
    random_node |= (uint64_t(1) << 40);
    return random_node;
}

std::string DeviceIdProvider::getDeviceId()
{
    // 1. MAC 주소 시도
    uint64_t node_id = get_mac_address_uint64();

    // 2. 실패시 UUID 생성
    if (node_id == 0)
    {
        node_id = generate_random_node_id();
    }

    // 3. 포맷팅 (BH_XXXXXXXXXXXX)
    // 16진수, UPPERCASE, 12자리, 빈공간 0으로 채움
    std::stringstream ss;
    ss << "BH_" << std::hex << std::uppercase << std::setw(12) << std::setfill('0') << node_id;
    return ss.str();
}

DeviceIdProvider.h

#pragma once

#include <string>
#include <cstdint>

class DeviceIdProvider
{
public:
    static std::string getDeviceId();

private:
    static uint64_t get_mac_address_uint64();
    static uint64_t mac_to_uint64(const unsigned char *mac_ptr);
    static uint64_t generate_random_node_id();
};

CMakeLists.txt

# bh_device_id라는 라이브러리 생성
add_library(bh_device_id
    DeviceIdProvider.cpp
    DeviceIdProvider.h
)

# CMAKE_CURRENT_SOURCE_DIR: 현재 폴더
target_include_directories(bh_device_id PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})

add_executable(get_device_id main_tool.cc) # python에서 사용할 수 있게 main 생성
target_link_libraries(get_device_id PRIVATE bh_device_id) # bh_device_id 라이브러리를 링크

# Python 스크립트에서 실행할 수 있게, 컴파일 위치 세팅
# main.spec안에 binaries=[('../../bin/get_device_id', '.')] 로 from.. to를 지정
set_target_properties(
    get_device_id 
    PROPERTIES # 속성 설정 명시
    RUNTIME_OUTPUT_DIRECTORY # 속성 이름 및 경로를 나타내는 키
    "${PROJECT_SOURCE_DIR}/bin" # 실제 값, 프로젝트 최상위 폴더 밑의 bin 폴더
)

main_tool.cc

#include "DeviceIdProvider.h"
#include <iostream>

int main() {
    std::cout << DeviceIdProvider::getDeviceId() << std::endl;
    return 0;
}

파이썬에서는


def get_base_dir():
    # When frozen (e.g., PyInstaller), resolve paths relative to the executable.
    if getattr(sys, "frozen", False):
        if hasattr(sys, "_MEIPASS"):
            return sys._MEIPASS
        return os.path.dirname(sys.executable)
    return os.path.dirname(os.path.abspath(__file__))


def get_robot_id():
    # Only runs in PyInstaller bundle where get_device_id is at root
    # get_base_dir는 frozen(실행파일) 상태면, sys._MEIPASS를 반환
    # 만약 스크립트로 그냥 실행하는 거라면, 현재 위치를 반환
    bin_path = os.path.join(get_base_dir(), "get_device_id")

    # C++ 코드 실행
    if os.path.exists(bin_path):
        try:
            output = subprocess.check_output([bin_path], text=True).strip()
            if output:
                return output
        except Exception as e:
            print(f"[WARNING] C++ Tool execution failed: {e}")

    raise Exception(
        "C++로 부터 로봇 ID를 가져오지 못했습니다."
    )

AWS IoT

Minwook Je — Wed, 03 Dec 2025 03:29:14 +0000

1. Policy

Robot Policy

[보안 (Security)] -> [정책 (Policies)] 선택

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:*:*:client/${iot:Connection.Thing.ThingName}"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Publish", "iot:Receive"],
      "Resource": "arn:aws:iot:*:*:topic/$aws/things/${iot:Connection.Thing.ThingName}/*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:*:*:topicfilter/$aws/things/${iot:Connection.Thing.ThingName}/*"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:UpdateThingShadow", "iot:GetThingShadow"],
      "Resource": "arn:aws:iot:*:*:thing/${iot:Connection.Thing.ThingName}"
    }
  ]
}

TabletPolicy

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:*:*:client/*"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Publish", "iot:Receive", "iot:Subscribe"],
      "Resource": "arn:aws:iot:*:*:topic/$aws/things/*/shadow/*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:*:*:topicfilter/$aws/things/*/shadow/*"
    }
  ]
}

2. Thing & Attributes

Robot 종류(HW&SW): Thing Type

하드웨어 모델은 절대 바뀌지 않으니, Type으로 정의

Type_G
Type_G
Type_Tablet_

Store: Thing Group
로봇과 태블릿은 '장소' 기준으로 묶어서 관리, Static Group 사용.

Store_* prefix를 사용하는 group생성

Store_Office
Store_Google
...

NUC: Thing

실제 물리적인 기기 하나하나, Type과 Group에 소속 시킴.

e.g., Robot_001 (Type: G_X, Group: Store_Office)

mac address로 등록하고, type으로 확인하면 된다.

$ cat /sys/class/net/wlp46s0/address

Certificate

Device Certificate (crt)
pub key (key)
private key (key)
Root CA certificate
- CA1 (RSA 2048
- CA2 ECC 256

SW 버전: Device Shadow

SW 버전은 계속 업데이트되면서 변하는 값, 그러니 상태값으로 관리되게한다.

3. Robot Client

# 1. 프로젝트 폴더 생성
mkdir -p ~/robot-agent/certs

# 2. 다운로드 받은 키 파일 3개를 여기로 옮기세요.
# (파일 이름을 헷갈리지 않게 바꾸는 걸 추천합니다)
# 예시:
#   xxxxxxxx-certificate.pem.crt  ->  device_cert.pem
#   xxxxxxxx-private.pem.key      ->  private.key
#   AmazonRootCA1.pem             ->  root_ca.pem

/home/유저명/robot-agent/
  ├── certs/
  │    ├── device_cert.pem
  │    ├── private.key
  │    └── root_ca.pem
  └── test_connect.py  (이제 만들 것)

# pip 업데이트
python3 -m pip install --upgrade pip

# AWS IoT SDK v2 설치 (최신 표준)
python3 -m pip install awsiotsdk

Simple Test

vim ~/robot-agent/test_connect.py

import time
from awscrt import io, mqtt
from awsiot import mqtt_connection_builder

# ============================================================
# [설정] 여기만 본인 환경에 맞춰 수정하세요
# ============================================================
ENDPOINT = "xxxxxxxxxxxx-ats.iot.ap-northeast-2.amazonaws.com" # Console > 설정 > 엔드포인트 복사
CLIENT_ID = "Robot_Naver_01" # 아까 등록한 Thing 이름 (또는 MAC주소)

# 인증서 경로 (폴더 위치 정확히!)
PATH_TO_CERT = "./certs/device_cert.pem"
PATH_TO_KEY = "./certs/private.key"
PATH_TO_ROOT = "./certs/root_ca.pem"
# ============================================================

def on_connection_interrupted(connection, error, **kwargs):
    print(f"연결 끊김! 에러: {error}")

def on_connection_resumed(connection, return_code, session_present, **kwargs):
    print(f"연결 복구됨! (Return Code: {return_code})")

if __name__ == '__main__':
    # 1. 이벤트 루프 생성
    event_loop_group = io.EventLoopGroup(1)
    host_resolver = io.DefaultHostResolver(event_loop_group)
    client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)

    # 2. MQTT 연결 빌더 설정
    mqtt_connection = mqtt_connection_builder.mtls_from_path(
        endpoint=ENDPOINT,
        cert_filepath=PATH_TO_CERT,
        pri_key_filepath=PATH_TO_KEY,
        client_bootstrap=client_bootstrap,
        ca_filepath=PATH_TO_ROOT,
        on_connection_interrupted=on_connection_interrupted,
        on_connection_resumed=on_connection_resumed,
        client_id=CLIENT_ID,
        clean_session=False,
        keep_alive_secs=30
    )

    print(f"AWS IoT Core ({ENDPOINT}) 에 연결 시도 중...")

    # 3. 연결 수행
    connect_future = mqtt_connection.connect()

    try:
        connect_future.result(timeout=10) # 10초 대기
        print("✅ 연결 성공! (Connected)")

        # 4. 메시지 한 번 쏴보기
        topic = f"test/topic/{CLIENT_ID}"
        print(f"메시지 발행 중... (Topic: {topic})")
        mqtt_connection.publish(
            topic=topic,
            payload="Hello AWS! I am Robot.",
            qos=mqtt.QoS.AT_LEAST_ONCE
        )
        print("메시지 전송 완료.")

        # 5초 대기 후 종료
        time.sleep(5)

        disconnect_future = mqtt_connection.disconnect()
        disconnect_future.result()
        print("연결 종료.")

    except Exception as e:
        print(f"❌ 연결 실패! 설정을 확인하세요.\n에러 내용: {e}")

cd ~/robot-agent
python3 test_connect.py

AWS IoT Core (xxxxx.amazonaws.com) 에 연결 시도 중...
✅ 연결 성공! (Connected)
메시지 발행 중...
메시지 전송 완료.
연결 종료.

scp -r c4_0f_08_a4_33_0f leo@192.168.0.3:/home/leo/

Motors

Minwook Je — Sat, 29 Nov 2025 09:00:33 +0000

PWM

pulse signal

There are multiple protocol how to interpret PWM signal itself.

DC Motor

Direct Current
Using PWM

Can

Motor itself: voltage -> rotate -> power
Direction control with H-Bridge (정방향, 반대 방향 모터 방향 2개)

Can't

Angle control
Stop with exact angle (angle 유지 못함)

PWM interpret

MCU (micro controller unit) can't send consecutive voltage (only turn on or turn off).

averaged voltage + duty cycle 사용 = 연속적인 voltage 표현
duty cycle: 1주기 동안 ON 비율
voltage가 연속적으로 표현 됨으로써, motor speed 조절가능

Servo Motor

e.g, MG996R, DSSERVO

Servo = DC motor + 기어 + 컨트롤러

Servo는 각도제어가 가능한 DC 모터, 단 jitter가 발생.

Servo는 pulse가 ON된 상태로 얼마나 유지되었는지의 width로 데이터를 읽는다.

|■■———|-------------------| (1.0 ms → 0도)
|■■■——|------------------| (1.5 ms → 90도)
|■■■■—|-------------------| (2.0 ms → 180도)

jitter란? "가만히 있어도 부르르 떨거나, ±1~2도 미세하게 흔들리는 현상"

jitter가 발생하는 이유?

1. PWM 신호 불안정
50hz(20ms) 주기마다 하나의 펄스(명령)을 받는데, 이 명령이 각도를 준다. 서보는 펄스폭 1.500ms를 받아야 90°에 간다고 치자. 그런데 MCU가
1.500ms → 1.505ms → 1.495ms → 1.505ms …
이렇게 조금씩 흔들리면 서보는, "어? 90도였다가 → 91도? → 89도?" 이렇게 오해해서 미세하게 계속 조정 = 떨림(jitter) 발생.

2. 전원 불안정

전류 부족 → 순간적으로 힘이 약해짐 → 다시 보정 → 떨림

Can

Angle 변경
Angle 유지

PWM Slice

https://pico.implrust.com/core-concepts/pwm/pwm-in-rp2350.html

The RP2350 has a PWM peripheral with 12 PWM generators called slices. Each slice contains two output channels (A and B), giving you a total of 24 PWM output channels.

RP2040, RP2350에는 PWM을 만들기 위한 전용 하드웨어 블록들이 있는데, 이 블록 하나를 PWM Slice라고 부른다.

Slice = 독립적인 PWM 엔진 하나, Channel은 같은 slice에서 tick을 공유하는 output 채널. 즉 RP2350은 24개의 PWM을 output할 수 있다.

Android Basic

Minwook Je — Wed, 19 Nov 2025 02:15:59 +0000

1. Components

Activity

사용자 진입점, 여러활동이 함께 작동 가능, 서로 독립적.

화면을 추적, 시스템은 Activity가 호스팅하는 프로세스를 계속 실행.

Service

백그라운드 계속 실행 진입점

시작된 서비스: 작업 완료 전까지 서비스 계속 실행하도록 시스템에 지시. (앱을 떠나도 유지)
바인딩된 서비스:

라이브 배경화면, 알림 리스너, 화면 보호기, 입력 방법, 접근성 서비스 및 기타 여러 핵심 시스템 기능 모두 애플리케이션이 구현하는 서비스로 빌드되며 애플리케이션이 실행될 때 시스템이 바인딩됨.

Broadcast receiver

시스템이 이벤트를 앱이 시스템 전체 브로드캐스트에 응답할 수 있도록, 앱 공지사항.

Broadcast는 Intent 객체로 전달

앱에서 알람을 예약하여 사용자에게 예정된 이벤트에 대해 알리는 알림을 게시 -- Intent --> BroadcastReceiver

Content Provider

데이터베이스 추상화
개발자가 저장할 수 있는 공유된 앱 데이터 집합을 관리.

파일시스템
SQLite DB
Web

2. Activate components

Intent라는 async 메시지는 4가지 구성요소 유형인 활동, 서비스, broadcast receiver를 분해.

런타임에 각 구성요소르르 서로 바인딩

3. Manifest

AndroidManifest.xml

앱 구성 요소를 시작하려면 먼저 구성요소의 존재 여부를 확인하는 방법은 앱의 매니페스트 파일

앱이 요구하는 모든 사용자 권한
API 최솟값
앱이 요구하는 하드웨어, 소프트웨어 기능
앱을 연결해야하는 API 라이브러리

Android::Bluetooth

Minwook Je — Fri, 14 Nov 2025 06:19:41 +0000

Terms

BLE: Bluetooth Low Energy (=Bluetooth LE)
RFCOMM: 블루투스 프로토콜
- RFCOMM (TCP 사용), Radio Frequency Communications
- 하위에서 L2CAP을 사용, TCP/IP 서비스
- L2CAP (UDP 비슷), Logical Link Control and Adaption Protocol
Profiles: Defined BLE devices specification, how a device works, A device could contain more than one profile.
GATT (Generic Attribute Profile): General specification for send/receive attributes(short pieces of data) over BLE link.
- All BLE application profiles are based on GATT.
ATT (Attribute protocol)
- GATT is built on top of the Attribute Protocol (ATT).
- (= GATT/ATT)
- Each attribute is uniquely identified by a UUID
- The attributes transported by ATT are formatted as characteristics and services.
Characteristic(type)
- single value: value itself
- 0-n descriptors: metadata of single value
Service: a collection of characteristics
- e.g., Heart Rate Monitor

상호작용

Central vs peripheral

중앙: 스캔하여 advertise를 찾음
주변(peripheral): advertise

peripheral role만 있는 두 기기, 또는 central role만 있는 두 기기 끼리는 talk 불가.

GATT server <-> GATT client

Establish 이후, 어떻게 talk할지 결정.

client: sends request
server: return

상호작용 예시

휴대폰(central), BLE device를 스캔한다. BLE device는 advertise하고 있다.
휴대폰과 BLE device가 establish conn
start transferring GATT metadata each other
이때 phone이 request를 보내니, GATT client, BLE device가 GATT server.

Basic

페어링 및 결합 프로세스
- 기기 검색
- 블루투스 요청 수락
- 보안키 교환 (결합 프로세스)
- 보안 키 캐시
통신 채널 형성
세션 완료 시, 요청 시작한 기기가 검색 가능한 기기에 연결한 채널을 해제
이 경우(두 기기 모두 연결을 삭제하지 않은 경우) 향후 세션 중에 자동으로 다시 연결할 수 있습니다.

BLE

android.bluetooth.le API

Usecase

근처 기기 사이에서 소량의 데이터를 전송합니다.
근접 센서와 상호작용하여 현재 위치에 따라 사용자에게 맞춤설정된 환경을 제공합니다.

! 사용자가 BLE를 통해 기기를 다른기기와 페어링하면, 두 기기 간에 교환된 데이터에 사용자 기기의 모든 앱이 액세스 가능. 민감 데이터는 앱 레이어에 보안 구현하여 보호해야 함.

Find BLE devices

`startScan(ScanCallback)`

find하면 반드시 asap stop scanning
Never scan on a loop, set a time limit on scan

handler: worker thread -> handler -> Main Thread(or worker thread)에 메시지 전달, Message Queue (FIFO)

CLion

Minwook Je — Tue, 11 Nov 2025 03:45:24 +0000

Keymap with vscode
ctrl + `: collapse terminal
ctrl + shift + e

M3 Java / Android

Minwook Je — Mon, 10 Nov 2025 07:03:56 +0000

brew install jenv

echo 'export PATH="$HOME/.jenv/bin:$PATH"' >> ~/.zshrc
echo 'eval "$(jenv init -)"' >> ~/.zshrc
source ~/.zshrc

# check
$ jenv --version
jenv 0.5.9

temurin jdk 설치

OpenJDK는 JDK 설치하면 JRE도 함께 설치된다.

I need jdk 1.8 ver

brew install --cask temurin@8

installer: This package requires Rosetta 2 to be installed.
                Please install Rosetta 2 and then try again.
                    `sudo softwareupdate --install-rosetta

rosetta setting

sudo softwareupdate --install-rosetta

brew install --cask temurin@8 # retry
🍺  temurin@8 was successfully installed!

Check dir

/Library/Java/JavaVirtualMachines ❯ ls                                                        
temurin-8.jdk

~ ❯ jenv add /Library/Java/JavaVirtualMachines/temurin-8.jdk/Contents/Home                      15:50:40
temurin64-1.8.0.472 added
1.8.0.472 added
1.8 added

$ jenv global temurin64-1.8.0.472

CMake Intro

Minwook Je — Mon, 03 Nov 2025 10:47:12 +0000

CMake is a configuration program, sometimes called a "meta" build system. CMake generates a build system based on project, environment, and user-provided configuration information.

Step 0

https://cmake.org/cmake/help/latest/guide/tutorial/Before%20You%20Begin.html

-D: Define, preprocessing time
-l: link time

$ cmake -G # select generator

$ cmake -S # source root dir, default current working dir

$ cmake -B # build dir, output generated build system

$ cmake --build # Build source by generated build system

# make
cmake -B build
cmake --build build && ./build/hello

Step 1

https://cmake.org/cmake/help/latest/guide/tutorial/Getting%20Started%20with%20CMake.html

CMakeLists.txt = CML = lists file

root CML

cmake_minimum_required(VERSION 3.23)

project(MyProjectName)

add_executable() or add_library(): output type
target_sources(): link input -> output
target_link_libraries(): link outputs

add_executable()

This command creates a target.

Targets themselves are simply names, a handle to these collection of properties.

The artifact kind (executable, lib, header collection, etc)
Source files
Include dirs
Output File name
Dependencies
Flags (Compiler, Linker)

add_executable(MyProgram)

target_sources()

Now that we have a name for our target, we can start associating properties with it like source files we want to build and link.

target_sources(MyProgram
  PRIVATE
    main.cxx
)

Above PRIVATE(scope keyword) means main.cxx property only belongs to MyProgram and is not inheritable.

add_library()

We need to know about header files in order to build other parts of a given target.

add_library(MyLibrary)

target_sources(MyLibrary
  PRIVATE
    library_implementation.cxx

  PUBLIC
    FILE_SET myHeaders
    TYPE HEADERS # HEADERS or MODULES
    BASE_DIRS
      include
    FILES
      include/library_header.h
)

PUBLIC: allows consumers of our library to "see" the library's header files.

Notably, if the FILE_SET name is the same as the type, we don't need to provide the TYPE field.

target_sources(MyLibrary
  PRIVATE
    library_implementation.cxx

  PUBLIC
    FILE_SET HEADERS
    BASE_DIRS
      include
    FILES
      include/library_header.h
)

target_link_libraries()

It describes relationships between targets generally.

target_link_libraries(MyProgram
  PRIVATE
    MyLibrary
)

scope

PRIVATE

INTERFACE

PUBLIC

The order here is only loosely relevant. That we call target_link_libraries() prior to defining MathFunctions with add_library() doesn't matter to CMake.

add_subdirectory()

allows us to incorporate CMLs located in subdirectories

When a CMakeLists.txt in a subdirectory is being processed by CMake all relative paths described in the subdirectory CML are relative to that subdirectory, not the top-level CML.

add_subdirectory(SubdirectoryName)

Step2

Data type

The only fundamental types in CMakeLang are strings and lists.

set(var "World!")
message("Hello ${var}")

List

Every object in CMake is a string, and lists are themselves strings which contain semicolons as separators.

set(list "A;B")
list(APPEND list "C")

message(${list}) # string
message("${list}") # list(string with ; format)

set(name "Minwook")
message(name) # name
message(${name}) # Minwook
message("${name}") # Minwook


set(arr "A;B;C")
message(arr) # arr
message(${arr}) # ABC
message("${arr}") # A;B;C

Boolean

Conditionals are entirely by convention of which strings are considered true and which are considered false

function() and macro()

In CMakeLang, both function() and macro() can "see" all the variables created in all the frames above them.

But, a macro() acts semantically like a text replacement, so side effects the macro creates are visible in their calling context. If we create or change a variable in a macro, the caller will see the change.

function() creates its own variable scope.

macro(MyMac)
  set(MacVar "GLOBAL")
endmacro()


function(MyFunc)
  set(FuncVar "LOCAL")
endfunction()

MyFunc()
MyMac()
message("${FuncVar}") # x
message("${MacVar}") # GLOBAL

Step3

The names created by -D flags and option() are not normal variables, they are cache variables.

option(COMPRESSION_SOFTWARE_USE_ZLIB "Support Zlib compression" ON)
option(COMPRESSION_SOFTWARE_USE_ZSTD "Support Zstd compression" ON)

if(COMPRESSION_SOFTWARE_USE_ZLIB)
  # Same as before
# ...

$ cmake -B build \
    -DCOMPRESSION_SOFTWARE_USE_ZLIB=OFF
...
I will use Zstd!

set() can also be used to manipulate cache variables, but will not change a variable which has already been created.

set(StickyCacheVariable "I will not change" CACHE STRING "") # Cache 1
set(StickyCacheVariable "Overwrite StickyCache" CACHE STRING "") # Cache 2

message("StickyCacheVariable: ${StickyCacheVariable}")

Because -D flags are processed before any other commands, they take precedence for setting the value of a cache variable.

$ cmake \
  -DStickyCacheVariable="Commandline always wins" \
  -P StickyCacheVariable.cmake
StickyCacheVariable: Commandline always wins

FYI, cache var cannot be changed, they can be shadowed by normal variables by set() and can be unset()

set(ShadowVariable "In the shadows" CACHE STRING "")
set(ShadowVariable "Hiding the cache variable")
message("ShadowVariable: ${ShadowVariable}")

unset(ShadowVariable)
unset(ShadowVariable)
unset(ShadowVariable)
message("ShadowVariable: ${ShadowVariable}")

$ cmake -P ShadowVariable.cmake
ShadowVariable: Hiding the cache variable
ShadowVariable: In the shadows

TODO

Minwook Je — Tue, 21 Oct 2025 07:14:28 +0000

ESP-IDF(ESP IoT Development Framework) / C++
MQTT
- EMQX
- AWS IoT
Kafka MQTT integration
Iceberg catalog

Fluent Bit on CloudWatch

Minwook Je — Tue, 30 Sep 2025 07:58:54 +0000

Architecture

Event format

The Fluent Bit wire protocol represents an Event as a two-element array with a nested array as the first element:

[[TIMESTAMP, METADATA], MESSAGE]

TIMESTAMP is a timestamp in seconds as an integer or floating point value (not a string).
METADATA is an object containing event metadata, and might be empty.
MESSAGE is an object containing the event body.

`record_accessor` syntax

If you need the log group or stream name to be based on the contents of the log record itself. Use record_accessor syntax

Fluent Bit on CloudWatch

If you don't already have a namespace called amazon-cloudwatch, create one by entering the following command:

kubectl apply -f https://raw.githubusercontent.com/aws-samples/amazon-cloudwatch-container-insights/latest/k8s-deployment-manifest-templates/deployment-mode/daemonset/container-insights-monitoring/cloudwatch-namespace.yaml

# create amazon-cloudwatch namespace
apiVersion: v1
kind: Namespace
metadata:
  name: amazon-cloudwatch
  labels:
    name: amazon-cloudwatch

Run the following command to create a ConfigMap named cluster-info with the cluster name and the Region to send logs to. Replace cluster-name and cluster-region with your cluster's name and Region.

ClusterName=dev 🔥
RegionName=ap-northeast-2 🔥
FluentBitHttpPort=''
FluentBitReadFromHead='Off'
[[ ${FluentBitReadFromHead} = 'On' ]] && FluentBitReadFromTail='Off'|| FluentBitReadFromTail='On'
[[ -z ${FluentBitHttpPort} ]] && FluentBitHttpServer='Off' || FluentBitHttpServer='On'

kubectl create configmap fluent-bit-cluster-info \
--from-literal=cluster.name=${ClusterName} \
--from-literal=http.server=${FluentBitHttpServer} \
--from-literal=http.port=${FluentBitHttpPort} \
--from-literal=read.head=${FluentBitReadFromHead} \
--from-literal=read.tail=${FluentBitReadFromTail} \
--from-literal=logs.region=${RegionName} -n amazon-cloudwatch

In this command, the FluentBitHttpServer for monitoring plugin metrics is on by default. To turn it off, change the third line in the command to FluentBitHttpPort='' (empty string) in the command.
Also by default, Fluent Bit reads log files from the tail, and will capture only new logs after it is deployed. If you want the opposite, set FluentBitReadFromHead='On' and it will collect all logs in the file system.

apply fluent bit daemonset to the cluster

kubectl apply -f https://raw.githubusercontent.com/aws-samples/amazon-cloudwatch-container-insights/latest/k8s-deployment-manifest-templates/deployment-mode/daemonset/container-insights-monitoring/fluent-bit/fluent-bit.yaml

apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluent-bit
  namespace: amazon-cloudwatch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluent-bit-role
rules:
  - nonResourceURLs:
      - /metrics
    verbs:
      - get
  - apiGroups: [""]
    resources:
      - namespaces
      - pods
      - pods/logs
      - nodes
      - nodes/proxy
    verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluent-bit-role-binding
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: fluent-bit-role
subjects:
  - kind: ServiceAccount
    name: fluent-bit
    namespace: amazon-cloudwatch
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: fluent-bit-config
  namespace: amazon-cloudwatch
  labels:
    k8s-app: fluent-bit
data:
  fluent-bit.conf: |
    [SERVICE]
        Flush                     5
        Grace                     30
        Log_Level                 error
        Daemon                    off
        Parsers_File              parsers.conf
        HTTP_Server               ${HTTP_SERVER}
        HTTP_Listen               0.0.0.0
        HTTP_Port                 ${HTTP_PORT}
        storage.path              /var/fluent-bit/state/flb-storage/
        storage.sync              normal
        storage.checksum          off
        storage.backlog.mem_limit 5M

    @INCLUDE application-log.conf
    @INCLUDE dataplane-log.conf
    @INCLUDE host-log.conf

  application-log.conf: |
    [INPUT]
        Name                tail
        Tag                 application.*
        Exclude_Path        /var/log/containers/cloudwatch-agent*, /var/log/containers/fluent-bit*, /var/log/containers/aws-node*, /var/log/containers/kube-proxy*, /var/log/containers/fluentd*
        Path                /var/log/containers/*.log
        multiline.parser    docker, cri
        DB                  /var/fluent-bit/state/flb_container.db
        Mem_Buf_Limit       50MB
        Skip_Long_Lines     On
        Refresh_Interval    10
        Rotate_Wait         30
        storage.type        filesystem
        Read_from_Head      ${READ_FROM_HEAD}

    [FILTER]
        Name                kubernetes
        Match               application.*
        Kube_URL            https://kubernetes.default.svc:443
        Kube_Tag_Prefix     application.var.log.containers.
        Merge_Log           On
        K8S-Logging.Parser  On
        K8S-Logging.Exclude On
        Labels              Off
        Annotations         On
        Use_Kubelet         On
        Kubelet_Port        10250
        Buffer_Size         0

    [FILTER]
        Name                grep
        Match               application.*
        Regex               $kubernetes['annotations']['fluentbit/enabled'] true

    [FILTER]
        Name                grep
        Match               application.*
        Regex               level .

    [FILTER]
        Name                nest
        Match               *
        Operation           lift
        Nested_under        kubernetes
        Add_prefix          k8s_

    [FILTER]
        Name                record_modifier
        Match               *
        Remove_key          stream
        Remove_key          _p
        Remove_key          log
        Remove_key          k8s_namespace_name
        Remove_key          k8s_pod_ip
        Remove_key          k8s_pod_name
        Remove_key          k8s_container_hash
        Remove_key          k8s_docker_id
        Remove_key          k8s_container_image
        Remove_key          k8s_container_name
        Remove_key          k8s_annotations
        Remove_key          kubernetes

    [FILTER]
        Name                nest
        Match               *
        Operation           nest
        Wildcard            k8s_*
        Nest_under          kubernetes
        Remove_prefix       k8s_

    [OUTPUT]
        Name                cloudwatch_logs
        Match               application.*
        region              ${AWS_REGION}
        log_group_name      /eks/${CLUSTER_NAME}/pipeline
        log_stream_prefix   ${HOST_NAME}-
        auto_create_group   true
        extra_user_agent    container-insights

  dataplane-log.conf: |
    [INPUT]
        Name                systemd
        Tag                 dataplane.systemd.*
        Systemd_Filter      _SYSTEMD_UNIT=docker.service
        Systemd_Filter      _SYSTEMD_UNIT=containerd.service
        Systemd_Filter      _SYSTEMD_UNIT=kubelet.service
        DB                  /var/fluent-bit/state/systemd.db
        Path                /var/log/journal
        Read_From_Tail      ${READ_FROM_TAIL}

    [INPUT]
        Name                tail
        Tag                 dataplane.tail.*
        Path                /var/log/containers/aws-node*, /var/log/containers/kube-proxy*
        multiline.parser    docker, cri
        DB                  /var/fluent-bit/state/flb_dataplane_tail.db
        Mem_Buf_Limit       50MB
        Skip_Long_Lines     On
        Refresh_Interval    10
        Rotate_Wait         30
        storage.type        filesystem
        Read_from_Head      ${READ_FROM_HEAD}

    [FILTER]
        Name                modify
        Match               dataplane.systemd.*
        Rename              _HOSTNAME                   hostname
        Rename              _SYSTEMD_UNIT               systemd_unit
        Rename              MESSAGE                     message
        Remove_regex        ^((?!hostname|systemd_unit|message).)*$

    [FILTER]
        Name                aws
        Match               dataplane.*
        imds_version        v2

    [OUTPUT]
        Name                cloudwatch_logs
        Match               dataplane.*
        region              ${AWS_REGION}
        log_group_name      /aws/containerinsights/${CLUSTER_NAME}/dataplane
        log_stream_prefix   ${HOST_NAME}-
        auto_create_group   true
        extra_user_agent    container-insights

  host-log.conf: |
    [INPUT]
        Name                tail
        Tag                 host.dmesg
        Path                /var/log/dmesg
        Key                 message
        DB                  /var/fluent-bit/state/flb_dmesg.db
        Mem_Buf_Limit       5MB
        Skip_Long_Lines     On
        Refresh_Interval    10
        Read_from_Head      ${READ_FROM_HEAD}

    [INPUT]
        Name                tail
        Tag                 host.messages
        Path                /var/log/messages
        Parser              syslog
        DB                  /var/fluent-bit/state/flb_messages.db
        Mem_Buf_Limit       5MB
        Skip_Long_Lines     On
        Refresh_Interval    10
        Read_from_Head      ${READ_FROM_HEAD}

    [INPUT]
        Name                tail
        Tag                 host.secure
        Path                /var/log/secure
        Parser              syslog
        DB                  /var/fluent-bit/state/flb_secure.db
        Mem_Buf_Limit       5MB
        Skip_Long_Lines     On
        Refresh_Interval    10
        Read_from_Head      ${READ_FROM_HEAD}

    [FILTER]
        Name                aws
        Match               host.*
        imds_version        v2

    [OUTPUT]
        Name                cloudwatch_logs
        Match               host.*
        region              ${AWS_REGION}
        log_group_name      /aws/containerinsights/${CLUSTER_NAME}/host
        log_stream_prefix   ${HOST_NAME}.
        auto_create_group   true
        extra_user_agent    container-insights

  parsers.conf: |
    [PARSER]
        Name                syslog
        Format              regex
        Regex               ^(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<host>[^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$
        Time_Key            time
        Time_Format         %b %d %H:%M:%S

    [PARSER]
        Name                container_firstline
        Format              regex
        Regex               (?<log>(?<="log":")\S(?!\.).*?)(?<!\\)".*(?<stream>(?<="stream":").*?)".*(?<time>\d{4}-\d{1,2}-\d{1,2}T\d{2}:\d{2}:\d{2}\.\w*).*(?=})
        Time_Key            time
        Time_Format         %Y-%m-%dT%H:%M:%S.%LZ

    [PARSER]
        Name                cwagent_firstline
        Format              regex
        Regex               (?<log>(?<="log":")\d{4}[\/-]\d{1,2}[\/-]\d{1,2}[ T]\d{2}:\d{2}:\d{2}(?!\.).*?)(?<!\\)".*(?<stream>(?<="stream":").*?)".*(?<time>\d{4}-\d{1,2}-\d{1,2}T\d{2}:\d{2}:\d{2}\.\w*).*(?=})
        Time_Key            time
        Time_Format         %Y-%m-%dT%H:%M:%S.%LZ

---
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluent-bit
  namespace: amazon-cloudwatch
  labels:
    k8s-app: fluent-bit
    version: v1
    kubernetes.io/cluster-service: "true"
spec:
  selector:
    matchLabels:
      k8s-app: fluent-bit
  template:
    metadata:
      labels:
        k8s-app: fluent-bit
        version: v1
        kubernetes.io/cluster-service: "true"
    spec:
      containers:
      - name: fluent-bit
        # https://gallery.ecr.aws/aws-observability/aws-for-fluent-bit
        # Fluent Bit v4.1.1
        image: public.ecr.aws/aws-observability/aws-for-fluent-bit:3.0.0
        imagePullPolicy: Always
        env:
            - name: AWS_REGION
              valueFrom:
                configMapKeyRef:
                  name: fluent-bit-cluster-info
                  key: logs.region
            - name: CLUSTER_NAME
              valueFrom:
                configMapKeyRef:
                  name: fluent-bit-cluster-info
                  key: cluster.name
            - name: HTTP_SERVER
              valueFrom:
                configMapKeyRef:
                  name: fluent-bit-cluster-info
                  key: http.server
            - name: HTTP_PORT
              valueFrom:
                configMapKeyRef:
                  name: fluent-bit-cluster-info
                  key: http.port
            - name: READ_FROM_HEAD
              valueFrom:
                configMapKeyRef:
                  name: fluent-bit-cluster-info
                  key: read.head
            - name: READ_FROM_TAIL
              valueFrom:
                configMapKeyRef:
                  name: fluent-bit-cluster-info
                  key: read.tail
            - name: HOST_NAME
              valueFrom:
                fieldRef:
                  fieldPath: spec.nodeName
            - name: HOSTNAME
              valueFrom:
                fieldRef:
                  apiVersion: v1
                  fieldPath: metadata.name
            - name: CI_VERSION
              value: "k8s/1.3.37"
        resources:
            limits:
              memory: 200Mi
            requests:
              cpu: 100m
              memory: 100Mi
        volumeMounts:
        # Please don't change below read-only permissions
        - name: fluentbitstate
          mountPath: /var/fluent-bit/state
        - name: varlog
          mountPath: /var/log
          readOnly: true
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
        - name: fluent-bit-config
          mountPath: /fluent-bit/etc/
        - name: runlogjournal
          mountPath: /run/log/journal
          readOnly: true
        - name: dmesg
          mountPath: /var/log/dmesg
          readOnly: true
      terminationGracePeriodSeconds: 10
      hostNetwork: true
      dnsPolicy: ClusterFirstWithHostNet
      volumes:
      - name: fluentbitstate
        hostPath:
          path: /var/fluent-bit/state
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
      - name: fluent-bit-config
        configMap:
          name: fluent-bit-config
      - name: runlogjournal
        hostPath:
          path: /run/log/journal
      - name: dmesg
        hostPath:
          path: /var/log/dmesg
      serviceAccountName: fluent-bit
      nodeSelector:
        kubernetes.io/os: linux

Python custom logger

"""
minwook 2025-10
Format logger for kubernetes
"""

import logging
import json
import sys
import atexit


class PipelineLogger:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._init()
        return cls._instance

    def _init(self):
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(_JsonFormatter())

        self.logger = logging.getLogger("perception")  # logger name "perception"
        self.logger.propagate = False
        self.logger.setLevel(logging.INFO)
        self.logger.addHandler(handler)
        self._context = {}

        # flush when exit
        atexit.register(self.flush)

    def flush(self):
        for handler in self.logger.handlers:
            try:
                handler.flush()
            except Exception:
                pass
        try:
            sys.stdout.flush()
        except Exception:
            pass

    def set_context(self, module: str, site: str, date: str, sync=None, cam_num=None):
        self._context = {"ai_module": module, "site": site, "date": date}
        if sync:
            self._context["sync"] = sync

        if cam_num:
            self._context["cam_num"] = cam_num

    def info(self, msg, *args, **kwargs):
        kwargs["extra"] = {**kwargs.pop("extra", {}), **self._context}
        self.logger.info(msg, *args, **kwargs)

    def warning(self, msg, *args, **kwargs):
        kwargs["extra"] = {**kwargs.pop("extra", {}), **self._context}
        self.logger.warning(msg, *args, **kwargs)

    def error(self, msg, *args, **kwargs):
        kwargs["extra"] = {**kwargs.pop("extra", {}), **self._context}
        self.logger.error(msg, *args, **kwargs)

    def exception(self, msg, *args, exc_info=True, **kwargs):
        kwargs["extra"] = {**kwargs.pop("extra", {}), **self._context}
        self.logger.exception(msg, *args, exc_info=exc_info, **kwargs)

    def critical(self, msg, *args, **kwargs):
        kwargs["extra"] = {**kwargs.pop("extra", {}), **self._context}
        self.logger.critical(msg, *args, **kwargs)


class _JsonFormatter(logging.Formatter):
    def format(self, record) -> str:
        log_data = {
            "level": record.levelname,
            "message": record.getMessage(),
        }

        for key, value in record.__dict__.items():
            if key not in [
                "name",
                "msg",
                "args",
                "created",
                "filename",
                "funcName",
                "levelname",
                "levelno",
                "lineno",
                "module",
                "msecs",
                "message",
                "pathname",
                "process",
                "processName",
                "relativeCreated",
                "thread",
                "threadName",
                "exc_info",
                "exc_text",
                "stack_info",
            ]:
                log_data[key] = value

        return json.dumps(log_data)

usage

def graceful_shutdown(signum, frame):
    global messages, main_pid, logger, is_processing

    cur_pid = os.getpid()
    if cur_pid != main_pid:
        return

    if is_processing:
        logger.warning(f"Process is busy. But SIGTERM Occured from main process.")

    logger.info(f"Received signal: {signum}. Shutting down gracefully.")
    for msg in messages[message_idx:]:
        reset_visibility(msg)

    logger.flush()
    sys.exit(0)


signal.signal(signal.SIGTERM, graceful_shutdown)

...

def main():
    global messages, message_idx, is_processing
    MaxNumberOfMessages = 3
    VisibilityTimeout = 720 * MaxNumberOfMessages
    while True:
        try:
            # Receive a message from the SQS queue
            response = sqs.receive_message(
                QueueUrl=queue_url,
                MaxNumberOfMessages=MaxNumberOfMessages,
                WaitTimeSeconds=10,  # Long polling
                VisibilityTimeout=VisibilityTimeout,
            )
            messages = response.get("Messages", [])
            for message_idx, message in enumerate(messages):
                    ...
                    logger.set_context(
                        module="pose",
                        site=site,
                        date=date,
                        sync=sync_nnn,
                        cam_num=cam_num,
                    )



if __name__ == "__main__":
    logger = PipelineLogger()

K8s annotation

apiVersion: keda.sh/v1alpha1
kind: ScaledJob
metadata:
  name: ...
spec:
  jobTargetRef:
    template:
      metadata:
        annotations:
          fluentbit/enabled: "true"

Kafka on local K8s (Kind)

Minwook Je — Sun, 21 Sep 2025 09:57:05 +0000

Today I'm going to set Strimzi kafka on kind cluster.

1. Kind

kind is a tool for running local Kubernetes clusters using Docker container “nodes”.
kind was primarily designed for testing Kubernetes itself, but may be used for local development or CI.

kind: Quick Start

Kind: Install

# For M1 / ARM Macs
$ [ $(uname -m) = arm64 ] && curl -Lo ./kind https://kind.sigs.k8s.io/dl/v0.30.0/kind-darwin-arm64
$ chmod +x ./kind
$ sudo mv ./kind /opt/homebrew/bin/kind

Kind: Creating a Cluster

# 1. First run docker desktop
# 2. create a cluster
$ kind create cluster # Default cluster context name is `kind`.
Creating cluster "kind" ...
⠈⠑ Ensuring node image (kindest/node:v1.34.0) 🖼
 ✓ Ensuring node image (kindest/node:v1.34.0) 🖼
 ✓ Preparing nodes 📦
 ✓ Writing configuration 📜
 ✓ Starting control-plane 🕹️
 ✓ Installing CNI 🔌
 ✓ Installing StorageClass 💾
Set kubectl context to "kind-kind"
You can now use your cluster with:

kubectl cluster-info --context kind-kind

$ k config current-context
kind-kind

After creating a cluster, you can use kubectl to interact with it by using the configuration file generated by kind

By default, the cluster access configuration is stored in ${HOME}/.kube/config if $KUBECONFIG environment variable is not set.

Kind: CLI

#$ Get
$ kind get clusters

# Delete
$ kind delete cluster

# Info
$ kubectl cluster-info --context kind-{KIND_CLUSTER_NAME}

Kind: Advanced

Kind cluster with

vim ~/./kind-config.yml

kind: Cluster
apiVersion: kind.x-k8s.io/v1alpha4
# nodes: 1 control plane node and 3 workers
nodes:
- role: control-plane
- role: worker
- role: worker
- role: worker

kind create cluster --config kind-config.yml
❯ k get no
NAME                 STATUS   ROLES           AGE   VERSION
kind-control-plane   Ready    control-plane   43s   v1.34.0
kind-worker          Ready    <none>          29s   v1.34.0
kind-worker2         Ready    <none>          29s   v1.34.0
kind-worker3         Ready    <none>          29s   v1.34.0

2. Kafka: Strimzi

Strimzi Quick Start with Kind

2.1. Deploy Strimzi

First deploying the Strimzi cluster operator

$ kubectl create namespace kafka
$ kubectl create -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka

Apply the Strimzi install files:

ClusterRoles
ClusterRoleBindings
some Custom Resource Definitions (CRDs)

CRDs define the schemas used for CRs such as Kafka, KafkaTopic and so on.

The YAML files for ClusterRoles and ClusterRoleBindings downloaded from strimzi.io contain a default namespace of kafka

$ kubectl logs deployment/strimzi-cluster-operator -n kafka -f

Strimzi Quick Look

docs

Cluster Operator
Manages Kafka clusters and related components

Topic Operator
Creates, configures, and deletes Kafka topics

User Operator
Manages Kafka users and their authentication credentials

The Cluster Operator can deploy the Entity Operator, which runs the Topic Operator and User Operator in a single pod. The Entity Operator can be configured to run one or both operators.

Operators within the Strimzi architecture

A standard Kafka deployment using Strimzi might include the following components:

Kafka: cluster of broker nodes as the core component
Kafka Connect: cluster for external data connections
Kafka MirrorMaker: cluster to mirror data to another Kafka cluster
Kafka Exporter: to extract additional Kafka metrics data for monitoring
Kafka Bridge: to enable HTTP-based communication with Kafka
Cruise Control: to rebalance topic partitions across brokers

2.2. Create an Apache Kafka cluster

kafka-persistent.yml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: controller
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - controller
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        kraftMetadata: shared
        deleteClaim: false
---

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: broker
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        kraftMetadata: shared
        deleteClaim: false
---

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: 4.0.0
    metadataVersion: 4.0-IV3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
  entityOperator:
    topicOperator: {}
    userOperator: {}

$ kubectl apply -f kafka-persistent.yml -n kafka

Check

$ k get svc -A

(opt) IF YOU NEED External port

kind with external

kind: Cluster
apiVersion: kind.x-k8s.io/v1alpha4
# nodes: 1 control plane node and 3 workers
nodes:
- role: control-plane
  extraPortMappings:
  - containerPort: 30092 # for bootstrap
    hostPort: 30092
  - containerPort: 30093 # for broker 0
    hostPort: 30093
  - containerPort: 30094 # for broker 1
    hostPort: 30094
  - containerPort: 30095 # for broker 2
    hostPort: 30095
    listenAddress: "0.0.0.0" # Optional, defaults to "0.0.0.0"
    protocol: tcp # Optional, defaults to tcp
- role: worker
- role: worker
- role: worker

kafka

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: controller
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - controller
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        kraftMetadata: shared
        deleteClaim: false
---

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: broker
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        kraftMetadata: shared
        deleteClaim: false
---

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: 4.0.0
    metadataVersion: 4.0-IV3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: external # 🔥
        port: 9094
        type: nodeport
        tls: true
        configuration:
          bootstrap:
            nodePort: 30092
          brokers:
            - broker: 0
              nodePort: 30093
            - broker: 1
              nodePort: 30094
            - broker: 2
              nodePort: 30095
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
  entityOperator:
    topicOperator: {}
    userOperator: {}

2.3. Pub / Sub

producer

$ kubectl -n kafka run kafka-producer -ti --image=quay.io/strimzi/kafka:0.47.0-kafka-4.0.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic my-topic

consumer

kubectl -n kafka run kafka-consumer -ti --image=quay.io/strimzi/kafka:0.47.0-kafka-4.0.0 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic my-topic --from-beginning

2.4. Kafka: delete

# full delete kafka
$ kubectl delete namespace kafka

DEV Community: Minwook Je

LeRobot with Custom Robot Arm

Bring Your Own Hardware

[C++] Get Unique serial ID

AWS IoT

1. Policy

Robot Policy

TabletPolicy

2. Thing & Attributes

Certificate

3. Robot Client

Motors

PWM

DC Motor

Can

Can't

PWM interpret

Servo Motor

Can

PWM Slice

Android Basic

1. Components

Activity

Service

Broadcast receiver

Content Provider

2. Activate components

3. Manifest

Android::Bluetooth

Terms

상호작용

Central vs peripheral

GATT server <-> GATT client

Basic

BLE

Usecase

Find BLE devices

startScan(ScanCallback)

CLion

M3 Java / Android

temurin jdk 설치

CMake Intro

Step 0

Step 1

root CML

add_executable()

target_sources()

add_library()

target_link_libraries()

add_subdirectory()

Step2

Data type

List

Boolean

function() and macro()

Step3

TODO

Fluent Bit on CloudWatch

Architecture

Event format

record_accessor syntax

Fluent Bit on CloudWatch

Kafka on local K8s (Kind)

1. Kind

Kind: Install

Kind: Creating a Cluster

Kind: CLI

Kind: Advanced

2. Kafka: Strimzi

2.1. Deploy Strimzi

Strimzi Quick Look

2.2. Create an Apache Kafka cluster

(opt) IF YOU NEED External port

2.3. Pub / Sub

2.4. Kafka: delete

`startScan(ScanCallback)`

`record_accessor` syntax