Add starting a Gear Guard live stream

src/rivian/parallax/__init__.py

@@ -17,6 +17,8 @@
         The RVM and payload of a signed lock or unlock command.
     decode_command_state_message(data) -> dict | None
         Reads a raw `subscribe_for_parallax_command_state` message.
+    encode_live_streaming(camera, device_id) -> bytes
+        A Gear Guard live stream start, for `start_gear_guard_live_stream`.
 
 Layout:
     core.py           - registry, payload parsing and shared helpers
@@ -65,6 +67,11 @@
     decode_parallax_subscription_message,
 )
 from .operation import encode_vehicle_operation
+from .security import (
+    GEAR_GUARD_CAMERAS,
+    START_LIVE_STREAMING_RVM,
+    encode_live_streaming,
+)
 
 PARALLAX_RVMS: list[str] = list(RVMDecoder.decoders)
 
@@ -85,14 +92,17 @@
     "CHARGING_RVMS",
     "CLIMATE_HOLD_SETTING_RVM",
     "CLIMATE_HOLD_STATUS_RVM",
+    "GEAR_GUARD_CAMERAS",
     "LOCK_RVM",
     "PARALLAX_RVMS",
+    "START_LIVE_STREAMING_RVM",
     "UNLOCK_RVM",
     "decode_command_state_message",
     "decode_parallax_message",
     "decode_parallax_subscription_message",
     "encode_climate_hold_setting",
     "encode_command",
+    "encode_live_streaming",
     "encode_lock",
     "encode_unlock",
     "encode_vehicle_operation",

src/rivian/parallax/proto/security.proto

@@ -129,3 +129,19 @@ enum TosAcceptance {
   TOS_NOT_ACCEPTED = 1;
   TOS_ACCEPTED = 2;
 }
+
+// security.video_monitoring.start_live_streaming
+message LiveStreamingRequest {
+  GearGuardCamera camera = 1;
+  string device_id = 2; // a label for the viewer; any stable string works
+}
+
+enum GearGuardCamera { // the app's camera selector
+  GEAR_GUARD_CAMERA_UNSPECIFIED = 0;
+  GEAR_GUARD_CAMERA_FRONT = 1;
+  GEAR_GUARD_CAMERA_BED = 2;
+  GEAR_GUARD_CAMERA_LEFT = 3;
+  GEAR_GUARD_CAMERA_RIGHT = 4;
+  GEAR_GUARD_CAMERA_REAR = 5;
+  GEAR_GUARD_CAMERA_INTERIOR = 6;
+}

src/rivian/parallax/security.py

@@ -7,6 +7,24 @@
 from .core import RVMDecoder, _enum, _present
 from .proto import security_pb2
 
+START_LIVE_STREAMING_RVM: Final = "security.video_monitoring.start_live_streaming"
+
+GEAR_GUARD_CAMERAS: Final[dict[str, security_pb2.GearGuardCamera]] = {
+    "front": security_pb2.GEAR_GUARD_CAMERA_FRONT,
+    "bed": security_pb2.GEAR_GUARD_CAMERA_BED,
+    "left": security_pb2.GEAR_GUARD_CAMERA_LEFT,
+    "right": security_pb2.GEAR_GUARD_CAMERA_RIGHT,
+    "rear": security_pb2.GEAR_GUARD_CAMERA_REAR,
+    "interior": security_pb2.GEAR_GUARD_CAMERA_INTERIOR,
+}
+
+
+def encode_live_streaming(camera: str, device_id: str) -> bytes:
+    """Encode a security.video_monitoring.start_live_streaming payload."""
+    return security_pb2.LiveStreamingRequest(
+        camera=GEAR_GUARD_CAMERAS[camera], device_id=device_id
+    ).SerializeToString()
+
 
 @RVMDecoder.register("security.access.passive_entry", security_pb2.PassiveEntry)
 def decode_passive_entry(_m: security_pb2.PassiveEntry) -> dict[str, Any]: