summaryrefslogtreecommitdiff
path: root/src/home/api
diff options
context:
space:
mode:
Diffstat (limited to 'src/home/api')
-rw-r--r--src/home/api/__init__.py11
-rw-r--r--src/home/api/__init__.pyi4
-rw-r--r--src/home/api/errors/__init__.py1
-rw-r--r--src/home/api/errors/api_response_error.py28
-rw-r--r--src/home/api/types/__init__.py6
-rw-r--r--src/home/api/types/types.py33
-rw-r--r--src/home/api/web_api_client.py225
7 files changed, 0 insertions, 308 deletions
diff --git a/src/home/api/__init__.py b/src/home/api/__init__.py
deleted file mode 100644
index 782a61e..0000000
--- a/src/home/api/__init__.py
+++ /dev/null
@@ -1,11 +0,0 @@
-import importlib
-
-__all__ = ['WebAPIClient', 'RequestParams']
-
-
-def __getattr__(name):
- if name in __all__:
- module = importlib.import_module(f'.web_api_client', __name__)
- return getattr(module, name)
-
- raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
diff --git a/src/home/api/__init__.pyi b/src/home/api/__init__.pyi
deleted file mode 100644
index 1b812d6..0000000
--- a/src/home/api/__init__.pyi
+++ /dev/null
@@ -1,4 +0,0 @@
-from .web_api_client import (
- RequestParams as RequestParams,
- WebAPIClient as WebAPIClient
-)
diff --git a/src/home/api/errors/__init__.py b/src/home/api/errors/__init__.py
deleted file mode 100644
index efb06aa..0000000
--- a/src/home/api/errors/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .api_response_error import ApiResponseError
diff --git a/src/home/api/errors/api_response_error.py b/src/home/api/errors/api_response_error.py
deleted file mode 100644
index 85d788b..0000000
--- a/src/home/api/errors/api_response_error.py
+++ /dev/null
@@ -1,28 +0,0 @@
-from typing import Optional, List
-
-
-class ApiResponseError(Exception):
- def __init__(self,
- status_code: int,
- error_type: str,
- error_message: str,
- error_stacktrace: Optional[List[str]] = None):
- super().__init__()
- self.status_code = status_code
- self.error_message = error_message
- self.error_type = error_type
- self.error_stacktrace = error_stacktrace
-
- def __str__(self):
- def st_formatter(line: str):
- return f'Remote| {line}'
-
- s = f'{self.error_type}: {self.error_message} (HTTP {self.status_code})'
- if self.error_stacktrace is not None:
- st = []
- for st_line in self.error_stacktrace:
- st.append('\n'.join(st_formatter(st_subline) for st_subline in st_line.split('\n')))
- s += '\nRemote stacktrace:\n'
- s += '\n'.join(st)
-
- return s
diff --git a/src/home/api/types/__init__.py b/src/home/api/types/__init__.py
deleted file mode 100644
index 9f27ff6..0000000
--- a/src/home/api/types/__init__.py
+++ /dev/null
@@ -1,6 +0,0 @@
-from .types import (
- BotType,
- TemperatureSensorDataType,
- TemperatureSensorLocation,
- SoundSensorLocation
-)
diff --git a/src/home/api/types/types.py b/src/home/api/types/types.py
deleted file mode 100644
index 981e798..0000000
--- a/src/home/api/types/types.py
+++ /dev/null
@@ -1,33 +0,0 @@
-from enum import Enum, auto
-
-
-class BotType(Enum):
- INVERTER = auto()
- PUMP = auto()
- SENSORS = auto()
- ADMIN = auto()
- SOUND = auto()
- POLARIS_KETTLE = auto()
- PUMP_MQTT = auto()
- RELAY_MQTT = auto()
-
-
-class TemperatureSensorLocation(Enum):
- BIG_HOUSE_1 = auto()
- BIG_HOUSE_2 = auto()
- BIG_HOUSE_ROOM = auto()
- STREET = auto()
- DIANA = auto()
- SPB1 = auto()
-
-
-class TemperatureSensorDataType(Enum):
- TEMPERATURE = auto()
- RELATIVE_HUMIDITY = auto()
-
-
-class SoundSensorLocation(Enum):
- DIANA = auto()
- BIG_HOUSE = auto()
- SPB1 = auto()
-
diff --git a/src/home/api/web_api_client.py b/src/home/api/web_api_client.py
deleted file mode 100644
index 6677182..0000000
--- a/src/home/api/web_api_client.py
+++ /dev/null
@@ -1,225 +0,0 @@
-import requests
-import json
-import threading
-import logging
-
-from collections import namedtuple
-from datetime import datetime
-from enum import Enum, auto
-from typing import Optional, Callable, Union, List, Tuple, Dict
-from requests.auth import HTTPBasicAuth
-
-from .errors import ApiResponseError
-from .types import *
-from ..config import config
-from ..util import stringify
-from ..media import RecordFile, MediaNodeClient
-
-logger = logging.getLogger(__name__)
-
-
-RequestParams = namedtuple('RequestParams', 'params, files, method')
-
-
-class HTTPMethod(Enum):
- GET = auto()
- POST = auto()
-
-
-class WebAPIClient:
- token: str
- timeout: Union[float, Tuple[float, float]]
- basic_auth: Optional[HTTPBasicAuth]
- do_async: bool
- async_error_handler: Optional[Callable]
- async_success_handler: Optional[Callable]
-
- def __init__(self, timeout: Union[float, Tuple[float, float]] = 5):
- self.token = config['api']['token']
- self.timeout = timeout
- self.basic_auth = None
- self.do_async = False
- self.async_error_handler = None
- self.async_success_handler = None
-
- if 'basic_auth' in config['api']:
- ba = config['api']['basic_auth']
- col = ba.index(':')
-
- user = ba[:col]
- pw = ba[col+1:]
-
- logger.debug(f'enabling basic auth: {user}:{pw}')
- self.basic_auth = HTTPBasicAuth(user, pw)
-
- # api methods
- # -----------
-
- def log_bot_request(self,
- bot: BotType,
- user_id: int,
- message: str):
- return self._post('log/bot_request/', {
- 'bot': bot.value,
- 'user_id': str(user_id),
- 'message': message
- })
-
- def log_openwrt(self,
- lines: List[Tuple[int, str]],
- access_point: int):
- return self._post('log/openwrt/', {
- 'logs': stringify(lines),
- 'ap': access_point
- })
-
- def get_sensors_data(self,
- sensor: TemperatureSensorLocation,
- hours: int):
- data = self._get('sensors/data/', {
- 'sensor': sensor.value,
- 'hours': hours
- })
- return [(datetime.fromtimestamp(date), temp, hum) for date, temp, hum in data]
-
- def add_sound_sensor_hits(self,
- hits: List[Tuple[str, int]]):
- return self._post('sound_sensors/hits/', {
- 'hits': stringify(hits)
- })
-
- def get_sound_sensor_hits(self,
- location: SoundSensorLocation,
- after: datetime) -> List[dict]:
- return self._process_sound_sensor_hits_data(self._get('sound_sensors/hits/', {
- 'after': int(after.timestamp()),
- 'location': location.value
- }))
-
- def get_last_sound_sensor_hits(self, location: SoundSensorLocation, last: int):
- return self._process_sound_sensor_hits_data(self._get('sound_sensors/hits/', {
- 'last': last,
- 'location': location.value
- }))
-
- def recordings_list(self, extended=False, as_objects=False) -> Union[List[str], List[dict], List[RecordFile]]:
- files = self._get('recordings/list/', {'extended': int(extended)})['data']
- if as_objects:
- return MediaNodeClient.record_list_from_serialized(files)
- return files
-
- def inverter_get_consumed_energy(self, s_from: str, s_to: str):
- return self._get('inverter/consumed_energy/', {
- 'from': s_from,
- 'to': s_to
- })
-
- def inverter_get_grid_consumed_energy(self, s_from: str, s_to: str):
- return self._get('inverter/grid_consumed_energy/', {
- 'from': s_from,
- 'to': s_to
- })
-
- @staticmethod
- def _process_sound_sensor_hits_data(data: List[dict]) -> List[dict]:
- for item in data:
- item['time'] = datetime.fromtimestamp(item['time'])
- return data
-
- # internal methods
- # ----------------
-
- def _get(self, *args, **kwargs):
- return self._call(method=HTTPMethod.GET, *args, **kwargs)
-
- def _post(self, *args, **kwargs):
- return self._call(method=HTTPMethod.POST, *args, **kwargs)
-
- def _call(self,
- name: str,
- params: dict,
- method: HTTPMethod,
- files: Optional[Dict[str, str]] = None):
- if not self.do_async:
- return self._make_request(name, params, method, files)
- else:
- t = threading.Thread(target=self._make_request_in_thread, args=(name, params, method, files))
- t.start()
- return None
-
- def _make_request(self,
- name: str,
- params: dict,
- method: HTTPMethod = HTTPMethod.GET,
- files: Optional[Dict[str, str]] = None) -> Optional[any]:
- domain = config['api']['host']
- kwargs = {}
-
- if self.basic_auth is not None:
- kwargs['auth'] = self.basic_auth
-
- if method == HTTPMethod.GET:
- if files:
- raise RuntimeError('can\'t upload files using GET, please use me properly')
- kwargs['params'] = params
- f = requests.get
- else:
- kwargs['data'] = params
- f = requests.post
-
- fd = {}
- if files:
- for fname, fpath in files.items():
- fd[fname] = open(fpath, 'rb')
- kwargs['files'] = fd
-
- try:
- r = f(f'https://{domain}/{name}',
- headers={'X-Token': self.token},
- timeout=self.timeout,
- **kwargs)
-
- if not r.headers['content-type'].startswith('application/json'):
- raise ApiResponseError(r.status_code, 'TypeError', 'content-type is not application/json')
-
- data = json.loads(r.text)
- if r.status_code != 200:
- raise ApiResponseError(r.status_code,
- data['error'],
- data['message'],
- data['stacktrace'] if 'stacktrace' in data['error'] else None)
-
- return data['response'] if 'response' in data else True
- finally:
- for fname, f in fd.items():
- # logger.debug(f'closing file {fname} (fd={f})')
- try:
- f.close()
- except Exception as exc:
- logger.exception(exc)
- pass
-
- def _make_request_in_thread(self, name, params, method, files):
- try:
- result = self._make_request(name, params, method, files)
- self._report_async_success(result, name, RequestParams(params=params, method=method, files=files))
- except Exception as e:
- logger.exception(e)
- self._report_async_error(e, name, RequestParams(params=params, method=method, files=files))
-
- def enable_async(self,
- success_handler: Optional[Callable] = None,
- error_handler: Optional[Callable] = None):
- self.do_async = True
- if error_handler:
- self.async_error_handler = error_handler
- if success_handler:
- self.async_success_handler = success_handler
-
- def _report_async_error(self, *args):
- if self.async_error_handler:
- self.async_error_handler(*args)
-
- def _report_async_success(self, *args):
- if self.async_success_handler:
- self.async_success_handler(*args) \ No newline at end of file
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-21 20:06:25 +0000