diff options
Diffstat (limited to 'bin')
| -rw-r--r-- | bin/web_kbn.py | 90 |
1 files changed, 78 insertions, 12 deletions
diff --git a/bin/web_kbn.py b/bin/web_kbn.py index 8b4ca6f..75437f1 100644 --- a/bin/web_kbn.py +++ b/bin/web_kbn.py @@ -2,16 +2,18 @@ import asyncio import jinja2 import aiohttp_jinja2 +import json import os +import re import __py_include from io import StringIO -from typing import Optional +from typing import Optional, Union from homekit.config import config, AppConfigUnit -from homekit.util import homekit_path +from homekit.util import homekit_path, filesize_fmt, seconds_to_human_readable_string from aiohttp import web from homekit import http -from homekit.modem import ModemsConfig +from homekit.modem import ModemsConfig, E3372, MacroNetWorkType class WebKbnConfig(AppConfigUnit): @@ -49,7 +51,7 @@ def get_css_link(file, version) -> str: def get_head_static() -> str: buf = StringIO() for file in STATIC_FILES: - v = 1 + v = 2 try: q_ind = file.index('?') v = file[q_ind+1:] @@ -64,19 +66,52 @@ def get_head_static() -> str: return buf.getvalue() +def get_modem_data(modem_cfg: dict, get_raw=False) -> Union[dict, tuple]: + cl = E3372(modem_cfg['ip'], legacy_token_auth=modem_cfg['legacy_auth']) + + signal = cl.device_signal + status = cl.monitoring_status + traffic = cl.traffic_stats + + if get_raw: + device_info = cl.device_information + dialup_conn = cl.dialup_connection + return signal, status, traffic, device_info, dialup_conn + else: + network_type_label = re.sub('^MACRO_NET_WORK_TYPE(_EX)?_', '', MacroNetWorkType(int(status['CurrentNetworkType'])).name) + return { + 'type': network_type_label, + 'level': int(status['SignalIcon']) if 'SignalIcon' in status else 0, + 'rssi': signal['rssi'], + 'sinr': signal['sinr'], + 'connected_time': seconds_to_human_readable_string(int(traffic['CurrentConnectTime'])), + 'downloaded': filesize_fmt(int(traffic['CurrentDownload'])), + 'uploaded': filesize_fmt(int(traffic['CurrentUpload'])) + } + + class WebSite(http.HTTPServer): + _modems_config: ModemsConfig + def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) + self._modems_config = ModemsConfig() + aiohttp_jinja2.setup( self.app, - loader=jinja2.FileSystemLoader(homekit_path('web', 'kbn_templates')) + loader=jinja2.FileSystemLoader(homekit_path('web', 'kbn_templates')), + autoescape=jinja2.select_autoescape(['html', 'xml']), ) + env = aiohttp_jinja2.get_env(self.app) + env.filters['tojson'] = lambda obj: json.dumps(obj, separators=(',', ':')) self.app.router.add_static('/assets/', path=homekit_path('web', 'kbn_assets')) self.get('/main.cgi', self.get_index) self.get('/modems.cgi', self.get_modems) + self.get('/modems/info.ajx', self.get_modems_ajax) + self.get('/modems/verbose.cgi', self.get_modems_verbose) async def render_page(self, req: http.Request, @@ -99,19 +134,50 @@ class WebSite(http.HTTPServer): title="Home web site") async def get_modems(self, req: http.Request): - mc = ModemsConfig() - print(mc) return await self.render_page(req, 'modems', title='Состояние модемов', - context=dict(modems=ModemsConfig())) + context=dict(modems=self._modems_config)) + + async def get_modems_ajax(self, req: http.Request): + modem = req.query.get('id', None) + if modem not in self._modems_config.getkeys(): + raise ValueError('invalid modem id') + + modem_cfg = self._modems_config.get(modem) + loop = asyncio.get_event_loop() + modem_data = await loop.run_in_executor(None, lambda: get_modem_data(modem_cfg)) + + html = aiohttp_jinja2.render_string('modem_data.j2', req, context=dict( + modem_data=modem_data, + modem=modem + )) + + return self.ok({'html': html}) + + async def get_modems_verbose(self, req: http.Request): + modem = req.query.get('id', None) + if modem not in self._modems_config.getkeys(): + raise ValueError('invalid modem id') + + modem_cfg = self._modems_config.get(modem) + loop = asyncio.get_event_loop() + signal, status, traffic, device, dialup_conn = await loop.run_in_executor(None, lambda: get_modem_data(modem_cfg, True)) + data = [ + ['Signal', signal], + ['Connection', status], + ['Traffic', traffic], + ['Device info', device], + ['Dialup connection', dialup_conn] + ] + + modem_name = self._modems_config.getfullname(modem) + return await self.render_page(req, 'modem_verbose', + title=f'Подробная информация о модеме "{modem_name}"', + context=dict(data=data, modem_name=modem_name)) if __name__ == '__main__': config.load_app(WebKbnConfig) - loop = asyncio.get_event_loop() - # print(config.app_config) - - print(config.app_config['listen_addr'].host) server = WebSite(config.app_config['listen_addr']) server.run() |