od_reference.py

# -*- coding: utf-8 -*-
from __future__ import annotations

from json import dumps

import click
from msgspec.json import encode

from pioreactor import types as pt
from pioreactor import whoami
from pioreactor.config import config
from pioreactor.logging import create_logger
from pioreactor.utils import local_persistant_storage
from pioreactor.utils import publish_ready_to_disconnected_state
from pioreactor.actions.od_blank import od_statistics

def od_reference(
    od_angle_channel1: pt.PdAngleOrREF,
    od_angle_channel2: pt.PdAngleOrREF,
    n_samples: int = 20,
    unit=None,
    experiment=None,
):
    action_name = "od_reference"
    logger = create_logger(action_name)
    unit = unit or whoami.get_unit_name()
    experiment = experiment or whoami.get_latest_experiment_name()
    testing_experiment = whoami.get_latest_testing_experiment_name()

    from pioreactor.background_jobs.od_reading import start_od_reading
    from pioreactor.background_jobs.stirring import start_stirring

    with publish_ready_to_disconnected_state(unit, experiment, action_name):
        
        with start_od_reading(
            od_angle_channel1,
            od_angle_channel2,
            unit=unit,
            interval=1.5,
            experiment=testing_experiment,
            fake_data=whoami.is_testing_env(),
        ) as od_stream, start_stirring(
            target_rpm=config.getfloat("stirring", "target_rpm"),
            unit=unit,
            experiment=testing_experiment,
        ) as st:
            # warm up OD reader
            for count, _ in enumerate(od_stream, start=0):
                if count == 5:
                    break

            st.block_until_rpm_is_close_to_target(timeout=30)

            means, variances = od_statistics(
                od_stream,
                action_name,
                unit=unit,
                experiment=experiment,
                n_samples=n_samples,
                logger=logger,
            )

            with local_persistant_storage("od_normalization_mean") as cache:
                cache[experiment] = dumps(means)

            with local_persistant_storage("od_normalization_variance") as cache:
                cache[experiment] = dumps(variances)

            logger.info("Finished reading reference OD.")

    return 


@click.command(name="od_reference")
@click.option(
    "--od-angle-channel1",
    default=config.get("od_config.photodiode_channel", "1", fallback=None),
    type=click.STRING,
    show_default=True,
    help="specify the angle(s) between the IR LED(s) and the PD in channel 1, separated by commas. Don't specify if channel is empty.",
)
@click.option(
    "--od-angle-channel2",
    default=config.get("od_config.photodiode_channel", "2", fallback=None),
    type=click.STRING,
    show_default=True,
    help="specify the angle(s) between the IR LED(s) and the PD in channel 2, separated by commas. Don't specify if channel is empty.",
)
@click.option(
    "--n-samples",
    default=30,
    show_default=True,
    help="Number of samples",
)
def click_od_reference(od_angle_channel1, od_angle_channel2, n_samples: int) -> None:
    """
    Compute statistics about a reference OD time series
    """
    od_reference(od_angle_channel1, od_angle_channel2, n_samples=n_samples)

od_reference.yaml

---
display_name: OD Reference
job_name: od_reference
display: true  # true to display on the /Pioreactors card
source: od_reference
description: Calculate a new reference value growth rate calculating and normalized OD
published_settings: []