<commit_before>#!/usr/bin/env python

import os
import json
import rospy
import requests

from std_msgs.msg import String
from interactivespaces_msgs.msg import GenericMessage
from rosapi import params


class DirectorAPIProxy:
    def __init__(self, director_api_url):
        """
        Class responsible for getting content from director api
        """
        rospy.loginfo("Initializing Attract Loooooooooooop")
        self.director_api_url = director_api_url
        rospy.loginfo("Using director API url: %s" % self.director_api_url)

    def get(self, uri):
        try:
            url = "%s%s" % (self.director_api_url, uri)
            return requests.get(url).content
        except Exception:
            rospy.logerr("Could not get content from URL: %s" % url)


def MockFunc(*args, **kwargs):
    pass


class AttractLoop:
    def __init__(self, api_proxy, director_scene_publisher,
                 director_presentation_publisher, stop_action,
                 earth_query_publisher, earth_planet_publisher,
                 default_presentation=None, default_planet='earth',
                 set_earth=MockFunc, default_duration=120):
        """
        Class responsible for playing back presentations/scenes that are marked as "attract_loop"
        in Liquid Galaxy content management system.

        """
        self.api_proxy = api_proxy
        self.earth_query_publisher = earth_query_publisher
        self.stop_action = stop_action
        self.default_presentation = default_presentation
        self.default_planet = default_planet
        self.earth_planet_publisher = earth_planet_publisher
        self.director_scene_publisher = director_scene_publisher
        self.director_presentation_publisher = director_presentation_publisher
        self.attract_loop_queue = []
        self.play_loop = False
        self.scene_timer = 0
        self.set_earth = set_earth
        self.default_duration = default_duration
        self.initialize_timer()

    def initialize_timer(self):
        """
        Each scene has a timer - countdown is using 1s resolution.
        """
        rospy.Timer(rospy.Duration(1), self._play_attract_loop)

    def _process_activity_state_change(self, message):
        """
        when message (that comes from /activity/active) equals True
        then we begin the playback
        otherwise we stop
        """
        self.scene_timer = 0

        if message.data is True and self.play_loop is True:
            rospy.loginfo("Director: Attract loop becoming inactive")
            self.play_loop = False
            self._stop_attract_loop()
        elif message.data is False and self.play_loop is False:
            self._switch_to_planet()
            rospy.loginfo("Director: Attract loop becoming active")
            self.play_loop = True
            rospy.sleep(2)
        else:
            rospy.logwarn("Activity message contained state %s and current state is %s - that's weird" % (message.data, self.play_loop))

    def _stop_attract_loop(self):
        """
        Emits ROS message on the basis of pre-configured action.
        This action is executed when system becomes active.
        """
        self.play_loop = False

        rospy.loginfo("Stopping scene timer")

        if self.stop_action == 'stop_playtour':
            self._stop_playtour()
        elif self.stop_action == 'go_blank':
            self._stop_playtour()
            self._publish_blank_scene()
        elif self.stop_action == 'go_blank_and_switch_to_planet':
            self._stop_playtour()
            self._publish_blank_scene()
            self._switch_to_planet()
        elif self.stop_action == 'load_presentation':
            self._stop_playtour()
            if self.default_presentation:
                pass
            else:
                rospy.logerr("No default presentation defined")
        else:
            pass

    def _switch_to_planet(self):
        """
        Emits a message with planet change taken from configuration
        """
        switch_to_planet_msg = String(data=self.default_planet)
        rospy.loginfo("Executing 'switch_to_planet' action")
        self.earth_planet_publisher.publish(switch_to_planet_msg)

    def _stop_playtour(self):
        """
        Emits empty message on /earth/query/tour to stop the tour

        **PUFF MAGIC**

        """
        stop_tour_msg = String(data='')
        rospy.loginfo("Executing 'stop_playtour' action")
        self.earth_query_publisher.publish(stop_tour_msg)

    def _publish_blank_scene(self):
        """
        Emits a scene with empty windows to clean up all assets from screens
        """
        rospy.loginfo("Playing blank scene")

        viewports = [viewport.split('/')[2] for viewport in params.get_param_names() if '/viewport/' in viewport]

        scene = {
            "description": "attract loop blank scene",
            "duration": 666,
            "name": "attract loop blank scene",
            "resource_uri": "no uri",
            "slug": "attract-loop-break",
            "windows": []
        }

        for viewport_name in viewports:
            window = {
                "assets": [],
                "y_coord": 666,
                "x_coord": 666,
                "height": 666,
                "width": 666,
                "activity": "no_activity",
                "presentation_viewport": viewport_name<commit_msg>Fix attract loop rosapi compatibility<commit_after>#!/usr/bin/env python

import os
import json
import rospy
import requests

from std_msgs.msg import String
from interactivespaces_msgs.msg import GenericMessage
from rosapi import params


class DirectorAPIProxy:
    def __init__(self, director_api_url):
        """
        Class responsible for getting content from director api
        """
        rospy.loginfo("Initializing Attract Loooooooooooop")
        self.director_api_url = director_api_url
        rospy.loginfo("Using director API url: %s" % self.director_api_url)

    def get(self, uri):
        try:
            url = "%s%s" % (self.director_api_url, uri)
            return requests.get(url).content
        except Exception:
            rospy.logerr("Could not get content from URL: %s" % url)


def MockFunc(*args, **kwargs):
    pass


class AttractLoop:
    def __init__(self, api_proxy, director_scene_publisher,
                 director_presentation_publisher, stop_action,
                 earth_query_publisher, earth_planet_publisher,
                 default_presentation=None, default_planet='earth',
                 set_earth=MockFunc, default_duration=120):
        """
        Class responsible for playing back presentations/scenes that are marked as "attract_loop"
        in Liquid Galaxy content management system.

        """
        self.api_proxy = api_proxy
        self.earth_query_publisher = earth_query_publisher
        self.stop_action = stop_action
        self.default_presentation = default_presentation
        self.default_planet = default_planet
        self.earth_planet_publisher = earth_planet_publisher
        self.director_scene_publisher = director_scene_publisher
        self.director_presentation_publisher = director_presentation_publisher
        self.attract_loop_queue = []
        self.play_loop = False
        self.scene_timer = 0
        self.set_earth = set_earth
        self.default_duration = default_duration
        self.initialize_timer()

    def initialize_timer(self):
        """
        Each scene has a timer - countdown is using 1s resolution.
        """
        rospy.Timer(rospy.Duration(1), self._play_attract_loop)

    def _process_activity_state_change(self, message):
        """
        when message (that comes from /activity/active) equals True
        then we begin the playback
        otherwise we stop
        """
        self.scene_timer = 0

        if message.data is True and self.play_loop is True:
            rospy.loginfo("Director: Attract loop becoming inactive")
            self.play_loop = False
            self._stop_attract_loop()
        elif message.data is False and self.play_loop is False:
            self._switch_to_planet()
            rospy.loginfo("Director: Attract loop becoming active")
            self.play_loop = True
            rospy.sleep(2)
        else:
            rospy.logwarn("Activity message contained state %s and current state is %s - that's weird" % (message.data, self.play_loop))

    def _stop_attract_loop(self):
        """
        Emits ROS message on the basis of pre-configured action.
        This action is executed when system becomes active.
        """
        self.play_loop = False

        rospy.loginfo("Stopping scene timer")

        if self.stop_action == 'stop_playtour':
            self._stop_playtour()
        elif self.stop_action == 'go_blank':
            self._stop_playtour()
            self._publish_blank_scene()
        elif self.stop_action == 'go_blank_and_switch_to_planet':
            self._stop_playtour()
            self._publish_blank_scene()
            self._switch_to_planet()
        elif self.stop_action == 'load_presentation':
            self._stop_playtour()
            if self.default_presentation:
                pass
            else:
                rospy.logerr("No default presentation defined")
        else:
            pass

    def _switch_to_planet(self):
        """
        Emits a message with planet change taken from configuration
        """
        switch_to_planet_msg = String(data=self.default_planet)
        rospy.loginfo("Executing 'switch_to_planet' action")
        self.earth_planet_publisher.publish(switch_to_planet_msg)

    def _stop_playtour(self):
        """
        Emits empty message on /earth/query/tour to stop the tour

        **PUFF MAGIC**

        """
        stop_tour_msg = String(data='')
        rospy.loginfo("Executing 'stop_playtour' action")
        self.earth_query_publisher.publish(stop_tour_msg)

    def _publish_blank_scene(self):
        """
        Emits a scene with empty windows to clean up all assets from screens
        """
        rospy.loginfo("Playing blank scene")

        viewports = [viewport.split('/')[2] for viewport in params.get_param_names(['/viewport/*'])]

        scene = {
            "description": "attract loop blank scene",
            "duration": 666,
            "name": "attract loop blank scene",
            "resource_uri": "no uri",
            "slug": "attract-loop-break",
            "windows": []
        }

        for viewport_name in viewports:
            window = {
                "assets": [],
                "y_coord": 666,
                "x_coord": 666,
                "height": 666,
                "width": 666,
                "activity": "no_activity",
                "presentation_viewport": viewport_name
{'fraction_non_alphanumeric': 0.08334102445777573, 'fraction_numerical': 0.004430087678818643, 'mean_word_length': 2.2106666666666666, 'pattern_counts': {'":': 26, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2762206', 'n_tokens_mistral': 3136, 'n_tokens_neox': 2951, 'n_words': 727}