ronantakizawa/codereview-bench huggingface dataset:
Repository: codenotary/immudb
File: pkg/server/remote_storage_test.go (Go)

Code before:
	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	err := s.Initialize()
	require.Error(t, ErrNoStorageForIdentifier, err)
}

func TestInitializeWithRemoteStorageWithoutIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.Error(t, ErrNoRemoteIdentifier, err)
}

func TestInitializeRemoteStorageWithIdentifierWhenLocalIsPresent(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	storeData(t, m, "immudb.identifier", []byte{1, 2, 3, 4, 5})

	_, err := getOrSetUUID(dir, dir, false)
	require.NoError(t, err)

	err = s.initializeRemoteStorage(m)
	require.ErrorIs(t, err, ErrUnexpectedLocalIdentifier)
}

func TestInitializeRemoteStorageWithoutLocalIdentifier(t *testing.T) {
	dir := t.TempDir()


Review comment (suggestion):
```suggestion
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
```

Code after:
	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	err := s.Initialize()
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeWithRemoteStorageWithoutIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeRemoteStorageWithoutLocalIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	storeData(t, m, "immudb.identifier", []byte{1, 2, 3, 4, 5})
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.NoError(t, err)

	uuidFilename := filepath.Join(dir, "immudb.identifier")

	require.FileExists(t, uuidFilename)

	id, err := ioutil.ReadFile(uuidFilename)
	require.NoError(t, err)
Repository: codenotary/immudb
File: pkg/server/remote_storage_test.go (Go)

Code before:
	storeData(t, m, "immudb.identifier", []byte{1, 2, 3, 4, 5})

	err := s.initializeRemoteStorage(m)
	require.NoError(t, err)

	uuidFilename := filepath.Join(dir, "immudb.identifier")

	require.FileExists(t, uuidFilename)

	id, err := ioutil.ReadFile(uuidFilename)
	require.NoError(t, err)
	require.Equal(t, []byte{1, 2, 3, 4, 5}, id)
}

func TestInitializeWithEmptyRemoteStorage(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	err := s.Initialize()
	require.Error(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeWithRemoteStorageWithoutIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.Error(t, err, ErrNoRemoteIdentifier)
}

func TestInitializeRemoteStorageWithoutLocalIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true


Review comment (suggestion):
```suggestion
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
```

Using `ErrorIs` to validate `err` is that specific type of error

Code after:
	storeData(t, m, "immudb.identifier", []byte{1, 2, 3, 4, 5})

	err := s.initializeRemoteStorage(m)
	require.NoError(t, err)

	uuidFilename := filepath.Join(dir, "immudb.identifier")

	require.FileExists(t, uuidFilename)

	id, err := ioutil.ReadFile(uuidFilename)
	require.NoError(t, err)
	require.Equal(t, []byte{1, 2, 3, 4, 5}, id)
}

func TestInitializeWithEmptyRemoteStorage(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	err := s.Initialize()
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeWithRemoteStorageWithoutIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeRemoteStorageWithoutLocalIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true
Repository: codenotary/immudb
File: pkg/server/remote_storage_test.go (Go)

Code before:
	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	err := s.Initialize()
	require.Error(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeWithRemoteStorageWithoutIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.Error(t, err, ErrNoRemoteIdentifier)
}

func TestInitializeRemoteStorageWithoutLocalIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	storeData(t, m, "immudb.identifier", []byte{1, 2, 3, 4, 5})
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.NoError(t, err)

	uuidFilename := filepath.Join(dir, "immudb.identifier")

	require.FileExists(t, uuidFilename)

	id, err := ioutil.ReadFile(uuidFilename)
	require.NoError(t, err)

Review comment (suggestion):
```suggestion
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
```

Using `ErrorIs` to validate `err` is that specific type of error

Code after:
	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	err := s.Initialize()
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeWithRemoteStorageWithoutIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.ErrorIs(t, err, ErrNoStorageForIdentifier)
}

func TestInitializeRemoteStorageWithoutLocalIdentifier(t *testing.T) {
	dir := t.TempDir()

	opts := DefaultOptions().WithDir(dir)
	opts.RemoteStorageOptions.S3ExternalIdentifier = true

	s := DefaultServer()

	s.WithOptions(opts)

	m := memory.Open()
	storeData(t, m, "immudb.identifier", []byte{1, 2, 3, 4, 5})
	s.remoteStorage = m

	err := s.initializeRemoteStorage(m)
	require.NoError(t, err)

	uuidFilename := filepath.Join(dir, "immudb.identifier")

	require.FileExists(t, uuidFilename)

	id, err := ioutil.ReadFile(uuidFilename)
	require.NoError(t, err)
Repository: codenotary/immudb
File: pkg/pgsql/server/pgsql_integration_test.go (Go)

Code before:
			err = db.QueryRow(fmt.Sprintf("SELECT id, amount, title FROM %s", table)).Scan(&id, &amount, &title)
			require.NoError(t, err)

		}(i)
	}
	wg.Wait()

}

func TestPgsqlServer_VersionStatement(t *testing.T) {
	td := t.TempDir()
	options := server.DefaultOptions().WithDir(td).WithPgsqlServer(true).WithPgsqlServerPort(0)
	bs := servertest.NewBufconnServer(options)

	bs.Start()
	defer bs.Stop()

	defer os.Remove(".state-")

	bs.WaitForPgsqlListener()

	db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d sslmode=disable user=immudb dbname=defaultdb password=immudb", bs.Server.Srv.PgsqlSrv.GetPort()))
	require.NoError(t, err)

	var version string
	err = db.QueryRow(("SELECT version()").Scan(&version)
	require.NoError(t, err)
	require.Equal(t, pgmeta.PgsqlProtocolVersionMessage, version)
}

func TestPgsqlServerSetStatement(t *testing.T) {
	td := t.TempDir()
	options := server.DefaultOptions().WithDir(td).WithPgsqlServer(true).WithPgsqlServerPort(0)
	bs := servertest.NewBufconnServer(options)

	bs.Start()
	defer bs.Stop()

	defer os.Remove(".state-")

	bs.WaitForPgsqlListener()

	db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d sslmode=disable user=immudb dbname=defaultdb password=immudb", bs.Server.Srv.PgsqlSrv.GetPort()))
	require.NoError(t, err)

	_, err = db.Query("SET test=val")
	require.NoError(t, err)
}

func TestPgsqlServer_SimpleQueryNilValues(t *testing.T) {
	td := t.TempDir()

Review comment (suggestion):
```suggestion
	err = db.QueryRow("SELECT version()").Scan(&version)
```

Code after:
			err = db.QueryRow(fmt.Sprintf("SELECT id, amount, title FROM %s", table)).Scan(&id, &amount, &title)
			require.NoError(t, err)

		}(i)
	}
	wg.Wait()

}

func TestPgsqlServer_VersionStatement(t *testing.T) {
	td := t.TempDir()
	options := server.DefaultOptions().WithDir(td).WithPgsqlServer(true).WithPgsqlServerPort(0)
	bs := servertest.NewBufconnServer(options)

	bs.Start()
	defer bs.Stop()

	defer os.Remove(".state-")

	bs.WaitForPgsqlListener()

	db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d sslmode=disable user=immudb dbname=defaultdb password=immudb", bs.Server.Srv.PgsqlSrv.GetPort()))
	require.NoError(t, err)

	var version string
	err = db.QueryRow("SELECT version()").Scan(&version)
	require.NoError(t, err)
	require.Equal(t, pgmeta.PgsqlProtocolVersionMessage, version)
}

func TestPgsqlServerSetStatement(t *testing.T) {
	td := t.TempDir()
	options := server.DefaultOptions().WithDir(td).WithPgsqlServer(true).WithPgsqlServerPort(0)
	bs := servertest.NewBufconnServer(options)

	bs.Start()
	defer bs.Stop()

	defer os.Remove(".state-")

	bs.WaitForPgsqlListener()

	db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d sslmode=disable user=immudb dbname=defaultdb password=immudb", bs.Server.Srv.PgsqlSrv.GetPort()))
	require.NoError(t, err)

	_, err = db.Query("SET test=val")
	require.NoError(t, err)
}

func TestPgsqlServer_SimpleQueryNilValues(t *testing.T) {
	td := t.TempDir()
Repository: commaai/openpilot
File: tools/mac_setup.sh (Shell)

Code before:
  echo "[ ] installed brew t=$SECONDS"

  # make brew available now
  if [[ $ARCH == "x86_64" ]]; then
    echo 'eval "$(/usr/local/bin/brew shellenv)"' >> $RC_FILE
    eval "$(/usr/local/bin/brew shellenv)"
  else
    echo 'eval "$(/opt/homebrew/bin/brew shellenv)"' >> $RC_FILE
    eval "$(/opt/homebrew/bin/brew shellenv)"
  fi
else
    brew up
fi

brew bundle --file=- <<-EOS
brew "git-lfs"
brew "zlib"
brew "capnp"
brew "coreutils"
brew "eigen"
brew "ffmpeg"
brew "glfw"
brew "libarchive"
brew "libusb"
brew "libtool"
brew "llvm@19"
brew "openssl@3.0"
brew "qt@5"
brew "zeromq"
cask "gcc-arm-embedded"
brew "portaudio"
brew "gcc@13"
EOS

echo "[ ] finished brew install t=$SECONDS"

BREW_PREFIX=$(brew --prefix)

# archive backend tools for pip dependencies
export LDFLAGS="$LDFLAGS -L${BREW_PREFIX}/opt/zlib/lib"
export LDFLAGS="$LDFLAGS -L${BREW_PREFIX}/opt/bzip2/lib"
export CPPFLAGS="$CPPFLAGS -I${BREW_PREFIX}/opt/zlib/include"
export CPPFLAGS="$CPPFLAGS -I${BREW_PREFIX}/opt/bzip2/include"

# pycurl curl/openssl backend dependencies
export LDFLAGS="$LDFLAGS -L${BREW_PREFIX}/opt/openssl@3/lib"
export CPPFLAGS="$CPPFLAGS -I${BREW_PREFIX}/opt/openssl@3/include"
export PYCURL_CURL_CONFIG=/usr/bin/curl-config
export PYCURL_SSL_LIBRARY=openssl

# install python dependencies

Review comment (suggestion):
```suggestion
brew "llvm@19"  # pinned for tinygrad
```

Code after:
  echo "[ ] installed brew t=$SECONDS"

  # make brew available now
  if [[ $ARCH == "x86_64" ]]; then
    echo 'eval "$(/usr/local/bin/brew shellenv)"' >> $RC_FILE
    eval "$(/usr/local/bin/brew shellenv)"
  else
    echo 'eval "$(/opt/homebrew/bin/brew shellenv)"' >> $RC_FILE
    eval "$(/opt/homebrew/bin/brew shellenv)"
  fi
else
    brew up
fi

brew bundle --file=- <<-EOS
brew "git-lfs"
brew "zlib"
brew "capnp"
brew "coreutils"
brew "eigen"
brew "ffmpeg"
brew "glfw"
brew "libarchive"
brew "libusb"
brew "libtool"
brew "llvm@19"  # pinned for tinygrad
brew "openssl@3.0"
brew "qt@5"
brew "zeromq"
cask "gcc-arm-embedded"
brew "portaudio"
brew "gcc@13"
EOS

echo "[ ] finished brew install t=$SECONDS"

BREW_PREFIX=$(brew --prefix)

# archive backend tools for pip dependencies
export LDFLAGS="$LDFLAGS -L${BREW_PREFIX}/opt/zlib/lib"
export LDFLAGS="$LDFLAGS -L${BREW_PREFIX}/opt/bzip2/lib"
export CPPFLAGS="$CPPFLAGS -I${BREW_PREFIX}/opt/zlib/include"
export CPPFLAGS="$CPPFLAGS -I${BREW_PREFIX}/opt/bzip2/include"

# pycurl curl/openssl backend dependencies
export LDFLAGS="$LDFLAGS -L${BREW_PREFIX}/opt/openssl@3/lib"
export CPPFLAGS="$CPPFLAGS -I${BREW_PREFIX}/opt/openssl@3/include"
export PYCURL_CURL_CONFIG=/usr/bin/curl-config
export PYCURL_SSL_LIBRARY=openssl

# install python dependencies
Repository: commaai/openpilot
File: system/micd.py (Python)

Code before:
#!/usr/bin/env python3
import numpy as np
from functools import cache
import threading

from cereal import messaging
from openpilot.common.realtime import Ratekeeper
from openpilot.common.retry import retry
from openpilot.common.swaglog import cloudlog

RATE = 10
FFT_SAMPLES = 4096
REFERENCE_SPL = 2e-5  # newtons/m^2
SAMPLE_RATE = 16000
SAMPLE_BUFFER = 800  # 50ms


@cache
def get_a_weighting_filter():
  # Calculate the A-weighting filter
  # https://en.wikipedia.org/wiki/A-weighting
  freqs = np.fft.fftfreq(FFT_SAMPLES, d=1 / SAMPLE_RATE)
  A = 12194 ** 2 * freqs ** 4 / ((freqs ** 2 + 20.6 ** 2) * (freqs ** 2 + 12194 ** 2) * np.sqrt((freqs ** 2 + 107.7 ** 2) * (freqs ** 2 + 737.9 ** 2)))
  return A / np.max(A)


def calculate_spl(measurements):
  # https://www.engineeringtoolbox.com/sound-pressure-d_711.html
  sound_pressure = np.sqrt(np.mean(measurements ** 2))  # RMS of amplitudes
  if sound_pressure > 0:
    sound_pressure_level = 20 * np.log10(sound_pressure / REFERENCE_SPL)  # dB
  else:
    sound_pressure_level = 0
  return sound_pressure, sound_pressure_level


def apply_a_weighting(measurements: np.ndarray) -> np.ndarray:
  # Generate a Hanning window of the same length as the audio measurements
  measurements_windowed = measurements * np.hanning(len(measurements))


Review comment (performance):
This might've been chosen for performance with calculating the noise level, I'd check nothing changes significantly (does micd now run fft at 20Hz instead of 10Hz?)

Code after:
#!/usr/bin/env python3
import numpy as np
from functools import cache
import threading

from cereal import messaging
from openpilot.common.realtime import Ratekeeper
from openpilot.common.retry import retry
from openpilot.common.swaglog import cloudlog

RATE = 10
FFT_SAMPLES = 1600 # 100ms
REFERENCE_SPL = 2e-5  # newtons/m^2
SAMPLE_RATE = 16000
SAMPLE_BUFFER = 800  # 50ms


@cache
def get_a_weighting_filter():
  # Calculate the A-weighting filter
  # https://en.wikipedia.org/wiki/A-weighting
  freqs = np.fft.fftfreq(FFT_SAMPLES, d=1 / SAMPLE_RATE)
  A = 12194 ** 2 * freqs ** 4 / ((freqs ** 2 + 20.6 ** 2) * (freqs ** 2 + 12194 ** 2) * np.sqrt((freqs ** 2 + 107.7 ** 2) * (freqs ** 2 + 737.9 ** 2)))
  return A / np.max(A)


def calculate_spl(measurements):
  # https://www.engineeringtoolbox.com/sound-pressure-d_711.html
  sound_pressure = np.sqrt(np.mean(measurements ** 2))  # RMS of amplitudes
  if sound_pressure > 0:
    sound_pressure_level = 20 * np.log10(sound_pressure / REFERENCE_SPL)  # dB
  else:
    sound_pressure_level = 0
  return sound_pressure, sound_pressure_level


def apply_a_weighting(measurements: np.ndarray) -> np.ndarray:
  # Generate a Hanning window of the same length as the audio measurements
  measurements_windowed = measurements * np.hanning(len(measurements))
Repository: commaai/openpilot
File: common/util.cc (C++)

Code before:
  FILE* fp = NULL;
  do {
    fp = fopen(filename, mode);
  } while ((nullptr == fp) && (errno == EINTR));
  return fp;
}

size_t safe_fwrite(const void* ptr, size_t size, size_t count, FILE* stream) {
  size_t written = 0;
  do {
    size_t ret = ::fwrite((void*)((char *)ptr + written * size), size, count - written, stream);
    if (ret == 0 && errno != EINTR) break;
    written += ret;
  } while (written != count);
  return written;
}

int safe_fflush(FILE *stream) {
  int ret = EOF;
  do {
    ret = fflush(stream);
  } while ((EOF == ret) && (errno == EINTR));
  return ret;
}

int safe_ioctl(int fd, unsigned long request, void *argp, const char* error_msg, std::function<void()> error_callback ) {
  int ret;
  do {
    ret = ioctl(fd, request, argp);
  } while ((ret == -1) && (errno == EINTR));

  if (ret == -1) {
    LOGE("safe_ioctl error: %s(%d) %s (fd: %d request: %lx argp: %p)", strerror(errno), errno, error_msg, fd, request, argp);
    if (error_callback) {
      error_callback();
    }
  }
  return ret;
}

std::string readlink(const std::string &path) {
  char buff[4096];
  ssize_t len = ::readlink(path.c_str(), buff, sizeof(buff)-1);
  if (len != -1) {
    buff[len] = '\0';
    return std::string(buff);
  }
  return "";
}

bool file_exists(const std::string& fn) {

Review comment (suggestion):
I propose changing the API slightly to
```suggestion
int safe_ioctl(int fd, unsigned long request, void *argp, const char* exception_msg) {
```

then just throw an `std::runtime_error` if exception_msg is defined. for the assert case, it should be functionally the same (aborts the program)

Code after:
  FILE* fp = NULL;
  do {
    fp = fopen(filename, mode);
  } while ((nullptr == fp) && (errno == EINTR));
  return fp;
}

size_t safe_fwrite(const void* ptr, size_t size, size_t count, FILE* stream) {
  size_t written = 0;
  do {
    size_t ret = ::fwrite((void*)((char *)ptr + written * size), size, count - written, stream);
    if (ret == 0 && errno != EINTR) break;
    written += ret;
  } while (written != count);
  return written;
}

int safe_fflush(FILE *stream) {
  int ret = EOF;
  do {
    ret = fflush(stream);
  } while ((EOF == ret) && (errno == EINTR));
  return ret;
}

int safe_ioctl(int fd, unsigned long request, void *argp, const char* exception_msg) {
  int ret;
  do {
    ret = ioctl(fd, request, argp);
  } while ((ret == -1) && (errno == EINTR));

  if (ret == -1 && exception_msg) {
    LOGE("safe_ioctl error: %s %s(%d) (fd: %d request: %lx argp: %p)", exception_msg, strerror(errno), errno, fd, request, argp);
    throw std::runtime_error(exception_msg);
  }
  return ret;
}

std::string readlink(const std::string &path) {
  char buff[4096];
  ssize_t len = ::readlink(path.c_str(), buff, sizeof(buff)-1);
  if (len != -1) {
    buff[len] = '\0';
    return std::string(buff);
  }
  return "";
}

bool file_exists(const std::string& fn) {
  struct stat st = {};
  return stat(fn.c_str(), &st) != -1;
Repository: commaai/openpilot
File: system/webrtc/device/video.py (Python)

Code before:
    "driver": "livestreamDriverEncodeData",
    "wideRoad": "livestreamWideRoadEncodeData",
    "road": "livestreamRoadEncodeData",
  }

  def __init__(self, camera_type: str):
    dt = DT_DMON if camera_type == "driver" else DT_MDL
    super().__init__(camera_type, dt)

    self._sock = messaging.sub_sock(self.camera_to_sock_mapping[camera_type], conflate=True)
    self._pts = 0
    self._t0_ns = time.monotonic_ns()

  async def recv(self):
    while True:
      msg = messaging.recv_one_or_none(self._sock)
      if msg is not None:
        break
      await asyncio.sleep(0.005)

    evta = getattr(msg, msg.which())

    packet = av.Packet(evta.header + evta.data)
    packet.time_base = self._time_base

    self._pts =  ((time.monotonic_ns()-self._t0_ns) * self._clock_rate) // 1_000_000_000
    packet.pts = self._pts
    self.log_debug("track sending frame %d", self._pts)

    return packet

  def codec_preference(self) -> str | None:
    return "H264"


Review comment (suggestion):
```suggestion
    self._pts =  ((time.monotonic_ns() - self._t0_ns) * self._clock_rate) // 1_000_000_000
```

Code after:
    "driver": "livestreamDriverEncodeData",
    "wideRoad": "livestreamWideRoadEncodeData",
    "road": "livestreamRoadEncodeData",
  }

  def __init__(self, camera_type: str):
    dt = DT_DMON if camera_type == "driver" else DT_MDL
    super().__init__(camera_type, dt)

    self._sock = messaging.sub_sock(self.camera_to_sock_mapping[camera_type], conflate=True)
    self._pts = 0
    self._t0_ns = time.monotonic_ns()

  async def recv(self):
    while True:
      msg = messaging.recv_one_or_none(self._sock)
      if msg is not None:
        break
      await asyncio.sleep(0.005)

    evta = getattr(msg, msg.which())

    packet = av.Packet(evta.header + evta.data)
    packet.time_base = self._time_base

    self._pts =  ((time.monotonic_ns() - self._t0_ns) * self._clock_rate) // 1_000_000_000
    packet.pts = self._pts
    self.log_debug("track sending frame %d", self._pts)

    return packet

  def codec_preference(self) -> str | None:
    return "H264"