#!/usr/bin/env python# gstfile.py# (c) 2005 Edward Hervey <edward at fluendo dot com># Discovers and prints out multimedia information of files# This example shows how to use gst-python:# _ in an object-oriented way (Discoverer class)# _ subclassing a gst.Pipeline# _ and overidding existing methods (do_iterate())import osimport sysimport gobjectimport gstdef time_to_string(value): """ transform a value in nanoseconds into a human-readable string """ ms = value / gst.MSECOND sec = ms / 1000 ms = ms % 1000 min = sec / 60 sec = sec % 60 return "%2dm %2ds %3d" % (min, sec, ms)class Discoverer(gst.Pipeline): """ Discovers information about files """ mimetype = None audiocaps = {} videocaps = {} videowidth = 0 videoheight = 0 videorate = 0 audiofloat = False audiorate = 0 audiodepth = 0 audiowidth = 0 audiochannels = 0 audiolength = 0L videolength = 0L is_video = False is_audio = False otherstreams = [] finished = False tags = {} def __init__(self, filename): gobject.GObject.__init__(self) self.mimetype = None self.audiocaps = {} self.videocaps = {} self.videowidth = 0 self.videoheight = 0 self.videorate = 0 self.audiofloat = False self.audiorate = 0 self.audiodepth = 0 self.audiowidth = 0 self.audiochannels = 0 self.audiolength = 0L self.videolength = 0L self.is_video = False self.is_audio = False self.otherstreams = [] self.finished = False self.tags = {} if not os.path.isfile(filename): self.finished = True return # the initial elements of the pipeline self.src = gst.element_factory_make("filesrc") self.src.set_property("location", filename) self.src.set_property("blocksize", 1000000) self.dbin = gst.element_factory_make("decodebin") self.add_many(self.src, self.dbin) self.src.link(self.dbin) self.typefind = self.dbin.get_by_name("typefind") # callbacks self.typefind.connect("have-type", self._have_type_cb) self.dbin.connect("new-decoded-pad", self._new_decoded_pad_cb) self.dbin.connect("unknown-type", self._unknown_type_cb) self.dbin.connect("found-tag", self._found_tag_cb) self.discover() def discover(self): """iterate on ourself to find the information on the given file""" if self.finished: return self.set_state(gst.STATE_PLAYING) while 1: if not self.iterate(): break self.set_state(gst.STATE_NULL) self.finished = True def print_info(self): """prints out the information on the given file""" if not self.finished: self.discover() if not self.mimetype: print "Unknown media type" return print "Mime Type :\t", self.mimetype if not self.is_video and not self.is_audio: return print "Length :\t", time_to_string(max(self.audiolength, self.videolength)) print "\tAudio:", time_to_string(self.audiolength), "\tVideo:", time_to_string(self.videolength) if self.is_video: print "Video :" print "\t%d x %d @ %.2f fps" % (self.videowidth, self.videoheight, self.videorate) if self.tags.has_key("video-codec"): print "\tCodec :", self.tags.pop("video-codec") if self.is_audio: print "Audio :" if self.audiofloat: print "\t%d channels(s) : %dHz @ %dbits (float)" % (self.audiochannels, self.audiorate, self.audiowidth) else: print "\t%d channels(s) : %dHz @ %dbits (int)" % (self.audiochannels, self.audiorate, self.audiodepth) if self.tags.has_key("audio-codec"): print "\tCodec :", self.tags.pop("audio-codec") for stream in self.otherstreams: if not stream == self.mimetype: print "Other unsuported Multimedia stream :", stream if self.tags: print "Additional information :" for tag in self.tags.keys(): print "%20s :\t" % tag, self.tags[tag] def _unknown_type_cb(self, dbin, pad, caps): self.otherstreams.append(caps.to_string()) def _have_type_cb(self, typefind, prob, caps): self.mimetype = caps.to_string() def _notify_caps_cb(self, pad, args): caps = pad.get_negotiated_caps() if not caps: return # the caps are fixed # We now get the total length of that stream length = pad.get_peer().query(gst.QUERY_TOTAL, gst.FORMAT_TIME) # We store the caps and length in the proper location if "audio" in caps.to_string(): self.audiocaps = caps self.audiolength = length self.audiorate = caps[0]["rate"] self.audiowidth = caps[0]["width"] self.audiochannels = caps[0]["channels"] if "x-raw-float" in caps.to_string(): self.audiofloat = True else: self.audiodepth = caps[0]["depth"] if (not self.is_video) or self.videocaps: self.finished = True elif "video" in caps.to_string(): self.videocaps = caps self.videolength = length self.videowidth = caps[0]["width"] self.videoheight = caps[0]["height"] self.videorate = caps[0]["framerate"] if (not self.is_audio) or self.audiocaps: self.finished = True def _new_decoded_pad_cb(self, dbin, pad, is_last): # Does the file contain got audio or video ? if "audio" in pad.get_caps().to_string(): self.is_audio = True elif "video" in pad.get_caps().to_string(): self.is_video = True if is_last and not self.is_video and not self.is_audio: self.finished = True return # we connect a fakesink to the new pad... fakesink = gst.element_factory_make("fakesink") self.add(fakesink) sinkpad = fakesink.get_pad("sink") # ... and connect a callback for when the caps are fixed sinkpad.connect("notify::caps", self._notify_caps_cb) pad.link(sinkpad) fakesink.set_state(gst.STATE_PLAYING) def _found_tag_cb(self, dbin, source, tags): self.tags.update(tags) def do_iterate(self): # this overrides the GstBin 'iterate' method # if we have finished discovering we stop the iteration if self.finished: return False # else we call the parent class method return gst.Pipeline.do_iterate(self)gobject.type_register(Discoverer)def main(args): if len(args) < 2: print 'usage: %s files...' % args[0] return 2 if len(args[1:]) > 1: for filename in args[1:]: print "File :", filename Discoverer(filename).print_info() print "\n" else: Discoverer(args[1]).print_info()if __name__ == '__main__': sys.exit(main(sys.argv))