Feat: Support CLI for Launching LeKiwiHost (#1614)

* Support CLI for LeKiwiHost

Signed-off-by: Xingdong Zuo <zuoxingdong@users.noreply.github.com>

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Signed-off-by: Xingdong Zuo <zuoxingdong@users.noreply.github.com>
Co-authored-by: Steven Palma <imstevenpmwork@ieee.org>
This commit is contained in:
Xingdong Zuo
2025-10-20 03:19:57 +09:00
committed by GitHub
parent 98662e5f24
commit a97d078d95

View File

@@ -18,14 +18,24 @@ import base64
import json import json
import logging import logging
import time import time
from dataclasses import dataclass, field
import cv2 import cv2
import draccus
import zmq import zmq
from .config_lekiwi import LeKiwiConfig, LeKiwiHostConfig from .config_lekiwi import LeKiwiConfig, LeKiwiHostConfig
from .lekiwi import LeKiwi from .lekiwi import LeKiwi
@dataclass
class LeKiwiServerConfig:
"""Configuration for the LeKiwi host script."""
robot: LeKiwiConfig = field(default_factory=LeKiwiConfig)
host: LeKiwiHostConfig = field(default_factory=LeKiwiHostConfig)
class LeKiwiHost: class LeKiwiHost:
def __init__(self, config: LeKiwiHostConfig): def __init__(self, config: LeKiwiHostConfig):
self.zmq_context = zmq.Context() self.zmq_context = zmq.Context()
@@ -47,17 +57,16 @@ class LeKiwiHost:
self.zmq_context.term() self.zmq_context.term()
def main(): @draccus.wrap()
def main(cfg: LeKiwiServerConfig):
logging.info("Configuring LeKiwi") logging.info("Configuring LeKiwi")
robot_config = LeKiwiConfig() robot = LeKiwi(cfg.robot)
robot = LeKiwi(robot_config)
logging.info("Connecting LeKiwi") logging.info("Connecting LeKiwi")
robot.connect() robot.connect()
logging.info("Starting HostAgent") logging.info("Starting HostAgent")
host_config = LeKiwiHostConfig() host = LeKiwiHost(cfg.host)
host = LeKiwiHost(host_config)
last_cmd_time = time.time() last_cmd_time = time.time()
watchdog_active = False watchdog_active = False