Skip to content

Show

ShowClient

Fetches the show info from LibreTime now-playing v2 endpoint.

Every show has a name, a start and endtime and an optional URL.

Source code in nowplaying/show/client.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class ShowClient:
    """Fetches the show info from LibreTime now-playing v2 endpoint.

    Every show has a name, a start and endtime and an optional URL.
    """

    __DEFAULT_SHOW_DURATION = 3  # 3 seconds
    __cleanup_show_name_regexp = re.compile(r"&(\w+?);")
    __show_datetime_format = "%Y-%m-%d %H:%M:%S"

    def __init__(self, current_show_url):
        self.current_show_url = current_show_url

        self.show = Show()
        self.showtz = None

    def get_show_info(self, force_update=False):
        """Return a Show object."""

        if force_update:
            self.update()
        else:
            self.lazy_update()

        return self.show

    def lazy_update(self):
        # only update the info if we expect that a new show has started
        if datetime.datetime.now(pytz.timezone("UTC")) > self.show.endtime:
            logger.info("Show expired, going to update show info")
            self.update()

        else:
            logger.debug("Show still running, won't update show info")

    def update(self):
        self.show = Show()  # Create a new show object

        # Set the show's default end time to now + 30 seconds to prevent updates
        # happening every second and hammering the web service if something
        # goes wrong later.
        self.show.set_endtime(
            datetime.datetime.now(pytz.timezone("UTC"))
            + datetime.timedelta(seconds=self.__DEFAULT_SHOW_DURATION)
        )

        try:
            # try to get the current show informations from loopy's cast web
            # service
            data = requests.get(self.current_show_url).json()

            logger.debug("Got show info: %s" % data)

        except Exception as e:
            logger.error("Unable to get current show informations")

            logger.exception(e)
            # LSB 2017: ignoring missing show update
            # raise ShowClientError('Unable to get show informations: %s' % e)
            return

        self.showtz = pytz.timezone(data["station"]["timezone"])

        # pick the current show
        show_data = self.__pick_current_show(data)

        if not show_data:
            logger.warning("Failed to find a current or upcoming show, bailing out.")
            return

        # get the name of the show, aka real_name
        # ex.: Stereo Freeze
        real_name = show_data["name"]

        if len(real_name) == 0:
            # keep the default show information
            logger.error("No show name found")
            raise ShowClientError("Missing show name")

        real_name = self.__cleanup_show_name(real_name)
        self.show.set_name(real_name)

        # get the show's end time in order to time the next lookup.
        # ex.: 2012-04-28 19:00:00 (missing a tzoffset and localized!)
        end_time = show_data["ends"]

        if len(end_time) == 0:
            logger.error("No end found")
            raise ShowClientError("Missing show end time")

        endtime = self.showtz.localize(
            datetime.datetime.strptime(end_time, self.__show_datetime_format)
        )

        # store as UTC datetime object
        self.show.set_endtime(endtime.astimezone(pytz.timezone("UTC")))

        # get the show's start time
        # ex.: 2012-04-28 18:00:00
        start_time = show_data["starts"]

        if len(start_time) == 0:
            logger.error("No start found")
            raise ShowClientError("Missing show start time")

        starttime = self.showtz.localize(
            datetime.datetime.strptime(start_time, self.__show_datetime_format)
        )

        # store as UTC datetime object
        self.show.set_starttime(starttime.astimezone(pytz.timezone("UTC")))

        # Check if the endtime is in the past
        # This prevents stale (wrong) show informations from beeing pushed to
        # the live stream and stops hammering the service every second
        if self.show.endtime < datetime.datetime.now(pytz.timezone("UTC")):
            logger.error("Show endtime %s is in the past" % self.show.endtime)

            raise ShowClientError(
                "Show end time (%s) is in the past" % self.show.endtime
            )

        # get the show's URL
        # ex.: http://www.rabe.ch/sendungen/entertainment/onda-libera.html
        url = show_data["url"]

        if len(url) == 0:
            logger.error("No url found")
        else:
            self.show.set_url(url)

        logger.info(
            'Show "%s" started and runs from %s till %s'
            % (self.show.name, starttime, endtime)
        )
        logger.debug(self.show)

    def __cleanup_show_name(self, name) -> str:
        """Cleanup name by undoing htmlspecialchars from libretime zf1 mvc."""

        def __entityref_decode(m):
            try:
                return entitydefs[m.group(1)]
            except KeyError:
                return m.group(0)

        return self.__cleanup_show_name_regexp.sub(__entityref_decode, name)

    def __pick_current_show(self, data):
        """Pick the current show from the data.

        If there is no current show and the next one starts reasonably soon, pick that.
        """
        if not data["shows"]["current"]:
            # ignore if no current show is playing
            logger.info("No current show is playing, checking next show")
            if data["shows"]["next"] and data["shows"]["next"][0]:
                show = data["shows"]["next"][0]
                logger.info("Next show is %s" % show["name"])
                next_start = self.showtz.localize(
                    datetime.datetime.strptime(
                        show["starts"], self.__show_datetime_format
                    )
                )
                logger.warning(
                    datetime.datetime.now(pytz.timezone("UTC"))
                    + datetime.timedelta(minutes=15)
                )
                logger.warning(next_start)
                if next_start < datetime.datetime.now(
                    pytz.timezone("UTC")
                ) + datetime.timedelta(minutes=15):
                    logger.info("Next show starts soon enough, using it")
                    return show
            return
        return data["shows"]["current"]

__cleanup_show_name(name)

Cleanup name by undoing htmlspecialchars from libretime zf1 mvc.

Source code in nowplaying/show/client.py 
158
159
160
161
162
163
164
165
166
167
def __cleanup_show_name(self, name) -> str:
    """Cleanup name by undoing htmlspecialchars from libretime zf1 mvc."""

    def __entityref_decode(m):
        try:
            return entitydefs[m.group(1)]
        except KeyError:
            return m.group(0)

    return self.__cleanup_show_name_regexp.sub(__entityref_decode, name)

__pick_current_show(data)

Pick the current show from the data.

If there is no current show and the next one starts reasonably soon, pick that.

Source code in nowplaying/show/client.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def __pick_current_show(self, data):
    """Pick the current show from the data.

    If there is no current show and the next one starts reasonably soon, pick that.
    """
    if not data["shows"]["current"]:
        # ignore if no current show is playing
        logger.info("No current show is playing, checking next show")
        if data["shows"]["next"] and data["shows"]["next"][0]:
            show = data["shows"]["next"][0]
            logger.info("Next show is %s" % show["name"])
            next_start = self.showtz.localize(
                datetime.datetime.strptime(
                    show["starts"], self.__show_datetime_format
                )
            )
            logger.warning(
                datetime.datetime.now(pytz.timezone("UTC"))
                + datetime.timedelta(minutes=15)
            )
            logger.warning(next_start)
            if next_start < datetime.datetime.now(
                pytz.timezone("UTC")
            ) + datetime.timedelta(minutes=15):
                logger.info("Next show starts soon enough, using it")
                return show
        return
    return data["shows"]["current"]

get_show_info(force_update=False)

Return a Show object.

Source code in nowplaying/show/client.py 
37
38
39
40
41
42
43
44
45
def get_show_info(self, force_update=False):
    """Return a Show object."""

    if force_update:
        self.update()
    else:
        self.lazy_update()

    return self.show

ShowClientError

Bases: Exception

ShowClient related exception.

Source code in nowplaying/show/client.py 
15
16
17
18
class ShowClientError(Exception):
    """ShowClient related exception."""

    pass