Source code for mycroft.skills.audioservice

# Copyright 2017 Mycroft AI Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from os.path import abspath

from mycroft.messagebus.message import Message


def ensure_uri(s):
    """Interprete paths as file:// uri's.

    Arguments:
        s: string to be checked

    Returns:
        if s is uri, s is returned otherwise file:// is prepended
    """
    if isinstance(s, str):
        if '://' not in s:
            return 'file://' + abspath(s)
        else:
            return s
    elif isinstance(s, (tuple, list)):
        if '://' not in s[0]:
            return 'file://' + abspath(s[0]), s[1]
        else:
            return s
    else:
        raise ValueError('Invalid track')


[docs]class AudioService: """AudioService class for interacting with the audio subsystem Arguments: bus: Mycroft messagebus connection """ def __init__(self, bus): self.bus = bus
[docs] def queue(self, tracks=None): """Queue up a track to playing playlist. Arguments: tracks: track uri or list of track uri's Each track can be added as a tuple with (uri, mime) to give a hint of the mime type to the system """ tracks = tracks or [] if isinstance(tracks, (str, tuple)): tracks = [tracks] elif not isinstance(tracks, list): raise ValueError tracks = [ensure_uri(t) for t in tracks] self.bus.emit(Message('mycroft.audio.service.queue', data={'tracks': tracks}))
[docs] def play(self, tracks=None, utterance=None, repeat=None): """Start playback. Arguments: tracks: track uri or list of track uri's Each track can be added as a tuple with (uri, mime) to give a hint of the mime type to the system utterance: forward utterance for further processing by the audio service. repeat: if the playback should be looped """ repeat = repeat or False tracks = tracks or [] utterance = utterance or '' if isinstance(tracks, (str, tuple)): tracks = [tracks] elif not isinstance(tracks, list): raise ValueError tracks = [ensure_uri(t) for t in tracks] self.bus.emit(Message('mycroft.audio.service.play', data={'tracks': tracks, 'utterance': utterance, 'repeat': repeat}))
[docs] def stop(self): """Stop the track.""" self.bus.emit(Message('mycroft.audio.service.stop'))
[docs] def next(self): """Change to next track.""" self.bus.emit(Message('mycroft.audio.service.next'))
[docs] def prev(self): """Change to previous track.""" self.bus.emit(Message('mycroft.audio.service.prev'))
[docs] def pause(self): """Pause playback.""" self.bus.emit(Message('mycroft.audio.service.pause'))
[docs] def resume(self): """Resume paused playback.""" self.bus.emit(Message('mycroft.audio.service.resume'))
[docs] def seek(self, seconds=1): """Seek X seconds. Arguments: seconds (int): number of seconds to seek, if negative rewind """ if seconds < 0: self.seek_backward(abs(seconds)) else: self.seek_forward(seconds)
[docs] def seek_forward(self, seconds=1): """Skip ahead X seconds. Arguments: seconds (int): number of seconds to skip """ self.bus.emit(Message('mycroft.audio.service.seek_forward', {"seconds": seconds}))
[docs] def seek_backward(self, seconds=1): """Rewind X seconds Arguments: seconds (int): number of seconds to rewind """ self.bus.emit(Message('mycroft.audio.service.seek_backward', {"seconds": seconds}))
[docs] def track_info(self): """Request information of current playing track. Returns: Dict with track info. """ info = self.bus.wait_for_response( Message('mycroft.audio.service.track_info'), reply_type='mycroft.audio.service.track_info_reply', timeout=5) return info.data if info else {}
[docs] def available_backends(self): """Return available audio backends. Returns: dict with backend names as keys """ msg = Message('mycroft.audio.service.list_backends') response = self.bus.wait_for_response(msg) return response.data if response else {}
@property def is_playing(self): """True if the audioservice is playing, else False.""" return self.track_info() != {}