PrivacyScanData/archive/new2/showroom-master/showroom/control.py

import os
import logging
import time
from queue import Empty as QueueEmpty
from multiprocessing import Process, Queue
from threading import Thread
from itertools import count

from .message import ShowroomMessage
from .settings import ShowroomSettings
from .index import ShowroomIndex
from .core import WatchManager
from .exceptions import ShowroomStopRequest

control_logger = logging.getLogger("showroom.control")


class BaseShowroomLiveController(object):
    def __init__(self, index: ShowroomIndex=None, settings: ShowroomSettings=None, record_all=False):
        # TODO: proper docstring
        super(BaseShowroomLiveController, self).__init__()

        self.command_queue = Queue()
        self.message_queue = Queue()

        if not settings:
            self.settings = ShowroomSettings.from_file()
        else:
            self.settings = settings

        # TODO: where is the best place for this? I put it here because
        # controller could potentially change output dir and so needs to be
        # the one to handle recreating the folders.
        # it could also be done in ShowroomSettings if I give that object more power
        # os.makedirs(self.settings.directory.temp, exist_ok=True)

        if not index:
            self.index = ShowroomIndex(self.settings.directory.index, record_all=record_all)
        else:
            self.index = index

        self._instance = None

        self.manager = WatchManager(self.index, self.settings)

        self.counter = count()
        # aliases
        self.send = self.send_command
        self.get = self.get_messages

        # TODO: make maintenance more intelligent

    # instance (thread or process) wrapping methods
    def start(self):
        # TODO: are there any other conditions required to restart the loop?
        raise NotImplementedError

    def is_alive(self):
        return self._instance.is_alive()

    def join(self, timeout=0):
        return self._instance.join(timeout=timeout)

    def run(self):
        control_logger.debug("Running ShowroomLiveController")

        # start index update tasks (runs in separate thread)
        self.index.start()
        while True:

            # TODO: check if time for maintenance, if so do maintenance then schedule next
            # if self.resume_time > self.time.time() > self.end_time:
            #     sleep_seconds = (datetime.datetime.combine(self.time, self.resume_time)
            #                      - self.time).total_seconds() + 1.0
            #     print('Time is {}, sleeping for {} seconds, until {}'.format(strftime(self.time, '%H:%M'),
            #                                                                  sleep_seconds,
            #                                                                  strftime(self.resume_time, '%H:%M')))
            #     self.scheduler.reset_ticks()
            #     time.sleep(sleep_seconds)

            self.manager.tick()

            while not self.command_queue.empty():
                control_logger.debug('Reading command queue')
                try:
                    ident, command, args, kwargs = self.command_queue.get(block=False)
                except QueueEmpty:
                    break
                else:
                    # TODO: check that command is valid and allowed
                    if command[0] == '_':
                        control_logger.warn('Forbidden command: {}'.format(command))
                        continue
                    # TODO: lookup command in a dictionary instead of this
                    msg = ShowroomMessage(ident, command)
                    cmd, *args2 = command.replace('/', '_').split('_')
                    try:
                        msg = getattr(self, '_' + cmd)(*(list(args) + args2), msg=msg, **kwargs)
                    except ShowroomStopRequest:
                        self.index.stop()
                        return
                    except AttributeError as e:
                        # invalid command
                        control_logger.debug('Unknown command: {}, {}'.format(command, e))
                    except TypeError as e:
                        # trying to call something besides a method
                        control_logger.debug('{} is not a command -- {}'.format(command, e))
                    else:
                        if msg is not None:
                            self.message_queue.put(msg)

            time.sleep(0.2)

    def send_command(self, command, *args, **kwargs):
        ident = next(self.counter)
        self.command_queue.put((ident, command, args, kwargs))
        return ident

    def get_messages(self):
        messages = []
        while not self.message_queue.empty():
            try:
                msg = self.message_queue.get(block=False)
            except QueueEmpty:
                break
            else:
                if msg:
                    messages.append(msg)
        return messages

    def stop(self):
        self.command_queue.put((next(self.counter), "stop", [], {}))

    def _stop(self, *args, msg=None, **kwargs):
        # TODO: log stopping
        self.manager.stop()
        self.manager.write_completed()
        raise ShowroomStopRequest

    # commands
    # all commands either return None or a message: either a dict or a showroom Message

    # index commands
    # do these need to be made thread-safe? they mutate rooms... but nothing else should.
    def _index(self, *args, msg=None, **kwargs):
        if not args or args[0] == 'list':
            if msg:
                # return index list in message
                pass
        elif args[0] == 'filter':
            self._index_filter(*args[1:], **kwargs)
        elif args[0] == 'update':
            self._index_update(*args[1:], **kwargs)

    def _index_filter(self, *args, msg=None, **kwargs):
        b_updated = False
        if not args or args[0] == 'list':
            if msg:
                """Returns a dict of all wanted and unwanted rooms, by name.
                    {"index_filters": {"wanted": [...], "unwanted": [...]} }
                """
                return msg.set_content(self.index.filter_get_list())
            else:
                # raise message needed exception
                pass

        if "all" in args:
            """Turns downloading on for all rooms."""
            self.index.filter_all()
            # TODO: check if this actually changed anything
            # have index.filter return number of modified rooms?
            b_updated = True

        elif "none" in args:
            """Turns downloading off for all rooms."""
            self.index.filter_none()
            b_updated = True

        if "add" in kwargs:
            """Sets downloading on for specific rooms."""
            self.index.filter_add(kwargs["add"])
            b_updated = True

        if "remove" in kwargs:
            """Sets downloading off for specific rooms."""
            self.index.filter_remove(kwargs["remove"])
            b_updated = True

        if b_updated:
            self.manager.update_flag.set()

    def _index_update(self, *args, **kwargs):
        """Updates the index from either the local filesystem or a web source"""
        if 'web' in args:
            if 'src_url' in kwargs:
                self.index.update_from_web(kwargs['src_url'])
            else:
                self.index.update_from_web()
        self.index.update()

    def _index_update_from_web(self, src_url=None):
        """Updates the index from a remote source.

        Source must be a json like
            https://wlerin.github.io/showroom-index/list.json
        that points to a set of jdex files to update from.
        """
        self.index.update_from_web(src_url)

    # TODO: Messages require a unique identifier given them by the caller
    # room list commands
    def _get_rooms_by_mode(self, mode):
        rooms = []
        for watch in self.manager.watchers.get_by_mode(mode):
            rooms.append(watch.get_info())
        return sorted(rooms, key=lambda x: (x['start_time'], x['name']))

    # TODO: get these working again
    # "endpoints"
    # take arbitrary args and kwargs and parse through them for meaningful instructions
    def _schedule(self, *args, msg=None, **kwargs):
        if msg is not None:
            if args:
                pass
            if kwargs:
                # TODO: take other options
                pass
            msg.set_content(self._get_rooms_by_mode("working"))
            return msg

    def _lives(self, *args, msg=None, **kwargs):
        if msg is not None:
            msg.set_content(self._get_rooms_by_mode("live"))
            return msg

    def _downloads(self, *args, msg=None, **kwargs):
        if msg is not None:
            # the caller can then filter these to get links if desired...?
            msg.set_content(self._get_rooms_by_mode("download"))
            return msg


class ShowroomLiveControllerThread(BaseShowroomLiveController):
    def start(self):
        # TODO: are there any other conditions required to restart the loop?
        if not self._instance or not self._instance.is_alive():
            self._instance = Thread(target=self.run, name="ShowroomLiveController")
            self._instance.start()


class ShowroomLiveControllerProcess(BaseShowroomLiveController):
    def start(self):
        # TODO: are there any other conditions required to restart the loop?
        if not self._instance or not self._instance.is_alive():
            self._instance = Process(target=self.run, name="ShowroomLiveController")
            self._instance.start()


Controller = ShowroomLiveControllerThread