Skip to main content

More about MQTT on the Pioreactor

You can view the MQTT data using

pio mqtt

And to follow a specific topic, use:

pio mqtt -t "pioreactor/{unit}/{experiment}/..."

How do you structure topics for MQTT?​

We suggest using the following prefix:

pioreactor/{unit name or $broadcast}/{experiment name or $experiment}/...

Bioreactor values and PWM duty cycles​

Bioreactor topics use pioreactor/<unit>/<experiment>/bioreactor/<variable>. Their JSON payload contains the worker's event timestamp and the numeric value, for example:

{"timestamp": "2026-10-14T12:00:00Z", "value": 14.0}

Read the value field when consuming these topics. In experiment profile expressions, use ${{ ::bioreactor:current_volume_ml.value }} to read the current unit's volume.

Jobs buffer up to 10,000 outgoing MQTT messages while the broker is unreachable, reducing lost readings during short network interruptions or leader reboots. This buffer is bounded and belongs to the running job; it is not a persistent archive.

How do I publish data to MQTT?​

  1. If you're writing code for Job, you can simply use Job's methods:

        def some_method(self, ...):
            ...

            self.publish(f"pioreactor/{self.unit}/{self.experiment}/{self.job_name}/more/topics", payload)

  2. If writing a script:

    from pioreactor.pubsub import publish


    publish(f"pioreactor/{unit}/{experiment}/more/topics", payload)

How do I subscribe to data on MQTT?​

  1. If you're writing code for Job, you can simply use Job's methods:

        def callback(self, msg):
            ...

        def some_method(self, ...):
            ...

            self.subscribe_and_callback(self.callback, f"pioreactor/{self.unit}/{self.experiment}/{self.job_name}/more/topics")

  2. If writing a script:

    from pioreactor.pubsub import subscribe, subscribe_and_callback