Basic porting of the Java code into Python

.vscode/settings.json

@@ -28,5 +28,6 @@
   "python.analysis.typeCheckingMode": "basic",
   "python.analysis.autoImportCompletions": true,
   "python.defaultInterpreterPath": ".venv/bin/python",
-  "python.terminal.activateEnvInCurrentTerminal": true
+  "python.terminal.activateEnvInCurrentTerminal": true,
+  "cSpell.words": ["MQTT"]
 }

README.md

@@ -1,7 +1,6 @@
 # Robot Library Python
 
-A modular Python library for building robot applications. It provides components
-for motion control, sensor access, communication, and indicators.
+A modular Python library for building robot applications. It provides components for motion control, sensor access, communication, and indicators.
 
 ## Installation
 
@@ -12,7 +11,26 @@ pip install -e .
 ## Usage
 
 ```python
-from robot.motion import MotionController
+from robot import MQTTSettings, VirtualRobot
+from robot.interfaces import RobotState
 
-controller = MotionController()
+
+class MyRobot(VirtualRobot):
+    def loop(self):
+        if self.state == RobotState.RUN:
+            print("running...")
+            self.delay(1000)
+
+
+MQTTSettings.server = "localhost"
+MQTTSettings.port = 1883
+MQTTSettings.user_name = ""
+MQTTSettings.password = ""
+MQTTSettings.channel = "v1"
+
+r = MyRobot(1, 0, 0, 90)
+r.start()
+r.run()  # or run via a thread
 ```
+
+See `examples/my_test_robot.py` for a complete example mirroring the Java demo.

examples/my_test_robot.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import threading
+import time
+
+from robot import MQTTSettings, VirtualRobot
+from robot.interfaces import RobotState
+
+
+class MyTestRobot(VirtualRobot):
+    def setup(self) -> None:  # type: ignore[override]
+        print("My Test Robot Started")
+        super().setup()
+
+    def loop(self) -> None:  # type: ignore[override]
+        super().loop()
+        if self.state == RobotState.RUN:
+            print("Test")
+            self.delay(1000)
+
+    def communication_interrupt(self, msg: str) -> None:  # type: ignore[override]
+        print(f"communicationInterrupt on {self.id} with msg:{msg}")
+
+
+if __name__ == "__main__":
+    # Configure MQTT (fill with your broker details)
+    MQTTSettings.server = "localhost"
+    MQTTSettings.port = 1883
+    MQTTSettings.user_name = ""
+    MQTTSettings.password = ""
+    MQTTSettings.channel = "v1"
+
+    robot = MyTestRobot(10, 0, 0, 90)
+    t = threading.Thread(target=robot.run, daemon=True)
+    t.start()
+
+    # Example to send start after a short delay
+    time.sleep(1)
+    robot.start()
+    t.join()

pyproject.toml

@@ -6,10 +6,13 @@ build-backend = "setuptools.build_meta"
 name = "py-swarm-robot"
 version = "0.1.0"
 description = "A modular robot control library"
-authors = [{name = "Nuwan Jaliyagoda", email = "nuwanjaliyagoda@gmail.com"},{name = "Nuwan Jaliyagoda", email = "kavindumethpura@gmail.com"},{name = "Kavindu Prabhath Methpura", email = "kavindumethpura@gmail.com"}]
+authors = [
+    { name = "Nuwan Jaliyagoda", email = "nuwanjaliyagoda@gmail.com" },
+    { name = "Kavindu Prabhath Methpura", email = "kavindumethpura@gmail.com" },
+]
 readme = "README.md"
 requires-python = ">=3.8"
-dependencies = []
+dependencies = ["paho-mqtt>=2.0.0"]
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/robot/__init__.py

@@ -1,11 +1,17 @@
-"""Top-level package for robot library."""
+"""Top-level package for robot library (Python port of robot-library-java)."""
 
 from .robot_base import Robot
-from .mqtt_client import RobotMqttClient
 from .motion import MotionController
+from .virtual_robot import VirtualRobot
+
+# Expose common subpackages
+from .configs.mqtt_settings import MQTTSettings
+from .configs.robot_settings import RobotSettings
 
 __all__ = [
     "Robot",
-    "RobotMqttClient",
     "MotionController",
+    "VirtualRobot",
+    "MQTTSettings",
+    "RobotSettings",
 ]

src/robot/communication/communication.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from robot.interfaces import IMqttHandler
+from robot.mqtt.robot_mqtt_client import RobotMqttClient
+
+
+class Communication(IMqttHandler):
+    def __init__(self, robot_id: int, mqtt_client: RobotMqttClient):
+        self.robot_mqtt_client = mqtt_client
+        self.robot_id = robot_id
+
+    def send_message(self, msg: str) -> None:  # abstract
+        raise NotImplementedError
+
+    def send_message_with_distance(self, msg: str, distance: int) -> None:  # abstract
+        raise NotImplementedError
+

src/robot/communication/directed_comm.py

@@ -1,7 +1,35 @@
-"""Directed communication module."""
+"""Directed communication module mirroring Java logic."""
 
+from __future__ import annotations
 
-class DirectedCommunication:
-    """Placeholder directed communication class."""
+import json
 
-    pass
+from robot.communication.communication import Communication
+from robot.mqtt.mqtt_msg import MqttMsg
+from robot.mqtt.robot_mqtt_client import RobotMqttClient
+
+
+class DirectedCommunication(Communication):
+    def __init__(self, robot_id: int, mqtt_client: RobotMqttClient):
+        super().__init__(robot_id, mqtt_client)
+        self._topics_sub: dict[str, str] = {}
+        self._subscribe("COMMUNICATION_IN_DIR", f"comm/in/direct/{robot_id}")
+
+    def _subscribe(self, key: str, topic: str) -> None:
+        self._topics_sub[key] = topic
+        self.robot_mqtt_client.subscribe(topic)
+
+    def send_message(self, msg: str) -> None:
+        obj = {"id": self.robot_id, "msg": msg}
+        self.robot_mqtt_client.publish("comm/out/direct", json.dumps(obj))
+
+    def send_message_with_distance(self, msg: str, distance: int) -> None:
+        obj = {"id": self.robot_id, "msg": msg, "dist": distance}
+        self.robot_mqtt_client.publish("comm/out/direct", json.dumps(obj))
+
+    def handle_subscription(self, robot, m: MqttMsg) -> None:
+        topic, msg = m.topic, m.message
+        if topic == self._topics_sub.get("COMMUNICATION_IN_DIR"):
+            robot.communication_interrupt(msg)
+        else:
+            print(f"Received (unknown dir): {topic}> {msg}")

src/robot/communication/simple_comm.py

@@ -1,7 +1,36 @@
-"""Simple communication module."""
+"""Simple communication module mirroring Java logic."""
 
+from __future__ import annotations
 
-class SimpleCommunication:
-    """Placeholder simple communication class."""
+import json
+from robot.mqtt.robot_mqtt_client import RobotMqttClient
+from robot.mqtt.mqtt_msg import MqttMsg
+from robot.communication.communication import Communication
 
-    pass
+
+class SimpleCommunication(Communication):
+    def __init__(self, robot_id: int, mqtt_client: RobotMqttClient):
+        super().__init__(robot_id, mqtt_client)
+        self._topics_sub: dict[str, str] = {}
+        self._subscribe("COMMUNICATION_IN_SIMP", f"comm/in/simple/{robot_id}")
+
+    def _subscribe(self, key: str, topic: str) -> None:
+        self._topics_sub[key] = topic
+        self.robot_mqtt_client.subscribe(topic)
+
+    def send_message(self, msg: str) -> None:
+        obj = {"id": self.robot_id, "msg": msg}
+        self.robot_mqtt_client.publish("comm/out/simple", json.dumps(obj))
+
+    def send_message_with_distance(self, msg: str, distance: int) -> None:
+        obj = {"id": self.robot_id, "msg": msg, "dist": distance}
+        self.robot_mqtt_client.publish("comm/out/simple", json.dumps(obj))
+
+    # IMqttHandler
+    def handle_subscription(self, robot, m: MqttMsg) -> None:
+        topic, msg = m.topic, m.message
+        if topic == self._topics_sub.get("COMMUNICATION_IN_SIMP"):
+            robot.communication_interrupt(msg)
+        else:
+            print(self._topics_sub.get("COMMUNICATION_IN_SIMP"))
+            print(f"Received (unknown simp): {topic}> {msg}")

src/robot/configs/mqtt_settings.py

@@ -0,0 +1,23 @@
+"""MQTT settings analogous to swarm.configs.MQTTSettings (Java).
+
+Set these values before starting a robot to connect to the broker.
+"""
+
+
+class MQTTSettings:
+    server: str | None = None
+    port: int = 1883
+    user_name: str | None = None
+    password: str | None = None
+    channel: str | None = None
+
+    @classmethod
+    def print(cls) -> None:
+        print(f"server: {cls.server}")
+        print(f"port: {cls.port}")
+        print(f"username: {cls.user_name}")
+        print(f"password: {cls.password}")
+        print(f"channel: {cls.channel}")
+
+
+__all__ = ["MQTTSettings"]

src/robot/configs/robot_settings.py

@@ -0,0 +1,17 @@
+"""Robot settings analogous to swarm.configs.RobotSettings (Java)."""
+
+
+class RobotSettings:
+    ROBOT_SPEED_MAX: int = 255
+    ROBOT_SPEED_MIN: int = 50
+
+    ROBOT_RADIUS: int = 6  # in cm
+    ROBOT_WIDTH: int = 12  # in cm
+    ROBOT_WHEEL_RADIUS: float = 3.5  # in cm
+
+    # 0: no logs (TODO: implement levels)
+    ROBOT_LOG_LEVEL: int = 0
+
+
+__all__ = ["RobotSettings"]
+

src/robot/exception/__init__.py

@@ -0,0 +1,42 @@
+"""Exception types mirroring Java exceptions (slim wrappers)."""
+
+
+class MotionControllerException(Exception):
+    def __init__(self, message: str):
+        super().__init__(message)
+        print(f"Motion Error: {message}")
+
+
+class MqttClientException(Exception):
+    def __init__(self, message: str):
+        super().__init__(message)
+        print(f"MQTT Error: {message}")
+
+
+class SensorException(Exception):
+    def __init__(self, message: str):
+        super().__init__(message)
+        print(f"Sensor Error: {message}")
+
+
+class ProximityException(Exception):
+    def __init__(self, message: str):
+        super().__init__(message)
+        print(f"Proximity reading error: {message}")
+
+
+class RGBColorException(Exception):
+    def __init__(self, R: int | None = None, G: int | None = None, B: int | None = None):
+        msg = f"Invalid RGB values: R={R}, G={G}, B={B}"
+        super().__init__(msg)
+        print(msg)
+
+
+__all__ = [
+    "MotionControllerException",
+    "MqttClientException",
+    "SensorException",
+    "ProximityException",
+    "RGBColorException",
+]
+

src/robot/helpers/__init__.py

@@ -2,8 +2,10 @@
 
 from .coordinate import Coordinate
 from .robot_mqtt import RobotMQTT
+from .motion_controller import MotionController
 
 __all__ = [
     "Coordinate",
     "RobotMQTT",
+    "MotionController",
 ]