path: root/p2pvr/daemon/schedules.cpp

#include <pch.hpp>
#include "schedules.h"
#include <rdbmsDataSource.h>
#include <logger.h>
#include <Ice/Ice.h>
#include <sqlVariableBinder.h>
#include <sqlMergeTask.h>
#include "p2Helpers.h"
#include "commonHelpers.h"
#include "containerIterator.h"
#include "resources.h"
#include <boost/date_time/posix_time/posix_time.hpp>
#include <instanceStore.impl.h>
#include "sqlSelectDeserializer.h"
#include <slicer/slicer.h>

ResourceString(Schedules_getCandidates, sql_Schedules_getCandidates);
ResourceString(Schedules_insert, sql_Schedules_insert);
ResourceString(Schedules_insertNewId, sql_Schedules_insertNewId);
ResourceString(Schedules_update, sql_Schedules_update);
ResourceString(Schedules_delete, sql_Schedules_delete);
ResourceString(Schedules_selectAll, sql_Schedules_selectAll);
ResourceString(Schedules_selectById, sql_Schedules_selectById);
ResourceString(Schedules_scheduledToRecord, sql_Schedules_scheduledToRecord);

std::string Schedules::SchedulerAlgorithm;

DECLARE_OPTIONS(Schedules, "P2PVR Scheduler options")
("p2pvr.scheduler.algorithm", Options::value(&SchedulerAlgorithm, "BitDumb"),
 "Implementation of episode group scheduler problem solver")
END_OPTIONS()

Showing::Showing(unsigned int e, unsigned int t, unsigned int sc, datetime start, datetime stop, int p, const Episode * ep) :
	episode(ep),
	eventUid(e),
	priority(p),
	scheduleId(sc),
	transportStreamId(t),
	startTime(start),
	stopTime(stop),
	period(start, stop)
{
}

Episode::Episode(const std::string & w) :
	priority(0),
	what(w)
{
}

EpisodeGroup::EpisodeGroup() :
	tuners(1),
	sumTimeToStart(0),
	score(0)
{
}

bool
EpisodeGroup::IsShowingListValid(const Showings & showings) const
{
	struct TransOffset {
		unsigned int trans;
		char offset;
	};
	typedef std::multimap<datetime, TransOffset> Periods;
	typedef std::map<int, unsigned char> Usage;

	Periods periods;
	Usage usage;

	for (const auto & s : showings) {
		if (s) {
			periods.insert(Periods::value_type(s->startTime, {s->transportStreamId, 1}));
			periods.insert(Periods::value_type(s->stopTime, {s->transportStreamId, -1}));
		}
	}
	bool result = true;
	for (const auto & p : periods) {
		auto & u = usage[p.second.trans];
		u += p.second.offset;
		if (std::count_if(usage.begin(), usage.end(), [](const Usage::value_type & uv) { return uv.second > 0;}) > tuners) {
			result = false;
			break;
		}
	}
	periods.clear();
	usage.clear();
	return result;
}

template<typename T>
inline
bool NotNull(const T & p)
{
	return p != NULL;
}

class SumTimeToStart {
	public:
		SumTimeToStart(time_t & t) :
			total(t),
			now(boost::posix_time::second_clock::universal_time())
		{
			total = 0;
		}
		inline void operator()(const ShowingPtr & s) const
		{
			if (s) {
				total += std::min<time_t>((now - s->startTime).seconds(), 0);
			}
		}
	public:
		time_t & total;
		datetime now;
};

const Showings &
EpisodeGroup::Solve()
{
	SelectShowings();
	return selected;
}

void
EpisodeGroup::Suggest(const Showings & showings)
{
	unsigned int c = 0;
	std::for_each(showings.begin(), showings.end(), [&c](const ShowingPtr & s) { if (s) c+= s->episode->priority; });
	if (c >= score) {
		time_t stt;
		std::for_each(showings.begin(), showings.end(), SumTimeToStart(stt));
		if (stt < sumTimeToStart || (stt == sumTimeToStart && c > score)) {
			if (IsShowingListValid(showings)) {
				selected = showings;
				score = c;
				sumTimeToStart = stt;
			}
		}
	}
}

EpisodeGroup::SuggestionResult
EpisodeGroup::SuggestWithFeedback(const Showings & showings)
{
	if (IsShowingListValid(showings)) {
		unsigned int c = 0;
		std::for_each(showings.begin(), showings.end(), [&c](const ShowingPtr & s) { if (s) c+= s->episode->priority; });
		if (c >= score) {
			time_t stt;
			std::for_each(showings.begin(), showings.end(), SumTimeToStart(stt));
			if (stt < sumTimeToStart) {
				selected = showings;
				score = c;
				sumTimeToStart = stt;
				return SuggestionValidAndAccepted;
			}
		}
		return SuggestionValid;
	}
	else {
		return SuggestionInvalid;
	}
}

void
Schedules::GetEpisodeIntersects(Episodes & all, Episodes & grouped)
{
	for (Episodes::iterator aei = all.begin(); aei != all.end(); aei++) {
		const auto & ae = *aei;
		for (const auto & ge : grouped) {
			for (const auto & gs : ge->showings) {
				for (const auto & as : ae->showings) {
					if (gs->period.intersects(as->period)) {
						Logger()->messagebf(LOG_DEBUG, "	added %s", ae->what);
						grouped.push_back(ae);
						all.erase(aei);
						GetEpisodeIntersects(all, grouped);
						return;
					}
				}
			}
		}
	}
}

void
Schedules::DoReschedule(const Ice::Current & ice)
{
	auto ic = ice.adapter->getCommunicator();
	auto devs = P2PVR::DevicesPrx::checkedCast(ice.adapter->createProxy(ic->stringToIdentity("GlobalDevices")));
	unsigned int tunerCount = devs->TunerCount();

	// Load list from database
	auto episodes = Slicer::DeserializeAny<SqlSelectDeserializer, P2PVR::ScheduleCandidates>(
			*Select(Schedules_getCandidates).second);

	Episodes scheduleList;
	Showings allShowings;
	EpisodePtr cur;
	int minPriority = 0;
	for (const auto & c : episodes) {
		if (!cur || cur->what != c->What) {
			cur = new Episode(c->What);
			scheduleList.push_back(cur);
		}
		ShowingPtr s = new Showing(c->EventUid, c->TransportStreamId, c->ScheduleId,
				*c->StartTime, *c->StopTime, c->Priority, cur.get());
		minPriority = std::min(minPriority, s->priority);
		cur->showings.push_back(s);
		allShowings.push_back(s);
	}
	Logger()->messagebf(LOG_DEBUG, "%d episodes created, %s showings", scheduleList.size(), allShowings.size());
	for (const auto & e : scheduleList) {
		Logger()->messagebf(LOG_DEBUG, "	%s", e->what);
		for (const auto & s : e->showings) {
			s->priority += 1 - minPriority;
			e->priority += s->priority;
		}
		e->priority /= e->showings.size();
	}

	P2PVR::ScheduledToRecordList records;
	// Solve
	while (!scheduleList.empty()) {
		auto work = scheduleList.begin();
		Logger()->messagebf(LOG_DEBUG, "start %s", (*work)->what);
		Episodes group;
		group.push_back(*work);
		scheduleList.erase(work);
		GetEpisodeIntersects(scheduleList, group);
		std::sort(group.begin(), group.end(), [](const EpisodePtr & a, const EpisodePtr & b) {
				if (a->priority > b->priority) return true;
				if (a->priority < b->priority) return false;
				return a->what < b->what;
			});

		Logger()->messagebf(LOG_DEBUG, "group created with %d episodes", group.size());
		double total = 1;
		// Measure and add the optional to not record
		for (const auto & e : group) {
			Logger()->messagebf(LOG_DEBUG, "	%d * %d:%s", e->showings.size(), e->priority, e->what);
			e->showings.push_back(NULL);
			total *= e->showings.size();
		}
		Logger()->messagebf(LOG_DEBUG, "group complexity of %d options", total);

		EpisodeGroupPtr sched = EpisodeGroupPtr(EpisodeGroupLoader::createNew(SchedulerAlgorithm, group));
		sched->tuners = tunerCount;
		std::set<ShowingPtr> selected;
		for (const auto & s : sched->Solve()) {
			if (s) selected.insert(s);
		}

		for (const auto & c : group) {
			Logger()->messagebf(LOG_DEBUG, "Episode %s, %d options", c->what, c->showings.size());
			for (const auto & i : c->showings) {
				if (selected.find(i) != selected.end()) {
					Logger()->messagebf(LOG_DEBUG, "	%s - %s (%d) <-", i->startTime, i->stopTime, i->transportStreamId);
				}
				else if (i) {
					Logger()->messagebf(LOG_DEBUG, "	%s - %s (%d)", i->startTime, i->stopTime, i->transportStreamId);
				}
			}
		}
		Logger()->message(LOG_DEBUG, "----------");
		for (const auto & c : group) {
			bool found = false;
			for (const auto & i : c->showings) {
				if (i && selected.find(i) != selected.end()) {
					found = true;
					break;
				}
			}
			for (const auto & i : c->showings) {
				if (i) {
					records.push_back(new P2PVR::ScheduledToRecord(i->eventUid,
									found ?
									selected.find(i) != selected.end() ? P2PVR::WillRecordThisShowing : P2PVR::WillRecordOtherShowing :
									P2PVR::CannotRecordAnyShowing, i->scheduleId));
				}
			}
		}
	}

	TxHelper tx(this);
	SqlMergeTask mergeRecords("postgres", "record");
	auto rcs = CreateColumns<P2PVR::ScheduledToRecordPtr>(boost::bind(&DatabaseClient::SqlMergeColumnsInserter, &mergeRecords, _1, _2));
	mergeRecords.sources.insert(new ContainerIterator<P2PVR::ScheduledToRecordList>(&records, rcs));
	mergeRecords.loadComplete(this);
	mergeRecords.execute(NULL);
	tx.Commit();

	auto recorder = P2PVR::RecorderPrx::checkedCast(ice.adapter->createProxy(ice.adapter->getCommunicator()->stringToIdentity("Recorder")));
	recorder->RefreshSchedules();
}

void
Schedules::DeleteSchedule(int id, const Ice::Current & ice)
{
	TxHelper tx(this);
	Modify(Schedules_delete, id).second->execute();
	DoReschedule(ice);
}

P2PVR::ScheduleList
Schedules::GetSchedules(const Ice::Current &)
{
	Logger()->message(LOG_DEBUG, __PRETTY_FUNCTION__);
	return Slicer::DeserializeAny<SqlSelectDeserializer, P2PVR::ScheduleList>(
			*Select(Schedules_selectAll).second);
}

P2PVR::SchedulePtr
Schedules::GetSchedule(int id, const Ice::Current &)
{
	Logger()->messagebf(LOG_DEBUG, "%s(%d)", __PRETTY_FUNCTION__, id);
	auto schedules = Slicer::DeserializeAny<SqlSelectDeserializer, P2PVR::ScheduleList>(
			*Select(Schedules_selectById, id).second);
	if (schedules.empty()) throw P2PVR::NotFound();
	return schedules.front();
}

P2PVR::ScheduledToRecordList
Schedules::GetScheduledToRecord(const Ice::Current &)
{
	Logger()->message(LOG_DEBUG, __PRETTY_FUNCTION__);
	return Slicer::DeserializeAny<SqlSelectDeserializer, P2PVR::ScheduledToRecordList>(
			*Select(Schedules_scheduledToRecord).second);
}

int
Schedules::UpdateSchedule(const P2PVR::SchedulePtr & s, const Ice::Current & ice)
{
	TxHelper tx(this);
	if (s->ScheduleId == 0) {
		Modify(Schedules_insert, s->ServiceId, s->EventUid, s->Title, s->Search, s->Priority, s->Early, s->Late, s->Repeats).second->execute();
		s->ScheduleId = SelectScalar<int>(Schedules_insertNewId);
	}
	else {
		Modify(Schedules_update, s->ServiceId, s->EventUid, s->Title, s->Search, s->Priority, s->Early, s->Late, s->Repeats, s->ScheduleId).second->execute();
	}
	DoReschedule(ice);
	return s->ScheduleId;
}

INSTANTIATESTORE(std::string, EpisodeGroupLoader);