First wave of refactoring

pyomxplayer/__init__.py

@@ -0,0 +1,78 @@
+import re
+from threading import Thread
+from time import sleep
+
+import pexpect
+
+from pyomxplayer.parser import OMXPlayerParser
+
+
+class OMXPlayer(object):
+    _STATUS_REGEX = re.compile(r'A:\s*[\d.]+\s-?([\d.]+).*')
+    _DONE_REGEX = re.compile(r'have a nice day.*')
+
+    _LAUNCH_CMD = 'omxplayer -s %s %s'
+    _PAUSE_CMD = 'p'
+    _TOGGLE_SUB_CMD = 's'
+    _QUIT_CMD = 'q'
+
+
+
+    def __init__(self, media_file, args=None, start_playback=False,
+                 _parser=OMXPlayerParser, _spawn=pexpect.spawn):
+        self.subtitles_visible = True
+        self._spawn = _spawn
+        self._launch_omxplayer(media_file, args)
+        self.parser = _parser(self._process)
+        self._monitor_play_position()
+
+        self.position = 0.0
+
+        # By default the process starts playing
+        self.paused = False
+        if not start_playback:
+            self.toggle_pause()
+        self.toggle_subtitles()
+
+    def _launch_omxplayer(self, media_file, args):
+        if not args:
+            args = ''
+        cmd = self._LAUNCH_CMD % (media_file, args)
+        self._process = self._spawn(cmd)
+
+    def _monitor_play_position(self):
+        self._position_thread = Thread(target=self._get_position)
+        self._position_thread.start()
+
+    def _get_position(self):
+        while True:
+            index = self._process.expect([self._STATUS_REGEX,
+                                          pexpect.TIMEOUT,
+                                          pexpect.EOF,
+                                          self._DONE_REGEX])
+            def timed_out():
+                return index == 1
+
+            def process_finished():
+                return index in (2, 3)
+
+            if timed_out():
+                continue
+            elif process_finished():
+                break
+            else:
+                # Process is still running (happy path)
+                self.position = float(self._process.match.group(1))
+            sleep(0.05)
+
+    def toggle_pause(self):
+        if self._process.send(self._PAUSE_CMD):
+            self.paused = not self.paused
+
+    def toggle_subtitles(self):
+        if self._process.send(self._TOGGLE_SUB_CMD):
+            self.subtitles_visible = not self.subtitles_visible
+
+    def stop(self):
+        self._process.send(self._QUIT_CMD)
+        self._process.terminate(force=True)

pyomxplayer/parser.py

@@ -0,0 +1,33 @@
+import re
+
+
+class OMXPlayerParser(object):
+    _VIDEO_PROPERTIES_REGEX = re.compile(r'.*Video codec ([\w-]+) width (\d+) height (\d+) profile (\d+) fps ([\d.]+).*')
+    _AUDIO_PROPERTIES_REGEX = re.compile(r'Audio codec (\w+) channels (\d+) samplerate (\d+) bitspersample (\d+).*')
+
+    def __init__(self, process):
+        self._process = process
+        self.video = dict()
+        self.audio = dict()
+        self._parse_properties()
+
+    def _parse_properties(self):
+        self._parse_video_properties()
+        self._parse_audio_properties()
+
+    def _parse_video_properties(self):
+        matches = self._VIDEO_PROPERTIES_REGEX.match(self._process.readline())
+        if matches:
+            video_props = matches.groups()
+            self.video['decoder'] = video_props[0]
+            self.video['dimensions'] = tuple(int(x) for x in video_props[1:3])
+            self.video['profile'] = int(video_props[3])
+            self.video['fps'] = float(video_props[4])
+
+    def _parse_audio_properties(self):
+        matches = self._AUDIO_PROPERTIES_REGEX.match(self._process.readline())
+        if matches:
+            audio_props = matches.groups()
+            self.audio['decoder'] = audio_props[0]
+            (self.audio['channels'], self.audio['rate'],
+             self.audio['bps']) = [int(x) for x in audio_props[1:]]

run_tests.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+nosetests tests

setup.py

@@ -1,7 +1,7 @@
 import distutils.core
 
 distutils.core.setup(
-    name="pyomxplayer",
-    packages = ["."],
-    requires = ['pexpect (>= 2.4)'],
-    )
+    name='pyomxplayer',
+    packages=['pyomxplayer'],
+    requires=['pexpect (>= 2.4)'],
+)

tests/acceptance_tests.py

@@ -0,0 +1,13 @@
+import unittest
+import time
+
+from pyomxplayer import OMXPlayer
+
+
+class AcceptanceTest(unittest.TestCase):
+    def test_opening_mp4_file(self):
+       player = OMXPlayer('./tests/test.mp4')
+       player.toggle_pause()
+       time.sleep(1)
+       player.stop()
+       self.assertTrue("Did not complete playing example without errors")

tests/omxplayer_tests.py

@@ -0,0 +1,20 @@
+import unittest
+from mock import Mock
+
+from pyomxplayer import OMXPlayer
+
+EXAMPLE_OMXPLAYER_OUTPUT = """Video codec omx-h264 width 1280 height 720 profile 77 fps 25.000000
+Audio codec aac channels 2 samplerate 48000 bitspersample 16
+Subtitle count: 0, state: off, index: 1, delay: 0
+V:PortSettingsChanged: 1280x720@0.04 interlace:0 deinterlace:0 par:1.00 layer:0  0k
+M:  545774 V:  5.20s   3120k/  4800k A:  5.24   5.09s/  6.07s Cv:     0k Ca:     0k
+"""
+
+
+class OMXPlayerTests(unittest.TestCase):
+    def test_paused_at_startup(self):
+        mock_process = Mock()
+        mock_process.readline = Mock(return_value=EXAMPLE_OMXPLAYER_OUTPUT)
+        mock_spawn = Mock(return_value=mock_process)
+        player = OMXPlayer("", _spawn=mock_spawn)
+        self.assertTrue(player.paused)

tests/parsing_tests.py

@@ -0,0 +1,42 @@
+import unittest
+from mock import Mock
+from nose_parameterized import parameterized
+
+from pyomxplayer.parser import OMXPlayerParser
+
+
+class VideoParsingTests(unittest.TestCase):
+    EXAMPLE_VIDEO_INFORMATION = 'Video codec omx-h264 width 1280 height 720 profile 77 fps 25.000000'
+
+    def test_file_with_no_file_data_doesnt_raise_an_error(self):
+        process = Mock()
+        process.readline = Mock(return_value='')
+        OMXPlayerParser(process)
+
+    @parameterized.expand([
+        ('omx-h264', 'decoder'),
+        ((1280, 720), 'dimensions'),
+        (77, 'profile'),
+        (25.0, 'fps'),
+    ])
+    def test_parsing_to_property_conversion(self, expected_value, property):
+        process = Mock()
+        process.readline = Mock(return_value=self.EXAMPLE_VIDEO_INFORMATION)
+        parser = OMXPlayerParser(process)
+        self.assertEqual(expected_value, parser.video[property])
+
+
+class AudioParsingTests(unittest.TestCase):
+    EXAMPLE_AUDIO_INFORMATION = "Audio codec aac channels 2 samplerate 48000 bitspersample 16"
+
+    @parameterized.expand([
+        ('aac', 'decoder'),
+        (2, 'channels'),
+        (48000, 'rate'),
+        (16, 'bps'),
+    ])
+    def test_parsing_to_property_conversion(self, expected_value, property):
+        process = Mock()
+        process.readline = Mock(return_value=self.EXAMPLE_AUDIO_INFORMATION)
+        parser = OMXPlayerParser(process)
+        self.assertEqual(expected_value, parser.audio[property])