summaryrefslogtreecommitdiff
path: root/include/py/homekit/openwrt/openwrt.py
blob: d5f949c42e474c5e6da47ab96080acbc6f8dab88 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
import requests
import logging

from datetime import datetime
from collections import namedtuple
from urllib.parse import quote_plus
from .config import OpenwrtConfig
from ..modem.config import ModemsConfig

DHCPLease = namedtuple('DHCPLease', 'time, time_s, mac, ip, hostname')
_config = OpenwrtConfig()
_modems_config = ModemsConfig()
_logger = logging.getLogger(__name__)


def ipset_list_all() -> list:
    args = ['ipset-list-all']
    args += _modems_config.keys()
    lines = _to_list(_call(args))
    sets = {}
    cur_set = None
    for line in lines:
        if line.startswith('>'):
            cur_set = line[1:]
            if cur_set not in sets:
                sets[cur_set] = []
            continue

        if cur_set is None:
            _logger.error('ipset_list_all: cur_set is not set')
            continue

        sets[cur_set].append(line)

    return sets


def ipset_add(set_name: str, ip: str):
    return _call(['ipset-add', set_name, ip])


def ipset_del(set_name: str, ip: str):
    return _call(['ipset-del', set_name, ip])


def set_upstream(ip: str):
    return _call(['homekit-set-default-upstream', ip])


def get_default_route():
    return _call(['get-default-route'])


def get_dhcp_leases() -> list[DHCPLease]:
    return list(map(lambda item: _to_dhcp_lease(item), _to_list(_call(['dhcp-leases']))))


def _call(arguments: list[str]) -> str:
    url = _get_link(arguments)
    r = requests.get(url)
    r.raise_for_status()
    return r.text.strip()


def _get_link(arguments: list[str]) -> str:
    url = f'http://{_config["ip"]}/cgi-bin/luci/command/{_config["command_id"]}'
    if arguments:
        url += '/'
        url += '%20'.join(list(map(lambda arg: quote_plus(arg.replace('/', '_')), arguments)))
    return url


def _to_list(s: str) -> list:
    return [] if s == '' else s.split('\n')


def _to_dhcp_lease(s: str) -> DHCPLease:
    words = s.split(' ')
    time = int(words.pop(0))
    mac = words.pop(0)
    ip = words.pop(0)
    words.pop()
    hostname = (' '.join(words)).strip()
    if not hostname or hostname == '*':
        hostname = '?'
    return DHCPLease(time=time,
                     time_s=datetime.fromtimestamp(time).strftime('%d %b, %H:%M:%S'),
                     mac=mac,
                     ip=ip,
                     hostname=hostname)