asw_component/__init__.py +3−0 View file@@ -0,0 +1,3 @@
1 +
from .component import execute 2 +
3 +
__all__ = ["execute"]REPOSITORY
via/asw-reference-ros2-kit Private
[참조 MVP] A-SW ROS2 참조 Kit — ROS2 토픽과 유사한 메시지 봉투로 인지→계획→제어 연결 방법을 보여주는 참조 Kit입니다.
https://git.agrithing.ai/via/asw-reference-ros2-kit.git 277c3f0 feat: adopt cpp TOOLING SDV reference implementation
d085d43 ci: add language-specific static and dynamic analysis
b7d2f98 docs: define component scope and interface contract
17e12a8 ci: verify contract tests and evidence package
0061666 feat: add executable reference MVP
9526ba0 docs: define component scope and interface contract
COMMIT DETAIL
feat: add executable reference MVP
006166654d45470e26cee1b898a8e6b2d55af0d9 asw_component/component.py +13−0 View file@@ -0,0 +1,13 @@
1 +
from __future__ import annotations 2 +
import hashlib 3 +
import json 4 +
5 +
REQUIRED_TOPICS = {"/vehicle/pose", "/planning/path", "/control/command"} 6 +
7 +
def execute(payload: dict) -> dict: 8 +
topic = payload.get("topic") 9 +
if topic not in REQUIRED_TOPICS: 10 +
raise ValueError(f"unsupported topic: {topic}") 11 +
envelope = {"topic": topic, "timestamp_ms": int(payload["timestamp_ms"]), "frame_id": payload.get("frame_id", ""), "data": payload.get("data", {})} 12 +
canonical = json.dumps(envelope, sort_keys=True, separators=(",", ":")).encode() 13 +
return {"accepted": True, "envelope": envelope, "message_id": hashlib.sha256(canonical).hexdigest()[:16], "qos": {"reliability": "reliable", "depth": 10}}examples/input.json +10−0 View file@@ -0,0 +1,10 @@
1 +
{ 2 +
"topic": "/vehicle/pose", 3 +
"timestamp_ms": 1000, 4 +
"frame_id": "field_local", 5 +
"data": { 6 +
"x": 1.2, 7 +
"y": 3.4, 8 +
"heading_rad": 0.1 9 +
} 10 +
}pyproject.toml +8−0 View file@@ -0,0 +1,8 @@
1 +
[project] 2 +
name = "asw_reference_ros2_kit" 3 +
version = "0.1.0" 4 +
description = "ROS2 토픽과 유사한 메시지 봉투로 인지→계획→제어 연결 방법을 보여주는 참조 Kit입니다." 5 +
requires-python = ">=3.11" 6 +
7 +
[tool.unittest] 8 +
test-path = "tests"run.py +10−0 View file@@ -0,0 +1,10 @@
1 +
from __future__ import annotations 2 +
import json 3 +
import sys 4 +
from asw_component import execute 5 +
6 +
if __name__ == "__main__": 7 +
path = sys.argv[1] if len(sys.argv) > 1 else None 8 +
with open(path, encoding="utf-8") if path else sys.stdin as source: 9 +
payload = json.load(source) 10 +
print(json.dumps(execute(payload), ensure_ascii=False, indent=2))