rm3100/tests/test_rm3100.py

"""Driver maths and register sequences -- no hardware, no USB."""

import pytest

import rm3100
from conftest import FakeBus


# --------------------------------------------------------------------------
# Gain and resolution
# --------------------------------------------------------------------------

@pytest.mark.parametrize("cycle_count, table_lsb_per_ut", [
    (50, 20), (100, 38), (200, 75),
])
def test_gain_fit_reproduces_table_3_1(cycle_count, table_lsb_per_ut):
    """The fit exists to reproduce Table 3-1's three points within a count."""
    fitted = rm3100.gain_lsb_per_tesla(cycle_count) / rm3100.UT_PER_TESLA
    assert abs(fitted - table_lsb_per_ut) < 1.0


def test_tesla_per_count_inverts_the_gain():
    for cc in (30, 100, 400, 65535):
        assert rm3100.tesla_per_count(cc) * rm3100.gain_lsb_per_tesla(cc) == pytest.approx(1.0)


def test_higher_cycle_count_means_finer_lsb():
    steps = [rm3100.tesla_per_count(cc) for cc in (50, 100, 200, 400)]
    assert steps == sorted(steps, reverse=True)


def test_expected_noise_follows_inverse_sqrt():
    """Table 3-1's 30/20/15 nT at 50/100/200 is a K/sqrt(cc) law."""
    assert rm3100.expected_noise_nt(50) == pytest.approx(29.4, abs=0.7)
    assert rm3100.expected_noise_nt(100) == pytest.approx(20.8, abs=0.9)
    assert rm3100.expected_noise_nt(200) == pytest.approx(14.7, abs=0.4)
    # Quadrupling the cycle count halves the noise.
    assert (rm3100.expected_noise_nt(100)
            == pytest.approx(2 * rm3100.expected_noise_nt(400)))


# --------------------------------------------------------------------------
# Timing model
# --------------------------------------------------------------------------

def test_sample_period_matches_the_documented_model():
    expected = 3 * (100 / rm3100.COUNTS_PER_SECOND + rm3100.AXIS_OVERHEAD_S)
    assert rm3100.sample_period(100) == pytest.approx(expected)
    # The default configuration is quoted at ~282 Hz throughout the docs.
    assert 1 / rm3100.sample_period(100) == pytest.approx(282, abs=1)


def test_integration_time_excludes_overhead():
    assert rm3100.integration_time(100) < rm3100.sample_period(100)
    assert rm3100.integration_time(100) == pytest.approx(3 * 100 / 90000.0)


def test_duty_rises_with_cycle_count():
    """Fixed per-axis overhead is a shrinking share as integration grows."""
    duties = [rm3100.integration_time(cc) / rm3100.sample_period(cc)
              for cc in (50, 100, 200, 800)]
    assert duties == sorted(duties)
    assert duties[-1] < 1.0


@pytest.mark.parametrize("rate", [0.5, 1, 8, 32, 128, 282, 500])
def test_cycle_count_for_rate_inverts_sample_period(rate):
    cc = rm3100.cycle_count_for_rate(rate)
    achieved = 1.0 / rm3100.sample_period(cc)
    # Rounding to an integer cycle count is the only error here.
    assert achieved == pytest.approx(rate, rel=0.01)


def test_cycle_count_for_rate_clamps_to_the_register_width():
    assert rm3100.cycle_count_for_rate(1e9) == 1
    assert rm3100.cycle_count_for_rate(1e-9) == rm3100.MAX_CYCLE_COUNT


# --------------------------------------------------------------------------
# plan(): every branch, since it is the only place configuration is resolved
# --------------------------------------------------------------------------

def test_plan_default_is_the_documented_configuration():
    cfg = rm3100.plan()
    assert cfg.cycle_count == rm3100.DEFAULT_CYCLE_COUNT
    assert cfg.tmrc == rm3100.TMRC_FASTEST
    assert cfg.governed_by == "cycle count"
    assert cfg.predicted_hz == pytest.approx(282, abs=1)
    assert cfg.duty == pytest.approx(0.94, abs=0.01)
    assert cfg.notes == []


def test_plan_from_rate_derives_the_cycle_count():
    cfg = rm3100.plan(rate_hz=50)
    assert cfg.predicted_hz == pytest.approx(50, rel=0.01)
    assert cfg.governed_by == "cycle count"
    assert cfg.notes == []


def test_plan_from_cycle_count_ignores_the_rate_knob():
    cfg = rm3100.plan(cycle_count=400)
    assert cfg.cycle_count == 400
    assert cfg.predicted_hz == pytest.approx(1 / rm3100.sample_period(400))


def test_plan_raises_a_cycle_count_below_the_quantisation_floor():
    cfg = rm3100.plan(cycle_count=5)
    assert cfg.cycle_count == rm3100.MIN_CYCLE_COUNT
    assert any("quantisation floor" in n for n in cfg.notes)


def test_plan_clamps_a_cycle_count_above_the_register_width():
    cfg = rm3100.plan(cycle_count=100_000)
    assert cfg.cycle_count == rm3100.MAX_CYCLE_COUNT


def test_plan_below_the_cycle_count_floor_hands_the_cadence_to_tmrc():
    """Under ~0.46 Hz the 16-bit register runs out and only TMRC can go slower."""
    rate = rm3100.MIN_RATE_BY_CYCLE_COUNT / 2
    cfg = rm3100.plan(rate_hz=rate)
    assert cfg.cycle_count == rm3100.MAX_CYCLE_COUNT
    assert cfg.governed_by == "TMRC"
    assert cfg.tmrc == min(rm3100.TMRC_RATES,
                           key=lambda t: abs(rm3100.TMRC_RATES[t] - rate))
    assert any("below the" in n for n in cfg.notes)


def test_plan_warns_when_tmrc_governs_and_the_sensor_idles():
    """cc=50 can run at 534 Hz; TMRC 0x99 asks for 4.5, so it idles ~99%."""
    cfg = rm3100.plan(cycle_count=50, tmrc=0x99)
    assert cfg.governed_by == "TMRC"
    assert cfg.predicted_hz == pytest.approx(4.5)
    assert cfg.duty < 0.02
    assert any("idle" in n for n in cfg.notes)


def test_plan_notes_a_rate_it_cannot_reach():
    cfg = rm3100.plan(rate_hz=5000)
    assert cfg.predicted_hz < 5000
    assert any("faster than this configuration can reach" in n for n in cfg.notes)


def test_plan_duty_is_integration_over_the_achieved_period():
    cfg = rm3100.plan(cycle_count=200)
    assert cfg.duty == pytest.approx(
        rm3100.integration_time(200) * cfg.predicted_hz)


def test_plan_rejects_an_unknown_tmrc():
    with pytest.raises(KeyError):
        rm3100.plan(tmrc=0x00)


# --------------------------------------------------------------------------
# Measurement decoding -- 24-bit two's complement, big-endian
# --------------------------------------------------------------------------

@pytest.mark.parametrize("raw, expected", [
    (b"\x00\x00\x00" * 3, (0, 0, 0)),
    (b"\x00\x00\x01" * 3, (1, 1, 1)),
    (b"\xff\xff\xff" * 3, (-1, -1, -1)),
    (b"\x7f\xff\xff" * 3, (8388607, 8388607, 8388607)),      # positive rail
    (b"\x80\x00\x00" * 3, (-8388608, -8388608, -8388608)),   # negative rail
])
def test_decode_measurements_sign_extends(raw, expected):
    assert rm3100.decode_measurements(raw) == expected


def test_decode_measurements_keeps_axes_in_order():
    assert rm3100.decode_measurements(
        bytes([0, 0, 1, 0, 0, 2, 255, 255, 255])) == (1, 2, -1)


@pytest.mark.parametrize("length", [0, 8, 10, 27])
def test_decode_measurements_rejects_a_wrong_length(length):
    with pytest.raises(ValueError, match="9 measurement bytes"):
        rm3100.decode_measurements(bytes(length))


# --------------------------------------------------------------------------
# Register access
# --------------------------------------------------------------------------

def test_set_cycle_counts_writes_all_three_axes_big_endian():
    bus = FakeBus()
    sensor = rm3100.RM3100(bus, 0x23)
    sensor.set_cycle_counts(0x0190)
    assert bus.writes == [(0x23, bytes([rm3100.REG_CCX]) + b"\x01\x90" * 3)]
    assert sensor.get_cycle_counts() == (400, 400, 400)


@pytest.mark.parametrize("bad", [-1, 0x10000, 999999])
def test_set_cycle_counts_rejects_out_of_range(bad):
    sensor = rm3100.RM3100(FakeBus(), 0x23)
    with pytest.raises(ValueError, match="outside 0..65535"):
        sensor.set_cycle_counts(bad)


def test_set_rate_rejects_a_value_outside_the_tmrc_table():
    sensor = rm3100.RM3100(FakeBus(), 0x23)
    with pytest.raises(ValueError, match="not one of"):
        sensor.set_rate(0x91)
    sensor.set_rate(rm3100.TMRC_FASTEST)      # a valid one must not raise


def test_configure_accepts_a_correct_hshake_readback():
    bus = FakeBus({rm3100.REG_HSHAKE: 0x1B})
    rm3100.RM3100(bus, 0x23).configure()
    assert bus.registers[rm3100.REG_HSHAKE] == rm3100.HSHAKE_DRDY_ON_READ_ONLY


def test_configure_ignores_the_read_only_nack_bits():
    """Bits 4-6 are NACK status, so they may read back set without meaning failure."""
    bus = FakeBus()
    sensor = rm3100.RM3100(bus, 0x23)

    def read_reg(reg, count=1):
        return bytes([rm3100.HSHAKE_DRDY_ON_READ_ONLY | 0x70]), 0.0, 0.0

    sensor.read_reg = read_reg
    sensor.configure()      # must not raise


def test_configure_raises_when_drc1_did_not_take():
    """Exactly-once sampling depends on DRC1, so a bad readback is fatal."""
    sensor = rm3100.RM3100(FakeBus(), 0x23)
    sensor.read_reg = lambda reg, count=1: (b"\x1b", 0.0, 0.0)
    with pytest.raises(IOError, match="HSHAKE did not take"):
        sensor.configure()


def test_read_reg_prefers_a_combined_transaction():
    bus = FakeBus({rm3100.REG_REVID: rm3100.EXPECTED_REVID})
    sensor = rm3100.RM3100(bus, 0x23)
    assert sensor.revid() == rm3100.EXPECTED_REVID
    # One transaction, not a separate write then read.
    assert len(bus.writes) == 1


def test_read_reg_falls_back_when_the_bus_cannot_do_repeated_start():
    bus = FakeBus({rm3100.REG_REVID: rm3100.EXPECTED_REVID}, combined=False)
    sensor = rm3100.RM3100(bus, 0x23)
    assert sensor.revid() == rm3100.EXPECTED_REVID


def test_read_raw_decodes_from_the_measurement_registers():
    registers = {rm3100.REG_MX + i: b for i, b in enumerate(
        [0x00, 0x00, 0x01, 0xFF, 0xFF, 0xFF, 0x7F, 0xFF, 0xFF])}
    sensor = rm3100.RM3100(FakeBus(registers), 0x23)
    counts, mono, wall = sensor.read_raw()
    assert counts == (1, -1, 8388607)
    assert mono > 0 and wall > mono      # the stamp travels with the data


def test_read_measurements_converts_with_the_active_cycle_count():
    registers = {rm3100.REG_MX + i: b for i, b in enumerate(
        [0x00, 0x00, 0x26, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])}
    sensor = rm3100.RM3100(FakeBus(registers), 0x23)
    sensor.cycle_count = 100
    counts, tesla = sensor.read_measurements()
    assert counts[0] == 0x26
    # 38 LSB/uT at cc=100, so 0x26 counts is very close to 1 uT.
    assert tesla[0] == pytest.approx(1e-6, rel=0.01)


def test_data_ready_reads_the_drdy_bit():
    sensor = rm3100.RM3100(FakeBus({rm3100.REG_STATUS: 0x80}), 0x23)
    assert sensor.data_ready() is True
    sensor = rm3100.RM3100(FakeBus({rm3100.REG_STATUS: 0x00}), 0x23)
    assert sensor.data_ready() is False


def test_wait_for_data_times_out_without_drdy():
    sensor = rm3100.RM3100(FakeBus({rm3100.REG_STATUS: 0x00}), 0x23)
    assert sensor.wait_for_data(timeout=0.01, interval=0.001) is False


def test_cmm_start_and_stop_write_the_documented_values():
    bus = FakeBus()
    sensor = rm3100.RM3100(bus, 0x23)
    sensor.start_cmm()
    sensor.stop_cmm()
    assert bus.writes == [
        (0x23, bytes([rm3100.REG_CMM, rm3100.CMM_ALL_AXES])),
        (0x23, bytes([rm3100.REG_CMM, rm3100.CMM_OFF])),
    ]


def test_addresses_cover_both_strap_pins():
    assert list(rm3100.RM3100.ADDRESSES) == [0x20, 0x21, 0x22, 0x23]