From 3e11e48405d82248c38c3a750afcbebe88740f23 Mon Sep 17 00:00:00 2001 From: Matthew Duffin Date: Wed, 28 Aug 2024 13:36:31 +0200 Subject: [PATCH] Create abstract base class for inverters (#183) * wip * wip * solarman * update token * update inverters * undo some changes * add docstring * mock inverter docstring * remove dotenv * update givenergy * enphase * try to fix tests running twice * delete event * pr comments * revert workflow changes * split inverters into separate modules * import * Revert "import" This reverts commit f9f3c5f40e984c9588497ce0f6eccc396b97a066. * Revert "split inverters into separate modules" This reverts commit 94a9e70db418a8331793d57e41df94b985d3c448. * add pydantic_settings * set config within settings classes * use named argument * fix access token issue * add a line to the docs --- .gitignore | 4 +- quartz_solar_forecast/data.py | 93 ++++---------------- quartz_solar_forecast/inverters/README.md | 1 + quartz_solar_forecast/inverters/enphase.py | 69 +++++++++++---- quartz_solar_forecast/inverters/givenergy.py | 41 +++++++-- quartz_solar_forecast/inverters/inverter.py | 14 +++ quartz_solar_forecast/inverters/mock.py | 12 +++ quartz_solar_forecast/inverters/solarman.py | 54 +++++++++--- quartz_solar_forecast/inverters/solis.py | 54 ++++++++---- quartz_solar_forecast/pydantic_models.py | 19 ++++ requirements.txt | 1 + 11 files changed, 229 insertions(+), 133 deletions(-) create mode 100644 quartz_solar_forecast/inverters/inverter.py create mode 100644 quartz_solar_forecast/inverters/mock.py diff --git a/.gitignore b/.gitignore index 97aecd44..262f2704 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,6 @@ quartz_solar_forecast.egg-info .env venv frontend/node_modules -frontend/.vite \ No newline at end of file +frontend/.vite +__pycache__/ +.cache.sqlite diff --git a/quartz_solar_forecast/data.py b/quartz_solar_forecast/data.py index 06815102..084f2b88 100644 --- a/quartz_solar_forecast/data.py +++ b/quartz_solar_forecast/data.py @@ -1,28 +1,19 @@ """ Function to get NWP data and create fake PV dataset""" import ssl -from datetime import datetime, timedelta -import os +from datetime import datetime +from typing import Optional + import numpy as np -import pandas as pd -import xarray as xr import openmeteo_requests +import pandas as pd import requests_cache -import asyncio - +import xarray as xr from retry_requests import retry -from typing import Optional from quartz_solar_forecast.pydantic_models import PVSite -from quartz_solar_forecast.inverters.enphase import get_enphase_data -from quartz_solar_forecast.inverters.solis import get_solis_data -from quartz_solar_forecast.inverters.givenergy import get_givenergy_data -from quartz_solar_forecast.inverters.solarman import get_solarman_data ssl._create_default_https_context = ssl._create_unverified_context -from dotenv import load_dotenv - -load_dotenv() def get_nwp(site: PVSite, ts: datetime, nwp_source: str = "icon") -> xr.Dataset: """ @@ -41,13 +32,13 @@ def get_nwp(site: PVSite, ts: datetime, nwp_source: str = "icon") -> xr.Dataset: # Define the variables we want. Visibility is handled separately after the main request variables = [ - "temperature_2m", - "precipitation", - "cloud_cover_low", - "cloud_cover_mid", - "cloud_cover_high", - "wind_speed_10m", - "shortwave_radiation", + "temperature_2m", + "precipitation", + "cloud_cover_low", + "cloud_cover_mid", + "cloud_cover_high", + "wind_speed_10m", + "shortwave_radiation", "direct_radiation" ] @@ -59,7 +50,7 @@ def get_nwp(site: PVSite, ts: datetime, nwp_source: str = "icon") -> xr.Dataset: # check whether the time stamp is more than 3 months in the past if (datetime.now() - ts).days > 90: print("Warning: The requested timestamp is more than 3 months in the past. The weather data are provided by a reanalyse model and not ICON or GFS.") - + # load data from open-meteo Historical Weather API url = "https://archive-api.open-meteo.com/v1/archive" @@ -104,7 +95,7 @@ def get_nwp(site: PVSite, ts: datetime, nwp_source: str = "icon") -> xr.Dataset: hourly_data["dswrf"] = hourly.Variables(6).ValuesAsNumpy() hourly_data["dlwrf"] = hourly.Variables(7).ValuesAsNumpy() - # handle visibility + # handle visibility if (datetime.now() - ts).days <= 90: # load data from open-meteo gfs model params = { @@ -144,61 +135,11 @@ def format_nwp_data(df: pd.DataFrame, nwp_source:str, site: PVSite): ) return data_xr -def fetch_enphase_data() -> Optional[pd.DataFrame]: - system_id = os.getenv('ENPHASE_SYSTEM_ID') - if not system_id: - print("Error: Enphase inverter ID is not provided in the environment variables.") - return None - return get_enphase_data(system_id) - -def fetch_solis_data() -> Optional[pd.DataFrame]: - try: - return asyncio.run(get_solis_data()) - except Exception as e: - print(f"Error retrieving Solis data: {str(e)}") - return None - -def fetch_givenergy_data() -> Optional[pd.DataFrame]: - try: - return get_givenergy_data() - except Exception as e: - print(f"Error retrieving GivEnergy data: {str(e)}") - return None - -def fetch_solarman_data() -> pd.DataFrame: - try: - end_date = datetime.now() - start_date = end_date - timedelta(weeks=1) - solarman_data = get_solarman_data(start_date, end_date) - - # Filter out rows with null power_kw values - valid_data = solarman_data.dropna(subset=['power_kw']) - - if valid_data.empty: - print("No valid Solarman data found.") - return pd.DataFrame(columns=['timestamp', 'power_kw']) - - return valid_data - except Exception as e: - print(f"Error retrieving Solarman data: {str(e)}") - return pd.DataFrame(columns=['timestamp', 'power_kw']) - -def fetch_live_generation_data(inverter_type: str) -> Optional[pd.DataFrame]: - if inverter_type == 'enphase': - return fetch_enphase_data() - elif inverter_type == 'solis': - return fetch_solis_data() - elif inverter_type == 'givenergy': - return fetch_givenergy_data() - elif inverter_type == 'solarman': - return fetch_solarman_data() - else: - return pd.DataFrame(columns=['timestamp', 'power_kw']) def process_pv_data(live_generation_kw: Optional[pd.DataFrame], ts: pd.Timestamp, site: 'PVSite') -> xr.Dataset: """ Process PV data and create an xarray Dataset. - + :param live_generation_kw: DataFrame containing live generation data, or None :param ts: Current timestamp :param site: PV site information @@ -231,7 +172,7 @@ def process_pv_data(live_generation_kw: Optional[pd.DataFrame], ts: pd.Timestamp return da -def make_pv_data(site: 'PVSite', ts: pd.Timestamp) -> xr.Dataset: +def make_pv_data(site: PVSite, ts: pd.Timestamp) -> xr.Dataset: """ Make PV data by combining live data from various inverters. @@ -239,7 +180,7 @@ def make_pv_data(site: 'PVSite', ts: pd.Timestamp) -> xr.Dataset: :param ts: the timestamp of the site :return: The combined PV dataset in xarray form """ - live_generation_kw = fetch_live_generation_data(site.inverter_type) + live_generation_kw = site.get_inverter().get_data(ts) # Process the PV data da = process_pv_data(live_generation_kw, ts, site) diff --git a/quartz_solar_forecast/inverters/README.md b/quartz_solar_forecast/inverters/README.md index dc63ac45..039b4be4 100644 --- a/quartz_solar_forecast/inverters/README.md +++ b/quartz_solar_forecast/inverters/README.md @@ -34,6 +34,7 @@ Open-Source-Quartz-Solar-Forecast/ * `pydantic_models.py`: Contains the PVSite class * `inverters/`: * This is the directory where you'd want to create a new file among the other `.py` files to add your inverter + * You will need to create a new inverter model that extends `AbstractInverter` which is defined in `inverter.py` * You will need to follow the appropriate authentication flow as mentioned in the documentation of the inverter you're trying to add * We need the past 7 days data formatted in intervals of 5 minutes for this model. Given below is an example with Enphase diff --git a/quartz_solar_forecast/inverters/enphase.py b/quartz_solar_forecast/inverters/enphase.py index e0434a0e..c54a633d 100644 --- a/quartz_solar_forecast/inverters/enphase.py +++ b/quartz_solar_forecast/inverters/enphase.py @@ -1,24 +1,50 @@ import http.client import os +from typing import Optional + import pandas as pd import json import base64 from datetime import datetime, timedelta, timezone -from dotenv import load_dotenv +from urllib.parse import urlencode + +from quartz_solar_forecast.inverters.inverter import AbstractInverter +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict -load_dotenv() -from urllib.parse import urlencode +class EnphaseSettings(BaseSettings): + model_config = SettingsConfigDict(env_file='.env', extra='ignore') + + client_id: str = Field(alias="ENPHASE_CLIENT_ID") + system_id: str = Field(alias="ENPHASE_SYSTEM_ID") + api_key: str = Field(alias="ENPHASE_API_KEY") + client_secret: str = Field(alias="ENPHASE_CLIENT_SECRET") + + +class EnphaseInverter(AbstractInverter): -def get_enphase_auth_url(): + def __init__(self, settings: EnphaseSettings): + self.__settings = settings + + def get_data(self, ts: pd.Timestamp) -> Optional[pd.DataFrame]: + return get_enphase_data(self.__settings) + + +def get_enphase_auth_url(settings: Optional[EnphaseSettings] = None): """ Generate the authorization URL for the Enphase API. - :param None + :param settings: the Enphase settings :return: Authentication URL """ - client_id = os.getenv('ENPHASE_CLIENT_ID') + if settings is None: + # Because this uses env variables we don't want to set it as a default argument, otherwise it will be evaluated + # even if the method is not called + settings = EnphaseSettings() + + client_id = settings.client_id redirect_uri = ( "https://api.enphaseenergy.com/oauth/redirect_uri" # Or your own redirect URI @@ -51,17 +77,23 @@ def get_enphase_authorization_code(auth_url): return code -def get_enphase_access_token(auth_code=None): +def get_enphase_access_token(auth_code: Optional[str] = None, settings: Optional[EnphaseSettings] = None): """ Obtain an access token for the Enphase API using the Authorization Code Grant flow. :param auth_code: Optional authorization code. If not provided, it will be obtained. + :param settings: Optional Enphase settings :return: Access Token """ - client_id = os.getenv('ENPHASE_CLIENT_ID') - client_secret = os.getenv('ENPHASE_CLIENT_SECRET') + if settings is None: + # Because this uses env variables we don't want to set it as a default argument, otherwise it will be evaluated + # even if the method is not called + settings = EnphaseSettings() + + client_id = settings.client_id + client_secret = settings.client_secret if auth_code is None: - auth_url = get_enphase_auth_url() + auth_url = get_enphase_auth_url(settings) auth_code = get_enphase_authorization_code(auth_url) credentials = f"{client_id}:{client_secret}" @@ -90,7 +122,7 @@ def get_enphase_access_token(auth_code=None): return access_token -def process_enphase_data(data_json: dict, start_at: int) -> pd.DataFrame: +def process_enphase_data(data_json: dict, start_at: int) -> pd.DataFrame: # Check if 'intervals' key exists in the response if 'intervals' not in data_json: return pd.DataFrame(columns=["timestamp", "power_kw"]) @@ -106,7 +138,7 @@ def process_enphase_data(data_json: dict, start_at: int) -> pd.DataFrame: timestamp = datetime.fromtimestamp(end_at, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S') # Append the data to the list - data_list.append({"timestamp": timestamp, "power_kw": interval['powr']/1000}) + data_list.append({"timestamp": timestamp, "power_kw": interval['powr'] / 1000}) # Convert the list to a DataFrame live_generation_kw = pd.DataFrame(data_list) @@ -120,18 +152,19 @@ def process_enphase_data(data_json: dict, start_at: int) -> pd.DataFrame: return live_generation_kw -def get_enphase_data(enphase_system_id: str) -> pd.DataFrame: + +def get_enphase_data(settings: EnphaseSettings) -> pd.DataFrame: """ Get live PV generation data from Enphase API v4 + :param settings: the Enphase settings :param enphase_system_id: System ID for Enphase API :return: Live PV generation in Watt-hours, assumes to be a floating-point number """ - api_key = os.getenv('ENPHASE_API_KEY') access_token = os.getenv('ENPHASE_ACCESS_TOKEN') # If access token is not in environment variables, get a new one if not access_token: - access_token = get_enphase_access_token() + access_token = get_enphase_access_token(settings=settings) # Set the start time to 1 week ago start_at = int((datetime.now() - timedelta(weeks=1)).timestamp()) @@ -142,11 +175,11 @@ def get_enphase_data(enphase_system_id: str) -> pd.DataFrame: conn = http.client.HTTPSConnection("api.enphaseenergy.com") headers = { "Authorization": f"Bearer {access_token}", - "key": api_key + "key": settings.api_key } # Add the system_id and duration parameters to the URL - url = f"/api/v4/systems/{enphase_system_id}/telemetry/production_micro?start_at={start_at}&granularity={granularity}" + url = f"/api/v4/systems/{settings.system_id}/telemetry/production_micro?start_at={start_at}&granularity={granularity}" conn.request("GET", url, headers=headers) res = conn.getresponse() @@ -161,4 +194,4 @@ def get_enphase_data(enphase_system_id: str) -> pd.DataFrame: # Process the data using the new function live_generation_kw = process_enphase_data(data_json, start_at) - return live_generation_kw \ No newline at end of file + return live_generation_kw diff --git a/quartz_solar_forecast/inverters/givenergy.py b/quartz_solar_forecast/inverters/givenergy.py index 02b2052c..b71bff63 100644 --- a/quartz_solar_forecast/inverters/givenergy.py +++ b/quartz_solar_forecast/inverters/givenergy.py @@ -1,19 +1,41 @@ -import os +from typing import Optional + import requests import pandas as pd from datetime import datetime -from dotenv import load_dotenv -# Load environment variables -load_dotenv() +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + +from quartz_solar_forecast.inverters.inverter import AbstractInverter + + +class GivEnergySettings(BaseSettings): + model_config = SettingsConfigDict(env_file='.env', extra='ignore') + + api_key: str = Field(alias="GIVENERGY_API_KEY") + -def get_inverter_serial_number(): +class GivEnergyInverter(AbstractInverter): + + def __init__(self, settings: GivEnergySettings): + self.__settings = settings + + def get_data(self, ts: pd.Timestamp) -> Optional[pd.DataFrame]: + try: + return get_givenergy_data(self.__settings) + except Exception as e: + print(f"Error retrieving GivEnergy data: {e}") + return None + + +def get_inverter_serial_number(settings: GivEnergySettings): """ Fetch the inverter serial number from the GivEnergy communication device API. :return: Inverter serial number as a string """ - api_key = os.getenv('GIVENERGY_API_KEY') + api_key = settings.api_key if not api_key: raise ValueError("GIVENERGY_API_KEY not set in environment variables") @@ -38,18 +60,19 @@ def get_inverter_serial_number(): inverter_serial_number = data[0]['inverter']['serial'] return inverter_serial_number -def get_givenergy_data(): + +def get_givenergy_data(settings: GivEnergySettings): """ Fetch the latest data from the GivEnergy API and return a DataFrame. :return: DataFrame with timestamp and power_kw columns """ - api_key = os.getenv('GIVENERGY_API_KEY') + api_key = settings.api_key if not api_key: raise ValueError("GIVENERGY_API_KEY not set in environment variables") - inverter_serial_number = get_inverter_serial_number() + inverter_serial_number = get_inverter_serial_number(settings) url = f'https://api.givenergy.cloud/v1/inverter/{inverter_serial_number}/system-data/latest' diff --git a/quartz_solar_forecast/inverters/inverter.py b/quartz_solar_forecast/inverters/inverter.py new file mode 100644 index 00000000..60fc72b9 --- /dev/null +++ b/quartz_solar_forecast/inverters/inverter.py @@ -0,0 +1,14 @@ +import abc +from typing import Optional + +import pandas as pd + + +class AbstractInverter(abc.ABC): + """ + An abstract base class representing an inverter which can provide a snapshot of live data. + """ + + @abc.abstractmethod + def get_data(self, ts: pd.Timestamp) -> Optional[pd.DataFrame]: + raise NotImplementedError diff --git a/quartz_solar_forecast/inverters/mock.py b/quartz_solar_forecast/inverters/mock.py new file mode 100644 index 00000000..923edd27 --- /dev/null +++ b/quartz_solar_forecast/inverters/mock.py @@ -0,0 +1,12 @@ +import pandas as pd + +from quartz_solar_forecast.inverters.inverter import AbstractInverter + + +class MockInverter(AbstractInverter): + """ + Provides mock data when live data is not available. + """ + + def get_data(self, ts: pd.Timestamp) -> pd.DataFrame: + return pd.DataFrame(columns=['timestamp', 'power_kw']) diff --git a/quartz_solar_forecast/inverters/solarman.py b/quartz_solar_forecast/inverters/solarman.py index 5b4ce491..7fd07074 100644 --- a/quartz_solar_forecast/inverters/solarman.py +++ b/quartz_solar_forecast/inverters/solarman.py @@ -1,23 +1,53 @@ -import os +from typing import Optional + import requests import pandas as pd -from datetime import timedelta -from dotenv import load_dotenv +from datetime import timedelta, datetime +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + +from quartz_solar_forecast.inverters.inverter import AbstractInverter + + +class SolarmanSettings(BaseSettings): + model_config = SettingsConfigDict(env_file='.env', extra='ignore') + + url: str = Field(alias="SOLARMAN_API_URL") + token: str = Field(alias="SOLARMAN_TOKEN") + id: str = Field(alias="SOLARMAN_ID") + + +class SolarmanInverter(AbstractInverter): + + def __init__(self, settings: SolarmanSettings): + self.__settings = settings + + def get_data(self, ts: pd.Timestamp) -> Optional[pd.DataFrame]: + try: + end_date = datetime.now() + start_date = end_date - timedelta(weeks=1) + solarman_data = get_solarman_data(start_date, end_date, self.__settings) + + # Filter out rows with null power_kw values + valid_data = solarman_data.dropna(subset=['power_kw']) + + if valid_data.empty: + print("No valid Solarman data found.") + return pd.DataFrame(columns=['timestamp', 'power_kw']) -# Load environment variables -load_dotenv() + return valid_data + except Exception as e: + print(f"Error retrieving Solarman data: {str(e)}") + return pd.DataFrame(columns=['timestamp', 'power_kw']) -# Constants -SOLARMAN_API_URL = os.getenv('SOLARMAN_API_URL') -SOLARMAN_TOKEN = os.getenv('SOLARMAN_TOKEN') -SOLARMAN_ID = os.getenv('SOLARMAN_ID') -def get_solarman_data(start_date, end_date): +def get_solarman_data(start_date, end_date, settings: SolarmanSettings): """ Fetch data from the Solarman API from start_date to end_date. :param start_date: Start date (datetime object) :param end_date: End date (datetime object) + :param settings: the Solarman settings :return: DataFrame with timestamp and power_kw columns """ all_data = [] @@ -29,10 +59,10 @@ def get_solarman_data(start_date, end_date): month = current_date.month day = current_date.day - url = f"{SOLARMAN_API_URL}/{SOLARMAN_ID}/record" + url = f"{settings.url}/{settings.id}/record" headers = { - 'Authorization': f'Bearer {SOLARMAN_TOKEN}', + 'Authorization': f'Bearer {settings.token}', 'User-Agent': 'Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.0.7) Gecko/2009021910 Firefox/3.0.7' } diff --git a/quartz_solar_forecast/inverters/solis.py b/quartz_solar_forecast/inverters/solis.py index 6c64661b..3b9d6c5b 100644 --- a/quartz_solar_forecast/inverters/solis.py +++ b/quartz_solar_forecast/inverters/solis.py @@ -1,6 +1,5 @@ from __future__ import annotations import asyncio -import os import pandas as pd from datetime import datetime, timedelta, timezone from aiohttp import ClientSession, ClientError @@ -11,16 +10,18 @@ from enum import Enum from http import HTTPStatus import json -from typing import Any +from typing import Any, Optional + +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + +from quartz_solar_forecast.inverters.inverter import AbstractInverter + try: import async_timeout except: print('Could not import `async_timeout`') -from dotenv import load_dotenv - -SOLIS_CLOUD_API_URL = os.environ.get('SOLIS_CLOUD_API_URL', 'https://www.soliscloud.com') -SOLIS_CLOUD_API_PORT = os.environ.get('SOLIS_CLOUD_API_PORT', '13333') # VERSION RESOURCE_PREFIX = '/v1/api/' @@ -30,6 +31,28 @@ # Endpoints INVERTER_LIST = RESOURCE_PREFIX + 'inverterList' INVERTER_DAY = RESOURCE_PREFIX + 'inverterDay' + + +class SolisSettings(BaseSettings): + model_config = SettingsConfigDict(env_file='.env', extra='ignore') + + api_url: str = Field(alias="SOLIS_CLOUD_API_URL", default='https://www.soliscloud.com') + port: str = Field(alias="SOLIS_CLOUD_API_PORT", default='13333') + api_key: str = Field(alias="SOLIS_CLOUD_API_KEY") + client_secret: str = Field(alias="SOLIS_CLOUD_API_KEY_SECRET") + + +class SolisInverter(AbstractInverter): + def __init__(self, settings: SolisSettings): + self.__settings = settings + + def get_data(self, ts: pd.Timestamp) -> Optional[pd.DataFrame]: + try: + return asyncio.run(get_solis_data(self.__settings)) + except Exception as e: + print(f"Error retrieving Solis data: {str(e)}") + return None + class SoliscloudAPI(): """Class with functions for reading data from the Soliscloud Portal.""" @@ -275,17 +298,13 @@ def _verify_date(format: SoliscloudAPI.DateFormat, date: str): return class SolisData: - def __init__(self, domain: str = None): - load_dotenv() - if domain is None: - domain = f"{SOLIS_CLOUD_API_URL}:{SOLIS_CLOUD_API_PORT}" - self.domain = domain - self.api_key = os.environ.get('SOLIS_CLOUD_API_KEY') - api_secret_str = os.environ.get('SOLIS_CLOUD_API_KEY_SECRET') + def __init__(self, settings: SolisSettings): + self.domain = f"{settings.api_url}:{settings.port}" + self.api_key = settings.api_key + api_secret_str = settings.client_secret if not self.api_key or not api_secret_str: raise ValueError("SOLIS_CLOUD_API_KEY or SOLIS_CLOUD_API_KEY_SECRET environment variable is not set") self.api_secret = api_secret_str.encode('utf-8') # Convert to binary string - self.domain = domain async def get_inverter_list(self, soliscloud: SoliscloudAPI): """Fetch the list of inverters""" @@ -387,7 +406,8 @@ async def get_solis_data(self) -> pd.DataFrame: processed_df = processed_df.reset_index(drop=True) return processed_df - -async def get_solis_data(): - solis_data = SolisData() + + +async def get_solis_data(settings: SolisSettings): + solis_data = SolisData(settings) return await solis_data.get_solis_data() diff --git a/quartz_solar_forecast/pydantic_models.py b/quartz_solar_forecast/pydantic_models.py index eb54ac80..e362e8c3 100644 --- a/quartz_solar_forecast/pydantic_models.py +++ b/quartz_solar_forecast/pydantic_models.py @@ -1,6 +1,13 @@ from pydantic import BaseModel, Field from typing import Optional +from quartz_solar_forecast.inverters.enphase import EnphaseInverter, EnphaseSettings +from quartz_solar_forecast.inverters.givenergy import GivEnergySettings, GivEnergyInverter +from quartz_solar_forecast.inverters.mock import MockInverter +from quartz_solar_forecast.inverters.solarman import SolarmanSettings, SolarmanInverter +from quartz_solar_forecast.inverters.solis import SolisSettings, SolisInverter + + class PVSite(BaseModel): latitude: float = Field(..., description="the latitude of the site", ge=-90, le=90) longitude: float = Field( @@ -25,6 +32,18 @@ class PVSite(BaseModel): json_schema_extra=["enphase", "solis", "givenergy", "solarman", None], ) + def get_inverter(self): + if self.inverter_type == 'enphase': + return EnphaseInverter(EnphaseSettings()) + elif self.inverter_type == 'solis': + return SolisInverter(SolisSettings()) + elif self.inverter_type == 'givenergy': + return GivEnergyInverter(GivEnergySettings()) + elif self.inverter_type == 'solarman': + return SolarmanInverter(SolarmanSettings()) + else: + return MockInverter() + class ForecastRequest(BaseModel): site: PVSite timestamp: Optional[str] = None diff --git a/requirements.txt b/requirements.txt index 3d9fd9c1..1039e439 100644 --- a/requirements.txt +++ b/requirements.txt @@ -15,3 +15,4 @@ streamlit async_timeout uvicorn fastapi +pydantic_settings