# Implementasi Robot Processor Anda Sendiri Dalam tutorial ini, Anda akan belajar cara mengimplementasikan Robot Processor Anda sendiri. Tutorial ini dimulai dengan mengeksplorasi kebutuhan akan prosesor kustom, kemudian menggunakan `NormalizerProcessorStep` sebagai contoh berjalan untuk menjelaskan cara mengimplementasikan, mengonfigurasi, dan menserialisasi prosesor. Untuk langkah-langkah prosesor yang disertakan dengan LeRobot, lihat [Common Processor Steps](./introduction_processors#common-processor-steps). ## Mengapa Anda membutuhkan prosesor kustom? Dalam kebanyakan kasus, saat membaca data mentah dari sensor atau saat model menghasilkan aksi, Anda perlu memproses data ini agar kompatibel dengan sistem target Anda. Misalnya, kebutuhan umum adalah menormalisasi rentang data agar sesuai untuk jaringan saraf. `NormalizerProcessorStep` LeRobot menangani tugas penting ini: ```python # Input: raw joint positions in [0, 180] degrees raw_action = torch.tensor([90.0, 45.0, 135.0]) # After processing: normalized to [-1, 1] range for model training normalizer = NormalizerProcessorStep(features=features, norm_map=norm_map, stats=dataset_stats) normalized_result = normalizer(transition) # ... ``` Kebutuhan pemrosesan umum lainnya meliputi: - **Penempatan perangkat**: Memindahkan tensor antara CPU/GPU dan mengonversi tipe data - **Konversi format**: Mengubah antara struktur data yang berbeda - **Batching**: Menambahkan/menghapus dimensi batch untuk kompatibilitas model - **Batasan keselamatan**: Menerapkan batasan pada perintah robot ```python # Example pipeline combining multiple processors pipeline = PolicyProcessorPipeline([ RenameObservationsProcessorStep(rename_map={}), AddBatchDimensionProcessorStep(), NormalizerProcessorStep(features=features, norm_map=norm_map, stats=stats), DeviceProcessorStep(device="cuda"), # ... ]) ``` LeRobot menyediakan mekanisme pipeline untuk mengimplementasikan urutan langkah pemrosesan untuk data input dan aksi output, sehingga memudahkan komposisi transformasi ini dalam urutan yang tepat untuk kinerja optimal. ## Bagaimana cara mengimplementasikan processor Anda sendiri? Kami akan menggunakan `NormalizerProcessorStep` sebagai contoh utama karena menunjukkan pola pemroses esensial, termasuk manajemen status, serialisasi konfigurasi, dan penanganan tensor yang sering Anda butuhkan. Siapkan urutan langkah pemrosesan yang diperlukan untuk masalah Anda. Langkah pemrosesan adalah kelas yang mengimplementasikan metode-metode berikut: - `__call__`: mengimplementasikan langkah pemrosesan untuk transisi input. - `get_config`: mengambil konfigurasi langkah pemroses. - `state_dict`: mendapatkan status langkah pemroses. - `load_state_dict`: memuat state dari langkah processor. - `reset`: mengatur ulang status langkah pemroses. - `transform_features`: menjelaskan bagaimana langkah prosesor memodifikasi ruang fitur. ### Implementasikan metode `__call__` Metode `__call__` adalah inti dari langkah pemroses Anda. Metode ini menerima `EnvTransition` dan mengembalikan `EnvTransition` yang telah dimodifikasi. Berikut cara kerja `NormalizerProcessorStep`: ```python @dataclass @ProcessorStepRegistry.register("normalizer_processor") class NormalizerProcessorStep(ProcessorStep): """Normalize observations/actions using dataset statistics.""" features: dict[str, PolicyFeature] norm_map: dict[FeatureType, NormalizationMode] stats: dict[str, dict[str, Any]] | None = None eps: float = 1e-8 _tensor_stats: dict = field(default_factory=dict, init=False, repr=False) def __post_init__(self): """Convert stats to tensors for efficient computation.""" self.stats = self.stats or {} self._tensor_stats = to_tensor(self.stats, device=self.device, dtype=torch.float32) def __call__(self, transition: EnvTransition) -> EnvTransition: new_transition = transition.copy() # Normalize observations # ... # Normalize action # ... return new_transition ``` Lihat implementasi lengkap di `src/lerobot/processor/normalize_processor.py` untuk detail lengkap. **Prinsip utama:** - **Selalu gunakan `transition.copy()`** untuk menghindari efek samping - **Atasi observasi dan aksi** secara konsisten - **Pisahkan konfigurasi dari state**: `get_config()` mengembalikan parameter yang dapat diserialisasi ke JSON, `state_dict()` mengembalikan tensor - **Konversi statistik menjadi tensor** di `__post_init__()` untuk komputasi yang efisien ### Konfigurasi dan Manajemen Status Processor mendukung serialisasi melalui tiga metode yang memisahkan konfigurasi dari status tensor. `NormalizerProcessorStep` mendemonstrasikan hal ini dengan sempurna - ia membawa statistik dataset (tensor) dalam statusnya, dan hiperparameter dalam konfigurasinya: ```python # Continuing the NormalizerProcessorStep example... def get_config(self) -> dict[str, Any]: """JSON-serializable configuration (no tensors).""" return { "eps": self.eps, "features": {k: {"type": v.type.value, "shape": v.shape} for k, v in self.features.items()}, "norm_map": {ft.value: nm.value for ft, nm in self.norm_map.items()}, # ... } def state_dict(self) -> dict[str, torch.Tensor]: """Tensor state only (e.g., dataset statistics).""" flat: dict[str, torch.Tensor] = {} for key, sub in self._tensor_stats.items(): for stat_name, tensor in sub.items(): flat[f"{key}.{stat_name}"] = tensor.cpu() # Always save to CPU return flat def load_state_dict(self, state: dict[str, torch.Tensor]) -> None: """Restore tensor state at runtime.""" self._tensor_stats.clear() for flat_key, tensor in state.items(): key, stat_name = flat_key.rsplit(".", 1) # Load to processor's configured device self._tensor_stats.setdefault(key, {})[stat_name] = tensor.to( dtype=torch.float32, device=self.device ) # ... ``` **Penggunaan:** ```python # Save (e.g., inside a policy) config = normalizer.get_config() tensors = normalizer.state_dict() # Restore (e.g., loading a pretrained policy) new_normalizer = NormalizerProcessorStep(**config) new_normalizer.load_state_dict(tensors) # Now new_normalizer has the same stats and configuration ``` ### Transformasikan fitur Metode `transform_features` mendefinisikan bagaimana prosesor Anda mengubah nama dan bentuk fitur. Hal ini sangat penting untuk konfigurasi kebijakan dan debugging. Untuk `NormalizerProcessorStep`, fitur biasanya dipertahankan tanpa perubahan karena normalisasi tidak mengubah kunci atau bentuk: ```python def transform_features(self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]: """Normalization preserves all feature definitions.""" return features # No changes to feature structure # ... ``` Ketika prosesor Anda mengubah nama atau bentuk data, implementasikan metode ini untuk mencerminkan pemetaan bagi komponen hilir. Sebagai contoh, prosesor pengubah nama yang sederhana: ```python def transform_features(self, features: dict[str, PolicyFeature]) -> dict[str, PolicyFeature]: # Simple renaming if "pixels" in features: features["observation.image"] = features.pop("pixels") # Pattern-based renaming for key in list(features.keys()): if key.startswith("env_state."): suffix = key[len("env_state."):] features[f"observation.{suffix}"] = features.pop(key) # ... return features ``` **Prinsip utama:** - Gunakan `features.pop(old_key)` untuk menghapus dan mendapatkan fitur lama - Gunakan `features[new_key] = old_feature` untuk menambahkan fitur yang telah diganti namanya - Selalu kembalikan kamus fitur yang telah dimodifikasi - Dokumentasikan transformasi dengan jelas di docstring ### Menggunakan penimpaan Anda dapat menimpa parameter langkah saat pemuatan menggunakan `overrides`. Ini berguna untuk objek yang tidak dapat diserialisasi atau pengaturan yang spesifik untuk situs tertentu. Cara ini berfungsi baik di pabrik kebijakan maupun dengan `DataProcessorPipeline.from_pretrained(...)`. **Adaptasi model fundamental**: Ini sangat berguna saat bekerja dengan kebijakan pra-latih fundamental di mana Anda jarang memiliki akses ke statistik pelatihan asli. Anda dapat menyuntikkan statistik dataset Anda sendiri untuk mengadaptasi normalizer ke data robot atau lingkungan spesifik Anda. Contoh: selama evaluasi kebijakan pada robot, timpa perangkat dan peta penamaan. Gunakan ini untuk menjalankan kebijakan yang dilatih di CUDA pada robot yang hanya memiliki CPU, atau untuk memetakan ulang kunci kamera ketika robot menggunakan nama yang berbeda dari dataset. Penggunaan langsung dengan `from_pretrained`: ```python from lerobot.datasets import LeRobotDataset from lerobot.processor import PolicyProcessorPipeline # Load a foundational policy trained on diverse robot data # but adapt normalization to your specific robot/environment new_stats = LeRobotDataset(repo_id="username/my-dataset").meta.stats processor = PolicyProcessorPipeline.from_pretrained( "huggingface/foundational-robot-policy", # Pretrained foundation model config_filename="policy_preprocessor.json", # The policy's saved preprocessor overrides={ "normalizer_processor": {"stats": new_stats}, # Inject your robot's statistics "device_processor": {"device": "cuda:0"}, # registry name for registered steps "rename_observations_processor": {"rename_map": robot_key_map}, # Map your robot's observation keys # ... }, ) ``` ## Praktik Terbaik Berdasarkan analisis dari semua implementasi prosesor LeRobot, berikut adalah pola dan praktik utama: ### 1. **Penanganan Data yang Aman** Selalu buat salinan data input untuk menghindari efek samping yang tidak diinginkan. Gunakan `transition.copy()` dan `observation.copy()` alih-alih memodifikasi data secara in-place. Hal ini mencegah prosesor Anda secara tidak sengaja memengaruhi komponen lain dalam pipeline. Periksa data yang diperlukan sebelum pemrosesan dan tangani data yang hilang dengan baik. Jika prosesor Anda mengharapkan kunci tertentu (seperti `"pixels"` untuk pemrosesan gambar), validasi kehadirannya terlebih dahulu. Untuk data opsional, gunakan pola akses aman seperti `transition.get()` dan tangani nilai `None` secara tepat. Ketika validasi data gagal, berikan pesan kesalahan yang jelas dan dapat ditindaklanjuti untuk membantu pengguna memahami apa yang salah dan bagaimana memperbaikinya. ### 2. **Pilih Kelas Dasar yang Tepat** LeRobot menyediakan kelas dasar khusus yang mengurangi kode boilerplate dan memastikan konsistensi. Gunakan `ObservationProcessorStep` ketika Anda hanya perlu memodifikasi observasi, `ActionProcessorStep` untuk pemrosesan aksi saja, dan `RobotActionProcessorStep` secara khusus untuk aksi robot berbasis kamus. Hanya warisi langsung dari `ProcessorStep` ketika Anda memerlukan kendali penuh atas seluruh transisi atau ketika memproses beberapa komponen transisi secara bersamaan. Kelas basis yang khusus menangani manajemen transisi untuk Anda dan menyediakan keamanan tipe. ### 3. **Registrasi dan Penamaan** Daftarkan prosesor Anda dengan nama yang deskriptif dan ber-namespace menggunakan `@ProcessorStepRegistry.register()`. Gunakan prefiks organisasi seperti `"robotics_lab/safety_clipper"` atau `"acme_corp/vision_enhancer"` untuk menghindari konflik penamaan. Hindari nama generik seperti `"processor"` atau `"step"` yang dapat bentrok dengan implementasi lain. Registrasi yang baik membuat prosesor Anda dapat ditemukan dan memungkinkan serialisasi/deserialisasi yang bersih saat menyimpan dan memuat pipeline. ### 4. **Pola Manajemen State** Bedakan antara parameter konfigurasi (nilai yang dapat diserialisasi JSON) dan status internal (tensor, buffer). Gunakan field dataclass dengan `init=False, repr=False` untuk status internal yang tidak boleh muncul dalam konstruktor atau representasi string. Implementasikan metode `reset()` untuk membersihkan status internal antar episode. Hal ini sangat penting untuk pemroses berstatus yang mengakumulasi data seiring waktu, seperti rata-rata bergerak atau filter temporal. Ingatlah bahwa `get_config()` seharusnya hanya mengembalikan konfigurasi yang dapat diserialisasi ke JSON, sementara `state_dict()` menangani status tensor secara terpisah. ### 5. **Validasi Input dan Penanganan Kesalahan** Validasi tipe dan bentuk input sebelum diproses. Periksa properti tensor seperti `dtype` dan dimensi untuk memastikan kompatibilitas dengan algoritma Anda. Untuk aksi robot, verifikasi bahwa komponen pose atau nilai sendi yang diperlukan ada dan berada dalam rentang yang diharapkan. Gunakan pengembalian dini untuk kasus tepi di mana tidak diperlukan pemrosesan. Sediakan pesan kesalahan yang jelas dan deskriptif, termasuk tipe data atau bentuk data yang diharapkan versus yang sebenarnya. Hal ini membuat proses debugging jauh lebih mudah bagi pengguna. ### 6. **Pengetahuan Perangkat dan Tipe Data** Desain prosesor Anda agar secara otomatis beradaptasi dengan perangkat dan dtype tensor input. Tensor internal (seperti statistik normalisasi) harus sesuai dengan perangkat dan dtype tensor input untuk memastikan kompatibilitas dengan pelatihan multi-GPU, presisi campuran, dan setup terdistribusi. Implementasikan metode `to()` yang memindahkan status internal prosesor Anda ke perangkat yang ditentukan. Periksa kompatibilitas perangkat/dtype saat runtime dan migrasikan status internal secara otomatis saat diperlukan. Pola ini memungkinkan operasi yang mulus di berbagai konfigurasi perangkat keras tanpa intervensi manual. ## Kesimpulan Anda kini memiliki semua alat untuk mengimplementasikan prosesor kustom di LeRobot! Langkah-langkah utamanya adalah: 1. **Definisikan prosesor Anda** sebagai dataclass dengan metode yang diperlukan (`__call__`, `get_config`, `state_dict`, `load_state_dict`, `reset`, `transform_features`) 2. **Daftarkan** menggunakan `@ProcessorStepRegistry.register("name")` agar dapat ditemukan 3. **Integrasikan** ke dalam `DataProcessorPipeline` dengan langkah pemrosesan lainnya 4. **Gunakan kelas dasar** seperti `ObservationProcessorStep` jika memungkinkan untuk mengurangi boilerplate 5. **Implementasi kesadaran perangkat/dtype** untuk mendukung konfigurasi multi-GPU dan presisi campuran Sistem prosesor dirancang agar modular dan dapat dikomposisikan, memungkinkan Anda membangun pipeline pemrosesan data yang kompleks dari komponen-komponen sederhana yang fokus. Baik Anda sedang melakukan pra-pemrosesan data sensor untuk pelatihan atau pasca-pemrosesan keluaran model untuk eksekusi robot, prosesor kustom memberikan fleksibilitas untuk menangani transformasi data apa pun yang dibutuhkan aplikasi robotika Anda. Prinsip-prinsip utama untuk prosesor yang robust: - **Adaptasi perangkat/dtype**: Tensor internal harus cocok dengan tensor input - **Pesan error yang jelas**: Membantu pengguna memahami apa yang salah - **Penggunaan kelas dasar**: Manfaatkan kelas dasar yang khusus untuk mengurangi boilerplate - **Kontrak fitur**: Nyatakan perubahan struktur data dengan `transform_features()` Mulai dengan sederhana, uji secara menyeluruh, dan pastikan prosesor Anda bekerja mulus di berbagai konfigurasi perangkat keras!