feat: implement abstract base classes for stats plugins and add example plugin for stats setup and download
All checks were successful
Build and Push Docker Image / build (push) Successful in 12m3s
All checks were successful
Build and Push Docker Image / build (push) Successful in 12m3s
This commit is contained in:
29
utils/plugins_base.py
Normal file
29
utils/plugins_base.py
Normal file
@@ -0,0 +1,29 @@
|
||||
import abc
|
||||
import json
|
||||
import socket
|
||||
import logging
|
||||
from selenium import webdriver
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Abstract Base Classes for Plugins
|
||||
class StatsSetupPlugin(abc.ABC):
|
||||
@abc.abstractmethod
|
||||
def setup_stats(self, driver: webdriver.Remote | webdriver.Chrome, url: str, retries: int = 5) -> webdriver.Remote | webdriver.Chrome:
|
||||
pass
|
||||
|
||||
class StatsDownloadPlugin(abc.ABC):
|
||||
@abc.abstractmethod
|
||||
def download_stats(self, driver: webdriver.Remote | webdriver.Chrome, peersDict: dict, socket_url: str, socket_port: int):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def saveStats(stats: list, socket_url: str, socket_port: int):
|
||||
try:
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
logger.debug(f'Saving stats: {json.dumps(stats, indent=4)}')
|
||||
sock.sendto(json.dumps(stats).encode(), (socket_url, socket_port))
|
||||
sock.close()
|
||||
logger.debug('Sent stats to socket.')
|
||||
except socket.error as e:
|
||||
logger.error(f'Got socket error: {e}')
|
Reference in New Issue
Block a user