Advanced Usage
Bring your own endpoint
The pyenphase package can be used to obtain Envoy data from endpoints not already collected. Access to these endpoint is enabled by the Authorization level set during authentication. Data is returned directly to the caller and not stored in the Envoy data model.
envoy = Envoy(host_ip_or_name)
await envoy.setup()
await envoy.authenticate(username=username, password=password, token=token)
myresponse: httpx.Response = await envoy.request('/my/own/endpoint')
status_code = response.status_code
if status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN):
# authentication error
content = myresponse.content
Register updater
The package can be extended by registering an additional updater
as a sub class of EnvoyUpdater
. Such an updater can serve as an alternative data source for existing data sources and provide requested data if other updaters don’t. The added updater can only store data in one of the existing data attributes of EnvoyData or store the raw data in Envoy’s raw
attribute.
An updater requires 2 methods. A probe
method which is used to initialize the updater and is only called once and signals capability to provide the data, and an update
method which is called repeatidly to collect the data. Each may collect the same or different data based on the needs. The updater will have to provide same data as other updaters for the data attributes in scope.
Example: Extend EnvoySystemProduction
The EnvoySystemProduction class provides overall production data reported by the Envoy. The data is sourced from various endpoints based on Envoy type and the firmware running in the Envoy. This package does not include reporting from Envoy Legacy HTML pages.[1]
Legacy Envoy SystemProduction
This example will get production data from legacy Envoy html production page and report it in the existing EnvoySystemProduction class also used for other Envoy versions. First step is to define a data model as a sub-class of EnvoySystemProduction and its method to obtain the data from the returned Envoy html. In below example the method ‘from_production_legacy’ provides this. The returned data should be the EnvoysSystemProduction class members.
from pyenphase import EnvoyData, EnvoySystemProduction, register_updater
from pyenphase.const import URL_PRODUCTION, SupportedFeatures
from pyenphase.envoy import get_updaters
from pyenphase.exceptions import ENDPOINT_PROBE_EXCEPTIONS
# regex to find production data in html page
_KEY_TO_REGEX = {
"watts_now": r"<td>Current.*</td>\s*<td>\s*(\d+|\d+\.\d+)\s*(W|kW|MW)</td>",
"watt_hours_last_7_days": r"<td>Past Week</td>\s*<td>\s*(\d+|\d+\.\d+)\s*(Wh|kWh|MWh)</td>",
"watt_hours_today": r"<td>Today</td>\s*<td>\s*(\d+|\d+\.\d+)\s*(Wh|kWh|MWh)</td>",
"watt_hours_lifetime": r"<td>Since Installation</td>\s*<td>\s*(\d+|\d+\.\d+)\s*(Wh|kWh|MWh)</td>",
}
class LegacyEnvoySystemProduction(EnvoySystemProduction):
"""Get production data from legacy Envoy html"""
def from_production_legacy(cls, text: str) -> EnvoySystemProduction:
"""Legacy parser."""
data: dict[str, int] = {
"watts_now": 0,
"watt_hours_today": 0,
"watt_hours_last_7_days": 0,
"watt_hours_lifetime": 0,
}
# extract the date from the html using regex
for key, regex in _KEY_TO_REGEX.items():
if match := re.search(regex, text, re.MULTILINE):
unit = match.group(2).lower()
value = float(match.group(1))
# scale units to w or wh
if unit.startswith("k"):
value *= 1000
elif unit.startswith("m"):
value *= 1000000
data[key] = int(value)
return cls(**data)
LegacyProductionScraper
Next define the actual updater as a subclass of EnvoyUpdater. The updater will collect the data and use above model to report the data.
class LegacyProductionScraper(EnvoyUpdater):
Probe
As described before, the probe
method is called once at initialization to detect and configure all that is needed. It is passed the bit mask (flags) of already SupportedFeatures
by other updaters. If the feature this updater provides is already provided by an other updater, ours should exit and leave it to the other updater. In this example the feature flag is SupportedFeatures.PRODUCTION
. If not set yet, the updater should configure and return SupportedFeatures.PRODUCTION
flag set to signal the Envoy class it should be used to obtain data or None if not. Returning a set SupportedFeatures flag will cause the update method to be used during data collection.
To collect the data the EnvoyUpdater class provides the methods _probe_request(endpoint)
and _json_probe_request(endpoint)
. These methods can be used retrieve text/html or json data.
async def probe(
self, discovered_features: SupportedFeatures
) -> SupportedFeatures | None:
"""Probe the Envoy for for Production HTML and return PRODUCTION SupportedFeature."""
if SupportedFeatures.PRODUCTION in discovered_features:
# Already discovered from another updater, leave alone
return None
try:
# get html data from the envoy using the probe_request
response = await self._probe_request(URL_PRODUCTION)
data = response.text
except ENDPOINT_PROBE_EXCEPTIONS:
return None
# check if response contains what we expect
if "Since Installation" not in data:
return None
# remember and return PRODUCTION as my supported feature.
self._supported_features |= SupportedFeatures.PRODUCTION
return self._supported_features
Update
The update
method is called at each update cycle to provide the actual data. It is passed the EnvoyData class to store the data to. The data colloction methods provided by the EnvoyUpdater class are _json_request(endpoint)
and _request(endpoint)
. Typically the method uses a data model to extract the data from the response.
async def update(self, envoy_data: EnvoyData) -> None:
"""Update the Envoy for this updater."""
# Get the HTML data from the Envoy
response = await self._request(URL_PRODUCTION)
production_data = response.text
# Store the data as is in the raw json of the EnvoyData
envoy_data.raw[URL_PRODUCTION] = production_data
# Store data in Envoy data using our data model.
envoy_data.system_production = (
LegacyEnvoySystemProduction.from_production_legacy(production_data)
)
Register updater
To make the updater available for use, it must be registered with the Envoy using register_updater
. Upon completion of the registration perform the usual setup, authentication and probe of the Envoy and start data collection.
# Initialize Envoy, setup and authenticate
envoy = Envoy(host)
# register our updater for legacy envoy
remove = register_updater(LegacyProductionScraper)
assert LegacyProductionScraper in get_updaters()
# setup and authenticate with Envoy
await envoy.setup()
await envoy.authenticate(username=username, password=password, token=token)
# probe what endpoints are available
await envoy.probe()
# get data, the production values now fill from html
data: EnvoyData = await envoy.update()
# remove our updater from the envoy
remove()
assert LegacyProductionScraper not in get_updaters()
Registering the updater inserts it at the end of the updaters giving priority to existing updaters to return production (in this example) data. If all prior ones fail, the newly registered one will be used. Adding a new one only makes sense for cases where the endpoint is not successfully accessed by the other ones. This is implemented by the use of the SupportedFeatures flags.
Example: New attribute EnvoyHomeInformation
The previous example Extend EnvoySystemProduction added a new data source for an existing attribute. Similarly a datasource for a new attribute can be added by registering an updater. The process is the same as the previous example with only difference being no existing EnvoyData attribute available and the EnvoyData.raw is to be used. This example will add retrieval of data from the Envoy Home endpoint /home.json.
EnvoyHomeInformation
The datamodel to use is new and designed towards the needs.
from pyenphase import EnvoyData, EnvoySystemProduction, register_updater
from pyenphase.const import URL_PRODUCTION, SupportedFeatures
from pyenphase.envoy import get_updaters
from pyenphase.exceptions import ENDPOINT_PROBE_EXCEPTIONS
@dataclass(slots=True)
class EnvoyHomeInformation():
"""Get home data from Envoy"""
software_build_epoch: int
timezone: str
@classmethod
def from_home(cls, data: dict[str, Any]):
"""Initialize from the Home API."""
return cls(
software_build_epoch=data["software_build_epoch"],
timezone=data["timezone"],
)
EnvoyHome
As described, the updater is a subclass of EnvoyUpdater and provides probe
and update
methods. As this is a new attribute no SupportedFeatures flags exists for it. The next higher flag is used to signal back this updater has data to provide. [2]
class EnvoyHome(EnvoyUpdater):
async def probe(
self, discovered_features: SupportedFeatures
) -> SupportedFeatures | None:
"""Probe the Envoy for home information."""
myflag = 1 << len(SupportedFeatures)
if myflag & discovered_features:
# Already discovered from another updater
return None
try:
home_json: dict[str, Any] = await self._json_probe_request("/home.json")
except ENDPOINT_PROBE_EXCEPTIONS:
return None
# our data not found in the page
if "software_build_epoch" not in home_json:
return None
# signal we can provide this data
self._supported_features |= myflag
return self._supported_features
async def update(self, envoy_data: EnvoyData) -> None:
"""Update the Envoy for this /home.json."""
home_data = await self._json_request("/home.json")
# No EnoyData attribute, only return raw as is
envoy_data.raw["/home.json"] = home_data
As there’s no EnvoyData attribute to store the EnvoyHome
data it should be obtained by the application using the model.
# Initialize Envoy, setup and authenticate
envoy = Envoy(host)
# register our updater for legacy envoy
remove = register_updater(EnvoyHome)
assert EnvoyHome in get_updaters()
# setup and authenticate with Envoy
await envoy.setup()
await envoy.authenticate(username=username, password=password, token=token)
# probe what endpoints are available
await envoy.probe()
# get data, the production values now fill from html
data: EnvoyData = await envoy.update()
# obtain our data from raw using the model
home_info: EnvoyHomeInformation = (
EnvoyHomeInformation.from_home(data.raw['/home.json'])
)
print(f'Home info: {home_info.timezone}')