Support for TL-SG1016DE

[andriej]

I've just found the component and connected it to my 5p switch with no problems. There is issue connecting to TL-SG1016DE (smart without PoE).

It does work to get stats via this code tho (found on github). Could you please take a look and see if there's some small fix to be done? [some tweaks done to push data:]

#!/usr/bin/python3
#
# Retrieve port statistics from a TP-Link TL-SG1016DE switch
#
import requests
import sys
import re
import json

from requests.exceptions import HTTPError

# Change to your needs
ipaddress = 'x.x.x.x'
# Default values for the switch
username = 'admin'
password = 'admin'

ha_headers = {
    'Content-type': 'application/json',
    'Authorization': 'Bearer <redacted>'
}

class tplink_stats:
    """Class for retrieving port statistics from a TP-Link switch"""

    # descriptive port labels; can be modified to your installation
    port_label = [
        'Port 1',
        'Port 2',
        'Port 3',
        'Port 4',
        'Port 5',
        'Port 6',
        'Port 7',
        'Port 8',
        'Port 9',
        'Port 10',
        'Port 11',
        'Port 12',
        'Port 13',
        'Port 14',
        'Port 15',
        'Port 16'
    ]
    # Remaining parts do not need to be modified
    port_state = [
        'disabled',
        'enabled'
    ]
    link_speed = [
        '0',
        '0',
        '10',
        '10',
        '100',
        '100',
        '1000'
    ]
    duplex_info = [
        'none',
        'auto',
        'half',
        'full',
        'half',
        'full',
        'full'
    ]

    def __init__(self, ipaddress, username, password, debug):
        self.ipaddress = ipaddress
        self.referer = 'http://' + self.ipaddress + '/Logout.htm'
        self.username = username
        self.password = password
        self.debug = debug

    def login(self):
        url = 'http://' + self.ipaddress + '/logon.cgi'

        try:
            if self.debug:
                print(f'POST {url}')
            response = requests.post(url, headers={'Referer' : self.referer}, data={'username': self.username,'password': self.password, 'logon': 'Login'})

            # If the response was successful, no Exception will be raised
            response.raise_for_status()
        except HTTPError as http_err:
            # For some weird reason the switch responds with 401 if authorisation suceeded
            if (http_err.response.status_code != 401):
                if (self.debug):
                    print(f'POST {url} failed with status {http_err.response.status_code}')
                return
        except Exception as err:
            if (self.debug):
                print(f'POST {url} exception {err}')
            return

        self.referer = url
        return url

    def port_stats(self):
        url = 'http://' + self.ipaddress + '/PortStatisticsRpm.htm'
        try:
            if (self.debug):
                print(f'GET {url}')
            response = requests.get(url, headers={'Referer' : self.referer})

            # If the response was successful, no Exception will be raised
            response.raise_for_status()
        except HTTPError as http_err:
            if (self.debug):
                print(f'GET {url} failed with status {http_err.response.status_code}')
            return
        except Exception as err:
            if (self.debug):
                print(f'GET {url} exception {err}')
            return

        if (self.debug):
            print(f'Body text: {response.text}')
        self.referer = url
        body = response.text.replace('\n','')
        body.replace('\r','')

        m = re.search(r'var all_info = {(.*?)};', body)
        if (m):
            s = re.search(r'state:\[(.*?)\]', m.group(1))
            state = s.group(1).split(',')
            l = re.search(r'link_status:\[(.*?)\]', m.group(1))
            link = l.group(1).split(',')
            p = re.search(r'pkts:\[(.*?)\]', m.group(1))
            pkts = p.group(1).split(',')
            output = []

            for i in range(0, 16):
                output.append({

                    "label":self.port_label[i],
                    "state":self.port_state[int(state[i])],
                    "link":self.link_speed[int(link[i])],
                    "duplex":self.duplex_info[int(link[i])],
                    "txgood":pkts[i * 4],
                    "txbad":pkts[i * 4 + 1],
                    "rxgood":pkts[i * 4 + 2],
                    "rxbad":pkts[i * 4 + 3]
                })

            #print(json.dumps(output))

            ha_data = {"topic": "util/tplink_switch", "retain": 'False',  "payload": json.dumps(output)}

            response = requests.post('http://homeassistant.local.hass.io:8123/api/services/mqtt/publish', headers=ha_headers, data=json.dumps(ha_data)).text

        return url

    def logout(self):
        url = 'http://' + self.ipaddress + '/Logout.htm'
        try:
            if (self.debug):
                print(f'GET {url}')
            response = requests.get(url, headers={'Referer' : self.referer})

            # If the response was successful, no Exception will be raised
            response.raise_for_status()
        except HTTPError as http_err:
            # 401 Unauthorized is the 'normal' response code here
            if (http_err.response.status_code != 401):
                if (self.debug):
                    print(f'GET {url} failed with status {http_err.response.status_code}')
                return
        except Exception as err:
            if (self.debug):
                print(f'GET {url} exception {err}')
            return

        return url

t = tplink_stats(ipaddress, username, password, 0)
login_url = t.login()
if login_url:
    t.port_stats()
    t.logout()

WebUI is pretty similiar to SG105E.

Of course best would be to implement the native protocol, I'll try to help if needed.

[cyberdreamandroid]

Would be great to add port statistics for this device. Currently I can only see if a port is linked or not.

[andriej]

I cannot see at all, so we have to be on different firmwares