Minidodo
84a5933490
generified the Worldboss module; removed the spammy-section as of currently, will be readded at a later stage.
228 lines
9.8 KiB
Python
228 lines
9.8 KiB
Python
import datetime
|
|
from typing import Union
|
|
|
|
from core.chat_blob import ChatBlob
|
|
from core.command_alias_service import CommandAliasService
|
|
from core.command_param_types import Any, Const
|
|
from core.decorators import instance, command, event
|
|
from core.igncore import IgnCore
|
|
from core.job_scheduler import JobScheduler
|
|
from core.logger import Logger
|
|
from core.message_hub_service import MessageHubService
|
|
from core.setting_service import SettingService
|
|
from core.setting_types import BooleanSettingType, TextSettingType
|
|
from core.text import Text
|
|
from core.util import Util
|
|
from modules.standard.datanet.ws_controller import WebsocketRelayController
|
|
|
|
|
|
class TimerObj:
|
|
name: str = ""
|
|
spawn: int = 0
|
|
mortal: int = 0
|
|
time: int = 0
|
|
spam: int = 0
|
|
|
|
def __init__(self, obj):
|
|
self.name = obj.name
|
|
self.spawn = obj.spawn
|
|
self.mortal = obj.mortal
|
|
self.time = obj.time
|
|
|
|
def update(self, obj):
|
|
self.spawn = obj.spawn
|
|
self.time = obj.time
|
|
|
|
def __str__(self):
|
|
return f"{self.name} " \
|
|
f"\n\tspawn: {datetime.datetime.fromtimestamp(self.spawn, tz=datetime.timezone.utc).strftime('%H:%M:%S')}" \
|
|
f"\n\tmortal delta: {self.mortal}" \
|
|
f"\n\tmortal: {datetime.datetime.fromtimestamp(self.time, tz=datetime.timezone.utc).strftime('%H:%M:%S')}" \
|
|
f"\n\tImmortal: {self.alive_immortal()}" \
|
|
f"\n\tMortal killable: {self.alive_mortal()}"\
|
|
f"\n\t{self.name} - {self.spawn} - {self.mortal} - {self.time} - {self.spam} - {self.getTime()}"
|
|
|
|
def alive_immortal(self, spam=False) -> bool:
|
|
if spam:
|
|
return self.spam < self.spawn < self.time and self.time > self.getTime()
|
|
return self.time > self.getTime()
|
|
|
|
def alive_mortal(self, spam=False) -> bool:
|
|
if spam:
|
|
return self.spam < self.time < (self.getTime()) and (self.time+60) > self.getTime()
|
|
return self.time < self.getTime()
|
|
|
|
@classmethod
|
|
def getTime(cls) -> int:
|
|
return int(datetime.datetime.now().timestamp())
|
|
|
|
def get_respawn_timer(self):
|
|
if self.name in ["Tarasque", "Loren Warr", "The Hollow Reaper", "Cerubin The Reborn"]:
|
|
return 9*60*60
|
|
elif self.name in ["Vizaresh"]:
|
|
return 17*60*60
|
|
elif self.name in ["Atma"]:
|
|
return 3 * 60 * 60
|
|
elif self.name in ["T.A.M.", "Zaal The Immortal"]:
|
|
return 6 * 60 * 60
|
|
elif self.name in ["Abmouth Indomitus"]:
|
|
return 0
|
|
|
|
|
|
@instance()
|
|
class WorldBossController:
|
|
timer_data = []
|
|
|
|
alerts = [480 * 60, 360 * 60, 240 * 60, 120 * 60, 60 * 60, 60 * 15,
|
|
60 * 5, 60 * 3, 60 * 2, 60, 30, 15, 5, 0]
|
|
jobs = {}
|
|
|
|
def inject(self, registry):
|
|
self.logger = Logger(__name__)
|
|
self.bot: IgnCore = registry.get_instance("bot")
|
|
self.text: Text = registry.get_instance("text")
|
|
self.util: Util = registry.get_instance("util")
|
|
self.command_alias_service: CommandAliasService = registry.get_instance("command_alias_service")
|
|
self.job_scheduler: JobScheduler = registry.get_instance("job_scheduler")
|
|
self.setting_service: SettingService = registry.get_instance("setting_service")
|
|
self.relay_hub: MessageHubService = registry.get_instance("message_hub_service")
|
|
|
|
def pre_start(self):
|
|
self.relay_hub.register_message_source("timers")
|
|
|
|
self.setting_service.register(self.module_name, 'timer_spam', False, BooleanSettingType(),
|
|
"should timers be spammed")
|
|
self.command_alias_service.add_alias("tara", "wb tara")
|
|
self.command_alias_service.add_alias("gaunt", "wb gaunt")
|
|
self.command_alias_service.add_alias("loren", "wb loren")
|
|
self.command_alias_service.add_alias("reaper", "wb reaper")
|
|
|
|
def spam_timer(self, time, mob, msg):
|
|
self.send_warn(msg)
|
|
self.jobs.pop(mob, 0)
|
|
|
|
@event(WebsocketRelayController.WS_RELAY, "save most current local_timers")
|
|
def get_timer(self, _, data):
|
|
# print(data)
|
|
if data.type != "timer":
|
|
return
|
|
local = {}
|
|
for x in data.payload:
|
|
found = False
|
|
for y in self.timer_data:
|
|
if y.name == x.name and not found:
|
|
y.update(x)
|
|
found = True
|
|
if not found:
|
|
self.timer_data.append(TimerObj(x))
|
|
# print("-"*50)
|
|
for x in self.timer_data:
|
|
x: TimerObj
|
|
# print(x.__str__())
|
|
if x.alive_immortal(True):
|
|
x.spam = x.getTime()
|
|
job = self.jobs.pop(x.name, None)
|
|
if job:
|
|
self.job_scheduler.cancel_job(job)
|
|
self.send_warn(f"<highlight>{x.name}</highlight> has spawned.")
|
|
i = self.job_scheduler.scheduled_job(self.spam_timer, x.time-1, mob=x.name, msg=f"<highlight>{x.name}</highlight> is now mortal.")
|
|
self.jobs[x.name] = i
|
|
elif x.alive_mortal(True):
|
|
x.spam = x.getTime()
|
|
job = self.jobs.pop(x.name, None)
|
|
if job:
|
|
self.job_scheduler.cancel_job(job)
|
|
self.send_warn(f"<highlight>{x.name}</highlight> is now mortal.")
|
|
if x.get_respawn_timer() != 0:
|
|
i = self.job_scheduler.scheduled_job(self.spam_timer, x.time+x.get_respawn_timer(), mob=x.name, msg=f"<highlight>{x.name}</highlight> should spawn soon.")
|
|
self.jobs[x.name] = i
|
|
# print("-"*50)
|
|
|
|
def send_warn(self, msg):
|
|
self.relay_hub.send_message("timers", None, f"[<red>WB</red>] {msg}", f"[<red>WB</red>] {msg}")
|
|
|
|
def get_next_alert(self, duration):
|
|
for alert in self.alerts:
|
|
if duration - 1 > alert:
|
|
return duration - alert - 1
|
|
return duration
|
|
|
|
@command(command="wb", params=[Any("worldboss", is_optional=True)],
|
|
description="Displays the next worldboss spawns", access_level="member")
|
|
def show_worldboss(self, request, boss: str):
|
|
return ChatBlob("Next Worldboss spawns", self.getWBTimer())
|
|
|
|
@command(command="wb", params=[Const("settings")],
|
|
description="Worldboss Settings", access_level="moderator", sub_command="mod")
|
|
def wb_settings_cmd(self, request, boss: str):
|
|
pass
|
|
|
|
def show_user(self, timer):
|
|
timer = self.getWBTimer(timer)
|
|
if not timer:
|
|
return "No timers cached; please try again later."
|
|
if timer.spawn < timer.getTime() < timer.time:
|
|
return f"<highlight>{timer.name}</highlight> :: mortal in {self.util.format_time(timer.time-timer.spawn)}"
|
|
elif timer.time < timer.getTime():
|
|
rt = timer.get_respawn_timer()
|
|
|
|
return f"<highlight>{timer.name}</highlight> :: spawn in {self.util.format_time(timer.time + timer.get_respawn_timer())}"
|
|
|
|
def getWBTimer(self, name=None) -> Union[TimerObj, list[TimerObj]]:
|
|
if not name:
|
|
blob = ""
|
|
for x in self.timer_data:
|
|
x: TimerObj
|
|
if x.alive_immortal():
|
|
blob += f" <highlight>{x.name}</highlight> :: mortal in {self.util.format_time(x.time - x.getTime())}\n"
|
|
else:
|
|
rt = x.get_respawn_timer()
|
|
if rt != 0:
|
|
if (x.time + rt) < x.getTime() < (x.time + 2 * rt):
|
|
blob += f" <highlight>{x.name}</highlight> :: should have spawned <highlight>{self.util.format_time(abs(x.getTime() - (x.time + rt)))}</highlight> ago\n"
|
|
continue
|
|
blob += f" <highlight>{x.name}</highlight> :: spawn in {self.util.format_time(x.time + rt - x.getTime())}\n"
|
|
return blob
|
|
for x in self.timer_data:
|
|
if x.name == name:
|
|
return x
|
|
|
|
def timer_alert(self, t, timer):
|
|
pass
|
|
# if self.setting_service.get_value("timer_spam") == "0":
|
|
# return
|
|
# for row in self.timer_data:
|
|
# if row["name"] == timer.name:
|
|
# timer = self.get_spawn(row)
|
|
# if not timer.at:
|
|
# return
|
|
# alert_duration = self.get_next_alert(timer.at - time.time())
|
|
# if timer.at - time.time() < 1:
|
|
# if timer.type == "mortal":
|
|
# self.send_warn(f"<highlight>{timer.name}</highlight> :: is now mortal")
|
|
# self.jobs = [x for x in self.jobs if x['name'] != timer.name]
|
|
# elif timer.type == "spawn":
|
|
# self.send_warn(f"<highlight>{timer.name}</highlight> :: has just spawned")
|
|
# self.jobs = [x for x in self.jobs if x['name'] != timer.name]
|
|
# else:
|
|
# if timer.type == "mortal":
|
|
# self.send_warn(f"<highlight>{timer.name}</highlight> :: mortal in {self.util.format_time(timer.time)}")
|
|
# elif timer.type == "spawn":
|
|
# if alert_duration > 60 * 2:
|
|
# self.send_warn(
|
|
# f"<highlight>{timer.name}</highlight> :: spawn in {self.util.format_time(timer.time)}")
|
|
# for row in self.timer_data:
|
|
# if row["name"] == timer.name:
|
|
# timer = self.get_spawn(row)
|
|
# job_id = self.job_scheduler.scheduled_job(self.timer_alert, time.time() + alert_duration, timer)
|
|
# for job in self.jobs:
|
|
# if job['name'] == timer.name:
|
|
# job['id'] = job_id
|
|
# return
|
|
# self.jobs.append({'name': timer.name, 'id': job_id})
|
|
|
|
# @setting('alert_times', '[480 * 60, 360 * 60, 240 * 60, 120 * 60, 60 * 60, 60 * 15, 60 * 5, 60 * 3, 60 * 2, 60, 30, 15, 5, 0]', 'Worldboss timer spam messages (ETA and actual)')
|
|
# def alert_times(self):
|
|
# return TextSettingType(['[480 * 60, 360 * 60, 240 * 60, 120 * 60, 60 * 60, 60 * 15, 60 * 5, 60 * 3, 60 * 2, 60, 30, 15, 5, 0]'])
|
|
|