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 # No respawn timer? 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 viza") 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, next=None): self.send_warn(msg) self.jobs.pop(mob.name, 0) if next == "spawn": if mob.get_respawn_timer() != 0: i = self.job_scheduler.scheduled_job(self.spam_timer, mob.time + mob.get_respawn_timer(), mob=mob.name, msg=f"{mob.name} should spawn soon.") self.jobs[mob.name] = i @event(WebsocketRelayController.WS_RELAY, "save most current local_timers") def get_timer(self, _, data): # print(data) if data.type != "timer": return 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)) for x in self.timer_data: x: TimerObj 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"{x.name} has spawned.") i = self.job_scheduler.scheduled_job(self.spam_timer, x.time-1, mob=x, msg=f"{x.name} is now mortal.", next="spawn") 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"{x.name} 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, msg=f"{x.name} should spawn soon.") self.jobs[x.name] = i elif (x.time + 60) < x.getTime() < (x.time + x.get_respawn_timer()): i = self.job_scheduler.scheduled_job(self.spam_timer, x.time + x.get_respawn_timer(), mob=x, msg=f"{x.name} should spawn soon.") self.jobs[x.name] = i def send_warn(self, msg): self.relay_hub.send_message("timers", None, f"[WB] {msg}", f"[WB] {msg}") def get_next_alert(self, duration): for alert in self.alerts: if duration - 1 > alert: return duration - alert - 1 return duration def getWorldbossName(self, name): name = name.lower() for x in self.timer_data: x.name: str if x.name.lower().__contains__(name): return x.name return None # if name in ["tara", "tarasque"]: # return "Tarasque" # elif name in ["lw", "loren", "warr", "loren warr"]: # return "Loren Warr" # elif name in ["thr", "reaper", "hollow reaper", "the hollow reaper"]: # return "The Hollow Reaper" @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): if boss: boss = self.getWorldbossName(boss) if boss: return self.getWBTimer(boss) return ChatBlob("Next Worldboss spawns", ""+self.getWBTimer()) def getWBTimer(self, name=None) -> Union[TimerObj, list[TimerObj]]: if not name: blob = "" for x in sorted(self.timer_data, key=lambda y: y.name): msg = self.formatMessage(x) blob += (msg + "\n") if msg else "" return blob for x in self.timer_data: if x.name == name: return self.formatMessage(x).strip() def formatMessage(self, x: TimerObj): if x.alive_immortal(): return f" {x.name} :: mortal in {self.util.format_time(x.time - x.getTime())}" else: rt = x.get_respawn_timer() if rt != 0: if (x.time + rt) < x.getTime() < (x.time + 2 * rt): return f" {x.name} :: should have spawned {self.util.format_time(abs(x.getTime() - (x.time + rt)))} ago" return f" {x.name} :: spawn in {self.util.format_time(x.time + rt - x.getTime())}"