تنفيذ معالج روبوت خاص بك

في هذا الشرح التعليمي، ستتعلم كيفية تنفيذ معالج روبوت خاص بك. يبدأ الشرح باستكشاف الحاجة إلى معالج مخصص، ثم يستخدم NormalizerProcessorStep كمثال تطبيقي لشرح كيفية تنفيذ معالج وتكوينه وتسلسله. للاطلاع على خطوات المعالج المرفقة مع LeRobot، راجع خطوات المعالج الشائعة.

لماذا قد تحتاج إلى معالج مخصص؟

في معظم الحالات، عند قراءة البيانات الخام من المستشعرات أو عند إخراج النماذج للأفعال، تحتاج إلى معالجة هذه البيانات لجعلها متوافقة مع نظامك المستهدف. على سبيل المثال، تعد الحاجة الشائعة هي تطبيع نطاقات البيانات لجعلها مناسبة للشبكات العصبية.

يتولى NormalizerProcessorStep في LeRobot معالجة هذه المهمة الحيوية:

# 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)
# ...

تشمل احتياجات المعالجة الشائعة الأخرى:

• توزيع الأجهزة: نقل الموترات بين المعالج المركزي/وحدة معالجة الرسوميات وتحويل أنواع البيانات
• تحويل التنسيق: التحويل بين هياكل بيانات مختلفة
• التجميع: إضافة/إزالة أبعاد الدفعة لتوافق النموذج
• قيود السلامة: تطبيق حدود على أوامر الروبوت

# Example pipeline combining multiple processors
pipeline = PolicyProcessorPipeline([
    RenameObservationsProcessorStep(rename_map={}),
    AddBatchDimensionProcessorStep(),
    NormalizerProcessorStep(features=features, norm_map=norm_map, stats=stats),
    DeviceProcessorStep(device="cuda"),
    # ...
])

توفّر LeRobot آلية خط أنابيب لتنفيذ تسلسلات من خطوات المعالجة لكلٍّ من البيانات المدخلة والإجراءات المخرجة، مما يسهّل تركيب هذه التحويلات بالترتيب الصحيح لتحقيق الأداء الأمثل.

كيف تنفّذ معالجك الخاص؟

سنستخدم NormalizerProcessorStep كمثال رئيسي لأنه يوضح أنماط المعالج الأساسية، بما في ذلك إدارة الحالة، وتسلسل التهيئة، ومعالجة الموترات، وهي عناصر ستحتاج إليها بشكل شائع.

جهّز تسلسل خطوات المعالجة اللازمة لمشكلتك. خطوة المعالج هي فئة تطبّق الطرق التالية:

• __call__: ينفّذ خطوة المعالجة للانتقال المدخل.
• get_config: يجلب إعدادات خطوة المعالج.
• state_dict: يحصل على حالة خطوة المعالجة.
• load_state_dict: يحمّل حالة خطوة المعالج.
• reset: يعيد ضبط حالة خطوة المعالج.
• transform_features: يصف كيف يعدّل خطوة المعالج مساحة الميزات.

تنفيذ طريقة __call__

تُعدّ طريقة __call__ جوهر خطوة المعالج لديك. تستقبل EnvTransition وتعيد EnvTransition معدّلة. وإليك كيف تعمل NormalizerProcessorStep:

@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

راجع التنفيذ الكامل في src/lerobot/processor/normalize_processor.py للحصول على التفاصيل الكاملة.

المبادئ الأساسية:

• استخدم دائمًا transition.copy() لتجنب الآثار الجانبية
• التعامل بشكل متسق مع الملاحظات والإجراءات معًا
• فصل الإعداد عن الحالة: يعيد get_config() معاملات قابلة للتسلسل بصيغة JSON، ويعيد state_dict() الموترات
• تحويل الإحصاءات إلى متجهات في __post_init__() لحساب أكثر كفاءة

إدارة التهيئة والحالة

تدعم المعالجات التسلسل عبر ثلاث طرق تفصل بين الإعدادات وحالة الموترات. يوضح NormalizerProcessorStep ذلك بشكل مثالي - فهو يحمل إحصاءات مجموعة البيانات (موترات) في حالته، ومعاملات ضبط في إعداده:

# 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
        )
        # ...

طريقة الاستخدام:

# 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

تحويل الميزات

تعرف طريقة transform_features كيفية تحويل المعالج لأسماء الميزات وأحجامها. وهذا أمر بالغ الأهمية لتكوين السياسة وتصحيح الأخطاء.

في NormalizerProcessorStep، تُحفظ الميزات عادةً دون تغيير، لأن عملية التطبيع لا تُغيّر المفاتيح أو الأشكال:

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
    # ...

عندما يعيد المعالج تسمية البيانات أو تغيير شكلها، نفّذ هذه الطريقة لتعكس عملية الإزاحة للمكوّنات اللاحقة. على سبيل المثال، معالج إعادة تسمية بسيط:

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

المبادئ الأساسية:

• استخدم features.pop(old_key) لإزالة الميزة القديمة والحصول عليها
• استخدم features[new_key] = old_feature لإضافة الميزة المُعاد تسميتها
• أعد دائمًا قاموس الميزات المُعدَّل
• وثّق التحويلات بوضوح في سلسلة التوثيق

استخدام التراكيب

يمكنك تجاوز معاملات الخطوة عند التحميل باستخدام overrides. وهذا مفيد للكائنات غير القابلة للتسلسل أو للإعدادات الخاصة بالموقع. ويعمل ذلك في مصانع السياسات ومع DataProcessorPipeline.from_pretrained(...).

تكييف النماذج التأسيسية: يكون هذا مفيدًا بشكل خاص عند العمل مع سياسات مسبقة التدريب التأسيسية حيث نادرًا ما يكون لديك وصول إلى إحصاءات التدريب الأصلية. يمكنك حق إحصاءات مجموعة بياناتك الخاصة لتكييف المعايِر (normalizer) مع بيانات روبوتك أو بيئتك المحددة.

مثال: أثناء تقييم السياسة على الروبوت، تجاوز إعداد الجهاز وخريطة إعادة التسمية. استخدم هذا لتشغيل سياسة مُدرَّبة على CUDA على روبوت يعمل بالمعالج المركزي فقط، أو لإعادة تعيين مفاتيح الكاميرا عندما يستخدم الروبوت أسماء مختلفة عن مجموعة البيانات.

الاستخدام المباشر مع from_pretrained:

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
        # ...
    },
)

أفضل الممارسات

بناءً على تحليل جميع تطبيقات معالجات LeRobot، فيما يلي الأنماط والممارسات الرئيسية:

1. معالجة البيانات بأمان

أنشئ دائمًا نسخًا من بيانات الإدخال لتجنب الآثار الجانبية غير المقصودة. استخدم transition.copy() وobservation.copy() بدلاً من تعديل البيانات في مكانها. يمنع هذا المعالج الخاص بك من التأثير عن غير قصد على مكوّنات أخرى في خط المعالجة.

تحقق من البيانات المطلوبة قبل المعالجة والتعامل مع البيانات الناقصة بسلاسة. إذا كان المعالج يتوقع مفاتيح معينة (مثل "pixels" لمعالجة الصور)، فتحقق من وجودها أولاً. بالنسبة للبيانات الاختيارية، استخدم أنماط وصول آمنة مثل transition.get() والتعامل مع قيم None بشكل مناسب.

عند فشل التحقق من صحة البيانات، قدّم رسائل خطأ واضحة وقابلة للتنفيذ تساعد المستخدمين على فهم ما حدث وكيف يمكنهم إصلاحه.

2. اختيار فئات الأساس المناسبة

توفّر LeRobot فئات أساسية متخصصة تقلّل من الكود المتكرر وتضمن الاتساق. استخدم ObservationProcessorStep عندما تحتاج فقط إلى تعديل الملاحظات، وActionProcessorStep لمعالجة الإجراءات فقط، وRobotActionProcessorStep خصيصًا لإجراءات الروبوت القائمة على القواميس.

لا ترث مباشرةً من ProcessorStep إلا عندما تحتاج إلى تحكم كامل في الانتقال بأكمله أو عند معالجة مكوّنات انتقال متعددة في وقت واحد. تتولى الفئات الأساسية المتخصصة إدارة الانتقال نيابةً عنك وتوفر أمان الأنواع.

3. التسجيل والتسمية

سجّل معالجاتك بأسماء وصفية ومساحات أسماء باستخدام @ProcessorStepRegistry.register(). استخدم بادئات المنظمات مثل "robotics_lab/safety_clipper" أو "acme_corp/vision_enhancer" لتجنب تعارض الأسماء. تجنّب الأسماء العامة مثل "processor" أو "step" التي قد تتعارض مع تطبيقات أخرى.

تُسهم عملية التسجيل الجيدة في جعل المعالجات قابلة للاكتشاف، وتُمكّن من التسلسل/إلغاء التسلسل النظيف عند حفظ خطوط الأنابيب وتحميلها.

4. أنماط إدارة الحالة

ميّز بين معاملات الإعداد (قيم قابلة للتسلسل بصيغة JSON) والحالة الداخلية (المُتجهات، المخازن). استخدم حقول dataclass مع init=False, repr=False للحالة الداخلية التي لا ينبغي أن تظهر في منشئ الكائن أو في التمثيل النصي.

نفّذ طريقة reset() لمسح الحالة الداخلية بين الحلقات. وهذا أمر بالغ الأهمية للمعالجات ذات الحالة التي تتراكم بياناتها مع مرور الوقت، مثل المتوسطات المتحركة أو المرشحات الزمنية.

تذكر أن get_config() يجب أن يعيد فقط إعدادات قابلة للتسلسل إلى JSON، بينما يتعامل state_dict() مع حالة الموترات بشكل منفصل.

5. التحقق من المدخلات ومعالجة الأخطاء

تحقق من أنواع المدخلات وأحجامها قبل المعالجة. افحص خصائص المصفوفات مثل dtype والأبعاد للتأكد من توافقها مع خوارزمياتك. فيما يتعلق بإجراءات الروبوت، تأكد من وجود مكونات الوضعية المطلوبة أو قيم المفاصل، وأن تكون ضمن النطاقات المتوقعة.

استخدم الإرجاع المبكر للحالات الحدّية التي لا تتطلب أي معالجة. قدّم رسائل خطأ واضحة وصفيّة تتضمن أنواع البيانات أو الأشكال المتوقَّعة مقابل الفعليّة. هذا يجعل تصحيح الأخطاء أسهل بكثير للمستخدمين.

6. الوعي بالجهاز ونوع البيانات

صمّم معالجاتك لتتكيف تلقائيًا مع الجهاز ونوع البيانات (dtype) للمصفوفات المدخلة. يجب أن تتطابق المصفوفات الداخلية (مثل إحصاءات التطبيع) مع جهاز المصفوفة المدخلة ونوع بياناتها لضمان التوافق مع التدريب متعدد وحدات معالجة الرسوميات (multi-GPU) والدقة المختلطة والإعدادات الموزعة.

نفّذ طريقة to() لنقل الحالة الداخلية لمعالجك إلى الجهاز المحدد. تحقق من توافق الجهاز/نوع البيانات في وقت التشغيل، وانقل الحالة الداخلية تلقائيًا عند الحاجة. تتيح هذه النمطية العمل بسلاسة عبر تكوينات الأجهزة المختلفة دون تدخل يدوي.

الخلاصة

لديك الآن جميع الأدوات اللازمة لتنفيذ المعالجات المخصصة في LeRobot! الخطوات الأساسية هي:

1. عرّف معالجك كـ dataclass مع الطرق المطلوبة (__call__، get_config، state_dict، load_state_dict، reset، transform_features)
2. سجّله باستخدام @ProcessorStepRegistry.register("name") ليتم اكتشافه
3. دمجه في DataProcessorPipeline مع خطوات معالجة أخرى
4. استخدم الفئات الأساسية مثل ObservationProcessorStep كلما أمكن ذلك لتقليل الأكواد المتكررة
5. تنفيذ الوعي بالجهاز/نوع البيانات لدعم إعدادات متعددة وحدات معالجة الرسوميات والدقة المختلطة

صُمم نظام المعالِج ليكون معيارياً وقابلاً للتركيب، مما يتيح لك بناء خطوط معالجة بيانات معقدة انطلاقاً من مكوّنات بسيطة ومركّزة. سواء كنت تقوم بمعالجة مسبقة لبيانات المستشعرات من أجل التدريب، أو بمعالجة لاحقة لمخرجات النموذج من أجل تنفيذ الروبوت، فإن المعالِجات المخصصة تمنحك المرونة اللازمة للتعامل مع أي تحويل بيانات تتطلبه تطبيقك في مجال الروبوتات.

المبادئ الأساسية للمعالجات القوية:

• ملاءمة الجهاز/نوع البيانات: يجب أن تتطابق المصفوفات الداخلية مع المصفوفات المدخلة
• رسائل خطأ واضحة: تساعد المستخدمين على فهم ما حدث
• استخدام الفئة الأساسية: الاستفادة من الفئات الأساسية المتخصصة لتقليل الأكواد المتكررة
• عقود الميزات: اعلن عن تغييرات بنية البيانات باستخدام transform_features()

ابدأ ببساطة، واختبر بشكل شامل، وتأكد من أن معالجاتك تعمل بسلاسة عبر تكوينات الأجهزة المختلفة!