/usr/share/streamtuner2/cli.py is in streamtuner2 2.2.0+dfsg-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 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 | # api: streamtuner2
# title: CLI interface
# description: allows to call streamtuner2 from the commandline
# status: experimental
# version: 0.3
#
# Returns JSON data when queried. Usually returns cache data, but the
# "category" module always loads fresh info from the directory servers.
#
# Not all channel plugins are gtk-free yet. And some workarounds are
# used here to not upset channel plugins about a missing parent window.
import sys
#from channels import *
import ahttp
import action
from config import *
import json
# CLI
class StreamTunerCLI (object):
# plugin info
title = "CLI interface"
version = 0.3
# channel plugins
channel_modules = ["shoutcast", "xiph", "icast", "jamendo", "radiobrowser"]# module_list()
current_channel = "cli"
plugins = {} # only populated sparsely by .stream()
# start
def __init__(self, actions):
# fake init
action.main = empty_parent()
action.main.current_channel = self.current_channel
# check if enough arguments, else help
if not actions:
a = self.help
# first cmdline arg == action
else:
command = actions[0]
if command in dir(self):
cmd = self.__getattribute__(command)
else:
log.ERR("No such command:", command)
return
# run
result = cmd(*actions[1:])
if result:
self.json(result)
# show help
def help(self, *args):
print("""
syntax: streamtuner2 action [channel] "stream title"
from cache:
streamtuner2 stream shoutcast frequence
streamtuner2 dump xiph
streamtuner2 play "..."
streamtuner2 url "..."
load fresh:
streamtuner2 category shoutcast "Top 40"
streamtuner2 categories xiph
""")
# prints stream data from cache
def stream(self, *args):
# optional channel name, title
if len(args) > 1:
(channel_list, title) = args
channel_list = channel_list.split(",")
else:
title = list(args).pop()
channel_list = self.channel_modules
# walk through channel plugins, categories, rows
title = title.lower()
for channel in channel_list:
self.current_channel = channel
c = self.channel(channel)
self.plugins[channel] = c
c.cache()
for cat in c.streams:
for row in c.streams[cat]:
if row and row.get("title","").lower().find(title)>=0:
return(row)
# just get url
def url(self, *args):
row = self.stream(*args)
if row.get("url"):
print(row["url"])
# run player
def play(self, *args):
row = self.stream(*args)
if row.get("url"):
#action.action.play(row["url"], audioformat=row.get("format","audio/mpeg"))
self.plugins[self.current_channel].play(row)
# return cache data 1:1
def dump(self, channel):
c = self.channel(channel)
c.cache()
return c.streams
# load from server
def category(self, module, cat):
c = self.channel(module)
r = c.update_streams(cat)
[c.postprocess(row) for row in r]
return r
# load from server
def categories(self, module):
c = self.channel(module)
c.cache()
r = c.update_categories()
if not r:
r = c.categories
if c.__dict__.get("empty"):
del r[0]
return r
# load module
def channel(self, module):
plugin = __import__("channels."+module, None, None, [""])
plugin_class = plugin.__dict__[module]
p = plugin_class(None)
p.parent = empty_parent()
return p
# load all channel modules
def channels(self, channels=None):
if channels:
channels = channels.split(",")
else:
channels = self.channel_modules
return channels#(self.channel(module) for module in channels)
# pretty print json
def json(self, dat):
print(json.dumps(dat, sort_keys=True, indent=2))
# trap for some main window calls
class empty_parent (object):
channel = {}
null = lambda *a: None
status = lambda *a: None
thread = lambda *a: None
|