Merge pull request #7 from kscalelabs/python

Fix python check

.github/workflows/publish.yml

@@ -27,9 +27,6 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v3
 
-      - name: Change directory
-        run: cd pykos
-
       - name: Set up Python
         uses: actions/setup-python@v4
         with:
@@ -41,7 +38,9 @@ jobs:
           pip install build wheel
 
       - name: Build package
+        working-directory: ./pykos
         run: python -m build --sdist --wheel --outdir dist/ .
 
       - name: Publish package
+        working-directory: ./pykos
         uses: pypa/gh-action-pypi-publish@release/v1

.github/workflows/test.yml

@@ -25,9 +25,6 @@ jobs:
       - name: Check out repository
         uses: actions/checkout@v3
 
-      - name: Change directory
-        run: cd pykos
-
       - name: Set up Python
         uses: actions/setup-python@v4
         with:
@@ -46,15 +43,18 @@ jobs:
             python-requirements-
 
       - name: Install package
+        working-directory: ./pykos
         run: |
           pip install --upgrade --upgrade-strategy eager --extra-index-url https://download.pytorch.org/whl/cpu -e '.[dev]'
 
       - name: Run static checks
+        working-directory: ./pykos
         run: |
           mkdir -p .mypy_cache
           make static-checks
 
       - name: Run unit tests
+        working-directory: ./pykos
         run: |
           make test
 

MANIFEST.in

@@ -1 +1 @@
-recursive-include kos/ *.py *.txt py.typed MANIFEST.in
+recursive-include kos/ *.py *.txt py.typed MANIFEST.in py.typed

pykos/Makefile

@@ -27,7 +27,12 @@ all:
 # ------------------------ #
 
 generate-proto:
-	python -m grpc_tools.protoc --python_out=. --grpc_python_out=. --proto_path=../kos_core/proto --proto_path=../proto/googleapis ../proto/kos/*
+	python -m grpc_tools.protoc \
+		--python_out=. \
+		--grpc_python_out=. \
+		--proto_path=../kos_core/proto/ \
+		--proto_path=../kos_core/proto/googleapis \
+		../kos_core/proto/kos/*.proto
 .PHONY: generate-proto
 
 
@@ -57,7 +62,7 @@ push-to-pypi: build-for-pypi
 #       Static Checks      #
 # ------------------------ #
 
-py-files := $(shell find . -name '*.py')
+py-files := $(shell find . -name '*.py' -not -path './kos/*' -not -path './pykos/kos/*')
 
 format:
 	@black $(py-files)
@@ -68,7 +73,7 @@ static-checks:
 	@black --diff --check $(py-files)
 	@ruff check $(py-files)
 	@mypy --install-types --non-interactive $(py-files)
-.PHONY: lint
+.PHONY: static-checks
 
 # ------------------------ #
 #        Unit tests        #

pykos/pykos/__init__.py

@@ -1,3 +1,3 @@
 __version__ = "0.1.1"
 
-from pykos.client import KOS
+from pykos.client import KOS

pykos/pykos/client.py

@@ -1,10 +1,13 @@
+"""KOS client."""
+
 import grpc
 
-from pykos.services.imu import IMUServiceClient
 from pykos.services.actuator import ActuatorServiceClient
+from pykos.services.imu import IMUServiceClient
+
+
 class KOS:
-    """
-    KOS client
+    """KOS client.
 
     Args:
         ip (str, optional): IP address of the robot running KOS. Defaults to localhost.
@@ -13,15 +16,14 @@ class KOS:
     Attributes:
         imu (IMUServiceClient): Client for the IMU service.
     """
-    def __init__(self, ip: str = "localhost", port: int = 50051):
+
+    def __init__(self, ip: str = "localhost", port: int = 50051) -> None:
         self.ip = ip
         self.port = port
         self.channel = grpc.insecure_channel(f"{self.ip}:{self.port}")
         self.imu = IMUServiceClient(self.channel)
         self.actuator = ActuatorServiceClient(self.channel)
 
-    def close(self):
-        """
-        Close the gRPC channel.
-        """
+    def close(self) -> None:
+        """Close the gRPC channel."""
         self.channel.close()

pykos/pykos/requirements-dev.txt

@@ -1,3 +1,11 @@
 # requirements-dev.txt
 
 grpcio-tools
+types-protobuf
+
+# python checks
+black
+darglint
+mypy
+pytest
+ruff

pykos/pykos/services/actuator.py

@@ -1,41 +1,48 @@
-from kos import actuator_pb2_grpc, actuator_pb2
-from google.protobuf.empty_pb2 import Empty
-from google.protobuf.any_pb2 import Any
+"""Actuator service client."""
+
+from typing import Any, Dict, List, Optional
+
+import grpc
 from google.longrunning import operations_pb2, operations_pb2_grpc
+from google.protobuf.any_pb2 import Any as AnyPb2
+
+from kos import actuator_pb2, actuator_pb2_grpc
 from kos.actuator_pb2 import CalibrateActuatorMetadata
 
+
 class CalibrationStatus:
     Calibrating = "calibrating"
     Calibrated = "calibrated"
     Timeout = "timeout"
 
+
 class CalibrationMetadata:
-    def __init__(self, metadata_any: Any):
-        self.actuator_id = None
-        self.status = None
+    def __init__(self, metadata_any: AnyPb2) -> None:
+        self.actuator_id: Optional[int] = None
+        self.status: Optional[str] = None
         self.decode_metadata(metadata_any)
 
-    def decode_metadata(self, metadata_any: Any):
+    def decode_metadata(self, metadata_any: AnyPb2) -> None:
         metadata = CalibrateActuatorMetadata()
         if metadata_any.Is(CalibrateActuatorMetadata.DESCRIPTOR):
             metadata_any.Unpack(metadata)
             self.actuator_id = metadata.actuator_id
-            self.status = metadata.status
+            self.status = metadata.status if metadata.HasField("status") else None
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"CalibrationMetadata(actuator_id={self.actuator_id}, status={self.status})"
-    
-    def __repr__(self):
+
+    def __repr__(self) -> str:
         return self.__str__()
 
+
 class ActuatorServiceClient:
-    def __init__(self, channel):
+    def __init__(self, channel: grpc.Channel) -> None:
         self.stub = actuator_pb2_grpc.ActuatorServiceStub(channel)
         self.operations_stub = operations_pb2_grpc.OperationsStub(channel)
 
-    def calibrate(self, actuator_id: int):
-        """
-        Calibrate an actuator.
+    def calibrate(self, actuator_id: int) -> CalibrationMetadata:
+        """Calibrate an actuator.
 
         Returns:
             Operation: The operation for the calibration.
@@ -44,14 +51,15 @@ def calibrate(self, actuator_id: int):
         metadata = CalibrationMetadata(response.metadata)
         return metadata
 
-    def get_calibration_status(self, actuator_id: int):
-        response = self.operations_stub.GetOperation(operations_pb2.GetOperationRequest(name=f"operations/calibrate_actuator/{actuator_id}"))
+    def get_calibration_status(self, actuator_id: int) -> Optional[str]:
+        response = self.operations_stub.GetOperation(
+            operations_pb2.GetOperationRequest(name=f"operations/calibrate_actuator/{actuator_id}")
+        )
         metadata = CalibrationMetadata(response.metadata)
         return metadata.status
 
-    def command_actuators(self, commands: list[dict]):
-        """
-        Command multiple actuators at once.
+    def command_actuators(self, commands: List[Dict[str, Any]]) -> List[actuator_pb2.ActionResult]:
+        """Command multiple actuators at once.
 
         Args:
             commands: List of dictionaries containing actuator commands.
@@ -66,9 +74,8 @@ def command_actuators(self, commands: list[dict]):
         response = self.stub.CommandActuators(request)
         return response.results
 
-    def configure_actuator(self, actuator_id: int, **kwargs):
-        """
-        Configure an actuator's parameters.
+    def configure_actuator(self, actuator_id: int, **kwargs: Dict[str, Any]) -> actuator_pb2.ActionResponse:
+        """Configure an actuator's parameters.
 
         Args:
             actuator_id: ID of the actuator to configure
@@ -83,9 +90,8 @@ def configure_actuator(self, actuator_id: int, **kwargs):
         request = actuator_pb2.ConfigureActuatorRequest(**config)
         return self.stub.ConfigureActuator(request)
 
-    def get_actuators_state(self, actuator_ids: list[int]):
-        """
-        Get the state of multiple actuators.
+    def get_actuators_state(self, actuator_ids: List[int]) -> List[actuator_pb2.ActuatorStateResponse]:
+        """Get the state of multiple actuators.
 
         Args:
             actuator_ids: List of actuator IDs to query
@@ -96,5 +102,3 @@ def get_actuators_state(self, actuator_ids: list[int]):
         request = actuator_pb2.GetActuatorsStateRequest(actuator_ids=actuator_ids)
         response = self.stub.GetActuatorsState(request)
         return response.states
-
-

pykos/pykos/services/imu.py

@@ -1,8 +1,13 @@
-from kos import imu_pb2_grpc, imu_pb2
+"""IMU service client."""
+
+import grpc
 from google.protobuf.empty_pb2 import Empty
 
+from kos import imu_pb2, imu_pb2_grpc
+
+
 class ImuValues:
-    def __init__(self, response: imu_pb2.IMUValuesResponse):
+    def __init__(self, response: imu_pb2.IMUValuesResponse) -> None:
         self.accel_x = response.accel_x
         self.accel_y = response.accel_y
         self.accel_z = response.accel_z
@@ -13,22 +18,29 @@ def __init__(self, response: imu_pb2.IMUValuesResponse):
         self.mag_y = response.mag_y if response.HasField("mag_y") else None
         self.mag_z = response.mag_z if response.HasField("mag_z") else None
         self.error = response.error if response.HasField("error") else None
-    def __str__(self):
-        return f"ImuValues(accel_x={self.accel_x}, accel_y={self.accel_y}, accel_z={self.accel_z}, gyro_x={self.gyro_x}, gyro_y={self.gyro_y}, gyro_z={self.gyro_z}, mag_x={self.mag_x}, mag_y={self.mag_y}, mag_z={self.mag_z}, error={self.error})"
-    def __repr__(self):
+
+    def __str__(self) -> str:
+        return (
+            f"ImuValues("
+            f"accel_x={self.accel_x}, accel_y={self.accel_y}, accel_z={self.accel_z}, "
+            f"gyro_x={self.gyro_x}, gyro_y={self.gyro_y}, gyro_z={self.gyro_z}, "
+            f"mag_x={self.mag_x}, mag_y={self.mag_y}, mag_z={self.mag_z}, "
+            f"error={self.error})"
+        )
+
+    def __repr__(self) -> str:
         return self.__str__()
 
+
 class IMUServiceClient:
-    def __init__(self, channel):
+    def __init__(self, channel: grpc.Channel) -> None:
         self.stub = imu_pb2_grpc.IMUServiceStub(channel)
 
-    def get_imu_values(self):
-        """
-        Get the latest IMU sensor values.
+    def get_imu_values(self) -> ImuValues:
+        """Get the latest IMU sensor values.
 
         Returns:
             ImuValuesResponse: The latest IMU sensor values.
         """
         response = self.stub.GetValues(Empty())
         return ImuValues(response)
-

pykos/pyproject.toml

@@ -29,14 +29,27 @@ warn_redundant_casts = true
 
 incremental = true
 namespace_packages = false
+exclude = ["kos/.*\\.py$"]
 
 [[tool.mypy.overrides]]
 
 module = [
     "setuptools",
     "setuptools_rust",
+    "grpc",
+    "grpc.*",
+    "google.protobuf.*",
+    "google.longrunning.*",
+    "kos.*",
+    "pykos.*",
+    "kos.actuator_pb2_grpc",
+    "kos.imu_pb2_grpc",
+    "kos.actuator_pb2",
+    "kos.imu_pb2"
 ]
 ignore_missing_imports = true
+follow_imports = "skip"
+disallow_untyped_defs = false
 
 [tool.isort]
 

pykos/setup.py

@@ -3,6 +3,7 @@
 """Setup script for the project."""
 
 import re
+from typing import List
 
 from setuptools import setup
 
@@ -11,11 +12,11 @@
 
 
 with open("pykos/requirements.txt", "r", encoding="utf-8") as f:
-    requirements: list[str] = f.read().splitlines()
+    requirements: List[str] = f.read().splitlines()
 
 
 with open("pykos/requirements-dev.txt", "r", encoding="utf-8") as f:
-    requirements_dev: list[str] = f.read().splitlines()
+    requirements_dev: List[str] = f.read().splitlines()
 
 
 with open("pykos/__init__.py", "r", encoding="utf-8") as fh:
@@ -37,9 +38,13 @@
     tests_require=requirements_dev,
     extras_require={"dev": requirements_dev},
     packages=["pykos"],
+    package_data={
+        "pykos": ["py.typed"],
+    },
+    include_package_data=True,
     entry_points={
         "console_scripts": [
             "pykos=pykos.cli:cli",
         ],
     },
-)
+)

tests/conftest.py → pykos/tests/conftest.py

@@ -1,6 +1,7 @@
 """Defines PyTest configuration for the project."""
 
 import random
+from typing import List
 
 import pytest
 from _pytest.python import Function
@@ -11,5 +12,5 @@ def set_random_seed() -> None:
     random.seed(1337)
 
 
-def pytest_collection_modifyitems(items: list[Function]) -> None:
+def pytest_collection_modifyitems(items: List[Function]) -> None:
     items.sort(key=lambda x: x.get_closest_marker("slow") is not None)