VersionWarning

CHANGELOG.md

@@ -5,14 +5,16 @@
   * Upgraded volume calculations to preserve relative positions when hitting the min or max setting via source volume bar
   * Update our spotify provider `go-librespot` to `0.7.3`
   * Upgrade from Logitech Media Server 8.5.2 to Lyrion Music Server 9.0.3
+  * Added in-place preamp recovery when I2C writes fail persistently
+* Web App
+  * Add warning for older versions of the webapp running on newer backends
 
 # 0.4.11
 * System
   * Update our spotify provider `go-librespot` to `0.7.1`
 * Web App
   * Changed caching rules to ensure that users don't get stuck with old versions of the webapp post update
 
-
 ## 0.4.10
 * Web App
   * Fixed internet radio search functionality

NEW_RELEASE.md

@@ -31,7 +31,7 @@ This project follows [Semantic Versioning](https://semver.org/). Here are some e
 - [ ] Update the API by running `scripts/create_spec` script.
 - [ ] Create & merge a branch/PR off `main` to bump the version in the CHANGELOG and also using `poetry version ${VERSION}`
 - [ ] Checkout main & create a detached HEAD: `git checkout main; git pull; git checkout --detach`
-- [ ] Build the webapp in `web` with `npm run build` and force add the changes with `git add -f web/dist; git commit -m "Build web app for release"`
+- [ ] CD into `web`, Run `echo "VITE_BACKEND_VERSION=$(poetry version -s)" > .env; npm run build; git add -f web/dist; git commit -m "Build web app for release"`. This will encode the version number into the frontend, build it, and add the build to git for release.
 - [ ] Tag the changes so we can make a release on GitHub: `git tag -as ${VERSION} -m '' && git push origin ${VERSION}`
 - [ ] Make a release using the GitHub interface
 - [ ] Use the AmpliPi updater to update to the release

amplipi/app.py

@@ -657,6 +657,12 @@ def debug() -> models.DebugResponse:
     logging.exception("couldn't load debug file: {e}")
     return models.DebugResponse()
 
+
+@api.patch("/api/info/alerts/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)
+
 # include all routes above
 
 

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
@@ -82,6 +82,13 @@
 
 MAX_ZONES = 6 * len(_DEV_ADDRS)
 
+_I2C_ALERT_MSG = (
+  "Writing data to the I2C bus has failed and automatic recovery was unsuccessful. "
+  "Please go to Settings -> Config -> Hardware Reset.\n"
+  "If you see this message again in a short time period, "
+  "contact AmpliPi Support at support@micro-nova.com"
+)
+
 
 class FanCtrl(Enum):
   MAX6644 = 0
@@ -145,6 +152,11 @@ class _Preamps:
 
   preamps: Dict[int, List[int]]  # Key: i2c address, Val: register values
 
+  # In-place preamp recovery — rate-limited so a benign I2C glitch
+  # never resets audio. See _recover_preamps() / write_byte_data().
+  _RECOVERY_COOLDOWN_S = 20.0
+  _last_recovery = 0.0
+
   def __init__(self, reset: bool = True, set_addr: bool = True, bootloader: bool = False, debug=True):
     self.preamps = dict()
     if not is_amplipi():
@@ -242,6 +254,47 @@ def new_preamp(self, addr: int):
       0x4F,
     ]
 
+  def _recover_preamps(self) -> bool:
+    """ Recover a wedged/hung preamp IN-PLACE.
+
+    The bare bus.write_byte_data retry in write_byte_data only reopens the
+    Linux SMBus handle — that recovers a transient bus glitch but NOT a hung
+    preamp microcontroller (which stops ACKing -> OSError 121 / EREMOTEIO).
+    The only thing that revives a hung preamp is pulsing its reset line,
+    which is exactly what a full reboot does. This does the same WITHOUT
+    rebooting: reset the preamp(s), re-assign I2C addresses, reopen the bus,
+    and re-flush every cached register so zone state (mute/source/vol)
+    survives the reset (self.preamps is the code's source of truth, updated
+    on every write).
+
+    Rate-limited to once per _RECOVERY_COOLDOWN_S so a benign one-off glitch
+    never resets audio. Returns True if a recovery was performed (caller may
+    retry the write), False if on cooldown or if the recovery itself failed —
+    in both cases an alert is surfaced to the user.
+    """
+    now = time.time()
+    if now - self._last_recovery < self._RECOVERY_COOLDOWN_S:
+      logger.warning('Preamp I2C write failed and recovery is still on cooldown — surfacing alert')
+      utils.add_alert(_I2C_ALERT_MSG)
+      return False
+
+    self._last_recovery = now
+    logger.warning('Preamp I2C wedged (EREMOTEIO) - attempting in-place recovery (reset + re-flush)')
+    try:
+      self.reset_preamps()
+      self.set_i2c_addr()
+      self.bus = SMBus(1)
+      for addr, regs in list(self.preamps.items()):
+        for reg, val in enumerate(regs):
+          time.sleep(0.001)
+          self.bus.write_byte_data(addr, reg, val)
+      logger.info('Preamp in-place recovery complete')
+      return True
+    except Exception as exc:
+      logger.error(f'Preamp in-place recovery failed: {exc}')
+      utils.add_alert(_I2C_ALERT_MSG)
+      return False
+
   def write_byte_data(self, preamp_addr, reg, data):
     assert preamp_addr in _DEV_ADDRS
     assert type(preamp_addr) == int
@@ -263,9 +316,21 @@ def write_byte_data(self, preamp_addr, reg, data):
         time.sleep(0.001)  # space out sequential calls to avoid bus errors
         self.bus.write_byte_data(preamp_addr, reg, data)
       except Exception:
-        time.sleep(0.001)
-        self.bus = SMBus(1)
-        self.bus.write_byte_data(preamp_addr, reg, data)
+        # Fallback 1: reopen the bus handle and retry (transient bus glitch).
+        try:
+          time.sleep(0.001)
+          self.bus = SMBus(1)
+          self.bus.write_byte_data(preamp_addr, reg, data)
+        except Exception:
+          # Fallback 2: a reopened fd can't revive a hung preamp MCU.
+          # Escalate to an in-place preamp reset + re-flush, then retry once
+          # more. If recovery is on cooldown or itself fails, an alert is
+          # surfaced to the user inside _recover_preamps().
+          if self._recover_preamps():
+            time.sleep(0.001)
+            self.bus.write_byte_data(preamp_addr, reg, data)
+          else:
+            raise
 
   def probe_preamp(self, addr: int):
     # Scan for preamps, and set source registers to be completely digital

amplipi/utils.py

@@ -495,3 +495,88 @@ 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]:
+  """Load the contents of the alerts file into memory as a dict. Automatically hides expired alerts before serving the list."""
+  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]:
+  """
+    Add an alert to the alerts file
+    If an existing alert has the same message as the new one, the new one will only be added if the previous same alert has expired
+    See amplipi.models.Alert for expiration flow
+  """
+  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]:
+  """Set the hidden bool of a given alert to True. Hidden alerts are not shown on the frontend."""
+  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]):
+  """Saves the given list of alerts to the alerts file at .config/amplipi/alerts.json"""
+  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)