85 lines
2.1 KiB
Python
85 lines
2.1 KiB
Python
import argparse
|
|
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from aiohttp import web
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class AlertServer:
|
|
shut_down: bool
|
|
shut_down_event: asyncio.Event
|
|
log: Any
|
|
app: Any
|
|
alert_file_path: Path
|
|
port: int
|
|
|
|
@staticmethod
|
|
async def create_alert_server(alert_file_path: Path, port):
|
|
self = AlertServer()
|
|
self.log = log
|
|
self.shut_down = False
|
|
self.app = web.Application()
|
|
self.shut_down_event = asyncio.Event()
|
|
self.port = port
|
|
routes = [
|
|
web.get("/status", self.status),
|
|
]
|
|
|
|
self.alert_file_path = alert_file_path
|
|
self.app.add_routes(routes)
|
|
|
|
return self
|
|
|
|
async def status(self, request):
|
|
file_text = self.alert_file_path.read_text()
|
|
return web.Response(body=file_text, content_type="text/plain")
|
|
|
|
async def stop(self):
|
|
self.shut_down_event.set()
|
|
|
|
async def run(self):
|
|
runner = web.AppRunner(self.app, access_log=None)
|
|
await runner.setup()
|
|
site = web.TCPSite(runner, None, self.port)
|
|
await site.start()
|
|
|
|
|
|
async def run_and_wait(file_path, port):
|
|
server = await AlertServer.create_alert_server(Path(file_path), port)
|
|
await server.run()
|
|
await server.shut_down_event.wait()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("-file_path", type=str, dest="file_path")
|
|
parser.add_argument("-port", type=str, dest="port")
|
|
|
|
port = None
|
|
file_path = None
|
|
|
|
for key, value in vars(parser.parse_args()).items():
|
|
if key == "port":
|
|
port = value
|
|
elif key == "file_path":
|
|
file_path = value
|
|
else:
|
|
print(f"Invalid argument {key}")
|
|
|
|
if port is None or file_path is None:
|
|
print(
|
|
"Missing arguments, example usage:\n\n"
|
|
"python chia/util/alert_server.py -p 4000 -file_path /home/user/alert.txt\n"
|
|
)
|
|
quit()
|
|
|
|
return asyncio.get_event_loop().run_until_complete(run_and_wait(file_path, port))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|