Add global alert bar, dynamic webapp version warning

Correct spelling error

Add initial form of global alert workflow

Author: SteveMicroNova

amplipi/app.py

@@ -635,6 +635,12 @@ def play_media(media: models.PlayMedia, ctrl: Api = Depends(get_ctrl)) -> models
     raise HTTPException(404, f'source id not found')
   return code_response(ctrl, ctrl.play_media(media))
 
+
+@api.patch("/api/alert/hide", response_model=List[models.Alert])
+def hide_alert(alert: models.Alert):
+  """Hide an Alert based on the Alert's message"""
+  return utils.hide_alert(message=alert.message)
+
 # Info
 
 

amplipi/ctrl.py

@@ -252,7 +252,8 @@ def reinit(self, settings: models.AppSettings = models.AppSettings(), change_not
       version=utils.detect_version(),
       stream_types_available=amplipi.streams.stream_types_available(),
       extra_fields=utils.load_extra_fields(),
-      serial=str(self._serial)
+      serial=str(self._serial),
+      global_alerts=utils.load_alerts()
     )
     for major, minor, ghash, dirty in self._rt.read_versions():
       fw_info = models.FirmwareInfo(version=f'{major}.{minor}', git_hash=f'{ghash:x}', git_dirty=dirty)
@@ -553,6 +554,7 @@ def _update_sys_info(self, throttled=True) -> None:
     self.status.info.connected_drives = self._connected_drives_cache.get(throttled)
     self.status.info.latest_release = self._latest_release_cache.get(throttled)
     self.status.info.access_key = auth.get_access_key("admin") if auth.user_access_key_set("admin") else ""
+    self.status.info.global_alerts = utils.load_alerts()
 
   def sync_stream_info(self) -> None:
     """Synchronize the stream list to the stream status"""

amplipi/models.py

@@ -25,6 +25,7 @@
 from types import SimpleNamespace
 from enum import Enum
 from pathlib import Path
+import datetime
 
 # pylint: disable=no-name-in-module
 from pydantic import BaseSettings, BaseModel, Field
@@ -1013,6 +1014,43 @@ class FirmwareInfo(BaseModel):
   git_dirty: bool = Field(default=False, description="True if local changes were made. Used for development.")
 
 
+class AlertLevel(Enum):
+  """What color should the alert be as per the Mui style guide: https://mui.com/material-ui/react-alert/#severity"""
+  WARNING = "warning"
+  ERROR = "error"
+  INFO = "info"
+  SUCCESS = "success"
+
+
+class Alert(BaseModel):
+  message: str
+  severity: AlertLevel = AlertLevel.ERROR
+  """What color should the alert be as per the Mui style guide: https://mui.com/material-ui/react-alert/#severity"""
+  hidden: bool = False
+  """Has this Alert been hidden by the user?"""
+  timestamp: datetime.datetime = Field(
+    default_factory=lambda: datetime.datetime.now(datetime.timezone.utc)
+  )
+
+  @property
+  def expired(self) -> bool:  # Used to limit alerts to have only a single instance per week. If the state that caused the alert is still valid after a week, the same alert will be made.
+    return (datetime.datetime.now(datetime.timezone.utc) - self.timestamp) > datetime.timedelta(weeks=1)
+
+  class Config:
+    schema_extra = {
+      'examples': {
+        'Example Alert': {
+          'value': {
+            "message": "Writing data to the I2C bus has failed multiple times, please contact AmpliPi Support at mailto:support@micro-nova.com",
+            "severity": "error",
+            "hidden": False,
+            "timestamp": "2026-05-26T19:28:57.907099+00:00"
+          }
+        },
+      }
+    }
+
+
 class Info(BaseModel):
   """ AmpliPi System information """
   version: str = Field(description="software version")
@@ -1033,6 +1071,7 @@ class Info(BaseModel):
     default=[], description='The stream types available on this particular appliance')
   extra_fields: Optional[Dict] = Field(default=None, description='Optional fields for customization')
   connected_drives: List[str] = Field(default=[], description='A list of all external drives connected')
+  global_alerts: List[Alert] = Field(default=[], description='A list of alerts to be shown to all users via the frontend global alert bar')
 
   class Config:
     schema_extra = {

amplipi/rt.py

@@ -30,7 +30,7 @@
 
 from smbus2 import SMBus
 from serial import Serial
-from amplipi import models  # TODO: importing this takes ~0.5s, reduce
+from amplipi import models, utils  # TODO: importing this takes ~0.5s, reduce
 
 # TODO: move constants like this to their own file
 DEBUG_PREAMPS = False  # print out preamp state after register write
@@ -242,6 +242,8 @@ def new_preamp(self, addr: int):
       0x4F,
     ]
 
+  write_byte_data_failures: int = 0
+
   def write_byte_data(self, preamp_addr, reg, data):
     assert preamp_addr in _DEV_ADDRS
     assert type(preamp_addr) == int
@@ -262,8 +264,13 @@ def write_byte_data(self, preamp_addr, reg, data):
       try:
         time.sleep(0.001)  # space out sequential calls to avoid bus errors
         self.bus.write_byte_data(preamp_addr, reg, data)
-      except Exception:
+      except Exception as e:
+        logger.exception(f"Writing preamp failed: {e}")
         time.sleep(0.001)
+        self.bus.close()
+        self.write_byte_data_failures += 1
+        if self.write_byte_data_failures >= 3:
+          utils.add_alert("Writing data to the I2C bus has failed multiple times, please contact AmpliPi Support at support@micro-nova.com")
         self.bus = SMBus(1)
         self.bus.write_byte_data(preamp_addr, reg, data)
 

amplipi/utils.py

@@ -495,3 +495,80 @@ def clear_custom_configs():
       os.remove(path)
     except Exception as e:
       logger.exception(f"failed to clear device configuration: {e}")
+
+
+# Every alert(s) function was in ctrl.py, but due to many files needing access to the add_alert flow they had to be here in utils
+
+
+def load_alerts() -> List[models.Alert]:
+  alert_file = f"{get_folder('config')}/alerts.json"
+  try:
+    with open(alert_file, 'r', encoding='utf-8') as file:
+      data = json.load(file)
+
+    alerts: List[models.Alert] = [models.Alert(**item) for item in data]
+    for alert in alerts:
+      if alert.expired:
+        alert.hidden = True  # Frontend can't see expired property, so autohide any expired alerts as to not have to close the same alert twice
+    return alerts
+
+  except (FileNotFoundError, json.JSONDecodeError):
+    return []
+
+  except Exception as e:
+    logger.exception(e)
+    return []
+
+
+def select_alert(message: str, alerts: Optional[List[models.Alert]] = None) -> Optional[models.Alert]:
+  """
+  Selects the most recent non-expired instance of a specific alert message. Takes two arguments:
+    message: the string that makes up the Alert's message
+
+    alerts: An optional list of alerts, for use when you want the returned alert to be a pointer to the same alert in that instance of the list.
+      Generally useful when mutating an alert before saving the full list.
+  """
+  if alerts is None:
+    alerts = load_alerts()
+  return next(
+    (
+      item for item in alerts
+      if item.message == message and not item.expired
+    ),
+    None
+  )
+
+
+def add_alert(message: str, severity: models.AlertLevel = models.AlertLevel.ERROR) -> List[models.Alert]:
+  alerts = load_alerts()
+  search = select_alert(message)
+  if search is None:
+    alert = models.Alert(message=message, severity=severity)
+    alerts.append(alert)
+    save_alerts(alerts)
+  return alerts  # Only returns anything to make the unit test for the hide endpoint easier
+
+
+def hide_alert(message: str) -> List[models.Alert]:
+  alerts = load_alerts()
+  selected_alert = select_alert(message, alerts)
+  if selected_alert is not None:
+    selected_alert.hidden = True
+    save_alerts(alerts)
+  else:
+    add_alert("Alert not found, could not be hidden!")
+    logger.exception("Alert not found, could not be hidden!")
+  return alerts
+
+
+def save_alerts(alerts: List[models.Alert]):
+  alert_file = f"{get_folder('config')}/alerts.json"
+  try:
+    with open(alert_file, 'w', encoding='utf-8') as file:
+      json.dump(
+        [json.loads(alert.json()) for alert in alerts],
+        file,
+        indent=2
+      )
+  except Exception as e:
+    logger.exception(e)