path: root/p2pvr/daemon/schedules.cpp

#include "schedules.h"
#include <logger.h>
#include <Ice/Ice.h>
#include "chronoHelpers.h"
#include <boost/date_time/posix_time/posix_time.hpp>
#include <factory.impl.h>
#include <slicer/db/sqlSelectDeserializer.h>
#include <slicer/db/sqlInsertSerializer.h>
#include <slicer/db/sqlUpdateSerializer.h>
#include <slicer/db/sqlTablePatchSerializer.h>
#include <tablepatch.h>
#include <p2pvr-int.h>

#include "sql/schedules/getCandidates.sql.h"
#include "sql/schedules/deleteById.sql.h"
#include "sql/schedules/selectAll.sql.h"
#include "sql/schedules/selectById.sql.h"
#include "sql/schedules/scheduledToRecord.sql.h"

namespace po = boost::program_options;
using namespace IceTray::Logging;

namespace P2PVR {
IceTray::Logging::LoggerPtr SchedulesI::logger(LOGMANAGER()->getLogger<SchedulesI>());

SchedulesI::Options::Options() :
	IceTray::Options("P2PVR Scheduler options")
{
}

ICETRAY_OPTIONS(SchedulesI::Options,
		("p2pvr.scheduler.algorithm", po::value(&SchedulerAlgorithm)->default_value("BitDumb"),
		 "Implementation of episode group scheduler problem solver")
)

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;
	}
}

SchedulesI::SchedulesI(DB::ConnectionPoolPtr db) :
	IceTray::AbstractDatabaseClient(db)
{
}

void
SchedulesI::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(LogLevel::DEBUG, "	added %s", ae->what);
						grouped.push_back(ae);
						all.erase(aei);
						GetEpisodeIntersects(all, grouped);
						return;
					}
				}
			}
		}
	}
}

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

	// Load list from database
	auto episodes = fetch<ScheduleCandidates>(P2PVR::sql::schedules::getCandidates);

	Episodes scheduleList;
	Showings allShowings;
	EpisodePtr cur;
	int minPriority = 0;
	for (const auto & c : episodes) {
		if (!cur || cur->what != c->What) {
			cur = std::make_shared<Episode>(c->What);
			scheduleList.push_back(cur);
		}
		ShowingPtr s = std::make_shared<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(LogLevel::DEBUG, "%d episodes created, %s showings", scheduleList.size(), allShowings.size());
	for (const auto & e : scheduleList) {
		logger->messagebf(LogLevel::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(LogLevel::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(LogLevel::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(LogLevel::DEBUG, "	%d * %d:%s", e->showings.size(), e->priority, e->what);
			e->showings.push_back(NULL);
			total *= e->showings.size();
		}
		logger->messagebf(LogLevel::DEBUG, "group complexity of %d options", total);

		EpisodeGroupPtr sched = EpisodeGroupPtr(EpisodeGroupFactory::createNew(options->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(LogLevel::DEBUG, "Episode %s, %d options", c->what, c->showings.size());
			for (const auto & i : c->showings) {
				if (selected.find(i) != selected.end()) {
					logger->messagebf(LogLevel::DEBUG, "	%s - %s (%d) <-", i->startTime, i->stopTime, i->transportStreamId);
				}
				else if (i) {
					logger->messagebf(LogLevel::DEBUG, "	%s - %s (%d)", i->startTime, i->stopTime, i->transportStreamId);
				}
			}
		}
		logger->message(LogLevel::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(std::make_shared<P2PVR::ScheduledToRecord>(i->eventUid,
									found ?
									selected.find(i) != selected.end() ? RecordStatuses::WillRecordThisShowing : RecordStatuses::WillRecordOtherShowing :
									RecordStatuses::CannotRecordAnyShowing, i->scheduleId));
				}
			}
		}
	}

	auto dbc = db->get();
	DB::TransactionScope tx(*dbc.get());
	DB::TablePatch mergeRecords;
	mergeRecords.dest = "record";
	Slicer::SerializeAny<Slicer::SqlTablePatchSerializer>(records, dbc.get(), mergeRecords);

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

void
SchedulesI::DeleteSchedule(int id, const Ice::Current & ice)
{
	auto dbc = db->get();
	auto del = P2PVR::sql::schedules::deleteById.modify(dbc.get());
	del->bindParamI(0, id);
	del->execute();
	DoReschedule(ice);
}

P2PVR::ScheduleList
SchedulesI::GetSchedules(const Ice::Current &)
{
	logger->message(LogLevel::DEBUG, __PRETTY_FUNCTION__);
	return fetch<P2PVR::ScheduleList>(P2PVR::sql::schedules::selectAll);
}

P2PVR::SchedulePtr
SchedulesI::GetSchedule(int id, const Ice::Current &)
{
	logger->messagebf(LogLevel::DEBUG, "%s(%d)", __PRETTY_FUNCTION__, id);
	auto schedules = fetch<P2PVR::ScheduleList>(P2PVR::sql::schedules::selectById, id);
	if (schedules.empty()) throw P2PVR::NotFound();
	return schedules.front();
}

P2PVR::ScheduledToRecordList
SchedulesI::GetScheduledToRecord(const Ice::Current &)
{
	logger->message(LogLevel::DEBUG, __PRETTY_FUNCTION__);
	return fetch<P2PVR::ScheduledToRecordList>(P2PVR::sql::schedules::scheduledToRecord);
}

Ice::Int
SchedulesI::NewSchedule(const P2PVR::SchedulePtr s, const Ice::Current & ice)
{
	auto dbc = db->get();
	Slicer::SerializeAny<Slicer::SqlFetchIdInsertSerializer>(s, dbc.get(), "schedules");
	DoReschedule(ice);
	return s->ScheduleId;
}

void
SchedulesI::UpdateSchedule(const P2PVR::SchedulePtr s, const Ice::Current & ice)
{
	auto dbc = db->get();
	Slicer::SerializeAny<Slicer::SqlUpdateSerializer>(s, dbc.get(), "schedules");
	DoReschedule(ice);
}
}

INSTANTIATEFACTORY(P2PVR::EpisodeGroup, const P2PVR::Episodes &);