Revert "webrtc: livestream bitrate controller (#38120)"

This reverts commit 04e2351246.

system/webrtc/webrtcd.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 
 from abc import abstractmethod
-import os
 import time
 import argparse
 import asyncio
@@ -23,7 +22,6 @@ if TYPE_CHECKING:
   from aiortc.rtcdatachannel import RTCDataChannel
 
 from openpilot.system.webrtc.schema import generate_field
-from openpilot.common.params import Params
 from cereal import messaging, log
 
 
@@ -129,78 +127,6 @@ class DynamicPubMaster(messaging.PubMaster):
           self.sock[service] = messaging.pub_sock(service)
 
 
-class LivestreamBitrateController(AsyncTaskRunner):
-  bitrates = [500_000, 1_500_000, int(os.environ.get("STREAM_BITRATE", 5_000_000))]
-  label_to_bitrate = { "high": bitrates[2], "med": bitrates[1], "low": bitrates[0]}
-  sample_interval = 0.2
-  high_level = 0.1 # drop immediately
-  med_level = 0.05 # drop after # of samples
-  low_level = 0 # raise after # of samples
-  down_samples = 5 # 1s
-  param_name = "LivestreamEncoderBitrate"
-
-  def __init__(self, peer_connection: Any):
-    super().__init__()
-    self.pc = peer_connection
-    self.params = Params()
-
-    self.level = 0
-    self.prev_lost, self.prev_sent = None, None
-    self.counter = 0
-    self.up_samples = 5 # 1s
-    self._auto = True
-
-  async def run(self):
-    while True:
-      await asyncio.sleep(self.sample_interval)
-      if not self._auto:
-        continue
-
-      loss_rate = await self._sample()
-      if loss_rate is None:
-        continue
-      if loss_rate >= self.med_level and self.level > 0:
-        self.counter += 1
-        if self.counter >= self.down_samples or loss_rate >= self.high_level:
-          self.level -= 1
-          self.up_samples *= 2 # exponential backoff before raising again
-          self.counter = 0
-          self._publish(self.bitrates[self.level])
-      elif loss_rate <= self.low_level and self.level < len(self.bitrates) - 1:
-        self.counter -= 1
-        if -self.counter >= self.up_samples:
-          self.level += 1
-          self.counter = 0
-          self._publish(self.bitrates[self.level])
-
-  async def _sample(self) -> float | None:
-    report = await self.pc.getStats()
-    packets_lost = packets_sent = 0
-    for s in report.values():
-      if s.type == "remote-inbound-rtp":
-        packets_lost += s.packetsLost
-      elif s.type == "outbound-rtp":
-        packets_sent += s.packetsSent
-
-    if self.prev_lost is None:
-      self.prev_lost, self.prev_sent = packets_lost, packets_sent
-      return None
-    lost_delta = max(0, packets_lost - self.prev_lost)
-    sent_delta = max(0, packets_sent - self.prev_sent)
-    self.prev_lost, self.prev_sent = packets_lost, packets_sent
-    return lost_delta / sent_delta if sent_delta else 0.0
-
-  def _publish(self, bitrate: float):
-    self.params.put(self.param_name, bitrate)
-
-  def set_quality(self, quality):
-    if quality in self.label_to_bitrate:
-      self._publish(self.label_to_bitrate[quality])
-      self._auto = False
-    elif quality == "auto":
-      self._auto = True
-
-
 class StreamSession:
   shared_pub_master = DynamicPubMaster([])
 
@@ -220,12 +146,10 @@ class StreamSession:
     self.incoming_bridge: CerealIncomingMessageProxy | None = None
     self.incoming_bridge_services = incoming_services
     self.outgoing_bridge: CerealOutgoingMessageProxy | None = None
-    self.bitrate_controller: LivestreamBitrateController | None = None
     if len(incoming_services) > 0:
       self.incoming_bridge = CerealIncomingMessageProxy(self.shared_pub_master)
     if len(outgoing_services) > 0:
       self.outgoing_bridge = CerealOutgoingMessageProxy(messaging.SubMaster(outgoing_services))
-    self.bitrate_controller = LivestreamBitrateController(self.stream.peer_connection)
 
     self.run_task: asyncio.Task | None = None
     self._cleanup_lock = asyncio.Lock()
@@ -257,23 +181,27 @@ class StreamSession:
       if isinstance(payload, dict):
         msg_type = payload.get("type")
 
-        match msg_type:
-          case "livestreamCameraSwitch":
-            self.video_track.switch_camera(payload["data"]["camera"])
-          case "livestreamSettings":
-            self.bitrate_controller.set_quality(payload["data"]["quality"])
-          case "clockSync":
-            pong = json.dumps({"type": "clockSync", "data": {
-              "action": "pong", "browserSendTime": payload["data"]["browserSendTime"], "deviceTime": time.time() * 1000, # noqa: TID251
-            }})
-            self.stream.get_messaging_channel().send(pong)
-          case "enableTimingSei":
-            if hasattr(self.video_track, 'timing_sei_enabled'):
-              self.video_track.timing_sei_enabled = bool(payload["data"]["enabled"])
-          case _:
-            if payload.get("type") not in self.incoming_bridge_services:
-              return
-            self.incoming_bridge.send(message)
+        if msg_type == "livestreamCameraSwitch":
+          self.video_track.switch_camera(payload["data"]["camera"])
+          return
+
+        if msg_type == "clockSync":
+          data = payload.get("data", {})
+          pong = json.dumps({"type": "clockSync", "data": {
+            "action": "pong", "browserSendTime": data.get("browserSendTime"), "deviceTime": time.time() * 1000, # noqa: TID251
+          }})
+          self.stream.get_messaging_channel().send(pong)
+          return
+
+        if msg_type == "enableTimingSei":
+          enabled = bool(payload.get("data", {}).get("enabled"))
+          if hasattr(self.video_track, 'timing_sei_enabled'):
+            self.video_track.timing_sei_enabled = enabled
+          return
+
+      if payload.get("type") not in self.incoming_bridge_services:
+        return
+      self.incoming_bridge.send(message)
     except Exception:
       self.logger.exception("Cereal incoming proxy failure")
 
@@ -288,8 +216,6 @@ class StreamSession:
           channel = self.stream.get_messaging_channel()
           self.outgoing_bridge.add_channel(channel)
           self.outgoing_bridge.start()
-      self.bitrate_controller.start()
-
       self.logger.info("Stream session (%s) connected", self.identifier)
 
       await self.stream.wait_for_disconnection()
@@ -305,8 +231,6 @@ class StreamSession:
       if self._cleanup_done:
         return
       self._cleanup_done = True
-      if self.bitrate_controller is not None:
-        await self.bitrate_controller.stop()
       if self.outgoing_bridge is not None:
         await self.outgoing_bridge.stop()
       await self.stream.stop()