Skip to content

Observer

Nowplaying input observer.

InputObserver

Bases: ABC

Abstract base for all InputObservers.

Source code in nowplaying/input/observer.py 
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
class InputObserver(ABC):
    """Abstract base for all InputObservers."""

    _SHOW_NAME_KLANGBECKEN = "Klangbecken"
    _SHOW_URL_KLANGBECKEN = "http://www.rabe.ch/sendungen/musik/klangbecken.html"

    def __init__(self: Self, current_show_url: str) -> None:
        """Create InputObserver."""
        self.show: Show
        self.track_handler: TrackEventHandler
        self.previous_saemubox_id: int = -1
        self.first_run = True
        self.previous_show_uuid = ""

        self.current_show_url = current_show_url

        self.showclient = client.ShowClient(current_show_url)
        self.show = self.showclient.get_show_info()

    def add_track_handler(self: Self, track_handler: TrackEventHandler) -> None:
        """Add Track handler."""
        self.track_handler = track_handler

    def update(self: Self, saemubox_id: int, event: CloudEvent | None = None) -> None:
        """Handle update."""
        # TODO(hairmare): v3-prep refactor to use :meth:`handles`
        #                 instead of :meth:`handle_id`
        # https://github.com/radiorabe/nowplaying/issues/180
        if self.handle_id(saemubox_id, event):
            self.handle(event)

    @abstractmethod
    # TODO(hairmare): v3 remove this method
    # https://github.com/radiorabe/nowplaying/issues/179
    def handle_id(
        self: Self,
        saemubox_id: int,
        event: CloudEvent | None = None,
    ) -> bool:  # pragma: no coverage
        """Handle ID."""

    @abstractmethod
    # TODO(hairmare): v3 remove this method
    # https://github.com/radiorabe/nowplaying/issues/179
    def handle(
        self: Self,
        event: CloudEvent | None = None,
    ) -> None:  # pragma: no coverage
        """Handle event."""

    @abstractmethod
    def handles(self: Self, event: CloudEvent) -> bool:  # pragma: no coverage
        """Handle event."""

    @abstractmethod
    def event(self: Self, event: CloudEvent) -> None:  # pragma: no coverage
        """Handle event."""

__init__(current_show_url)

Create InputObserver.

Source code in nowplaying/input/observer.py 
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self: Self, current_show_url: str) -> None:
    """Create InputObserver."""
    self.show: Show
    self.track_handler: TrackEventHandler
    self.previous_saemubox_id: int = -1
    self.first_run = True
    self.previous_show_uuid = ""

    self.current_show_url = current_show_url

    self.showclient = client.ShowClient(current_show_url)
    self.show = self.showclient.get_show_info()

add_track_handler(track_handler)

Add Track handler.

Source code in nowplaying/input/observer.py 
51
52
53
def add_track_handler(self: Self, track_handler: TrackEventHandler) -> None:
    """Add Track handler."""
    self.track_handler = track_handler

event(event) abstractmethod

Handle event.

Source code in nowplaying/input/observer.py 
86
87
88
@abstractmethod
def event(self: Self, event: CloudEvent) -> None:  # pragma: no coverage
    """Handle event."""

handle(event=None) abstractmethod

Handle event.

Source code in nowplaying/input/observer.py 
73
74
75
76
77
78
79
80
@abstractmethod
# TODO(hairmare): v3 remove this method
# https://github.com/radiorabe/nowplaying/issues/179
def handle(
    self: Self,
    event: CloudEvent | None = None,
) -> None:  # pragma: no coverage
    """Handle event."""

handle_id(saemubox_id, event=None) abstractmethod

Handle ID.

Source code in nowplaying/input/observer.py 
63
64
65
66
67
68
69
70
71
@abstractmethod
# TODO(hairmare): v3 remove this method
# https://github.com/radiorabe/nowplaying/issues/179
def handle_id(
    self: Self,
    saemubox_id: int,
    event: CloudEvent | None = None,
) -> bool:  # pragma: no coverage
    """Handle ID."""

handles(event) abstractmethod

Handle event.

Source code in nowplaying/input/observer.py 
82
83
84
@abstractmethod
def handles(self: Self, event: CloudEvent) -> bool:  # pragma: no coverage
    """Handle event."""

update(saemubox_id, event=None)

Handle update.

Source code in nowplaying/input/observer.py 
55
56
57
58
59
60
61
def update(self: Self, saemubox_id: int, event: CloudEvent | None = None) -> None:
    """Handle update."""
    # TODO(hairmare): v3-prep refactor to use :meth:`handles`
    #                 instead of :meth:`handle_id`
    # https://github.com/radiorabe/nowplaying/issues/180
    if self.handle_id(saemubox_id, event):
        self.handle(event)

KlangbeckenInputObserver

Bases: InputObserver

Observe when Sämu Box says Klangbecken we have now-playing.xml input.

Source code in nowplaying/input/observer.py 
 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
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
class KlangbeckenInputObserver(InputObserver):
    """Observe when Sämu Box says Klangbecken we have now-playing.xml input."""

    def __init__(
        self: Self,
        current_show_url: str,
        input_file: str | None = None,
    ) -> None:  # pragma: no coverage
        """Create KlangbeckenInputObserver."""
        # TODO(hairmare): test once input file is replaced with api
        # https://github.com/radiorabe/nowplaying/issues/180
        if input_file:
            warnings.warn(
                "The now-playing.xml format from Loopy/Klangbecken "
                "will be replaced in the future",
                PendingDeprecationWarning,
                stacklevel=2,
            )
            self.input_file = input_file
            self.last_modify_time = Path(self.input_file).stat().st_mtime

        self.track: Track
        super().__init__(current_show_url)

    def handles(self: Self, event: CloudEvent | None) -> bool:
        """Check if we need to handle the event."""
        # TODO(hairmare): v3-prep call :meth:`handle_id` from here
        #                 needs saemubox_id compat workaround
        # https://github.com/radiorabe/nowplaying/issues/180
        # TODO(hairmare): v3 remove call to :meth:`handle_id`
        # https://github.com/radiorabe/nowplaying/issues/179
        # TODO(hairmare): make magic string configurable
        # https://github.com/radiorabe/nowplaying/issues/179
        # TODO(hairmare): check if source is currently on-air
        # https://github.com/radiorabe/nowplaying/issues/179
        if not event:  # pragma: no coverage
            # TODO(hairmare): remove checking for None once only events exist
            # https://github.com/radiorabe/nowplaying/issues/179
            return False
        return event["source"] == "https://github/radiorabe/klangbecken"

    def event(self: Self, event: CloudEvent) -> None:
        """Handle event."""
        self._handle(event)

    def handle_id(
        self: Self,
        saemubox_id: int,
        event: CloudEvent | None = None,
    ) -> bool:
        """Handle ID."""
        # only handle Klangbecken output
        if saemubox_id == 1:
            return True

        # TODO(hairmare): v3-prep make this get called from :meth:`handles`
        # https://github.com/radiorabe/nowplaying/issues/180
        return self.handles(event)

    def handle(self: Self, event: CloudEvent | None = None) -> None:
        """Handle RaBe CloudEevent."""
        self._handle(event)

    def _handle(self: Self, event: CloudEvent | None = None) -> None:
        """Handle actual RaBe CloudEevent.

        TODO(hairmare): v3: move into :meth:`event`
                        once :meth:`handle` and :meth:`handle_id` have been yeeted
        https://github.com/radiorabe/nowplaying/issues/179
        TODO(hairmare): v3: remove all refs to input_file and it's modify time
                        once we use event handlers
        https://github.com/radiorabe/nowplaying/issues/179
        """
        if not event:
            # @TODO(hairmare): replace the stat method with inotify
            # https://github.com/radiorabe/nowplaying/issues/179
            modify_time = Path(self.input_file).stat().st_mtime

        # @TODO(hairmare): Need to check if we have a stale file and send default
        #                  track infos in this case. This might happend if loopy
        #                  went out for lunch...
        #                  pseudo code: now > modify_time + self.track.get_duration()
        # https://github.com/radiorabe/nowplaying/issues/179

        if self.first_run or event or modify_time > self.last_modify_time:
            logger.info("Now playing file changed")

            self.show = self.showclient.get_show_info()
            if not event:
                self.last_modify_time = modify_time

            if event:
                self.track = self.parse_event(event)
                self.first_run = False

            logger.info("First run: %s", self.first_run)

            if not self.first_run:  # pragma: no coverage
                # TODO(hairmare): test once we don't have to care about
                #                 mtime/inotify because it's an api
                # https://github.com/radiorabe/nowplaying/issues/179
                logger.info("calling track_finished")
                self.track_handler.track_finished(self.track)

            if not event:
                # TODO(hairmare): remove once legacy xml is gone
                # https://github.com/radiorabe/nowplaying/issues/179
                self.track = self.get_track_info()

            # Klangbecken acts as a failover and last resort input, if other
            # active inputs are silent or have problems.
            # Therefore the show's name should always be Klangbecken, regardless
            # of what loopy thinks.
            if self.show.name != self._SHOW_NAME_KLANGBECKEN:
                logger.info(
                    "Klangbecken Input active, overriding current show '%s' with '%s'",
                    self.show.name,
                    self._SHOW_NAME_KLANGBECKEN,
                )

                self.show = Show()
                self.show.set_name(self._SHOW_NAME_KLANGBECKEN)
                self.show.set_url(self._SHOW_URL_KLANGBECKEN)

                # Set the show's end time to the one of the track, as we have
                # no idea for how long the Klangbecken input will be active.
                # The show's start time is initially set to now.
                self.show.set_endtime(self.track.endtime)

            self.track.set_show(self.show)

            # TODO(hairmare): or finished?
            # https://github.com/radiorabe/nowplaying/issues/179
            self.track_handler.track_started(self.track)

            self.first_run = False

    def get_track_info(self: Self) -> Track:
        """Get Track info."""
        # TODO(hairmare): v3 remove method once legacy xml is gone
        # https://github.com/radiorabe/nowplaying/issues/179
        dom = xml.dom.minidom.parse(self.input_file)  # noqa: S318

        # default track info
        track_info = {
            "artist": DEFAULT_ARTIST,
            "title": DEFAULT_TITLE,
            "album": "",
            "track": "",
            "time": "",
        }

        songs = dom.getElementsByTagName("song")

        if len(songs) == 0 or songs[0].hasChildNodes() is False:  # pragma: no coverage
            # TODO(hairmare): replace with non generic exception and test
            # https://github.com/radiorabe/nowplaying/issues/179
            raise Exception(_EXCEPTION_INPUT_MISSING_SONG_TAG)  # noqa: TRY002

        song = songs[0]

        for name in list(track_info.keys()):
            elements = song.getElementsByTagName(name)

            if len(elements) == 0:  # pragma: no coverage
                # TODO(hairmare): replace with non generic exception and test
                # https://github.com/radiorabe/nowplaying/issues/179
                raise Exception("No <%s> tag found" % name)  # noqa: TRY002, UP031
            if elements[0].hasChildNodes():
                element_data = elements[0].firstChild.data.strip()  # type: ignore[attr-defined,union-attr]

                if element_data != "":
                    track_info[name] = element_data
                else:  # pragma: no coverage
                    logger.info("Element %s has empty value, ignoring", name)

        if not song.hasAttribute("timestamp"):  # pragma: no coverage
            # TODO(hairmare): replace with non generic exception and test
            # https://github.com/radiorabe/nowplaying/issues/179
            raise Exception(_EXCEPTION_INPUT_MISSING_TIMESTAMP)  # noqa: TRY002

        # set the start time and append the missing UTC offset
        # @TODO: The UTC offset should be provided by the now playing XML
        #        generated by Thomas
        # ex.: 2012-05-15T09:47:07+02:00
        track_info["start_timestamp"] = song.getAttribute("timestamp") + time.strftime(
            "%z",
        )

        current_track = Track()

        current_track.set_artist(track_info["artist"])
        current_track.set_title(track_info["title"])
        current_track.set_album(track_info["album"])

        # Store as UTC datetime object
        current_track.set_starttime(
            isodate.parse_datetime(track_info["start_timestamp"]).astimezone(
                pytz.timezone("UTC"),
            ),
        )

        current_track.set_duration(int(track_info["time"]))

        return current_track

    def parse_event(self: Self, event: CloudEvent) -> Track:
        """Parse event."""
        track = Track()
        logger.info("Parsing event: %s", event)

        track.set_artist(event.data["item.artist"])
        track.set_title(event.data["item.title"])

        event_time = isodate.parse_datetime(event["time"])
        if event["type"] == "ch.rabe.api.events.track.v1.trackStarted":
            track.set_starttime(event_time)
        elif event["type"] == "ch.rabe.api.events.track.v1.trackFinished":
            track.set_endtime(event_time)

        if "item.length" in event.data:
            track.set_duration(event.data["item.length"])

        logger.info("Track: %s", track)
        return track

__init__(current_show_url, input_file=None)

Create KlangbeckenInputObserver.

Source code in nowplaying/input/observer.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def __init__(
    self: Self,
    current_show_url: str,
    input_file: str | None = None,
) -> None:  # pragma: no coverage
    """Create KlangbeckenInputObserver."""
    # TODO(hairmare): test once input file is replaced with api
    # https://github.com/radiorabe/nowplaying/issues/180
    if input_file:
        warnings.warn(
            "The now-playing.xml format from Loopy/Klangbecken "
            "will be replaced in the future",
            PendingDeprecationWarning,
            stacklevel=2,
        )
        self.input_file = input_file
        self.last_modify_time = Path(self.input_file).stat().st_mtime

    self.track: Track
    super().__init__(current_show_url)

event(event)

Handle event.

Source code in nowplaying/input/observer.py 
132
133
134
def event(self: Self, event: CloudEvent) -> None:
    """Handle event."""
    self._handle(event)

get_track_info()

Get Track info.

Source code in nowplaying/input/observer.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def get_track_info(self: Self) -> Track:
    """Get Track info."""
    # TODO(hairmare): v3 remove method once legacy xml is gone
    # https://github.com/radiorabe/nowplaying/issues/179
    dom = xml.dom.minidom.parse(self.input_file)  # noqa: S318

    # default track info
    track_info = {
        "artist": DEFAULT_ARTIST,
        "title": DEFAULT_TITLE,
        "album": "",
        "track": "",
        "time": "",
    }

    songs = dom.getElementsByTagName("song")

    if len(songs) == 0 or songs[0].hasChildNodes() is False:  # pragma: no coverage
        # TODO(hairmare): replace with non generic exception and test
        # https://github.com/radiorabe/nowplaying/issues/179
        raise Exception(_EXCEPTION_INPUT_MISSING_SONG_TAG)  # noqa: TRY002

    song = songs[0]

    for name in list(track_info.keys()):
        elements = song.getElementsByTagName(name)

        if len(elements) == 0:  # pragma: no coverage
            # TODO(hairmare): replace with non generic exception and test
            # https://github.com/radiorabe/nowplaying/issues/179
            raise Exception("No <%s> tag found" % name)  # noqa: TRY002, UP031
        if elements[0].hasChildNodes():
            element_data = elements[0].firstChild.data.strip()  # type: ignore[attr-defined,union-attr]

            if element_data != "":
                track_info[name] = element_data
            else:  # pragma: no coverage
                logger.info("Element %s has empty value, ignoring", name)

    if not song.hasAttribute("timestamp"):  # pragma: no coverage
        # TODO(hairmare): replace with non generic exception and test
        # https://github.com/radiorabe/nowplaying/issues/179
        raise Exception(_EXCEPTION_INPUT_MISSING_TIMESTAMP)  # noqa: TRY002

    # set the start time and append the missing UTC offset
    # @TODO: The UTC offset should be provided by the now playing XML
    #        generated by Thomas
    # ex.: 2012-05-15T09:47:07+02:00
    track_info["start_timestamp"] = song.getAttribute("timestamp") + time.strftime(
        "%z",
    )

    current_track = Track()

    current_track.set_artist(track_info["artist"])
    current_track.set_title(track_info["title"])
    current_track.set_album(track_info["album"])

    # Store as UTC datetime object
    current_track.set_starttime(
        isodate.parse_datetime(track_info["start_timestamp"]).astimezone(
            pytz.timezone("UTC"),
        ),
    )

    current_track.set_duration(int(track_info["time"]))

    return current_track

handle(event=None)

Handle RaBe CloudEevent.

Source code in nowplaying/input/observer.py 
150
151
152
def handle(self: Self, event: CloudEvent | None = None) -> None:
    """Handle RaBe CloudEevent."""
    self._handle(event)

handle_id(saemubox_id, event=None)

Handle ID.

Source code in nowplaying/input/observer.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def handle_id(
    self: Self,
    saemubox_id: int,
    event: CloudEvent | None = None,
) -> bool:
    """Handle ID."""
    # only handle Klangbecken output
    if saemubox_id == 1:
        return True

    # TODO(hairmare): v3-prep make this get called from :meth:`handles`
    # https://github.com/radiorabe/nowplaying/issues/180
    return self.handles(event)

handles(event)

Check if we need to handle the event.

Source code in nowplaying/input/observer.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def handles(self: Self, event: CloudEvent | None) -> bool:
    """Check if we need to handle the event."""
    # TODO(hairmare): v3-prep call :meth:`handle_id` from here
    #                 needs saemubox_id compat workaround
    # https://github.com/radiorabe/nowplaying/issues/180
    # TODO(hairmare): v3 remove call to :meth:`handle_id`
    # https://github.com/radiorabe/nowplaying/issues/179
    # TODO(hairmare): make magic string configurable
    # https://github.com/radiorabe/nowplaying/issues/179
    # TODO(hairmare): check if source is currently on-air
    # https://github.com/radiorabe/nowplaying/issues/179
    if not event:  # pragma: no coverage
        # TODO(hairmare): remove checking for None once only events exist
        # https://github.com/radiorabe/nowplaying/issues/179
        return False
    return event["source"] == "https://github/radiorabe/klangbecken"

parse_event(event)

Parse event.

Source code in nowplaying/input/observer.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def parse_event(self: Self, event: CloudEvent) -> Track:
    """Parse event."""
    track = Track()
    logger.info("Parsing event: %s", event)

    track.set_artist(event.data["item.artist"])
    track.set_title(event.data["item.title"])

    event_time = isodate.parse_datetime(event["time"])
    if event["type"] == "ch.rabe.api.events.track.v1.trackStarted":
        track.set_starttime(event_time)
    elif event["type"] == "ch.rabe.api.events.track.v1.trackFinished":
        track.set_endtime(event_time)

    if "item.length" in event.data:
        track.set_duration(event.data["item.length"])

    logger.info("Track: %s", track)
    return track

NonKlangbeckenInputObserver

Bases: InputObserver

Observer for input not from klangbecken ie. w/o track information.

Uses the show's name instead of the actual track infos

Source code in nowplaying/input/observer.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
class NonKlangbeckenInputObserver(InputObserver):
    """Observer for input not from klangbecken ie. w/o track information.

    Uses the show's name instead of the actual track infos
    """

    def handles(self: Self, _: CloudEvent) -> bool:  # pragma: no coverage
        """Do not handle events yet.

        TODO implement this method
        TODO v3-prep call :meth:`handle_id` from here
             (needs saemubox_id compat workaround)
        TODO v3 remove call to :meth:`handle_id`:
        """
        return False

    def event(self: Self, event: CloudEvent) -> None:  # pragma: no coverage
        """Do not handle events yet.

        TODO implement this method
        """

    def handle_id(self: Self, saemubox_id: int, _: CloudEvent | None = None) -> bool:
        """Handle new ID from Saemubox."""
        if saemubox_id != self.previous_saemubox_id:
            # If sämubox changes, force a show update, this acts as
            # a self-healing measurement in case the show web service provides
            # nonsense ;)
            self.show = self.showclient.get_show_info(force_update=True)

        self.previous_saemubox_id = saemubox_id

        # only handle non-Klangbecken
        return saemubox_id != 1

    def handle(self: Self, _: CloudEvent | None = None) -> None:
        """Handle Track."""
        self.show = self.showclient.get_show_info()

        # only handle if a new show has started
        if self.show.uuid != self.previous_show_uuid:
            logger.info("Show changed")
            self.track_handler.track_started(self.get_track_info())
            self.previous_show_uuid = self.show.uuid

    def get_track_info(self: Self) -> Track:
        """Get Track info."""
        current_track = Track()

        current_track.set_artist(DEFAULT_ARTIST)
        current_track.set_title(DEFAULT_TITLE)

        # Set the track's start/end time to the start/end time of the show
        current_track.set_starttime(self.show.starttime)
        current_track.set_endtime(self.show.endtime)

        current_track.set_show(self.show)

        return current_track

event(event)

Do not handle events yet.

TODO implement this method

Source code in nowplaying/input/observer.py 
334
335
336
337
338
def event(self: Self, event: CloudEvent) -> None:  # pragma: no coverage
    """Do not handle events yet.

    TODO implement this method
    """

get_track_info()

Get Track info.

Source code in nowplaying/input/observer.py 
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def get_track_info(self: Self) -> Track:
    """Get Track info."""
    current_track = Track()

    current_track.set_artist(DEFAULT_ARTIST)
    current_track.set_title(DEFAULT_TITLE)

    # Set the track's start/end time to the start/end time of the show
    current_track.set_starttime(self.show.starttime)
    current_track.set_endtime(self.show.endtime)

    current_track.set_show(self.show)

    return current_track

handle(_=None)

Handle Track.

Source code in nowplaying/input/observer.py 
353
354
355
356
357
358
359
360
361
def handle(self: Self, _: CloudEvent | None = None) -> None:
    """Handle Track."""
    self.show = self.showclient.get_show_info()

    # only handle if a new show has started
    if self.show.uuid != self.previous_show_uuid:
        logger.info("Show changed")
        self.track_handler.track_started(self.get_track_info())
        self.previous_show_uuid = self.show.uuid

handle_id(saemubox_id, _=None)

Handle new ID from Saemubox.

Source code in nowplaying/input/observer.py 
340
341
342
343
344
345
346
347
348
349
350
351
def handle_id(self: Self, saemubox_id: int, _: CloudEvent | None = None) -> bool:
    """Handle new ID from Saemubox."""
    if saemubox_id != self.previous_saemubox_id:
        # If sämubox changes, force a show update, this acts as
        # a self-healing measurement in case the show web service provides
        # nonsense ;)
        self.show = self.showclient.get_show_info(force_update=True)

    self.previous_saemubox_id = saemubox_id

    # only handle non-Klangbecken
    return saemubox_id != 1

handles(_)

Do not handle events yet.

TODO implement this method TODO v3-prep call :meth:handle_id from here (needs saemubox_id compat workaround) TODO v3 remove call to :meth:handle_id:

Source code in nowplaying/input/observer.py 
324
325
326
327
328
329
330
331
332
def handles(self: Self, _: CloudEvent) -> bool:  # pragma: no coverage
    """Do not handle events yet.

    TODO implement this method
    TODO v3-prep call :meth:`handle_id` from here
         (needs saemubox_id compat workaround)
    TODO v3 remove call to :meth:`handle_id`:
    """
    return False