Merge pull request #3 from eseglem/develop

Overhaul for async and ssh / telnet.

.editorconfig

@@ -0,0 +1,8 @@
+[*]
+end_of_line = lf
+insert_final_newline = true
+
+[*.py]
+charset = utf-8
+indent_style = space
+indent_size = 4

.github/workflows/release.yml

@@ -0,0 +1,45 @@
+name: "Release"
+
+on:
+  release:
+    types:
+      - "published"
+
+permissions: {}
+
+jobs:
+  release:
+    name: "Release"
+    runs-on: "ubuntu-latest"
+    environment: release
+    permissions:
+      contents: write
+      id-token: write
+    steps:
+      - name: "Checkout the repository"
+        uses: "actions/checkout@v3.5.2"
+
+      - name: Install poetry
+        run: pipx install poetry
+
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.11
+          cache: "poetry"
+
+      - name: Install dependencies
+        run: poetry install --no-dev
+
+      - name: Build package
+        run: poetry build
+
+      - name: Publish package
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          password: ${{ secrets.PYPI_API_TOKEN }}
+
+      - name: "Upload the files to the release"
+        uses: softprops/action-gh-release@v0.1.15
+        with:
+          files: ${{ github.workspace }}/dist/*

.pre-commit-config.yaml

@@ -0,0 +1,42 @@
+ci:
+  autoupdate_schedule: monthly
+
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.5.0
+    hooks:
+      - id: check-added-large-files
+      - id: check-toml
+      - id: check-yaml
+        args: ["--unsafe"]
+      - id: debug-statements
+      - id: end-of-file-fixer
+      - id: mixed-line-ending
+        args: ["--fix=lf"]
+      - id: trailing-whitespace
+  - repo: https://github.com/asottile/pyupgrade
+    rev: v3.15.0
+    hooks:
+      - id: pyupgrade
+        args: ["--py38-plus", "--keep-runtime-typing"]
+  - repo: https://github.com/charliermarsh/ruff-pre-commit
+    rev: v0.1.6
+    hooks:
+      - id: ruff
+        args: ["--fix"]
+  - repo: https://github.com/psf/black
+    rev: 23.11.0
+    hooks:
+      - id: black
+        language_version: python
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.7.0
+    hooks:
+      - id: mypy
+        language_version: python
+        args: [--config-file=pyproject.toml, pywattbox/]
+        pass_filenames: false
+        additional_dependencies:
+          - scrapli
+          - httpx
+          - types-beautifulsoup4

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2020 Erik Seglem
+Copyright (c) 2023 Erik Seglem
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -1,5 +1,19 @@
-# pywattbox
-
-The documentation for the API is available at: https://www.snapav.com/wcsstore/ExtendedSitesCatalogAssetStore/attachments/documents/PowerManagement/SupportDocuments/Wattbox%20API%20v2.0.pdf
-
-I suggest not using the admin account to run this code. Log into your WattBox and create a new user account.
+# pywattbox
+
+For usage see: [eseglem/hass-wattbox][hass_wattbox]
+
+Python wrapper for [WattBox][wattbox]
+
+The documentation for the HTTP API was found [Here][http_api]
+The documentation for Telnet / SSH API was found [Here][ssh_api]
+
+I suggest not using the admin account to run this code. Log into your WattBox and create a new user account.
+
+<!---->
+
+***
+
+[wattbox]: https://www.snapav.com/shop/en/snapav/wattbox
+[hass_wattbox]: https://github.com/eseglem/hass-wattbox
+[http_api]: https://www.snapav.com/wcsstore/ExtendedSitesCatalogAssetStore/attachments/documents/PowerManagement/SupportDocuments/Wattbox%20API%20v2.0.pdf
+[ssh_api]: https://www.snapav.com/wcsstore/ExtendedSitesCatalogAssetStore/attachments/documents/PowerManagement/ProtocolsAndDrivers/SnapAV_Wattbox_API_V2.4.pdf

pyproject.toml

@@ -0,0 +1,84 @@
+[tool.poetry]
+name = "pywattbox"
+version = "0.7.0"
+description = "A python wrapper for WattBox APIs."
+license = "MIT"
+readme = "README.md"
+authors = ["Erik Seglem <erik.seglem@gmail.com>"]
+repository = "https://github.com/eseglem/pywattbox"
+classifiers = [
+    "Development Status :: 4 - Beta",
+    "Programming Language :: Python",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3 :: Only",
+    "License :: OSI Approved :: MIT License",
+    "Operating System :: OS Independent",
+    "Intended Audience :: Developers",
+    "Topic :: Software Development :: Libraries :: Python Modules",
+    "Topic :: Home Automation",
+]
+
+[tool.poetry.dependencies]
+python = "^3.8"
+httpx = { version = ">=0.23.0", optional = true }
+beautifulsoup4 = { version = ">=4.11.0", optional = true }
+lxml = { version = ">=4.9.0", optional = true }
+h11 = { version = ">=0.14.0", optional = true }
+scrapli = { version = ">=2022.7.30", optional = true }
+ssh2-python = { version = ">=1.0.0", optional = true }
+asyncssh = { version = ">=2.12.0", optional = true }
+
+[tool.poetry.extras]
+ip = ["scrapli", "ssh2-python", "asyncssh"]
+http = ["httpx", "beautifulsoup4", "lxml", "h11"]
+
+[tool.poetry.group.dev.dependencies]
+black = ">=23.1.0"
+ruff = ">=0.0.254"
+mypy = ">=1.1.0"
+pre-commit = ">=3.1.1"
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.black]
+target-version = ["py38"]
+
+[tool.isort]
+profile = "black"
+
+[tool.ruff]
+target-version = "py38"
+select = [
+    "F",   # pyflakes
+    "E",   # pycodestyle errors
+    "W",   # pycodestyle warnings
+    "I",   # isort
+    "YTT", # flake8-2020
+    "C4",  # flake8-comprehensions
+    "B",   # flake8-bugbear
+]
+fix = true
+ignore = [
+    "E501", # Line too long. Handled by black.
+
+]
+
+[tool.ruff.isort]
+combine-as-imports = true
+
+[tool.mypy]
+python_version = 3.8
+show_error_codes = true
+check_untyped_defs = true
+no_implicit_reexport = true
+disallow_incomplete_defs = true
+disallow_untyped_calls = true
+disallow_untyped_defs = true
+warn_unused_configs = true
+warn_unused_ignores = true
+warn_redundant_casts = true

pywattbox/__init__.py

@@ -1 +0,0 @@
-from .wattbox import Commands, Outlet, WattBox

pywattbox/base.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import logging
+from abc import ABC, abstractmethod
+from enum import IntEnum
+from typing import Dict, Optional, Type, TypeVar
+
+logger = logging.getLogger("pywattbox")
+
+
+class Commands(IntEnum):
+    """Commands Enum for Convenience.
+
+    HTTP API uses the values.
+    Integration Protocol uses the names.
+    """
+
+    OFF = 0
+    ON = 1
+    RESET = 3
+    # Only used for HTTP
+    AUTO_REBOOT_ON = 4
+    AUTO_REBOOT_OFF = 5
+    # Only used for Integration Protocol
+    TOGGLE = 6
+
+
+class BaseWattBox(ABC):
+    """Base WattBox that defines the"""
+
+    def __init__(self, host: str, user: str, password: str, port: int) -> None:
+        self.host: str = host
+        self.port: Optional[int] = port
+        self.user: str = user
+        self.password: str = password
+
+        # Info, set once
+        self.hardware_version: Optional[str] = None
+        self.firmware_version: Optional[str] = None
+        self.has_ups: bool = False
+        self.hostname: str = ""
+        self.number_outlets: int = 0
+        self.serial_number: str = ""
+
+        # Status values
+        self.audible_alarm: bool = False
+        self.auto_reboot: bool = False
+        self.cloud_status: Optional[bool] = False
+        self.mute: bool = False
+        self.power_lost: bool = False
+
+        # Power values
+        self.current_value: float = 0.0  # In Amps
+        self.power_value: float = 0.0  # In watts
+        self.safe_voltage_status: bool = True
+        self.voltage_value: float = 0.0  # In volts
+
+        # Battery values
+        self.battery_charge: int = 0  # In percent
+        self.battery_health: bool = False
+        self.battery_load: int = 0  # In percent
+        self.battery_test: Optional[bool] = False
+        self.est_run_time: int = 0  # In minutes
+
+        # Outlets list
+        self.outlets: Dict[int, Outlet] = {}
+        self.master_outlet: Optional[Outlet] = None
+
+    @abstractmethod
+    def get_initial(self) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def async_get_initial(self) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def update(self) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def async_update(self) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def send_command(self, outlet: int, command: Commands) -> None:
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def async_send_command(self, outlet: int, command: Commands) -> None:
+        raise NotImplementedError()
+
+
+_T_WattBox = TypeVar("_T_WattBox", bound=BaseWattBox)
+
+
+def _create_wattbox(
+    type_: Type[_T_WattBox], host: str, user: str, password: str, port: int
+) -> _T_WattBox:
+    wattbox = type_(host=host, user=user, password=password, port=port)
+    wattbox.get_initial()
+    wattbox.update()
+    return wattbox
+
+
+async def _async_create_wattbox(
+    type_: Type[_T_WattBox], host: str, user: str, password: str, port: int
+) -> _T_WattBox:
+    wattbox = type_(host=host, user=user, password=password, port=port)
+    await wattbox.async_get_initial()
+    await wattbox.async_update()
+    return wattbox
+
+
+class Outlet:
+    def __init__(self, index: int, wattbox: BaseWattBox) -> None:
+        self.index: int = index
+        self.method: Optional[bool] = None
+        self.name: Optional[str] = ""
+        self.status: Optional[bool] = None
+        # Power values
+        self.current_value: Optional[float] = None  # In Amps
+        self.power_value: Optional[float] = None  # In watts
+        self.voltage_value: Optional[float] = None  # In volts
+        # The WattBox
+        self.wattbox: BaseWattBox = wattbox
+
+    def turn_on(self) -> None:
+        self.wattbox.send_command(self.index, Commands.ON)
+
+    async def async_turn_on(self) -> None:
+        await self.wattbox.async_send_command(self.index, Commands.ON)
+
+    def turn_off(self) -> None:
+        self.wattbox.send_command(self.index, Commands.OFF)
+
+    async def async_turn_off(self) -> None:
+        await self.wattbox.async_send_command(self.index, Commands.OFF)
+
+    def reset(self) -> None:
+        self.wattbox.send_command(self.index, Commands.RESET)
+
+    async def async_reset(self) -> None:
+        await self.wattbox.async_send_command(self.index, Commands.RESET)
+
+    def __str__(self) -> str:
+        return f"{self.name} ({self.index}): {self.status}"

pywattbox/driver/__init__.py

@@ -0,0 +1,10 @@
+from typing import Final
+
+PROMPTS: Final[str] = (
+    r"^"  # Start of the string
+    r"(.*Successfully Logged In!)|"  # After Login
+    r"(\?\w+=\S+)|"  # Response to `?` request message
+    r"(OK)|"  # Response to `!` control message
+    r"(#Error)"  # Error Message
+    r"\n$"  # Newline / End of String
+)