feat(transport): deferred TransportSpec + .spec() migration

[spomichter]

Summary

Extracts and rebases Sam's transport config work (#2499) onto current main.

Adds TransportSpec dataclass for deferred transport construction in blueprints. Transports are now declared via Cls.spec(...) so nothing is instantiated at blueprint-definition time — the coordinator materializes specs at build time once CLI/env/config overrides resolve.

Changes

Infrastructure:

• blueprints.py: TransportSpec, transport_config_name(), Blueprint type updates, dynamic transport config model generation
• module_coordinator.py: _materialize_transports(), updated _connect_streams() to accept pre-built transports
• stream.py: Transport._config_cls class-level default (None)
• transport.py: PubSubTransport.spec() classmethod

Mechanical migration:

• All blueprint files: LCMTransport(...) → LCMTransport.spec(...)
• All blueprint files: ROSTransport(...) → ROSTransport.spec(...)

Context

This is a prerequisite for the WebRTC transport PR (#2048). The WebRTC transport will layer on top of this once merged.

Co-authored-by: Sam Bull git@sambull.org

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
1904	2	1902	159

View the top 2 failed test(s) by shortest run time

dimos.perception.fiducial.blueprints.test_desk_marker_tf::test_desk_marker_tf_blueprint_declares_static_tf_module

Stack Traces | 0.001s run time

def test_desk_marker_tf_blueprint_declares_static_tf_module() -> None:
        assert isinstance(desk_marker_tf, Blueprint)
        assert desk_marker_tf.blueprints[0].module is DeskStaticTfModule
        assert desk_marker_tf.blueprints[1].module is CameraModule
        assert desk_marker_tf.blueprints[1].kwargs["hardware"] is create_desk_webcam
        assert desk_marker_tf.blueprints[1].kwargs["transform"] is None
        assert desk_marker_tf.blueprints[2].module is MarkerDetectionStreamModule
        assert desk_marker_tf.blueprints[2].kwargs["marker_length_m"] == DESK_MARKER_LENGTH_M
        assert desk_marker_tf.blueprints[2].kwargs["aruco_dictionary"] == DESK_MARKER_ARUCO_DICTIONARY
        assert desk_marker_tf.blueprints[2].kwargs["camera_info"].frame_id == DESK_CAMERA_FRAME_ID
        assert desk_marker_tf.blueprints[3].module is MarkerTfModule
        assert (
            desk_marker_tf.blueprints[3].kwargs["marker_namespace_prefix"]
            == DESK_MARKER_NAMESPACE_PREFIX
        )
        assert (
>           desk_marker_tf.transport_map[("detections", MarkerDetectionStreamModule)].topic.topic
            == "/marker_detection/detections"
        )
E       AttributeError: 'TransportSpec' object has no attribute 'topic'


.../fiducial/blueprints/test_desk_marker_tf.py:50: AttributeError

dimos.robot.unitree.go2.blueprints.smart.test_unitree_go2_markers::test_unitree_go2_markers_uses_detector_backed_tf_stack

Stack Traces | 0.001s run time

def test_unitree_go2_markers_uses_detector_backed_tf_stack() -> None:
        assert isinstance(unitree_go2_markers, Blueprint)
    
        modules = [bp.module for bp in unitree_go2_markers.blueprints]
        assert MarkerDetectionStreamModule in modules
        assert MarkerTfModule in modules
    
        detector = next(
            bp for bp in unitree_go2_markers.blueprints if bp.module is MarkerDetectionStreamModule
        )
        assert detector.kwargs["marker_length_m"] == 0.1
        assert detector.kwargs["camera_info"].frame_id == "camera_optical"
        assert (
>           unitree_go2_markers.transport_map[("detections", MarkerDetectionStreamModule)].topic.topic
            == "/marker_detection/detections"
        )
E       AttributeError: 'TransportSpec' object has no attribute 'topic'

detector   = BlueprintAtom(kwargs={'marker_length_m': 0.1, 'camera_info': CameraInfo(height=720, width=1280, distortion_model='equi...etections', type=<class 'dimos.msgs.vision_msgs.Detection3DArray.Detection3DArray'>, direction='out')), module_refs=())
modules    = [<class 'dimos.visualization.rerun.bridge.RerunBridgeModule'>, <class 'dimos.visualization.rerun.websocket_server.Reru...on.GO2Connection'>, <class 'dimos.mapping.voxels.VoxelGridMapper'>, <class 'dimos.mapping.costmapper.CostMapper'>, ...]

.../blueprints/smart/test_unitree_go2_markers.py:34: AttributeError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}