#include "QueueManager.h" QM::QueueManager::QueueManager(wxEvtHandler *eventHandler, std::string jobsdir) { // need to send events into the mainwindow by the threads... this->eventHandler = eventHandler; this->eventHandler->Bind(wxEVT_THREAD, &QueueManager::OnThreadMessage, this); this->jobsDir = jobsdir; this->QueueList = std::map(); this->LoadJobListFromDir(); } QM::QueueManager::~QueueManager() { } int QM::QueueManager::AddItem(QM::QueueItem item) { if (item.id == 0) { item.id = this->GetAnId(); } if (item.created_at == 0) { item.created_at = this->GetCurrentUnixTimestamp(); } this->QueueList[item.id] = item; this->SendEventToMainWindow(QM::QueueEvents::ITEM_ADDED, item); if (this->isRunning == false && item.status == QM::QueueStatus::PENDING) { this->SendEventToMainWindow(QM::QueueEvents::ITEM_START, item); this->isRunning = true; } this->SaveJobToFile(item); return item.id; } int QM::QueueManager::AddItem(sd_gui_utils::SDParams *params) { QM::QueueItem item; item.params = *params; return this->AddItem(item); } int QM::QueueManager::AddItem(sd_gui_utils::SDParams params) { QM::QueueItem item; item.params = params; return this->AddItem(item); } QM::QueueItem QM::QueueManager::GetItem(int id) { if (this->QueueList.find(id) == this->QueueList.end()) { return QM::QueueItem(); } else { return this->QueueList[id]; } } QM::QueueItem QM::QueueManager::GetItem(QM::QueueItem item) { return this->GetItem(item.id); } const std::map QM::QueueManager::getList() { return this->QueueList; } int QM::QueueManager::Duplicate(QM::QueueItem item) { if (this->QueueList.find(item.id) == this->QueueList.end()) { return -1; } else { return this->AddItem(this->QueueList.at(item.id).params); } } int QM::QueueManager::Duplicate(int id) { return this->AddItem(this->GetItem(id).params); } void QM::QueueManager::SetStatus(QM::QueueStatus status, int id) { if (this->QueueList.find(id) != this->QueueList.end()) { this->QueueList[id].status = status; this->QueueList[id].updated_at = this->GetCurrentUnixTimestamp(); if (status == QM::QueueStatus::DONE) { this->QueueList[id].finished_at = this->GetCurrentUnixTimestamp(); } this->SaveJobToFile(this->QueueList[id]); this->SendEventToMainWindow(QM::QueueEvents::ITEM_STATUS_CHANGED, this->QueueList[id]); } } void QM::QueueManager::PauseAll() { for (auto [key, value] : this->QueueList) { if (value.status == QM::QueueStatus::PENDING) { this->SetStatus(QM::PAUSED, key); this->SendEventToMainWindow(QM::QueueEvents::ITEM_STATUS_CHANGED); } } } void QM::QueueManager::SendEventToMainWindow(QM::QueueEvents eventType, QM::QueueItem item) { wxThreadEvent *e = new wxThreadEvent(); e->SetString(wxString::Format("QUEUE:%d", (int)eventType)); e->SetPayload(item); wxQueueEvent(this->eventHandler, e); } void QM::QueueManager::OnThreadMessage(wxThreadEvent &e) { if (e.GetSkipped() == false) { e.Skip(); } auto msg = e.GetString().ToStdString(); std::string token = msg.substr(0, msg.find(":")); std::string content = msg.substr(msg.find(":") + 1); // only numbers here... // only handle the QUEUE messages, what this class generate if (token == "QUEUE") { QM::QueueEvents event = (QM::QueueEvents)std::stoi(content); auto payload = e.GetPayload(); if (event == QM::QueueEvents::ITEM_START) { this->SetStatus(QM::QueueStatus::RUNNING, payload.id); this->isRunning = true; return; } if (event == QM::QueueEvents::ITEM_FINISHED) { this->SetStatus(QM::QueueStatus::DONE, payload.id); this->isRunning = false; // jump to the next item in queue // find waiting jobs for (auto job : this->QueueList) { if (job.second.status == QM::QueueStatus::PENDING) { if (this->isRunning == false) { this->SendEventToMainWindow(QM::QueueEvents::ITEM_START, job.second); this->isRunning = true; } break; } } return; } } if (token == "MODEL_LOAD_START") { auto payload = e.GetPayload(); this->SetStatus(QM::QueueStatus::MODEL_LOADING, payload.id); } // this state can not usable at here, because the payload is the sd_ctx* pointer here.. // we can't identify the current running job here... (we can maybe guess it, but not needed) // see GENERATION_START if (token == "MODEL_LOAD_DONE") { // auto payload = e.GetPayload(); // this->SetStatus(QM::QueueStatus::RUNNING, payload.id); } if (token == "MODEL_LOAD_ERROR" || token == "GENERATION_ERROR") { auto payload = e.GetPayload(); this->SetStatus(QM::QueueStatus::FAILED, payload.id); this->isRunning = false; } if (token == "GENERATION_START") { auto payload = e.GetPayload(); this->SetStatus(QM::QueueStatus::RUNNING, payload.id); this->isRunning = true; } // nothing to todo here, the payload is the generated image list, we can't find whitch item was it... // TODO: use struct to payload, and store multiple items in it... if (token == "GENERATION_DONE") { } } void QM::QueueManager::SaveJobToFile(int id) { auto item = this->GetItem(id); this->SaveJobToFile(item); } void QM::QueueManager::SaveJobToFile(QM::QueueItem item) { nlohmann::json jsonfile(item); std::string filename = this->jobsDir + "/" + std::to_string(item.id) + ".json"; std::ofstream file(filename); file << jsonfile; file.close(); } int QM::QueueManager::GetCurrentUnixTimestamp() { const auto p1 = std::chrono::system_clock::now(); return static_cast(std::chrono::duration_cast(p1.time_since_epoch()).count()); } void QM::QueueManager::LoadJobListFromDir() { if (!std::filesystem::exists(this->jobsDir)) { std::filesystem::create_directories(this->jobsDir); } int i = 0; for (auto const &dir_entry : std::filesystem::recursive_directory_iterator(this->jobsDir)) { if (!dir_entry.exists() || !dir_entry.is_regular_file() || !dir_entry.path().has_extension()) { continue; } std::filesystem::path path = dir_entry.path(); std::string ext = path.extension().string(); if (ext != ".json") { continue; } std::string name = path.filename().replace_extension("").string(); std::ifstream f(path.string()); try { nlohmann::json data = nlohmann::json::parse(f); auto item = data.get(); if (item.status == QM::QueueStatus::RUNNING) { item.status = QM::QueueStatus::FAILED; } if (item.status == QM::QueueStatus::MODEL_LOADING) { item.status = QM::QueueStatus::FAILED; } this->AddItem(item); } catch (const std::exception &e) { std::cerr << e.what() << '\n'; } } } int QM::QueueManager::GetAnId() { int id = this->GetCurrentUnixTimestamp(); while (id <= this->lastId) { id++; } this->lastId = id; return id; } void QM::QueueManager::onItemAdded(QM::QueueItem item) { // this->parent->m_joblist // auto dataTable = this->parent->m_joblist; }