new mqtt config

Author: CamDavidsonPilon

CHANGELOG.md

@@ -1,6 +1,13 @@
 ### Upcoming
  - fixed `datum` bug in the Overview that was crashing the UI. Sorry!
  - introducing some outlier protection in growth rate calculations. This is tunable with the `ekf_outlier_std_threshold` value under `[growth_rate_calculating.config]`. The default value, 5.0, is pretty conservative, and will still allow some spikes to permeate to the growth rate and nOD. Try a value of 3.5 or 4.0 if you want to see more. Don't put it less than 3.0 - that's silly.
+ - added options alternative mqtt brokers with the new configuration
+ ```
+[mqtt]
+mqtt_username=pioreactor
+mqtt_password=raspberry
+broker_address=
+ ```
 
 
 ### 24.2.11

config.dev.ini

@@ -111,6 +111,13 @@ send_errors_to_Pioreactor=1
 send_od_statistics_to_Pioreactor=1
 
 
+[mqtt]
+username=pioreactor
+password=raspberry
+broker_address=localhost
+broker_ws_port=9001
+broker_port=1883
+ws_protocol=ws
 
 [growth_rate_kalman]
 acc_std=0.0008

migration_scripts/import_data.sh

@@ -16,6 +16,9 @@ ARCHIVE_HOSTNAME=$(echo "$ARCHIVE_NAME" | cut -d'_' -f 2)
 # Get the current hostname of the system
 CURRENT_HOSTNAME=$(hostname)
 
+PIO_DIR=/home/pioreactor/.pioreactor
+
+
 # the hostname of this system and the archive file should be the same. Exit if not.
 if [ "$ARCHIVE_HOSTNAME" != "$CURRENT_HOSTNAME" ]; then
   echo "Error: Hostname of the archive does not match this hostname."
@@ -30,32 +33,32 @@ sudo systemctl stop lighttpd.service || true
 sudo systemctl stop huey.service || true
 
 # blow away the old .pioreactor
-rm -rf /home/pioreactor/.pioreactor/
+rm -rf $PIO_DIR/
 
 # create the new .pioreactor/
 tar -xzf $ARCHIVE_NAME
 
 
-leader_hostname=$(crudini --get /home/pioreactor/.pioreactor/config.ini cluster.topology leader_hostname)
+leader_hostname=$(crudini --get $PIO_DIR/config.ini cluster.topology leader_hostname)
 
 if [ "$leader_hostname" = "$CURRENT_HOSTNAME" ]; then
   # rename the sqlite .backup, if leader
-  mv /home/pioreactor/.pioreactor/storage/pioreactor.sqlite.backup /home/pioreactor/.pioreactor/storage/pioreactor.sqlite
-  touch /home/pioreactor/.pioreactor/storage/pioreactor.sqlite-shm
-  touch /home/pioreactor/.pioreactor/storage/pioreactor.sqlite-wal
+  mv $PIO_DIR/storage/pioreactor.sqlite.backup $PIO_DIR/storage/pioreactor.sqlite
+  touch $PIO_DIR/storage/pioreactor.sqlite-shm
+  touch $PIO_DIR/storage/pioreactor.sqlite-wal
 
   # check integrity, quickly
-  DB_CHECK=$(sqlite3 /home/pioreactor/.pioreactor/storage/pioreactor.sqlite "PRAGMA quick_check;")
+  DB_CHECK=$(sqlite3 $PIO_DIR/storage/pioreactor.sqlite "PRAGMA quick_check;")
   if [[ "$DB_CHECK" != "ok" ]]; then
       echo "Database integrity check failed: $DB_CHECK"
   fi
 
 fi
 
 # confirm permissions
-chmod -R 770 /home/pioreactor/.pioreactor/storage/
-chown -R pioreactor:www-data /home/pioreactor/.pioreactor/storage/
-chmod g+s /home/pioreactor/.pioreactor/storage/
+chmod -R 770 $PIO_DIR/storage/
+chown -R pioreactor:www-data $PIO_DIR/storage/
+chmod g+s $PIO_DIR/storage/
 
 echo "Done! Rebooting..."
 

pioreactor/actions/od_calibration.py

@@ -39,6 +39,7 @@
 from pioreactor.whoami import is_testing_env
 
 
+
 def green(string):
     return style(string, fg="green")
 
@@ -75,6 +76,7 @@ def get_name_from_user() -> str:
                 default=f"od-cal-{current_utc_datestamp()}",
                 show_default=False,
             ).strip()
+
             if name == "":
                 echo("Name cannot be empty")
                 continue
@@ -88,7 +90,7 @@ def get_name_from_user() -> str:
                 return name
 
 
-def get_metadata_from_user():
+def get_metadata_from_user() -> tuple[pt.OD600, pt.OD600, pt.mL, pt.PdAngle, pt.PdChannel]:
     initial_od600 = prompt(
         green("Provide the OD600 measurement of your initial, high density, culture"),
         type=click.FloatRange(min=0.01, clamp=False),
@@ -105,6 +107,7 @@ def get_metadata_from_user():
             type=click.FloatRange(min=0, max=initial_od600, clamp=False),
         )
 
+    assert isinstance(minimum_od600, )
     if minimum_od600 == 0:
         minimum_od600 = 0.01
 
@@ -128,6 +131,7 @@ def get_metadata_from_user():
 
     ref_channel = config["od_config.photodiode_channel_reverse"]["REF"]
     signal_channel = "1" if ref_channel == "2" else "2"
+    #assert isinstance(signal_channel, pt.PdChannel)
 
     confirm(
         green(
@@ -136,7 +140,8 @@ def get_metadata_from_user():
         abort=True,
         default=True,
     )
-    angle = str(config["od_config.photodiode_channel"][signal_channel])
+    angle = config["od_config.photodiode_channel"][signal_channel]
+    #assert isinstance(angle, pt.PdAngle)
     return initial_od600, minimum_od600, dilution_amount, angle, signal_channel
 
 

pioreactor/background_jobs/base.py

@@ -16,7 +16,6 @@
 from pioreactor import structs
 from pioreactor import types as pt
 from pioreactor.config import config
-from pioreactor.config import leader_address
 from pioreactor.config import leader_hostname
 from pioreactor.logging import create_logger
 from pioreactor.pubsub import Client
@@ -269,7 +268,6 @@ def __init__(self, unit: str, experiment: str, source: str = "app") -> None:
             unit=self.unit,
             experiment=self.experiment,
             source=self._source,
-            mqtt_hostname=leader_address,
         )
 
         self._check_for_duplicate_activity()
@@ -576,7 +574,6 @@ def _check_published_settings(published_settings: dict[str, pt.PublishableSettin
     def _create_pub_client(self) -> Client:
         # see note above as to why we split pub and sub.
         client = create_client(
-            hostname=leader_address,
             client_id=f"{self.job_name}-pub-{self.unit}-{self.experiment}",
             keepalive=15 * 60,
         )
@@ -609,7 +606,6 @@ def on_disconnect(client, userdata, rc: int) -> None:
         }
 
         client = create_client(
-            hostname=leader_address,
             client_id=f"{self.job_name}-sub-{self.unit}-{self.experiment}",
             last_will=last_will,
             keepalive=60,

pioreactor/background_jobs/monitor.py

@@ -21,7 +21,7 @@
 from pioreactor.background_jobs.base import BackgroundJob
 from pioreactor.config import config
 from pioreactor.config import get_config
-from pioreactor.config import leader_address
+from pioreactor.config import mqtt_address
 from pioreactor.config import leader_hostname
 from pioreactor.hardware import GPIOCHIP
 from pioreactor.hardware import is_HAT_present
@@ -397,7 +397,7 @@ def check_for_mqtt_connection_to_leader(self) -> None:
 
             self.logger.warning(
                 f"""Not able to connect MQTT clients to leader.
-1. Is the leader, {leader_hostname} at {leader_address}, in config.ini correct?
+1. Is the {mqtt_address=}, in config.ini correct?
 2. Is the Pioreactor leader online and responsive?
 """
             )  # remember, this doesn't get published to leader...

pioreactor/config.py

@@ -15,6 +15,10 @@ def __getattr__(attr):  # type: ignore
         return get_leader_hostname()
     elif attr == "leader_address":
         return get_leader_address()
+    elif attr == "mqtt_address":
+        return get_mqtt_address()
+    elif attr == "mqtt_port":
+        return get_mqtt_port()
     else:
         raise AttributeError
 
@@ -172,6 +176,14 @@ def get_leader_hostname() -> str:
 def get_leader_address() -> str:
     return get_config().get("cluster.topology", "leader_address", fallback="localhost")
 
+@cache
+def get_mqtt_address() -> str:
+    return get_config().get("mqtt", "broker_address", fallback=get_leader_address())
+
+@cache
+def get_mqtt_port() -> str:
+    return get_config().getint("mqtt", "broker_port", fallback=1883)
+
 
 def check_firstboot_successful() -> bool:
     from pioreactor.whoami import is_testing_env

pioreactor/logging.py

@@ -130,7 +130,6 @@ def create_logger(
     experiment: Optional[str] = None,
     source: str = "app",
     to_mqtt: bool = True,
-    mqtt_hostname: Optional[str] = None,
 ) -> CustomLogger:
     """
 
@@ -195,7 +194,6 @@ def create_logger(
 
     if to_mqtt:
         pub_client = create_client(
-            hostname=mqtt_hostname,
             client_id=f"{name}-logging-{unit}-{experiment}",
             max_connection_attempts=2,
             keepalive=15 * 60,

pioreactor/pubsub.py

@@ -12,7 +12,9 @@
 
 from paho.mqtt.client import Client as PahoClient
 
-from pioreactor.config import leader_address
+from pioreactor.config import mqtt_address
+from pioreactor.config import mqtt_port
+from pioreactor.config import config
 from pioreactor.types import MQTTMessage
 
 
@@ -70,7 +72,7 @@ class QOS:
 
 
 def create_client(
-    hostname: Optional[str] = None,
+    hostname: str = mqtt_address,
     last_will: Optional[dict] = None,
     client_id: str = "",
     keepalive=60,
@@ -80,6 +82,7 @@ def create_client(
     on_disconnect: Optional[Callable] = None,
     on_message: Optional[Callable] = None,
     userdata: Optional[dict] = None,
+    port: int = mqtt_port
 ):
     """
     Create a MQTT client and connect to a host.
@@ -98,7 +101,10 @@ def default_on_connect(client: Client, userdata, flags, rc: int, properties=None
         clean_session=clean_session,
         userdata=userdata,
     )
-    client.username_pw_set("pioreactor", "raspberry")
+    client.username_pw_set(
+            config.get("mqtt", "username", fallback="pioreactor"),
+            config.get("mqtt", "password", fallback="raspberry")
+    )
 
     if on_connect:
         client.on_connect = on_connect  # type: ignore
@@ -114,12 +120,9 @@ def default_on_connect(client: Client, userdata, flags, rc: int, properties=None
     if last_will is not None:
         client.will_set(**last_will)
 
-    if hostname is None:
-        hostname = leader_address
-
     for retries in range(1, max_connection_attempts + 1):
         try:
-            client.connect(hostname, keepalive=keepalive)
+            client.connect(hostname, port, keepalive=keepalive)
         except (socket.gaierror, OSError):
             if retries == max_connection_attempts:
                 break
@@ -131,7 +134,7 @@ def default_on_connect(client: Client, userdata, flags, rc: int, properties=None
     return client
 
 
-def publish(topic: str, message, hostname: str = leader_address, retries: int = 10, **mqtt_kwargs) -> None:
+def publish(topic: str, message, hostname: str = mqtt_address, retries: int = 10, port: int = mqtt_port, **mqtt_kwargs) -> None:
     from paho.mqtt import publish as mqtt_publish
     import socket
 
@@ -141,6 +144,7 @@ def publish(topic: str, message, hostname: str = leader_address, retries: int =
                 topic,
                 payload=message,
                 hostname=hostname,
+                port=port,
                 auth={"username": "pioreactor", "password": "raspberry"},
                 **mqtt_kwargs,
             )
@@ -164,11 +168,12 @@ def publish(topic: str, message, hostname: str = leader_address, retries: int =
 
 def subscribe(
     topics: str | list[str],
-    hostname: str = leader_address,
+    hostname: str = mqtt_address,
     retries: int = 5,
     timeout: Optional[float] = None,
     allow_retained: bool = True,
     name: Optional[str] = None,
+    port: int = mqtt_port,
     **mqtt_kwargs,
 ) -> Optional[MQTTMessage]:
     """
@@ -222,7 +227,7 @@ def on_message(client: Client, userdata, message: MQTTMessage) -> None:
             client.username_pw_set("pioreactor", "raspberry")
             client.on_connect = on_connect  # type: ignore
             client.on_message = on_message  # type: ignore
-            client.connect(hostname)
+            client.connect(hostname, port=port)
 
             if timeout is None:
                 client.loop_forever()
@@ -255,11 +260,12 @@ def on_message(client: Client, userdata, message: MQTTMessage) -> None:
 def subscribe_and_callback(
     callback: Callable[[MQTTMessage], Any],
     topics: str | list[str],
-    hostname: str = leader_address,
+    hostname: str = mqtt_address,
     last_will: Optional[dict] = None,
     name: Optional[str] = None,
     allow_retained: bool = True,
     client: Optional[Client] = None,
+    port: int = mqtt_port,
     **mqtt_kwargs,
 ) -> Client:
     """
@@ -313,6 +319,7 @@ def on_connect(client: Client, userdata: dict, *args):
             on_connect=on_connect,
             on_message=wrap_callback(callback),
             userdata=userdata,
+            port=port,
             **mqtt_kwargs,
         )
 
@@ -325,7 +332,7 @@ def on_connect(client: Client, userdata: dict, *args):
     return client
 
 
-def prune_retained_messages(topics_to_prune: str = "#", hostname=leader_address):
+def prune_retained_messages(topics_to_prune: str = "#", hostname=mqtt_address):
     topics = []
 
     def on_message(message):

pioreactor/types.py

@@ -100,6 +100,7 @@ class PublishableSetting(t.TypedDict, total=False):
 AnalogValue = t.Union[int, float]
 Voltage = float  # maybe should be non-negative?
 OD = t.Annotated[float, Meta(ge=0)]
+OD600 = t.Annotated[float, Meta(ge=0)]
 
 AdcChannel = t.Literal[0, 1, 2, 3]
 

update_scripts/upcoming/update.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+set -x
+set -e
+
+export LC_ALL=C
+
+PIO_DIR=/home/pioreactor/.pioreactor
+
+
+# update MQTT configuration
+LEADER_ADDRESS=$(crudini --get $PIO_DIR/config.ini cluster.inventory leader_address)
+
+crudini --ini-options=nospace --set  $PIO_DIR/config.ini mqtt username pioreactor \
+                              --set  $PIO_DIR/config.ini mqtt password raspberry \
+                              --set  $PIO_DIR/config.ini mqtt broker_address "$LEADER_ADDRESS" \
+                              --set  $PIO_DIR/config.ini mqtt broker_port 9001