New plugin: Autarco PV telemetry direct from inverter using Solis S3 logger

Python and python framework

Moderator: leecollings

Post Reply
User avatar
Domoberry
Posts: 132
Joined: Tuesday 30 May 2017 19:00
Target OS: Raspberry Pi / ODroid
Domoticz version: 2026.2
Contact:

New plugin: Autarco PV telemetry direct from inverter using Solis S3 logger

Post by Domoberry »

Hi Domotians,
Some time ago, I had realized a dzVents script that retrieved telemetry data (current power, total energy, etc.) from the Autarco cloud for my Autarco LX6000 inverter. Unfortunately, Autarco went out of business earlier this year and per consequence the cloud service and mobile app stopped working (there is a paid alternative from Invrida).
Below please find information I used to tackle the issue, leading to a Domoticz plugin showing locally downloaded telemetry data.
If you have a suitable data logger, the plugin should work instantly. See documentation and plugin attached.

Local telemetry download
I searched for a way to retrieve the data locally, direct from the inverter. I found several.
The simple and straightforward one used for this plugin is to use a (‘hidden’) endpoint on the Wi-Fi stick (logger). This endpoint provides telemetry data (including serial number, FW version, inverter model, temperature, current power, total power, and inverter alerts). This endpoint is available in several of the commonly used loggers.
An alternative is using direct Modbus communication to the inverter. This requires a Modbus gateway hardware interface. Some of the logger devices have something similar as a (again ‘hidden’) feature. This option typically requires the pysolarmanv5 library. The Modbus interface provides much more data.

Solis S3-WIFI-ST logger
The Autarco branded logger included in my setup did not support the semi-Modbus feature, nor the hidden endpoint. I learned there are multiple loggers used for this inverter (both Inverter and logger are Ginlong Solis rebranded products). I got a Solis S3-WIFI-ST logger from Marktplaats (not costly), which still did not have the semi-Modbus feature, but did offer the endpoint! Relacing the logger with the S3 means I would lose cloud connectivity, but the cloud was offline anyhow, so not an issue.
With help of fora and AI, I put together a Domoticz Python Plugin, which takes care of the data retrieval using http GET on the endpoint. No cloud needed.

Will my logger work?
As this plugin fully relies on the availability of the ‘hidden’ endpoint in the logger, you should check this first. I assume you have the logger installed, you know it’s local IP address and the login username / password. Pointing your browser to this address should give you the logger WebUI and a pop up to login. The attached documentation gives details on how to install the logger.
1. Point your browser to the “/inverter.cgi” endpoint:
http://192.168.0.55/inverter.cgi (use the correct IP)
2. Login using the credentials you provided during installation of the logger
3. The reply you see should be a simple line of semicolon delimited text:
xxxxxxxxxxxxxxxx;2E0330;41;37.7;4412;108;53781;NO;
This text represents the telemetry data you need.
Great: the logger has what it needs for this plugin to work.
If you have a similar model inverter / logger but still get the data via this endpoint, it is likely that this plugin will work as well. I did not test that.
Note: to check if your logger supports the Semi-Modbus gateway function, other tests can be done. See Google. Modbus is not supported in this plugin.

Figure 1 - Solis S3-WIFI-ST Logger
Picture1 w=275.jpg
Picture1 w=275.jpg (7.4 KiB) Viewed 70 times


Figure 2 - Autarco original Logger
Picture2.png
Picture2.png (11.68 KiB) Viewed 70 times
Anything else needed?
With the following in place, you could try and install the Plugin:
  • A logger that responds to the ‘inverter.cgi’ endpoint as tested above.
    You need to know its IP address and login credentials.

    A running Domoticz setup.
    I tried this on an RPI4 with Bullseye and Domoticz 2026.3 and on an RPI5 with Bullseye running Domoticz 2026.3 on Docker. It will likely work on other systems as well.

    Ability to access your RPI to install the plugin, I used PuTTY and WinSCP.

    No specific additional Python libraries are needed, everything is installed as part of the Domoticz installation

How to install the plugin
Please take a look at the documentation: Results
Please take a look at the documentation

Plugin code

Code: Select all

"""
<plugin key="SolisLocal" name="Solis S3-WIFI-TS logger for Autarco Local Telemetry" author="Domoberry" version="2.1.1">
    <description>
        <h3>Solis S3-WIFI-TS logger for Autarco Local Telemetry</h3>
        <p><b>v2.1.1</b></p>
        <ul>
            <li>Reads inverter telemetry from Solis S3-WIFI-ST datalogger via HTTP.</li>
            <li>Includes sleep-mode detection (night-time) and automatic wake-up handling.</li>
            <li>Note: If Debug is enabled, ensure the global Domoticz -loglevel includes 'debug' (or set to 15), and remember to disable it afterwards to protect your log file size.</li>
            <li>Note: Logger credentials are sent over unencrypted http connection, use on a trusted network only.)</li>
        </ul>
    </description>
    <params>
        <param field="Address" label="Logger IP Address" width="200px" required="true" default="127.0.0.1"/>
        <param field="Username" label="Username" width="200px" required="true" default="admin"/>
        <param field="Password" label="Password" width="200px" required="true" default="123456789"/>
        <param field="Mode1" label="Polling interval (seconds)" width="100px" required="true" default="300"/>
        <param field="Mode6" label="Debug" width="200px">
            <options>
                <option label="None (0)" value="0" default="true"/>
                <option label="Standard (62 Basic)" value="62"/>
                <option label="Python Framework Only (2)" value="2"/>
                <option label="Verbose (-1 Everything)" value="-1"/>
            </options>
         </param>
    </params>
</plugin>
"""

import Domoticz
import time
import base64 # needed for user credentials base64 encoding

class BasePlugin:
    def __init__(self):
        self.poll_interval = 10 # 10 secs, 1 sec min. Default is 300 secs, overwritten by what user enters
        self.retry_interval = 9 # retries should be quicker than polling, to determine 'sleep' mode quickly
        # just under 10 secs (heartbeat) to allow for some time to do calculation yet ensure retries every 10 secs
        self.next_poll = time.time()# current time in epoch notation
        self.http_conn = None # keeps track of an active connection, 'None' if no connection
        self.retry_attempt = 0 # means (if not 0: "Am I in a connection-failure retry sequence?"
        self.max_retries = 3
        self.is_fetching = False # means: "Am I waiting for a network response?"
        self.is_disconnecting = False # means: "Did we deliberately request a disconnect?"

    def onStart(self):
        try:
            Domoticz.Log("Solis / Autarco Local Telemetry plugin starting...")
            
            # Check if the Debug Level parameter is present and not set to "None" (0)
            # Debugging can have bitwise values like 0, 1, 2, 16, 18, 62, 126, -1, see documentation
            if "Mode6" in Parameters and Parameters["Mode6"] != "0":
                try:
                    # Convert the string value from the XML option to an integer bitmask
                    debug_level = int(Parameters["Mode6"])
                    Domoticz.Debugging(debug_level)
                    Domoticz.Log(f"Debugging enabled with bitmask level: {debug_level}")
                except ValueError:
                    # Fallback safety case in case of an invalid string conversion
                    Domoticz.Debugging(1)
                    Domoticz.Log("Invalid debug level provided. Defaulting to standard debugging (1).")
            else:
                # Explicitly turn off debugging if "0" or not set
                Domoticz.Debugging(0)
            
            # Using standard global variable injected by Domoticz
            self.poll_interval = max(1, int(Parameters["Mode1"])) # enfore minimum and integer number only
            
            # Create Domoticz devices if missing
            if 1 not in Devices:
                Domoticz.Log("Device not found. Creating 'Inverter Temperature' device...")
                Domoticz.Device(Name="Inverter Temperature", Unit=1, TypeName="Temperature").Create()
                Domoticz.Log("Device successfully created.")
            else:
                Domoticz.Log("Device 'Inverter Temperature' (Unit 1) already exists")
            #
            #
            if 2 not in Devices:
                Domoticz.Log("Device not found. Creating 'Solar Production' device...")
                # a. Structural creation (Run once)
                Domoticz.Device(Name="Solar Production", Unit=2, TypeName="kWh", Switchtype=4).Create()
                # b. Apply metadata configuration options (Run once)
                solar_options = {
                    "EnergyMeterMode": "0", 
                    "ValueQuantity": "Energy", 
                    "ValueUnits": "kWh"
                } # options for the device to optimize for use as kWh counter for generated energy
                # c. Initialize database state with a valid "0;0" data schema shape
                Devices[2].Update(nValue=0, sValue="0;0", Options=solar_options)
                Domoticz.Log("Device successfully created and configured.")
            else:
                Domoticz.Log("Device 'Solar Production' (Unit 2) already exists")
            # this “kWh” device takes two values: Current Power and Total Energy.
            #
            #
            if 3 not in Devices:
                Domoticz.Log("Device not found. Creating 'Today Production' device...")
                # a. Structural creation
                Domoticz.Device(Name="Today Production", Unit=3, TypeName="Custom").Create()
                # b. Assign the string suffix label for the UI
                daily_options = {"Custom": "1;kWh"} # '1;' acts as an axis format multiplier flag
                # c. Initialize the database row
                Devices[3].Update(nValue=0, sValue="0.0", Options=daily_options)
                Domoticz.Log("Device successfully created and configured.")
            else:
                Domoticz.Log("Device 'Today Production' (Unit 3) already exists")
            #
            #
            if 4 not in Devices:
                Domoticz.Log("Device not found. Creating 'Inverter Alerts' device...")
                Domoticz.Device(Name="Inverter Alerts", Unit=4, TypeName="Text").Create()
                Domoticz.Log("Device successfully created.")
            else:
                Domoticz.Log("Device 'Inverter Alerts' (Unit 4) already exists")
            #
            #
            if 5 not in Devices:
                Domoticz.Log("Device not found. Creating 'Inverter Status' device...")
                Domoticz.Device(Name="Inverter Status", Unit=5, TypeName="Text").Create()
                Domoticz.Log("Device successfully created.")
            else:
                Domoticz.Log("Device 'Inverter Status' (Unit 5) already exists")
            Devices[5].Update(nValue=0, sValue="Starting...") # Assing an initial value
            #
            #
            if 6 not in Devices:
                Domoticz.Log("Device not found. Creating 'Lifetime Production' device...")
                # a. Structural creation
                Domoticz.Device(Name="Lifetime Production", Unit=6, TypeName="Custom").Create()
                # b. Assign the string suffix label for the UI
                daily_options = {"Custom": "1;kWh"} # '1;' acts as an axis format multiplier flag
                # c. Initialize the database row
                Devices[6].Update(nValue=0, sValue="0.0", Options=daily_options)
                Domoticz.Log("Device successfully created and configured.")
            else:
                Domoticz.Log("Device 'Lifetime Production' (Unit 6) already exists")
            
            Domoticz.Log("Devices checked and initialized successfully")            
            # Calculate the encoded header token ONCE and save it to the class instance
            creds = f"{Parameters['Username']}:{Parameters['Password']}"
            encoded = base64.b64encode(creds.encode("utf-8")).decode("utf-8")
            self.auth_header = f"Basic {encoded}" 
            
            Domoticz.Log("User credentials encoded successfully")
            
            # Initialize the persistent, asynchronous Domoticz connection object
            self.http_conn = Domoticz.Connection(
                Name="SolisConn", 
                Transport="TCP/IP", 
                Protocol="HTTP", 
                Address=Parameters["Address"], # holds the IP addrss of the logger 
                Port="80"
            ) # self.http_con is initialized as 'None'. If connection ok, it contains the connection object
            
            Domoticz.Status("Plugin startup and Device creation completed")
        except Exception as e:
            Domoticz.Error(f"Fatal error during plugin setup (onStart) phase: {str(e)}")

    def onHeartbeat(self):
        """Call triggerFetch() unless the below guard clauses kick-in."""
        Domoticz.Debug("onHeartbeat called")
        # Guard clause to prevent actions if onStart failed to initialize the connection object
        if self.http_conn is None:
            Domoticz.Debug("Return on Gaurd Clause: no active connection")        
            return
        
        # Guard clause to respect poll intervals
        if time.time() < self.next_poll:
            Domoticz.Debug("Return on Gaurd Clause: not yet time for next poll")
            return
        # Guard clause to ensure we aren't overlaying a network request if one is already active
        if self.is_fetching:
            Domoticz.Debug("Return on Guard Clause: Previous fetch still active, skipping heartbeat tick.")
            return
        
        # Use normal polling interval for a new normal telemetry cycle,
        # Use retry interval for subsequent attempts.
        if self.retry_attempt == 0:
            self.next_poll = time.time() + self.poll_interval
            Domoticz.Debug(f"Next poll time based on poll_interval ({self.poll_interval} secs)")
        
        # continue with (trying to) fetch data
        self.triggerFetch()

    def triggerFetch(self):
        """Initiates the connection or sends the HTTP command if already open."""
        Domoticz.Debug(f"triggerFetch called, fetching telemetry (attempt {self.retry_attempt}/{self.max_retries}) ...")
        self.is_fetching = True # that what we are doing now
        if not self.http_conn.Connected():
            Domoticz.Debug("Connection not established yet, setting up connection ...")
            self.http_conn.Connect() # establishes the underlying network connection to the server (async approach)
            # next step is wait for the connection result, which is handled by onConnect()
        else:
            # Send HTTP GET command asynchronously
            Domoticz.Debug("Connection already established, send the GET...")
            self.http_conn.Send({
                'Verb': 'GET',
                'URL': '/inverter.cgi',
                'Headers': {
                    'Authorization': self.auth_header, # As defined in onStart
                    'Connection': 'keep-alive' # keep the underlying network connection open for the next GET
                }
            }) # uses the IP address used when setting up http_con object
            # next step is wait for GET reply, whic is handled by onMessage() (async approach)

    def onConnect(self, Connection, Status, Description):
        """called by Domoticz once a connection attempt is made, if Status is ok, send GET request."""
        Domoticz.Debug(f"onConnect called: Status={Status}, Description={Description}")
        self.last_status = Status # store for later use
        if Status == 0: # "0" means all is ok
            Domoticz.Status("Connected successfully to logger, send telemetry data request")
            Connection.Send({
                'Verb': 'GET',
                'URL': '/inverter.cgi',
                'Headers': {
                    'Authorization': self.auth_header, # As defined in onStart
                    'Connection': 'keep-alive' # keep the underlying network connection open for the next GET
                }
            })
        else:
            # A failed connection is an expected condition when the logger
            # is sleeping/offline, so don't log it as an Error.
            Domoticz.Debug("Status != 0, call handleFailure")
            self.handleFailure()

    def onMessage(self, Connection, Data):
        """Check and process returned data and -if all ok- update devices """
        # checks on returned data, decode it, more checks, update wake/sleep, parse it and if all ok, update devices
        Domoticz.Debug("onMessage called")
        try:
            Domoticz.Debug("Data received from logger")
            
            # A response has arrived, so the network fetch itself is finished.
            self.is_fetching = False
            
            # A response was received, so terminate any connection-retry sequence
            # and wait for the next normal polling cycle unless valid telemetry
            # is processed below.
            self.retry_attempt = 0
            self.next_poll = time.time() + self.poll_interval
            
            # Disconnect gracefully to keep the connection clean for the next cycle
            if Connection.Connected():
                self.is_disconnecting = True
                Connection.Disconnect()
                Domoticz.Debug("Data received, disconnect from logger")
            
           # --- Detect HTTP status errors ---
            if "Status" in Data:
                status = str(Data["Status"]) # str() to avoid runtime error if Status would for some reason be a number
                if "401" in status or "Unauthorized" in status:
                    Domoticz.Error("Authentication failed: incorrect username and/or password.")
                    return
            
            # --- Detect missing or empty payload ---
            if "Data" not in Data or not Data["Data"]:
                Domoticz.Error("Logger returned no data. Possible incorrect credentials or logger not ready.")
                return
            
            # --- Decode payload ---
            try:
                raw = Data["Data"].decode("utf-8").strip()
                # better not log 'raw' during a happy flow as it is very bulky. If needed, always use 'raw!r' to
                # convert non-printables to characters, e.g. a \x00 (NUL) is converted to 4 characters: "/x00"
                # leaving 'real' NUL's in the data to log is confusing the Domoticz logging system and nothing log..
            except Exception as e:
                Domoticz.Error(f"Failed to decode response payload: {e}")
                return
            
            # --- Detect HTML (wrong credentials or logger login page) ---
            if "<html" in raw.lower():
                Domoticz.Error("Logger returned HTML instead of telemetry. Incorrect credentials likely.")
                Domoticz.Debug(f"Raw HTML was: {raw!r}")
                return
            
            # --- Detect plain-text unauthorized responses ---
            if "unauthorized" in raw.lower():
                Domoticz.Error("Logger indicates authentication failure.")
                Domoticz.Debug(f"Raw response was: {raw!r}")
                return
            
            # --- Detect malformed telemetry ---
            parts = raw.split(";")
            if len(parts) < 8:
                Domoticz.Error("Telemetry format incomplete. Logger may be waking up, wrong IP, etc.")
                Domoticz.Debug(f"Raw response was: '{raw!r}'")
                return
            
            # --- Parse and populate telemetry parameters ---
            data = self.parseTelemetry(raw)
            if not data: # check if data would be empty/nothing
                return
            
            # --- Wake-up detection ---
            Domoticz.Debug("Passed all guard checks on received data, inverter status is obviously onLine")
            if Devices[5].sValue != "Online":
                Devices[5].Update(nValue=0, sValue="Online")
                Domoticz.Status("Logger woke up or was already online")
            
            Domoticz.Debug(f"Parsed data is: {data}")
            
            # --- Update Domoticz Devices safely with current numbers ---
            
            # Update inverter temperature
            Devices[1].Update(nValue=0, sValue=str(data["temp"]))
            
            # Update current power / total energy device under a condition
            # If watts are 0, that's fine. But if lifetime energy is 0, 
            # it means the logger has lost its place or hasn't booted fully yet.
            if data["total"] == 0:
                Domoticz.Status("Warning: Received a 0 value for Total Lifetime Energy. Ignoring to protect historical logs")
                # do not update the Domoticz device
            else:
                # Otherwise, update with current power and total energy
                Devices[2].Update(nValue=0, sValue=str(data["power"])+";"+str(int(data["total"]*1000)))
                # int() was added to ensure there is no decimal send to the device
                # added a device to show Total Lifetime Energy as the above device does not display the total
                Devices[6].Update(nValue=0, sValue=f"{data['total']}") #this device accepts kWh
            
            # Update todays production
            Devices[3].Update(nValue=0, sValue=f"{data['today']}") #this device accepts kWh
            
            # Update alerts
            Devices[4].Update(nValue=0, sValue=str(data["alerts"]))
            
            Domoticz.Debug(f"onMessage: updated devices, retry_attempt = {self.retry_attempt}")
            Domoticz.Status("Devices succesfully updated")
        except Exception as e:
            Domoticz.Error(f"Error handling message payload processing: {str(e)}")

    def onDisconnect(self, Connection):
        Domoticz.Debug("onDisconnect called")
        
        if self.is_disconnecting:
            self.is_disconnecting = False
            Domoticz.Debug("Connection closed cleanly after response")
        elif self.is_fetching:
            self.handleFailure()

    def handleFailure(self):
        """Manages retry loops safely across ticks without spamming the network layer."""
        Domoticz.Debug("handleFailure called")
        # Clear fetching flag so the next heartbeat step triggers the next retry cleanly
        self.is_fetching = False
        if self.retry_attempt < self.max_retries:
            # at least one more attempt allowed
            self.retry_attempt += 1
            self.next_poll = time.time() + self.retry_interval # set the shorter retry interval
            Domoticz.Debug(f"poll_interval set to +{self.retry_interval} secs later")
            Domoticz.Debug(
                f"Connection unavailable "
                f"(status={self.last_status}). "
                f"Retry connecting"
            )
        
        else:
            # All retries failed safely -> execute sleep mode parameters
            self.retry_attempt = 0
            self.next_poll = time.time() + self.poll_interval # solis sleeping, no need continue quick retries
            Domoticz.Debug(f"Max retries reached, Solis likely sleeping (unreachable), retry counter reset to {self.retry_attempt} and next_poll set to +{self.poll_interval} secs later")
            if Devices[5].sValue != "Sleeping":
                Devices[5].Update(nValue=0, sValue="Sleeping")
                # Reset Today Production to 0 when entering sleep mode. This prevents Domoticz from 
                # drawing a slanted line from yesterday's final value to the next day's initial value
                Devices[3].Update(nValue=0, sValue="0.0")
                Domoticz.Status("Inverter to sleeping mode (logger offline)")
            else:
                Domoticz.Debug("Inverter already in sleeping mode (logger offline)")

    def parseTelemetry(self, raw):
        """Parse semicolon-separated raw string into fields."""
        # The Solis logger adds a lot of NUL-padding before and after the data, remove that first
        data = raw.strip("\x00\r\n") # strip out NUL, new-line and CR
        # returns a Python dictionary (key-value pairs)
        # Index:   0                1        2    3      4     5      6      7
        # Value:   xxxxxxxxxxxxxxxxx;2E0330  ;41  ;27.5  ;211  ;83    ;53560 ;NO;
        # Meaning: [Serial]        [FW vers][Mod][Temp] [Watt][today] [Total][Alert]
        try:
            parts = data.split(";")
            # Scale daily yield from integer to decimal (e.g., 83 -> 8.3 kWh)
            # 83 kWh in one day would be to good to be true!)
            # Total 53560 is scaled when updating Domoticz device: Wh expected, not kWh.
            scaled_today = float(parts[5]) / 10.0  
            
            return {
                "serial": parts[0],
                "fw": parts[1],
                "model": parts[2],
                "temp": float(parts[3]),
                "power": int(parts[4]),
                "today": scaled_today,
                "total": float(parts[6]),
                "alerts": parts[7]
            }
        except Exception as e:
            Domoticz.Error(f"Telemetry parse error: {e}")
            return None

# ==============================================================================
# GLOBAL DOMOTICZ HOOK ENGINES
# ==============================================================================
_plugin = BasePlugin()

def onStart():
    _plugin.onStart()

def onStop():
    pass

def onHeartbeat():
    _plugin.onHeartbeat()

def onConnect(Connection, Status, Description):
    _plugin.onConnect(Connection, Status, Description)

def onMessage(Connection, Data):
    _plugin.onMessage(Connection, Data)

def onDisconnect(Connection):
    _plugin.onDisconnect(Connection)
Let me know!
Post Reply