# (c) 2020-2024 Martin Wendt and contributors; see https://github.com/mar10/stressor# Licensed under the MIT license: https://www.opensource.org/licenses/mit-license.php""""""importioimportjsonimportloggingimportosimportsocketserverimportwebbrowserfromhttp.serverimportHTTPStatus,SimpleHTTPRequestHandlerfromthreadingimportThreadfromurllibimportparsefromstressorimport__version__# from stressor.util import loggerlogger=logging.getLogger("stressor.monitor")
[docs]classHandler(SimpleHTTPRequestHandler):server_version=("stressor/"+__version__+" "+SimpleHTTPRequestHandler.server_version)# Custom attributes, set by `MonitorServer`:DIRECTORY=Nonerun_manager=Nonedef__init__(self,*args,**kwargs):super().__init__(*args,directory=self.DIRECTORY,**kwargs)
[docs]deflog_request(self,code="-",size="-"):# Overide base implementation (writing to stderr)ifisinstance(code,HTTPStatus)andnot(200<=code.value<400):logger.warning('"%s" %s%s',self.requestline,str(code),str(size))else:logger.debug('"%s" %s%s',self.requestline,str(code),str(size))
[docs]deflog_error(self,format,*args):# Overide base implementation (writing to stderr)logger.error(format,*args)
[docs]deflog_message(self,format,*args):# Overide base implementation (writing to stderr)logger.debug(format,*args)
[docs]classMonitorServer(Thread):""" Run a web server in a separate thread, so it does not block """def__init__(self,run_manager,bind="",port=8081):super().__init__(name="stressor.monitor",daemon=None)Handler.DIRECTORY=os.path.join(os.path.dirname(__file__),"htdocs")Handler.run_manager=run_managerself.run_manager=run_managerself.bind=bindself.port=portself.httpd=None
[docs]defrun(self):withsocketserver.TCPServer((self.bind,self.port),Handler)ashttpd:self.httpd=httpdlogger.info("Monitor serving at http://{}:{}...".format(self.bindor"localhost",self.port))httpd.serve_forever()self.httpd=Nonelogger.info("Monitor server stopped.")
[docs]defopen_browser(self):assertself.bind==""monitor_url=f"http://localhost:{self.port}/"# monitor_url = "http://127.0.0.1:{}/".format(self.port)logger.info(f"Opening web browser at {monitor_url}")webbrowser.open_new_tab(monitor_url)