MQTT Connection

[rschoebel]

Hi,

i found some api endpoint to get certificates for the mqtt server,
my current problem is i can't trigger an update to receive new mqtt data.

but if i run this script and connect via the app on my smart phone i got some data :)

needs pip install paho-mqtt

#!/usr/bin/env python
"""Example exec for mqtt."""

import asyncio
from datetime import datetime
import json
import logging
from pathlib import Path

from aiohttp import ClientSession
from api import api  # pylint: disable=no-name-in-module
from api.apitypes import SolixParmType  # pylint: disable=no-name-in-module
import common

import paho.mqtt.client as mqtt
import ssl
import time
import tempfile
import os

_LOGGER: logging.Logger = logging.getLogger(__name__)
_LOGGER.addHandler(logging.StreamHandler())
# _LOGGER.setLevel(logging.DEBUG)    # enable for detailed API output
CONSOLE: logging.Logger = common.CONSOLE

TOPIC = ""

def _out(jsondata):
    CONSOLE.info(json.dumps(jsondata, indent=2))


def on_connect(client, userdata, flags, rc):
    """Callback for when the client receives a CONNACK response from the server."""
    if rc == 0:
        print("Connected successfully to MQTT broker")
        client.subscribe(TOPIC)
    else:
        print(f"Failed to connect, return code {rc}")

def on_message(client, userdata, msg):
    """Callback for when a PUBLISH message is received from the server."""
    print(f"Received message: {msg.payload.decode()} on topic: {msg.topic}")

def on_disconnect(client, userdata, rc):
    """Callback for when the client disconnects from the server."""
    print("Disconnected from MQTT broker")

def create_mqtt_client(cid,cert,key,ca):
    """Create and configure MQTT client with SSL/TLS certificates from strings."""
    
    # Create client instance
    client = mqtt.Client(client_id=cid, clean_session=True)
    
    # Set callbacks
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    
    # Create temporary files for certificates
    ca_cert_file = None
    client_cert_file = None
    client_key_file = None
    
    try:
        # Create temporary file for CA certificate
        ca_cert_file = tempfile.NamedTemporaryFile(mode='w', suffix='.crt', delete=False)
        ca_cert_file.write(ca)
        ca_cert_file.flush()
        
        # Create temporary file for client certificate
        client_cert_file = tempfile.NamedTemporaryFile(mode='w', suffix='.crt', delete=False)
        client_cert_file.write(cert)
        client_cert_file.flush()
        
        # Create temporary file for client private key
        client_key_file = tempfile.NamedTemporaryFile(mode='w', suffix='.key', delete=False)
        client_key_file.write(key)
        client_key_file.flush()
        
        # Configure SSL/TLS using temporary files
        client.tls_set(ca_certs=ca_cert_file.name,
                      certfile=client_cert_file.name,
                      keyfile=client_key_file.name,
                      cert_reqs=ssl.CERT_REQUIRED,
                      tls_version=ssl.PROTOCOL_TLS,
                      ciphers=None)
        
        # Alternative method using SSL context (commented out):
        # context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        # context.check_hostname = False
        # context.load_verify_locations(ca_cert_file.name)
        # context.load_cert_chain(client_cert_file.name, client_key_file.name)
        # client.tls_set_context(context)
        
        # Store temp file references for cleanup
        client._temp_cert_files = [ca_cert_file.name, client_cert_file.name, client_key_file.name]
        
    except Exception as e:
        # Clean up temp files if there was an error
        for temp_file in [ca_cert_file, client_cert_file, client_key_file]:
            if temp_file and os.path.exists(temp_file.name):
                temp_file.close()
                os.unlink(temp_file.name)
        raise e
    finally:
        # Close file handles (files remain on disk for SSL to use)
        if ca_cert_file:
            ca_cert_file.close()
        if client_cert_file:
            client_cert_file.close()
        if client_key_file:
            client_key_file.close()
    
    return client

def cleanup_temp_files(client):
    """Clean up temporary certificate files."""
    if hasattr(client, '_temp_cert_files'):
        for temp_file_path in client._temp_cert_files:
            try:
                if os.path.exists(temp_file_path):
                    os.unlink(temp_file_path)
            except Exception as e:
                print(f"Warning: Could not delete temp file {temp_file_path}: {e}")

async def main() -> None:
    """Create the aiohttp session and run the example."""

    CONSOLE.info("Testing Solix API:")
    async with ClientSession() as websession:
        myapi = api.AnkerSolixApi(
            common.user(),
            common.password(),
            common.country(),
            websession,
            _LOGGER,
        )
        # show login response
        new = await myapi.async_authenticate(
            restart=True
        )  # enforce new login data from server
        new = (
            await myapi.async_authenticate()
        )  # receive new or load cached login data
        if new:
            CONSOLE.info("Received Login response:")
        else:
            CONSOLE.info("Cached Login response:")
        _out(
            myapi.apisession._login_response  # noqa: SLF001
        )  # show used login response for API requests
        mqtt = await myapi.apisession.request("post","v1/openapi/devicemanage/get_user_mqtt_info")
        _out(
            mqtt
        )
        #_out(
        #    mqtt["data"]
        #)
        #await myapi.update_sites()
        await myapi.update_device_details()
        #CONSOLE.info("Account Overview:")
        #_out(myapi.account)
        CONSOLE.info("Device Overview:")
        _out(myapi.devices)
        #await test_api_from_json_files(myapi)
       #get first device
        device = next(iter(myapi.devices))
        global TOPIC
        TOPIC = f"dt/{mqtt['data']['app_name']}/{myapi.devices[device]['device_pn']}/{myapi.devices[device]['device_sn']}/param_info"
        TOPIC = f"dt/{mqtt['data']['app_name']}/{myapi.devices[device]['device_pn']}/{myapi.devices[device]['device_sn']}/#"
        #client.subscribe(f"dt/{mqtt['data']['app_name']}/{myapi.devices[device]['device_pn']}/{myapi.devices[device]['device_sn']}/param_info")
        _out(TOPIC) 
        client = create_mqtt_client(mqtt["data"]["thing_name"],mqtt["data"]["certificate_pem"],mqtt["data"]["private_key"],mqtt["data"]["aws_root_ca1_pem"])
        # Connect to broker
        print(f"Connecting to MQTT broker at {mqtt['data']['endpoint_addr']}:8883")
        client.connect(mqtt["data"]["endpoint_addr"], 8883, 20)
        
        # Start the loop to process network traffic and callbacks
        client.loop_start()
        try:
            while True:
                time.sleep(20)
                #publish mqtt messag to trigger update
        except KeyboardInterrupt:
            print("\nDisconnecting...")
            client.loop_stop()
            client.disconnect()
            cleanup_temp_files(client)

# run async main
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as err:  # pylint: disable=broad-exception-caught  # noqa: BLE001
        CONSOLE.exception("%s: %s", type(err), err)

[thomluther]

Hi, I tried your code with SB1 and I receive data without need to trigger something with the App.
The problem is, that the payload is encrypted and the question is how it can be decrypted?

The mqtt response contains pkcs12 data beside the certificates for the server connection. I assume this pkcs12 data is used for the payload encryption. The question is how to derive the private key and certificate from pkcs12 file without knowing its password?
I tried with my account password, but is not valid for the pkcs12 data. It could be something account specific, or a common secret used by the Anker App to get the private key from pkcs12 data...

[thomluther]

Looking at the timestamps of the messages, I get them in 60 second intervals, the normal rate as data is also available on power cloud.
So I assume without finding the App trigger to decrease publish interval, you would see messages only every 5 minutes for SB2 and SB3.
But again, this is not the problem, first it will be important to decrypt the data field in the payload, otherwise the MQTT interface is not really useful.

[rschoebel]

As far as i can tell, it is not encrypted.
I found a string representing my model A1780P
and one byte wich seems to represent the charge level of my F2000.
i hope the data packages are somewhat like the Bluetooth packages.
https://www.reddit.com/r/bluetoothlowenergy/comments/193s2v2/anker_767_bluetooth_le/

And i found the topic to send the update command, but i don't know what data i have to publish there.

f"cmd/{mqtt['data']['app_name']}/{myapi.devices[device]['device_pn']}/{myapi.devices[device]['device_sn']/req"

[thomluther]

You are probably right. After base64decode of the data field it shows the SN and couple of bytes, starting with a marker of 'ff09' like the BT hex fields.
So it will be a matter of reverse engineering each specific data field, which might be different for each device type...
You may find this helpful from the BT project.

[thomluther]

Hi @rschoebel
Die you find the payload that can trigger updates with the publish command?
If you can sniff that traffic, it may be an unencrypted binary payload for the device, like for the subscribed topics...

BTW, I'm going to implement an MQTT client and a monitor tool into the library. My goal is to provide a standardized way to decode the binary payload in a consumable way to simplify mapping of fields to device model values.
Since each device will have different field mappings, that will have to be done and documented by the community for each of the devices. The good part is that this format seems to be identical with the BT transmission format. So sharing with other projects is possible.

I think first step is to understand which data can be taken from the MQTT stats that are not available via Api yet. Those could also be merged with the HA integration.
Using more frequent data updates however may be difficult, even after finding the required publish payload. This is because device data may be just a subset of cloud Api data, since some of the system related values are probably just calculated in the cloud and not provided by the device itself. Therefore frequent updates for a subset of HA Integration values will become a mess, since the values don't correlate anymore.

[rschoebel]

Hi, i had no time this weekend, to analyze the traffic.
and i hope i have enough data saved,
because i left my anker device at another place connected to wifi,
only to realize the F Series will seemingly drop wifi after some time.

[rschoebel]

I had some time today to analyze my data, and the main message format seems identical to the Bluetooth one.

and i think i found some data if the bytes[2:9] -> 'f8 01 03 01 0f 04 05'
0xa5 = Temp in °C
0xb9 = Output in Watt

data types, current assumptions first byte of the data fields:
00 -> ASCII / String
01 -> Byte
02 -> Int?
03 -> Unknown
04 -> Bitmask?
05 -> Unknown

i need to invest more time, and collect more data

import base64
import binascii

coded_string = '''<input mqtt  payload data>'''
bytes=base64.b64decode(coded_string)


def decode(pos):
    start=int(pos)
    section=hex(bytes[start])
    length=bytes[start+1]
    data=start + 2
    next=data + length
    print("Section: ",section)
    #print("Start: ",start)
    print("Length:  ",hex(length))
    #print("next: ",next)
    #in the first section the length dose not include the section and length field
    if (section == "0xff"):
        next = next - 2
    if(length > 1):
        print("Data:\t ",binascii.hexlify(bytes[data:next], ' '))
    else:
        print("Data:\t ",int.from_bytes(bytes[data:next]))
    return next

print("=========================")
pos = 0
while len(bytes) > pos:
    
    if pos + 1 >= len(bytes):
        print("Checksum: ", bytes[pos])
        break
    pos = decode(pos)

[rschoebel]

And another update, i found a way to trigger mqtt updates.

updated code below.

next part wire data to sqlite, in try to identify more fields

#!/usr/bin/env python
"""Example exec for mqtt."""

import asyncio
from datetime import datetime
import json
import logging
from pathlib import Path

from aiohttp import ClientSession
from api import api  # pylint: disable=no-name-in-module
from api.apitypes import SolixParmType  # pylint: disable=no-name-in-module
import common

import paho.mqtt.client as mqtt
import ssl
import time
import tempfile
import os

import re

_LOGGER: logging.Logger = logging.getLogger(__name__)
_LOGGER.addHandler(logging.StreamHandler())
# _LOGGER.setLevel(logging.DEBUG)    # enable for detailed API output
CONSOLE: logging.Logger = common.CONSOLE

def _out(jsondata):
    CONSOLE.info(json.dumps(jsondata, indent=2))


def on_connect(client, userdata, flags, rc):
    """Callback for when the client receives a CONNACK response from the server."""
    if rc == 0:
        print("Connected successfully to MQTT broker")
    else:
        print(f"Failed to connect, return code {rc}")

def on_message(client, userdata, msg):
    """Callback for when a PUBLISH message is received from the server."""
    print(f"Received message: {msg.payload.decode()} on topic: {msg.topic}")

def on_disconnect(client, userdata, rc):
    """Callback for when the client disconnects from the server."""
    print("Disconnected from MQTT broker")

def create_mqtt_client(cid,cert,key,ca):
    """Create and configure MQTT client with SSL/TLS certificates from strings."""

    # Create client instance
    client = mqtt.Client(client_id=cid, clean_session=True)

    # Set callbacks
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect

    # Create temporary files for certificates
    ca_cert_file = None
    client_cert_file = None
    client_key_file = None

    try:
        # Create temporary file for CA certificate
        ca_cert_file = tempfile.NamedTemporaryFile(mode='w', suffix='.crt', delete=False)
        ca_cert_file.write(ca)
        ca_cert_file.flush()

        # Create temporary file for client certificate
        client_cert_file = tempfile.NamedTemporaryFile(mode='w', suffix='.crt', delete=False)
        client_cert_file.write(cert)
        client_cert_file.flush()

        # Create temporary file for client private key
        client_key_file = tempfile.NamedTemporaryFile(mode='w', suffix='.key', delete=False)
        client_key_file.write(key)
        client_key_file.flush()

        # Configure SSL/TLS using temporary files
        client.tls_set(ca_certs=ca_cert_file.name,
                       certfile=client_cert_file.name,
                       keyfile=client_key_file.name,
                       cert_reqs=ssl.CERT_REQUIRED,
                       tls_version=ssl.PROTOCOL_TLS,
                       ciphers=None)

        # Alternative method using SSL context (commented out):
        # context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        # context.check_hostname = False
        # context.load_verify_locations(ca_cert_file.name)
        # context.load_cert_chain(client_cert_file.name, client_key_file.name)
        # client.tls_set_context(context)

        # Store temp file references for cleanup
        client._temp_cert_files = [ca_cert_file.name, client_cert_file.name, client_key_file.name]

    except Exception as e:
        # Clean up temp files if there was an error
        for temp_file in [ca_cert_file, client_cert_file, client_key_file]:
            if temp_file and os.path.exists(temp_file.name):
                temp_file.close()
                os.unlink(temp_file.name)
        raise e
    finally:
        # Close file handles (files remain on disk for SSL to use)
        if ca_cert_file:
            ca_cert_file.close()
        if client_cert_file:
            client_cert_file.close()
        if client_key_file:
            client_key_file.close()

    return client

def cleanup_temp_files(client):
    """Clean up temporary certificate files."""
    if hasattr(client, '_temp_cert_files'):
        for temp_file_path in client._temp_cert_files:
            try:
                if os.path.exists(temp_file_path):
                    os.unlink(temp_file_path)
            except Exception as e:
                print(f"Warning: Could not delete temp file {temp_file_path}: {e}")

async def main() -> None:
    """Create the aiohttp session and run the example."""

    CONSOLE.info("Testing Solix API:")
    async with ClientSession() as websession:
        myapi = api.AnkerSolixApi(
            common.user(),
            common.password(),
            common.country(),
            websession,
            _LOGGER,
        )
        # show login response
        new = await myapi.async_authenticate(
            restart=True
        )  # enforce new login data from server
        new = (
            await myapi.async_authenticate()
        )  # receive new or load cached login data
        if new:
            CONSOLE.info("Received Login response:")
        else:
            CONSOLE.info("Cached Login response:")
        _out(
            myapi.apisession._login_response  # noqa: SLF001
        )  # show used login response for API requests
        mqtt = await myapi.apisession.request("post","v1/openapi/devicemanage/get_user_mqtt_info")
        _out(
            mqtt
        )
        await myapi.update_device_details()
        CONSOLE.info("Device Overview:")
        device = next(iter(myapi.devices))
        TOPIC = f"dt/{mqtt['data']['app_name']}/{myapi.devices[device]['device_pn']}/{myapi.devices[device]['device_sn']}/param_info"
        TOPIC_SEND = f"cmd/{mqtt['data']['app_name']}/{myapi.devices[device]['device_pn']}/{myapi.devices[device]['device_sn']}/req"
        client = create_mqtt_client(mqtt["data"]["thing_name"],mqtt["data"]["certificate_pem"],mqtt["data"]["private_key"],mqtt["data"]["aws_root_ca1_pem"])
        print(f"Connecting to MQTT broker at {mqtt['data']['endpoint_addr']}:8883")
        client.connect(mqtt["data"]["endpoint_addr"], 8883, 20)
        client.subscribe(TOPIC)
        # Start the loop to process network traffic and callbacks
        client.loop_start()
        try:
            while True:
                f = {
                    "client_id": f"android-{mqtt['data']['app_name']}-{mqtt['data']['user_id']}-{mqtt['data']['certificate_id']}",
                    "device_sn": myapi.devices[device]['device_sn'],
                    "device_pn": myapi.devices[device]['device_pn'],
                    "data": "/wkfAAMADwBXoQEiogIBAaMFAywBAAD+BQPI17ZoIQ==",
                    "account_id": mqtt['data']['user_id'],
                    "timestamp": int(time.time())
                }
                data = '''                
{{
  "head": {{
    "version": "1.0.0.1",
    "client_id": "{client_id}",
    "sess_id": "5681-3252",
    "msg_seq": "1",
    "seed": "1",
    "timestamp": {timestamp},
    "cmd_status": 2,
    "cmd": 17,
    "sign_code": 1,
    "device_pn": "{device_pn}",
    "device_sn": "{device_sn}"
  }},
  "payload": "{{\\"account_id\\": \\"{account_id}\\",\\"device_sn\\": \\"{device_sn}\\",\\"data\\": \\"{data}\\"}}
}}'''
                client.publish(TOPIC_SEND, re.sub(r"[\n\t\s]*", "", data.format(**f)))
                time.sleep(60)
        except KeyboardInterrupt:
            print("\nDisconnecting...")
            client.loop_stop()
            client.disconnect()
            cleanup_temp_files(client)

# run async main
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as err:  # pylint: disable=broad-exception-caught  # noqa: BLE001
        CONSOLE.exception("%s: %s", type(err), err)

[thomluther]

I just uploaded the latest updates to main branch
there is now a mqtt module and the new mqtt_monitor tool that you can use. You can optionally dump the decoded output to files for later review and documentation.
It is not documented yet because there will still be changes.
But I have data classes to structure the binary data and help to print various decoded formats.
To get this in a generic usage model, there is a mapping constant SOLIXMQTTMAP that will have to be expanded with field names for the validated Hex fields in the response. This is different for each model and each message type (param_info or stat info). So the mapping definition will get quite nested and large over time, but I see no other way how to structure a common decoding and mapping of the binary fields to utilize data in a common way. With such a mapping constant, code can be used to traverse reveived topic data once model and message pattern is defined, in order to pick the correct byte fields and convert the value as required for usage. The mapping will also need to include conversion parameters if required, like a factor for certain values.
I defined the field format in DeviceHexDataTypes Enum.
Format '03' is pretty flexible and can be anything. Its used for firmware for example, where each byte can stand for a version digit, or the whole byte is the version, e.g. int 215 is for Version 2.1.5 in my case. So you need to think around the corned when decoding fields.

I can see if I can also build in the update topic payload that you found, so this might be helpful for general purpose.

The question is whether you are able to sniff payloads for device control and apply settings. How is the payload structured and is it just sending the 'correct' field name with format and value like in the binary response?

[thomluther]

Here is an example how the decoding may look like. Any existing mappings will be added to the decoding output if the model is passed to the dataclass that structures the hexdata:

------------- Smartmeter / A17X7 / 0405 / Header -------------
ff 09   : 2 Byte Anker Solix message marker (supposed 'ff 09')
64 00   : 2 Byte total message length (100) in Bytes (Little Endian format)
03 01 0f: 3 Byte fixed message pattern (supposed `03 01 0f`)
04 05   : 2 Byte message type pattern (varies per device model and message type)
        : 1 Byte optional message increment (  0)
-- Fields --|- Value (Hex/Decode Options)---------------------------------------
Fld Len Typ   uIntLe/var     sIntLe      floatLe     dblLe/4int 
a1 01  --   34
└->   1 unk            52           52                          
a2 11  00   XX:XX:XX:XX:XX:XX:XX:XX:XX:XX:XX:XX:XX:XX:XX:XX
└->  17 str  b'XXXXXXXXXXXXXXXX'                                        --> device_sn
a3 02  01   00
└->   2 ui              0            0                          
a5 02  01   00
└->   2 ui              0            0                          
a6 05  03   00:00:00:28
└->   5 var       0;10240    671088640         0.00     0;0;0;40 --> fw_device
a7 05  03   00:00:00:28
└->   5 var       0;10240    671088640         0.00     0;0;0;40
a8 05  03   00:00:00:00
└->   5 var           0;0            0         0.00      0;0;0;0 --> export_power
a9 05  03   00:00:00:00
└->   5 var           0;0            0         0.00      0;0;0;0 --> import_power
aa 05  03   86:11:00:00
└->   5 var        4486;0         4486         0.00   134;17;0;0 --> export_energy
ab 05  03   7d:16:01:00
└->   5 var        5757;1        71293         0.00   125;22;1;0 --> import_energy
ac 05  03   00:00:00:00
└->   5 var           0;0            0         0.00      0;0;0;0
ad 05  03   00:00:00:00
└->   5 var           0;0            0         0.00      0;0;0;0
fd 02  01   01
└->   2 ui              1            1                          
--------------------------------------------------------------------------------

If you have a hex string, you can simply do this for decoding.

            hexstr = "ff09......."
            data = apitypes.DeviceHexData(model="A17X7",hexbytes=hexstr)
            CONSOLE.info(str(data))
            CONSOLE.info(data.decode())

For general purpose use I'm just concerned how Anker implements changes. If they add fields in between which will shift the namings, the whole mapping definitions will become garbage...

[thomluther]

Thanks for the publish data. I'm trying to figure out how to compose it in general to add this to the mqtt module.
I think an easier way to format the data is using json methods, this avoids usage of re.

            f1 = {
                "head": {
                    "version": "1.0.0.1",
                    "client_id": clientid,
                    "sess_id": "5681-3252",  # => can this be fix, or can it be obtained from client connection?
                    "msg_seq": 1,
                    "seed": 1,
                    "timestamp": int(datetime.now().timestamp()),
                    "cmd_status": 2,
                    "cmd": 17,
                    "sign_code": 1,
                    "device_pn": devicePn,
                    "device_sn": deviceSn,
                },
                "payload": json.dumps({
                    "account_id": accountId,
                    "device_sn": deviceSn,
                    "data": "/wkfAAMADwBXoQEiogIBAaMFAywBAAD+BQPI17ZoIQ==",
                },separators=(',', ':'))
            }
            client.publish(TOPIC_SEND, json.dumps(f1,separators=(',', ':')))

But when I look to your hex data, I'm puzzeled what the fields actually mean. I guess the hex data must be composed too and base64 encoded. Any clue how to compose the field values?

------------------------------------ Header ------------------------------------
ff 09   : 2 Byte Anker Solix message marker (supposed 'ff 09')
1f 00   : 2 Byte total message length (31) in Bytes (Little Endian format)
03 00 0f: 3 Byte fixed message pattern (supposed `03 01 0f`)
00 57   : 2 Byte message type pattern (varies per device model and message type)
        : 1 Byte optional message increment (  0)
-- Fields --|- Value (Hex/Decode Options)---------------------------------------
Fld Len Typ    uIntLe/var     sIntLe      floatLe     dblLe/4int 
a1  01  --  22
└->   1 unk           34           34
a2  02  01  01
└->   2 ui            1            1
a3  05  03  2c:01:00:00
└->   5 var        300;0          300         0.00     44;1;0;0
fe  05  03  c8:d7:b6:68
└->   5 var -10296;26806   1756813256 6907609291015444544618496.00 200;215;182;104
--------------------------------------------------------------------------------

The message pattern is slightly different, OK. But what does the a1-a3 field values mean?
1 could be requested publish interval, 300 the regular publish interval, but 34? Maybe timeout to return to regular interval if no follow on publish received?

The fe field seems to be just a finish pattern which is also in other messages, but I have no clue whether that is fixed or what it means. however it seems to follow the common field structure with name, length, type and 4 bytes of value.

[rschoebel]

as far as i can tell in the messages with the header 03:00:0f:00:57
have always the same data contained
0xa1 -> 0x22
0xa2 -> 1 ; enabled/disable just a hunch
0xa3 -> 300; might be the timeout (5min after the last 03:00:0f:00:57 i'm getting no more 03:01:0f:04:0a (status updates))
0xfe -> is a unix timestamp and is every message

and i had no problems leaving the msg_seq and sess_id fixed.

message header i got till today from byte 4

03:00:0f:00:5a
03:00:0f:00:67
03:01:0f:04:00
03:00:0f:00:40
03:01:0f:04:05
03:00:0f:00:57 (tigger with timeout)
03:00:0f:00:68
03:01:0f:04:0a (status updates)

[4] 0x03 -> seems fixed
[5] 0x01/0x00 -> migth be the dircetion
[6] 0x0f -> seems also fixed
[7] message disabled fields?
[8] message disabled fields?

for packages with header 03:01:0f:04:05
0xa5 = Temp in °C
0xb9 = Output in Watt
0xbd = Grid Connected Output
0xbe = Grid Connected Input
0xd5 = MPPT Maximum Input
0xfe = timestamp

[thomluther]

have always the same data contained
0xa1 -> 0x22
0xa2 -> 1 ; enabled/disable just a hunch
0xa3 -> 300; might be the timeout (5min after the last 03:00:0f:00:57 i'm getting no more 03:01:0f:04:0a (status updates))
0xfe -> is a unix timestamp and is every message

OK, makes sense. Confirmation for 0xa3 field. It is the duration in seconds how long publish messages are send. I used the update command just once, and messages stopped after 300 seconds. So we can probably influence, how long frequent updates should be done. I tested updates with a smartmeter, which is a good candidate for this publish, as they usually do not send data frequently, and only have one topic. Here I see, that the topic gets updates every 5 seconds, which stops after the requested 300 seconds.

If you start documenting the field mappings, we need to make that in a common format per device model and the 2 byte message type pattern. For the field names I will still review them and make them consistent with names used in the power Api so far, to avoid too much different names for the same thing.
I also enhanced the mqtt session to compose the publish call automatically. Still need to implement a method to compose the hex data for the publish, which might be useful if someone finds the structure/fields for device controls.
I plan to use the update publish as default, where duration could be used as parameter.
The timestamp can also be automatically composed, not sure if that is relevant however, since command is also accepted with old timestamp...

[thomluther]

OK, I uploaded latest updates. Further enhancements and fixed to the dataclasses for the MQTT hex data. They can now be used also to compose headers, fields etc.
So the monitor can now compose the update trigger command as following:

                    message, response = mqtt_session.publish(
                        deviceDict=device_selected,
                        hexbytes=mqtt_session.get_command_data(
                            command="update_trigger", parameters={"timeout": 120}
                        ),
                    )
                    CONSOLE.info(f"Published message: {response!s}\n{message!s}")

I kept the mqtt_session.get_command_data quite generic. So once new hexdata for other commands are known, they can be implemented accordingly. Here is an example how the hexdata is built, showing also different methods to build the data fields:

        if command == "update_trigger":
            hexdata = DeviceHexData(msg_header=DeviceHexDataHeader(cmd_msg="0057"))
            hexdata.update_field(DeviceHexDataField(hexbytes="a10122"))
            hexdata.update_field(DeviceHexDataField(hexbytes="a2020101"))
            hexdata.update_field(
                DeviceHexDataField(
                    f_name=bytes.fromhex("a3"),
                    f_type=DeviceHexDataTypes.var.value,
                    f_value=int(parameters.get("timeout") or 60).to_bytes(
                        length=4, byteorder="little"
                    ),
                )
            )
            hexdata.add_timestamp_field()
        if hexdata:
            self._logger.info("Generated hexdata for device mqtt command '%s':\n%s", command, hexdata.hex(":"))
            return hexdata.hex()

Any value parameters could be passed with the mapping, and used properly in the get_command_data method.
I'm still thinking about a method to convert any value type automatically into the required bytes based on field format and other common conversion parameters that might required and may be defined in the message mapping constant. Generic conversion methods would be usefull in either direction, for parsing the values as well as composing the bytes for command parameters.

[thomluther]

Making more progress on MQTT integration, but there is still lot to do on my list:

• Discovered another field format 0x06 for various byte length fields, which can be mix of strings and byte values. This is used for Solarbank Expansion batteries, showing serial, temp, Soc etc. I will need to add this format to the decoder and think of a proper description format in the field Mapper, so fields can be extracted properly.
• I added a background executer to the mqtt session class. This can be used to run the update trigger and subscription changes in background. It will also be used in the monitor to simplify its code
• I plan to enhance the mqtt monitor with more options upon key press events.
  • This will also help to test various code routines. Possible actions are
  • [s]ubscribe to device root topic
  • [u]nsubscribe (which will also stop the update trigger publish for the device, if there are no more subscribed topics left for the device),
  • [t]oggle through individual topic subscriptions as they are found from root subscription (still need to implement code to track found message topics)
  • [v]iewer toggle, to switch output to defined value mapping extracts with a clear screen refresh, similar to normal monitor display. This requires a value extractor method and change of the message callback routine. It will be helpful to test proper value extractions.
• Implement method to (recursively) get the value from a HexDataField. Complex field formats will have to be described in the field mapping with base field formats and start/length of bytes to extract only a subset of bytes with described base format
• Implement a generic mqtt data extractor method, that can be used for the value extraction, based on the field mapper description.
  • It should extract all defined mappings and return a dict with defined value name and value
  • Its then up on data wrapper methods how and which of the extracted fields to me matched eventually with Api data, that may reflect another time stamp and therefore does not correlate 'system relevant' values
• Document the whole MQTT approach, mqtt_monitor examples, how to document field mappings etc....

I will upload a new main branch update once I made for progress on the first part of the list. For the time being, field mapping documentation is easiest with output from the mqtt_monitor. The model, message pattern and clear description of required byte conversion must be provided for non basic field formats. Ideally with an example value if conversion factors are needed.

[thomluther]

Hi @rschoebel I'm done with most of the items I wanted to include and updated the main branch.
I also added documentation and decoding guidelines.
You may now use the mqtt_monitor tool and describe your device message fields as I did for my device.

In the future, I will see how to make use of the MQTT data and how to integrate them into the Api cache structures.

[rschoebel]

Hi,
i'll test and document as soon as i'm able to.

and at this point in time, i think all change requests to the device are done via the api and not mqtt.
but i need to fix my testing setup, and permanently setup my anker device.

[thomluther]

and at this point in time, i think all change requests to the device are done via the api and not mqtt.
but i need to fix my testing setup, and permanently setup my anker device.

That seems weird if the device is not configured into a system. Someone else did endpoint sniffing with testing the Smart plug changes, and none of them used the power server Api, but only MQTT server (you can check that already with outgoing connections at the time of change on your mobile.
There could be Api calls that goes to the power cloud and inturn are published to the MQTT server accordingly, since this is the only interface to control the device remotely. I guess each device is subscribed to its command topic and any MQTT client can publish commands to that topic which is listened by the device (where the update trigger is also sent).

Using the Api for device commands may be easier, but you need to know the endpoints and parameters which is encrypted by the App.
The MQTT commands are not encrypted, but have to be decoded for each and every device type and command, which is difficult as well. Based on the decoding, you need to build code that can compose the command hex data with parameters as required.

[Rsclub22]

and at this point in time, i think all change requests to the device are done via the api and not mqtt.
but i need to fix my testing setup, and permanently setup my anker device.

That seems weird if the device is not configured into a system. Someone else did endpoint sniffing with testing the Smart plug changes, and none of them used the power server Api, but only MQTT server (you can check that already with outgoing connections at the time of change on your mobile. There could be Api calls that goes to the power cloud and inturn are published to the MQTT server accordingly, since this is the only interface to control the device remotely. I guess each device is subscribed to its command topic and any MQTT client can publish commands to that topic which is listened by the device (where the update trigger is also sent).

Using the Api for device commands may be easier, but you need to know the endpoints and parameters which is encrypted by the App. The MQTT commands are not encrypted, but have to be decoded for each and every device type and command, which is difficult as well. Based on the decoding, you need to build code that can compose the command hex data with parameters as required.

Did anyone try to get anything from the app itself? Otherwise i could try to get something from my rooted Device and look into the endpoints.

[thomluther]

Did anyone try to get anything from the app itself? Otherwise i could try to get something from my rooted Device and look into the endpoints.

If you get to a point where you can see the App endpoints and payload, that would be a good start. It does not help much for Api calls, but the MQTT payload is only base64 encoded hex data. So any MQTT packet that is being sent for changes could be decoded to understand the publish data structure. Maybe that is a common structure for any command, but data fields could be completely different on structure per command and device type...
For decoding of the mqtt payload, you can use this example