Dask lcm multiprocess

[leshy]

pretty big but separate

artisinal hand crafted multiprocess + pluggable transport between nodes

• builds a generic structure for services and generic Pubsub on top, with LCM, redis and in-memory implementations (for grid testing of my spec) - allows us to be independant from transport protocols (can implement other types of services in the future like KV stores, RPC etc)

• builds a bunch of ROS/LCM native types - this is a bit time consuming and awkward - likely will pivot into fixing autogen vs rewriting so much

• builds a multiprocessing architecture with (per stream) configurable transport strategy - so we can use zenoh, dask, lcm, even within the same deployment

core works well, but API will get syntax sugar on top - I have some basics in mind, pls also give suggestions for follow up PRs

Dask manages multiprocesing but we are not tied to dask for this, can switch this out without rewriting anything built on top.

Supports convinient way to delegate actual classes to separate processes and talk to them, but nothing stops you from creating ros-style separate python scripts either.

API example (paste from dimos/core/test_core.py)

def test_deployment(dimos):
    robot = dimos.deploy(RobotClient)
    target_stream = RemoteOut[Vector](Vector, "target")

    print("\n")
    print("lidar stream", robot.lidar)
    print("target stream", target_stream)
    print("odom stream", robot.odometry)

    nav = dimos.deploy(Navigation)

    # transport override - this stream encodes proper LCM messages
    robot.lidar.transport = LCMTransport("/lidar", LidarMessage)

    # transport override - odometry is transfered using a pickle over LCM
    robot.odometry.transport = pLCMTransport("/odom")

    # move messages we are sending using the (default) dask transport so we are not specifying mov transport
    # nav.mov.transport = pLCMTransport("/mov")

    nav.lidar.connect(robot.lidar)
    nav.odometry.connect(robot.odometry)
    robot.mov.connect(nav.mov)

    print("\n" + robot.io().result() + "\n")
    print("\n" + nav.io().result() + "\n")
    robot.start().result()
    nav.start().result()

    time.sleep(1)
    robot.stop().result()

    print("robot.mov_msg_count", robot.mov_msg_count)
    print("nav.odom_msg_count", nav.odom_msg_count)
    print("nav.lidar_msg_count", nav.lidar_msg_count)

    assert robot.mov_msg_count >= 9
    assert nav.odom_msg_count >= 9
    assert nav.lidar_msg_count >= 9

output:

[spomichter]

Probably cafe not be in the system data root maybe as data/cafe.jpg maybe data/test where we put all of the test assets like videos, image, pcd, lidar, etc.

[leshy]

issue is that we don't support dir trees for data/ (there was a good reason why but need to think, obviously would be nice to have)

[spomichter]

Isn't threading handled via Dask why do we have threading events set via python threading? @leshy

[leshy]

dask runs you in a separate process, once there you can do whatever you want. in this case I need some separate loop to run message broadcast loop, two things though:

• I don't like "threading" too much, and prefer asyncio, in this case cursor vibecodeded a thread and I said "ok fine"

• In the case above given we don't do anything apart from the message broadcast loop - I might not need a "thread" - will test

[spomichter]

This type does not look unitree specific?

[leshy]

it is, for now, I've moved out some types that started unitree specific to dimos types/ once they feel ready and need to be shared so that's my policy, once we are implementing a new lidar, we can see if we can make this type universal

[spomichter]

No better way to do this?

[leshy]

I've spent a bit of time on it and decided to do this. mypy is a typecheck that only I use for now, so I've been keeping it's config loose, it's also possible that this is the best way unfortunately, not sure.

[spomichter]

I will add to pyproject build

[leshy]

thanks