Add Victron inverter (#202)

* 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 f9f3c5f.

* Revert "split inverters into separate modules"

This reverts commit 94a9e70.

* add pydantic_settings

* set config within settings classes

* add dependency

* add victron

* use named argument

* start and end times

* add a line to the documentation

* dependencies

* add test

* add test

* update script

* add username and password to example env variable file

* move vrmapi dependency to optional-dependencies

* comment

* update to use ocf_vrmapi

.env.example

@@ -15,6 +15,10 @@ SOLIS_CLOUD_API_PORT = '13333'
 # User needs to add their GivEnergy API details
 GIVENERGY_API_KEY = 'user_givenergy_api_key'
 
+# To connect to a Victron system use the environment variables below to set the username and password
+#VICTRON_USER=username
+#VICTRON_PASS=password
+
 # User needs to add their GivEnergy API details
 SOLARMAN_API_URL = 'https://home.solarmanpv.com/maintain-s/history/power'
 SOLARMAN_TOKEN = 'user_solarman_token'

examples/inverter_example.py

@@ -14,7 +14,7 @@ def main(save_outputs: bool = False):
     ts = pd.to_datetime(timestamp_str)
 
     # make input data with live enphase, solis, givenergy, or solarman data
-    site_live = PVSite(latitude=51.75, longitude=-1.25, capacity_kwp=1.25, inverter_type="enphase") # inverter_type="enphase", "solis", "givenergy", or "solarman"
+    site_live = PVSite(latitude=51.75, longitude=-1.25, capacity_kwp=1.25, inverter_type="enphase") # inverter_type="enphase", "solis", "givenergy", "solarman" or "victron"
 
     # make input data with nan data
     site_no_live = PVSite(latitude=51.75, longitude=-1.25, capacity_kwp=1.25)

pyproject.toml

@@ -43,6 +43,10 @@ package-data = { "quartz_solar_forecast" = ["*"] }
 
 [project.optional-dependencies]
 dev = []
+# additional vendor-specific dependencies for connecting to inverter APIs
+inverters = [
+    "ocf_vrmapi" # victron
+]
 
 [tool.mypy]
 

quartz_solar_forecast/inverters/victron.py

@@ -0,0 +1,46 @@
+from typing import Callable
+
+import pandas as pd
+from datetime import datetime, timedelta
+from quartz_solar_forecast.inverters.inverter import AbstractInverter
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from ocf_vrmapi.vrm import VRM_API
+
+
+class VictronSettings(BaseSettings):
+    model_config = SettingsConfigDict(env_file='.env', extra='ignore')
+
+    username: str = Field(alias="VICTRON_USER")
+    password: str = Field(alias="VICTRON_PASS")
+
+
+class VictronInverter(AbstractInverter):
+
+    def __init__(self, get_sites: Callable, get_kwh_stats: Callable):
+        self.__get_sites = get_sites
+        self.__get_kwh_stats = get_kwh_stats
+
+    @classmethod
+    def from_settings(cls, settings: VictronSettings):
+        api = VRM_API(username=settings.username, password=settings.password)
+        get_sites = lambda: api.get_user_sites(api.user_id)
+        end = datetime.now()
+        start = end - timedelta(weeks=1)
+        get_kwh_stats = lambda site_id: api.get_kwh_stats(site_id, start=start, end=end)
+        return cls(get_sites, get_kwh_stats)
+
+    def get_data(self, ts: pd.Timestamp) -> pd.DataFrame:
+        sites = self.__get_sites()
+        # get first site (bit of a guess)
+        first_site_id = sites["records"][0]["idSite"]
+
+        stats = self.__get_kwh_stats(first_site_id)
+
+        kwh = stats["records"]["kwh"]
+
+        df = pd.DataFrame(kwh)
+
+        df[0] = pd.to_datetime(df[0], unit='ms')
+        df.columns = ["timestamp", "power_kw"]
+        return df

quartz_solar_forecast/pydantic_models.py

@@ -6,6 +6,7 @@
 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
+from quartz_solar_forecast.inverters.victron import VictronSettings, VictronInverter
 
 
 class PVSite(BaseModel):
@@ -41,6 +42,8 @@ def get_inverter(self):
             return GivEnergyInverter(GivEnergySettings())
         elif self.inverter_type == 'solarman':
             return SolarmanInverter(SolarmanSettings())
+        elif self.inverter_type == 'victron':
+            return VictronInverter.from_settings(VictronSettings())
         else:
             return MockInverter()