feat(STAR): read recorded TADM pressure traces from a PIP channel

[rickwierenga]

Summary

Adds methods on PIPChannel to read a recorded TADM (Total Aspiration and Dispense Monitoring) pressure trace back into a list of floats.

• request_tadm_recording_length(slot=0) — number of recorded samples in a measurement slot via Px:QL.
• request_tadm_recording(num_samples=None, start=0, batch_size=50, slot=0) — read the trace via Px:QN. num_samples=None queries the length and reads the whole trace, batched (≤50 samples/command, the firmware cap). Returns List[float].
• request_tadm_recording_value(index) — single sample.

Notes

• QN sample fields are sign + 4 digits, so they parse with qn#### (n) (the 5-# form drops the last value and breaks single-sample reads).
• Values are the raw signed firmware pressure samples cast to float; the QN firmware command does not document a physical unit.

Testing

Hardware-validated on a STAR: a 150 µL single-channel aspiration with recording_mode=2 recorded 106 samples; QL (length) and QN (read-back) agree on the count, and the trace shows the expected underpressure dip during aspiration (0 → ~−200 → 0). Reproduced across two runs.

Recording requires the firmware prelude AQ→AY→BG→AF before aspirating (so tadm_algorithm=True has a valid limit curve); the read-back methods themselves are independent of that.

🤖 Generated with Claude Code

[rickwierenga]

Usage samples

Recording a TADM trace needs only tadm_recording_mode="all" — the BG (begin monitoring) step is issued automatically inside aspirate/dispense.

1. Simple read-back (await the aspiration, then read the finalized trace):

import asyncio
from pylabrobot.hamilton.liquid_handlers.star.star import STAR
from pylabrobot.hamilton.liquid_handlers.star.pip_backend import STARPIPBackend
from pylabrobot.resources.hamilton.tip_carriers import TIP_CAR_480_A00
from pylabrobot.resources.hamilton.tip_racks import hamilton_96_tiprack_1000uL_filter
from pylabrobot.resources.hamilton.plate_carriers import PLT_CAR_L5AC_A00
from pylabrobot.resources.corning.plates import Cor_96_wellplate_360ul_Fb

async def main():
    star = STAR()
    tip_carrier = TIP_CAR_480_A00(name="tip carrier")
    tip_carrier[0] = tips = hamilton_96_tiprack_1000uL_filter(name="tips")
    plate_carrier = PLT_CAR_L5AC_A00(name="plate carrier")
    plate_carrier[0] = plate = Cor_96_wellplate_360ul_Fb(name="plate")
    star.deck.assign_child_resource(tip_carrier, rails=1)
    star.deck.assign_child_resource(plate_carrier, rails=15)
    await star.setup()

    ch = star.driver.pip.channels[0]
    await star.pip.pick_up_tips(tips["A1"], use_channels=[0])

    # tadm_recording_mode="all" is the whole TADM setup; BG is issued automatically.
    params = STARPIPBackend.AspirateParams(tadm_recording_mode="all")
    await star.pip.aspirate(plate["A1"], vols=[150.0], use_channels=[0], backend_params=params)

    trace = await ch.request_tadm_recording()   # list[float], one pressure sample per firmware tick
    print(f"recorded {len(trace)} samples; first 8: {[round(x) for x in trace[:8]]}")

    await star.pip.dispense(plate["A1"], vols=[150.0], use_channels=[0])
    await star.pip.drop_tips(tips["A1"], use_channels=[0])
    await star.stop()

asyncio.run(main())

2. Live stream (run the aspiration as a background task and iterate stream_tadm(), which yields the trace the moment it finalizes — mid-command, ~6 s before aspirate returns):

import asyncio
from pylabrobot.hamilton.liquid_handlers.star.star import STAR
from pylabrobot.hamilton.liquid_handlers.star.pip_backend import STARPIPBackend
from pylabrobot.resources.hamilton.tip_carriers import TIP_CAR_480_A00
from pylabrobot.resources.hamilton.tip_racks import hamilton_96_tiprack_1000uL_filter
from pylabrobot.resources.hamilton.plate_carriers import PLT_CAR_L5AC_A00
from pylabrobot.resources.corning.plates import Cor_96_wellplate_360ul_Fb

async def main():
    star = STAR()
    tip_carrier = TIP_CAR_480_A00(name="tip carrier")
    tip_carrier[0] = tips = hamilton_96_tiprack_1000uL_filter(name="tips")
    plate_carrier = PLT_CAR_L5AC_A00(name="plate carrier")
    plate_carrier[0] = plate = Cor_96_wellplate_360ul_Fb(name="plate")
    star.deck.assign_child_resource(tip_carrier, rails=1)
    star.deck.assign_child_resource(plate_carrier, rails=15)
    await star.setup()

    ch = star.driver.pip.channels[0]
    await star.pip.pick_up_tips(tips["A1"], use_channels=[0])

    params = STARPIPBackend.AspirateParams(tadm_recording_mode="all")
    # background task so we can read while it runs (Q* queries are lock-free)
    asp = asyncio.create_task(star.pip.aspirate(
        plate["A1"], vols=[150.0], use_channels=[0], backend_params=params))

    async for index, value in ch.stream_tadm():
        print(index, value)
    await asp

    await star.pip.dispense(plate["A1"], vols=[150.0], use_channels=[0])
    await star.pip.drop_tips(tips["A1"], use_channels=[0])
    await star.stop()

asyncio.run(main())

Both return identical data; #1 gets it after the command returns, #2 gets it as soon as the recording phase finalizes. Validated on a STAR. The buffer is overwritten per measurement (not appended), so consecutive aspirations don't accumulate.