simplify services api

protos/farm_ng/canbus/canbus.proto

@@ -24,9 +24,6 @@ service CanbusService {
   rpc sendCanbusMessage(stream SendCanbusMessageRequest)
       returns (SendCanbusMessageReply) {}
   rpc getServiceState(GetServiceStateRequest) returns (GetServiceStateResult) {}
-  rpc startService(StartServiceRequest) returns (StartServiceResult) {}
-  rpc stopService(StopServiceRequest) returns (StopServiceResult) {}
-  rpc pauseService(PauseServiceRequest) returns (PauseServiceResult) {}
 }
 
 enum ReplyStatus {
@@ -43,6 +40,7 @@ enum CanbusServiceState {
   ERROR = 5;
 }
 
+// NOTE: `message` not really necessary
 message StreamCanbusRequest {
   string message = 1;
 }
@@ -74,33 +72,6 @@ message RawCanbusMessages {
   repeated RawCanbusMessage messages = 1;
 }
 
-message StartServiceRequest {
-  string message = 1;
-}
-
-message StopServiceRequest {
-  string message = 1;
-}
-
-message PauseServiceRequest {
-  string message = 1;
-}
-
-message StartServiceResult {
-  string message = 1;
-  ReplyStatus status = 2;
-}
-
-message StopServiceResult {
-  string message = 1;
-  ReplyStatus status = 2;
-}
-
-message PauseServiceResult {
-  string message = 1;
-  ReplyStatus status = 2;
-}
-
 message GetServiceStateRequest {
   string message = 1;
 }

protos/farm_ng/oak/oak.proto

@@ -22,9 +22,6 @@ service OakService {
   rpc cameraControl(CameraControlRequest) returns (CameraControlReply) {}
   rpc streamFrames(StreamFramesRequest) returns (stream StreamFramesReply) {}
   rpc getServiceState(GetServiceStateRequest) returns (GetServiceStateResult) {}
-  rpc startService(StartServiceRequest) returns (StartServiceResult) {}
-  rpc stopService(StopServiceRequest) returns (StopServiceResult) {}
-  rpc pauseService(PauseServiceRequest) returns (PauseServiceResult) {}
   rpc getCalibration(GetCalibrationRequest) returns (GetCalibrationResult) {}
 }
 
@@ -50,33 +47,6 @@ message GetCalibrationResult {
   ReplyStatus status = 2;
 }
 
-message StopServiceRequest {
-  string message = 1;
-}
-
-message StopServiceResult {
-  string message = 1;
-  ReplyStatus status = 2;
-}
-
-message StartServiceRequest {
-  string message = 1;
-}
-
-message StartServiceResult {
-  string message = 1;
-  ReplyStatus status = 2;
-}
-
-message PauseServiceRequest {
-  string message = 1;
-}
-
-message PauseServiceResult {
-  string message = 1;
-  ReplyStatus status = 2;
-}
-
 message GetServiceStateRequest {
   string message = 1;
 }

py/farm_ng/canbus/canbus_client.py

@@ -11,7 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import asyncio
 import logging
 from dataclasses import dataclass
 
@@ -66,26 +65,11 @@ def __init__(self, config: CanbusClientConfig) -> None:
         self.channel = grpc.aio.insecure_channel(self.server_address)
         self.stub = canbus_pb2_grpc.CanbusServiceStub(self.channel)
 
-        self._state = CanbusServiceState()
-
-    @property
-    def state(self) -> CanbusServiceState:
-        return self._state
-
     @property
     def server_address(self) -> str:
         """Returns the composed address and port."""
         return f"{self.config.address}:{self.config.port}"
 
-    async def poll_service_state(self) -> None:
-        while True:
-            try:
-                self._state = await self.get_state()
-                await asyncio.sleep(0.1)
-            except asyncio.CancelledError:
-                self.logger.info("Got Cancelled Error")
-                break
-
     async def get_state(self) -> CanbusServiceState:
         state: CanbusServiceState
         try:
@@ -98,28 +82,6 @@ async def get_state(self) -> CanbusServiceState:
         self.logger.debug("CanbusServiceStub: port -> %i state is: %s", self.config.port, state.name)
         return state
 
-    async def connect_to_service(self) -> None:
-        """Starts the canbus streaming.
-
-        The service state will go to `RUNNING`.
-        """
-        state: CanbusServiceState = await self.get_state()
-        if state.value == canbus_pb2.CanbusServiceState.UNAVAILABLE:
-            return
-        await self.stub.startService(canbus_pb2.StartServiceRequest())
-
-    async def disconnect_from_service(self) -> None:
-        state: CanbusServiceState = await self.get_state()
-        if state.value == canbus_pb2.CanbusServiceState.UNAVAILABLE:
-            return
-        await self.stub.stopService(canbus_pb2.StopServiceRequest())
-
-    async def pause_service(self) -> None:
-        """Pauses the canbus streaming.
-
-        The service state will go from `RUNNING` to `IDLE`.
-        """
-        state: CanbusServiceState = await self.get_state()
-        if state.value == canbus_pb2.CanbusServiceState.UNAVAILABLE:
-            return
-        await self.stub.pauseService(canbus_pb2.PauseServiceRequest())
+    def stream(self):
+        """Return the async streaming object."""
+        return self.stub.streamCanbusMessages(canbus_pb2.StreamCanbusRequest())

py/farm_ng/oak/camera_client.py

@@ -125,17 +125,11 @@ def __init__(self, config: OakCameraClientConfig) -> None:
         self.channel = grpc.aio.insecure_channel(self.server_address)
         self.stub = oak_pb2_grpc.OakServiceStub(self.channel)
 
-        self._state = OakCameraServiceState()
-
         self._mono_camera_settings = oak_pb2.CameraSettings(auto_exposure=True)
         self._rgb_camera_settings = oak_pb2.CameraSettings(auto_exposure=True)
 
         self.needs_update = False
 
-    @property
-    def state(self) -> OakCameraServiceState:
-        return self._state
-
     @property
     def server_address(self) -> str:
         """Returns the composed address and port."""
@@ -154,15 +148,6 @@ def settings_reply(self, reply) -> None:
             self._mono_camera_settings.CopyFrom(reply.stereo_settings)
             self._rgb_camera_settings.CopyFrom(reply.rgb_settings)
 
-    async def poll_service_state(self) -> None:
-        while True:
-            try:
-                self._state = await self.get_state()
-                await asyncio.sleep(0.1)
-            except asyncio.CancelledError:
-                self.logger.info("Got CancellededError")
-                break
-
     async def get_state(self) -> OakCameraServiceState:
         """Async call to retrieve the state of the connected service."""
         state: OakCameraServiceState
@@ -176,28 +161,6 @@ async def get_state(self) -> OakCameraServiceState:
         self.logger.debug("OakServiceStub: port -> %i state is: %s", self.config.port, state.name)
         return state
 
-    async def connect_to_service(self) -> None:
-        """Start the camera streaming.
-
-        The service state will go from `IDLE` to `RUNNING`.
-        """
-        state: OakCameraServiceState = await self.get_state()
-        if state.value == oak_pb2.OakServiceState.UNAVAILABLE:
-            return
-        reply = await self.stub.cameraControl(oak_pb2.CameraControlRequest())
-        self.settings_reply(reply)
-        await self.stub.startService(oak_pb2.StartServiceRequest())
-
-    async def pause_service(self) -> None:
-        """Pauses the camera streaming.
-
-        The service state will go from `RUNNING` to `IDLE`.
-        """
-        state: OakCameraServiceState = await self.get_state()
-        if state.value == oak_pb2.OakServiceState.UNAVAILABLE:
-            return
-        await self.stub.pauseService(oak_pb2.PauseServiceRequest())
-
     async def send_settings(self) -> oak_pb2.CameraControlReply:
         request = oak_pb2.CameraControlRequest()
         request.stereo_settings.CopyFrom(self._mono_camera_settings)