feat: define specific exception classes for command rejections and state errors

openevsehttp/__init__.py

@@ -16,11 +16,15 @@
 from .exceptions import (
     AlreadyListening,
     AuthenticationError,
+    CommandFailedError,
+    FirmwareResolutionError,
     InvalidType,
     MissingMethod,
     MissingSerial,
+    OpenEVSEError,
     ParseJSONError,
     UnknownError,
+    UnknownStateError,
     UnsupportedFeature,
 )
 from .websocket import (
@@ -43,15 +47,19 @@
     "UPDATE_TRIGGERS",
     "AlreadyListening",
     "AuthenticationError",
+    "CommandFailedError",
     "ContentTypeError",
+    "FirmwareResolutionError",
     "InvalidType",
     "MissingMethod",
     "MissingSerial",
     "OpenEVSE",
+    "OpenEVSEError",
     "OpenEVSEWebsocket",
     "ParseJSONError",
     "ServerTimeoutError",
     "UnknownError",
+    "UnknownStateError",
     "UnsupportedFeature",
     "divert_mode",
     "states",

openevsehttp/commands.py

@@ -13,7 +13,12 @@
 from awesomeversion.exceptions import AwesomeVersionCompareException
 
 from .const import MAX_AMPS, MIN_AMPS, RAPI_ERRORS, SUCCESS_ANSWERS, divert_mode
-from .exceptions import UnknownError, UnsupportedFeature
+from .exceptions import (
+    CommandFailedError,
+    FirmwareResolutionError,
+    UnknownStateError,
+    UnsupportedFeature,
+)
 from .utils import get_awesome_version
 
 _LOGGER = logging.getLogger(__name__)
@@ -90,12 +95,12 @@ async def set_charge_mode(self, mode: str = "fast") -> None:
         msg = response.get("msg") if isinstance(response, Mapping) else None
         if msg not in SUCCESS_ANSWERS:
             _LOGGER.error("Problem issuing command: %s", response)
-            raise UnknownError
+            raise CommandFailedError(f"Problem issuing command: {response}")
 
     async def divert_mode(self) -> Mapping[str, Any] | list[Any]:
         """Set the divert mode to either Normal or Eco modes."""
         if not self._config:
-            raise RuntimeError("Missing configuration: self._config is required")
+            raise UnknownStateError("Missing configuration: self._config is required")
 
         if not self._version_check("2.9.1"):
             _LOGGER.debug("Feature not supported for older firmware.")
@@ -213,7 +218,7 @@ async def toggle_override(self) -> None:
                 or response.get("msg") not in SUCCESS_ANSWERS
             ):
                 _LOGGER.error("Problem toggling override: %s", response)
-                raise RuntimeError(f"Failed to toggle override: {response}")
+                raise CommandFailedError(f"Failed to toggle override: {response}")
         else:
             # Older firmware use RAPI commands
             _LOGGER.debug("Toggling manual override via RAPI")
@@ -222,7 +227,9 @@ async def toggle_override(self) -> None:
 
             if "state" not in self._status:
                 _LOGGER.error("Cannot toggle override: unknown charger state.")
-                raise RuntimeError("Cannot toggle override: unknown charger state.")
+                raise UnknownStateError(
+                    "Cannot toggle override: unknown charger state."
+                )
 
             command = "$FE" if self._status.get("state") == 254 else "$FS"
             response, msg = await self.send_command(command)
@@ -231,7 +238,7 @@ async def toggle_override(self) -> None:
                 isinstance(msg, str) and (msg.startswith("$NK") or msg in RAPI_ERRORS)
             ):
                 _LOGGER.error("Problem toggling override via RAPI: %s", msg)
-                raise RuntimeError(f"Failed to toggle override via RAPI: {msg}")
+                raise CommandFailedError(f"Failed to toggle override via RAPI: {msg}")
 
     async def clear_override(self) -> None:
         """Clear the manual override status."""
@@ -247,7 +254,7 @@ async def clear_override(self) -> None:
         _LOGGER.debug("Clear override response: %s", msg)
         if msg not in SUCCESS_ANSWERS:
             _LOGGER.error("Problem clearing override: %s", response)
-            raise RuntimeError(f"Failed to clear override: {response}")
+            raise CommandFailedError(f"Failed to clear override: {response}")
 
     async def set_current(self, amps: int = 6) -> None:
         """Set the soft current limit."""
@@ -276,7 +283,7 @@ async def set_current(self, amps: int = 6) -> None:
                 or response.get("msg") not in SUCCESS_ANSWERS
             ):
                 _LOGGER.error("Problem setting current limit: %s", response)
-                raise UnknownError
+                raise CommandFailedError(f"Problem setting current limit: {response}")
 
         else:
             # RAPI commands
@@ -291,7 +298,7 @@ async def set_current(self, amps: int = 6) -> None:
                 isinstance(msg, str) and (msg.startswith("$NK") or msg in RAPI_ERRORS)
             ):
                 _LOGGER.error("Problem setting current via RAPI: %s", msg)
-                raise UnknownError
+                raise CommandFailedError(f"Problem setting current via RAPI: {msg}")
 
     async def set_service_level(self, level: int | str = 2) -> None:
         """Set the service level of the EVSE."""
@@ -311,7 +318,7 @@ async def set_service_level(self, level: int | str = 2) -> None:
         msg = response.get("msg") if isinstance(response, Mapping) else None
         if msg not in SUCCESS_ANSWERS:
             _LOGGER.error("Problem issuing command: %s", response)
-            raise UnknownError
+            raise CommandFailedError(f"Problem issuing command: {response}")
 
     # Restart OpenEVSE WiFi
     async def restart_wifi(self) -> None:
@@ -346,7 +353,7 @@ async def restart_wifi(self) -> None:
 
         if not success:
             _LOGGER.error("Problem restarting WiFi: %s", response)
-            raise RuntimeError(f"Failed to restart WiFi: {msg}")
+            raise CommandFailedError(f"Failed to restart WiFi: {msg}")
 
     # Restart EVSE module
     async def restart_evse(self) -> None:
@@ -364,7 +371,9 @@ async def restart_evse(self) -> None:
                 or reply.get("error")
             ):
                 _LOGGER.error("Problem restarting EVSE module via HTTP: %s", reply)
-                raise RuntimeError(f"Failed to restart EVSE module via HTTP: {reply}")
+                raise CommandFailedError(
+                    f"Failed to restart EVSE module via HTTP: {reply}"
+                )
 
             response = (
                 reply.get("msg", "Unknown error")
@@ -381,7 +390,7 @@ async def restart_evse(self) -> None:
                 and (response.startswith("$NK") or response in RAPI_ERRORS)
             ):
                 _LOGGER.error("Problem restarting EVSE module via RAPI: %s", response)
-                raise RuntimeError(
+                raise CommandFailedError(
                     f"Failed to restart EVSE module via RAPI: {response}"
                 )
 
@@ -563,7 +572,7 @@ async def update_firmware(
                 _LOGGER.error(
                     "Could not resolve latest firmware download URL from GitHub."
                 )
-                raise RuntimeError(
+                raise FirmwareResolutionError(
                     "Could not resolve latest firmware download URL from GitHub."
                 )
             firmware_url = check_result["browser_download_url"]
@@ -605,7 +614,7 @@ async def set_led_brightness(self, level: int) -> None:
         msg = response.get("msg") if isinstance(response, Mapping) else None
         if msg not in SUCCESS_ANSWERS:
             _LOGGER.error("Problem issuing command: %s", response)
-            raise UnknownError
+            raise CommandFailedError(f"Problem issuing command: {response}")
 
     async def set_divert_mode(self, mode: str = "fast") -> None:
         """Set the divert mode."""
@@ -630,7 +639,7 @@ async def set_divert_mode(self, mode: str = "fast") -> None:
 
         if not success:
             _LOGGER.error("Problem issuing command: %s", response)
-            raise UnknownError
+            raise CommandFailedError(f"Problem issuing command: {response}")
 
         self._status["divertmode"] = new_mode
 
@@ -650,7 +659,7 @@ async def set_shaper(self, enable: bool = True) -> None:
         msg = response.get("msg") if isinstance(response, Mapping) else None
         if msg not in SUCCESS_ANSWERS and msg != "Current Shaper state changed":
             _LOGGER.error("Problem issuing command: %s", response)
-            raise UnknownError
+            raise CommandFailedError(f"Problem issuing command: {response}")
 
         self._status["shaper"] = mode
 
@@ -663,7 +672,7 @@ async def toggle_shaper(self) -> None:
 
         if shaper_active is None:
             _LOGGER.error("Cannot toggle shaper: unknown shaper state.")
-            raise RuntimeError("Cannot toggle shaper: unknown shaper state.")
+            raise UnknownStateError("Cannot toggle shaper: unknown shaper state.")
 
         new_state = not bool(shaper_active)
         await self.set_shaper(new_state)
@@ -686,4 +695,4 @@ async def set_mqtt_vehicle_range_miles(self, enable: bool = True) -> None:
         msg = response.get("msg") if isinstance(response, Mapping) else None
         if msg not in SUCCESS_ANSWERS:
             _LOGGER.error("Problem issuing command: %s", response)
-            raise UnknownError
+            raise CommandFailedError(f"Problem issuing command: {response}")

openevsehttp/exceptions.py

@@ -1,33 +1,49 @@
 """Exceptions."""
 
 
-class AuthenticationError(Exception):
+class OpenEVSEError(Exception):
+    """Base exception for python-openevse-http."""
+
+
+class AuthenticationError(OpenEVSEError):
     """Exception for authentication errors."""
 
 
-class ParseJSONError(Exception):
+class ParseJSONError(OpenEVSEError):
     """Exception for JSON parsing errors."""
 
 
-class UnknownError(Exception):
+class UnknownError(OpenEVSEError):
     """Exception for Unknown errors."""
 
 
-class MissingMethod(Exception):
+class MissingMethod(OpenEVSEError):
     """Exception for missing method variable."""
 
 
-class AlreadyListening(Exception):
+class AlreadyListening(OpenEVSEError):
     """Exception for already listening websocket."""
 
 
-class MissingSerial(Exception):
+class MissingSerial(OpenEVSEError):
     """Exception for missing serial number."""
 
 
-class UnsupportedFeature(Exception):
+class UnsupportedFeature(OpenEVSEError):
     """Exception for firmware that is too old."""
 
 
-class InvalidType(Exception):
+class InvalidType(OpenEVSEError):
     """Exception for invalid types."""
+
+
+class CommandFailedError(OpenEVSEError):
+    """Exception for command rejections or failures."""
+
+
+class UnknownStateError(OpenEVSEError):
+    """Exception when charger state cannot be determined."""
+
+
+class FirmwareResolutionError(OpenEVSEError):
+    """Exception when firmware download URL cannot be resolved."""

tests/test_client.py

@@ -21,6 +21,7 @@
 from openevsehttp.exceptions import (
     AlreadyListening,
     AuthenticationError,
+    CommandFailedError,
     MissingMethod,
     MissingSerial,
     ParseJSONError,
@@ -77,6 +78,39 @@ async def test_public_api_export():
     assert PublicOpenEVSE is OpenEVSE
 
 
+async def test_exceptions_hierarchy():
+    """Verify library exceptions inherit from OpenEVSEError."""
+    import openevsehttp as pkg
+
+    exception_classes = [
+        pkg.AuthenticationError,
+        pkg.ParseJSONError,
+        pkg.UnknownError,
+        pkg.MissingMethod,
+        pkg.AlreadyListening,
+        pkg.MissingSerial,
+        pkg.UnsupportedFeature,
+        pkg.InvalidType,
+        pkg.CommandFailedError,
+        pkg.UnknownStateError,
+        pkg.FirmwareResolutionError,
+    ]
+
+    for exc_cls in exception_classes:
+        assert issubclass(exc_cls, pkg.OpenEVSEError)
+        assert issubclass(exc_cls, Exception)
+
+    # Test catching with base class
+    with pytest.raises(pkg.OpenEVSEError):
+        raise pkg.CommandFailedError("Command rejected")
+
+    with pytest.raises(pkg.OpenEVSEError):
+        raise pkg.UnknownStateError("State unknown")
+
+    with pytest.raises(pkg.OpenEVSEError):
+        raise pkg.FirmwareResolutionError("Download URL resolution failed")
+
+
 async def test_get_status_auth(test_charger_auth):
     """Test authenticated status update."""
     await test_charger_auth.update()
@@ -968,7 +1002,7 @@ async def test_send_command_rapi_rejection(test_charger, mock_aioclient):
     mock_aioclient.post(TEST_URL_RAPI, status=200, body=json.dumps(value))
 
     with pytest.raises(
-        RuntimeError, match=r"Failed to toggle override via RAPI: \$NK\^21"
+        CommandFailedError, match=r"Failed to toggle override via RAPI: \$NK\^21"
     ):
         await test_charger.toggle_override()
 
@@ -977,7 +1011,8 @@ async def test_send_command_rapi_rejection(test_charger, mock_aioclient):
     mock_aioclient.post(TEST_URL_RAPI, status=200, body=json.dumps(value))
 
     with pytest.raises(
-        RuntimeError, match="Failed to toggle override via RAPI: RAPI_RESPONSE_TIMEOUT"
+        CommandFailedError,
+        match="Failed to toggle override via RAPI: RAPI_RESPONSE_TIMEOUT",
     ):
         await test_charger.toggle_override()
 
@@ -1000,7 +1035,7 @@ async def test_restart_evse_rapi_failure(test_charger, mock_aioclient, caplog):
     )
     with caplog.at_level(logging.ERROR):
         with pytest.raises(
-            RuntimeError, match="Failed to restart EVSE module via RAPI:"
+            CommandFailedError, match="Failed to restart EVSE module via RAPI:"
         ):
             await test_charger.restart_evse()
     assert "Problem restarting EVSE module via RAPI: $NK^21" in caplog.text