Internet radio browser GUI for music/video streams from various directory services.

⌈⌋ ⎇ branch:  streamtuner2


Diff

Differences From Artifact [fc0650246a]:

To Artifact [a3c14ebdbb]:

  • File action.py — part of check-in [1beab0563e] at 2014-04-10 04:31:02 on branch py3 — * Fixed gtk_list_store_get_value: assertion `column < list_store->n_columns' by removing {width:20} reference from treeview datamap. * row.setdefault() for absent search_col/set and deleted state * More __print__/dbg colorization * Disabled pson.filter_data in favour of str casting in mygtk.columns() * Removed streamactions.popup PY2/PY3 workaround with named args * More .iteritems() removal (user: mario, size: 10387) [annotate] [blame] [check-ins using] [more...]

19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
#
#
#


import re
import os
import http
from config import conf, __print__, dbg
import platform


main = None









|







19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
#
#
#


import re
import os
import ahttp as http
from config import conf, __print__, dbg
import platform


main = None


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
        # media formats
        mf = {"mp3":"audio/mp3", "ogg":"audio/ogg", "aac":"audio/aac"}
        
        
        # web
        @staticmethod
        def browser(url):
            __print__( conf.browser )
            action.run(conf.browser + " " + action.quote(url))


            
        # os shell cmd escaping
        @staticmethod
        def quote(s):
            if conf.windows:
                return s   # should actually be "\\\"%s\\\"" % s
            else:
                return "%r" % s


        # calls player for stream url and format
        @staticmethod
        def play(url, audioformat="audio/mp3", listformat="text/x-href"):
            if (url):
                url = action.url(url, listformat)
            if (audioformat):
                if audioformat == "audio/mpeg":
                    audioformat = "audio/mp3"  # internally we use the more user-friendly moniker
                cmd = conf.play.get(audioformat, conf.play.get("*/*", "vlc %u"))
                __print__( "play", url, cmd )
            try:
                action.run( action.interpol(cmd, url) )
            except:
                pass

        
        # exec wrapper
        @staticmethod
        def run(cmd):
            if conf.windows:
 	        os.system("start \"%s\"")
 	    else:
                os.system(cmd + " &")


        # streamripper
        @staticmethod
        def record(url, audioformat="audio/mp3", listformat="text/x-href", append="", row={}):
            __print__( "record", url )
            cmd = conf.record.get(audioformat, conf.record.get("*/*", None))
            try: action.run( action.interpol(cmd, url, row) + append )
            except: pass


        # save as .m3u
        @staticmethod







|








|

|











|










|
|






|







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
        # media formats
        mf = {"mp3":"audio/mp3", "ogg":"audio/ogg", "aac":"audio/aac"}
        
        
        # web
        @staticmethod
        def browser(url):
            __print__( dbg.CONF, conf.browser )
            action.run(conf.browser + " " + action.quote(url))


            
        # os shell cmd escaping
        @staticmethod
        def quote(s):
            if conf.windows:
                return str(s)   # should actually be "\\\"%s\\\"" % s
            else:
                return "%r" % str(s)


        # calls player for stream url and format
        @staticmethod
        def play(url, audioformat="audio/mp3", listformat="text/x-href"):
            if (url):
                url = action.url(url, listformat)
            if (audioformat):
                if audioformat == "audio/mpeg":
                    audioformat = "audio/mp3"  # internally we use the more user-friendly moniker
                cmd = conf.play.get(audioformat, conf.play.get("*/*", "vlc %u"))
                __print__( dbg.PROC,"play", url, cmd )
            try:
                action.run( action.interpol(cmd, url) )
            except:
                pass

        
        # exec wrapper
        @staticmethod
        def run(cmd):
            if conf.windows:
                os.system("start \"%s\"")
            else:
                os.system(cmd + " &")


        # streamripper
        @staticmethod
        def record(url, audioformat="audio/mp3", listformat="text/x-href", append="", row={}):
            __print__( dbg.PROC, "record", url )
            cmd = conf.record.get(audioformat, conf.record.get("*/*", None))
            try: action.run( action.interpol(cmd, url, row) + append )
            except: pass


        # save as .m3u
        @staticmethod
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
            return url

            
        # download a .pls resource and extract urls
        @staticmethod
        def pls(url):
            text = http.get(url)
            __print__( "pls_text=", text )
            return re.findall("\s*File\d*\s*=\s*(\w+://[^\s]+)", text, re.I)
            # currently misses out on the titles            
            
        # get a single direct ICY stream url (extract either from PLS or M3U)
        @staticmethod
        def srv(url):
            return action.extract_urls(url)[0]







|







187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
            return url

            
        # download a .pls resource and extract urls
        @staticmethod
        def pls(url):
            text = http.get(url)
            __print__( dbg.DATA, "pls_text=", text )
            return re.findall("\s*File\d*\s*=\s*(\w+://[^\s]+)", text, re.I)
            # currently misses out on the titles            
            
        # get a single direct ICY stream url (extract either from PLS or M3U)
        @staticmethod
        def srv(url):
            return action.extract_urls(url)[0]
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
            # use shoutcast unique stream id if available
            stream_id = re.search("http://.+?/.*?(\d+)", pls, re.M)
            stream_id = stream_id and stream_id.group(1) or "XXXXXX"
            try:
                channelname = main.current_channel
            except:
                channelname = "unknown"
            return (conf.tmp + os.sep + "streamtuner2."+channelname+"."+stream_id+".m3u", len(stream_id) > 3 and stream_id != "XXXXXX")
        
        # check if there are any urls in a given file
        @staticmethod
        def has_urls(tmp_fn):
            if os.path.exists(tmp_fn):
                return open(tmp_fn, "r").read().find("http://") > 0
            
        
        # create a local .m3u file from it
        @staticmethod
        def m3u(pls):
        
            # temp filename
            (tmp_fn, unique) = action.tmp_fn(pls)
            # does it already exist?
            if tmp_fn and unique and conf.reuse_m3u and action.has_urls(tmp_fn):
                return tmp_fn

            # download PLS
            __print__( "pls=",pls )
            url_list = action.extract_urls(pls)
            __print__( "urls=", url_list )

            # output URL list to temporary .m3u file
            if (len(url_list)):
                #tmp_fn = 
                f = open(tmp_fn, "w")
                f.write("#M3U\n")
                f.write("\n".join(url_list) + "\n")
                f.close()
                # return path/name of temporary file
                return tmp_fn
            else:
                __print__( "error, there were no URLs in ", pls )
                raise "Empty PLS"

        # open help browser                
        @staticmethod
        def help(*args):
        
            action.run("yelp /usr/share/doc/streamtuner2/help/")
            #or action.browser("/usr/share/doc/streamtuner2/")

#class action









|



















|

|











|












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
            # use shoutcast unique stream id if available
            stream_id = re.search("http://.+?/.*?(\d+)", pls, re.M)
            stream_id = stream_id and stream_id.group(1) or "XXXXXX"
            try:
                channelname = main.current_channel
            except:
                channelname = "unknown"
            return (str(conf.tmp) + os.sep + "streamtuner2."+channelname+"."+stream_id+".m3u", len(stream_id) > 3 and stream_id != "XXXXXX")
        
        # check if there are any urls in a given file
        @staticmethod
        def has_urls(tmp_fn):
            if os.path.exists(tmp_fn):
                return open(tmp_fn, "r").read().find("http://") > 0
            
        
        # create a local .m3u file from it
        @staticmethod
        def m3u(pls):
        
            # temp filename
            (tmp_fn, unique) = action.tmp_fn(pls)
            # does it already exist?
            if tmp_fn and unique and conf.reuse_m3u and action.has_urls(tmp_fn):
                return tmp_fn

            # download PLS
            __print__( dbg.DATA, "pls=",pls )
            url_list = action.extract_urls(pls)
            __print__( dbg.DATA, "urls=", url_list )

            # output URL list to temporary .m3u file
            if (len(url_list)):
                #tmp_fn = 
                f = open(tmp_fn, "w")
                f.write("#M3U\n")
                f.write("\n".join(url_list) + "\n")
                f.close()
                # return path/name of temporary file
                return tmp_fn
            else:
                __print__( dbg.ERR, "error, there were no URLs in ", pls )
                raise "Empty PLS"

        # open help browser                
        @staticmethod
        def help(*args):
        
            action.run("yelp /usr/share/doc/streamtuner2/help/")
            #or action.browser("/usr/share/doc/streamtuner2/")

#class action