jaseg/python-mpv
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix MpvNode logic to use pröper unions

Browse source 
...instead of lots manual ctypes pointer casting
This commit is contained in:
jaseg 2017-12-24 22:43:13 +01:00
parent c2616054a1
commit 8f5f5c645b
2 changed files with 53 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

22
mpv-test.py
View file

@ -42,7 +42,7 @@ class TestProperties(MpvTestCase):
while self.m.core_idle:
time.sleep(0.05)
for name in sorted(self.m.property_list):
name = name.replace('-', '_')
name = name.replace('-', '_')
with self.subTest(property_name=name), self.swallow_mpv_errors([
mpv.ErrorCode.PROPERTY_UNAVAILABLE,
mpv.ErrorCode.PROPERTY_ERROR,
@ -132,7 +132,6 @@ class TestProperties(MpvTestCase):
def test_property_decoding_invalid_utf8(self):
invalid_utf8 = b'foo\xc3\x28bar'
self.m.alang = invalid_utf8
self.assertEqual(self.m.alang, [invalid_utf8])
self.assertEqual(self.m.raw.alang, [invalid_utf8])
with self.assertRaises(UnicodeDecodeError):
self.m.strict.alang
@ -186,7 +185,7 @@ class ObservePropertyTest(MpvTestCase):
time.sleep(0.1)
m.play(TESTVID)
time.sleep(0.1) #couple frames
time.sleep(0.5) #couple frames
m.unobserve_property('vid', handler)
time.sleep(0.1) #couple frames
@ -228,7 +227,7 @@ class ObservePropertyTest(MpvTestCase):
m.mute = True
m.loop = 'inf'
self.assertEqual(m.mute, True)
self.assertEqual(m.loop, 'inf')
self.assertEqual(m.loop, True)
time.sleep(0.05)
foo.unobserve_mpv_properties()
@ -240,7 +239,7 @@ class ObservePropertyTest(MpvTestCase):
m.terminate() # needed for synchronization of event thread
handler.assert_has_calls([
mock.call('mute', True),
mock.call('loop', 'inf')],
mock.call('loop', True)],
any_order=True)
class KeyBindingTest(MpvTestCase):
@ -373,7 +372,7 @@ class TestLifecycle(unittest.TestCase):
mpv.MPV(this_option_does_not_exists=23)
m = mpv.MPV(osd_level=0, loop='inf', deinterlace=False)
self.assertEqual(m.osd_level, 0)
self.assertEqual(m.loop, 'inf')
self.assertEqual(m.loop, True)
self.assertEqual(m.deinterlace, False)
m.terminate()
@ -401,7 +400,12 @@ class TestLifecycle(unittest.TestCase):
m.play(TESTVID)
m.wait_for_playback()
m.terminate()
handler.assert_any_call('info', 'cplayer', 'Playing: test.webm')
for call in handler.mock_calls:
_1, (a, b, c), _2 = call
if a == 'info' and b == 'cplayer' and c.startswith('Playing: '):
break
else:
self.fail('"Playing: foo..." call not found in log handler calls: '+','.join(repr(call) for call in handler.mock_calls))
class RegressionTests(MpvTestCase):
@ -449,7 +453,7 @@ class RegressionTests(MpvTestCase):
# events. This is a limitation of the upstream API.
time.sleep(0.01)
m.loop = 'inf'
self.assertEqual(m.loop, 'inf')
self.assertEqual(m.loop, True)
time.sleep(0.02)
m.unobserve_property('loop', t.t)
@ -457,7 +461,7 @@ class RegressionTests(MpvTestCase):
m.loop = False
m.loop = 'inf'
m.terminate() # needed for synchronization of event thread
handler.assert_has_calls([mock.call('loop', False), mock.call('loop', 'inf')])
handler.assert_has_calls([mock.call('loop', False), mock.call('loop', True)])
if __name__ == '__main__':

67
mpv.py
View file

@ -191,27 +191,37 @@ class MpvByteArray(Structure):
return cast(self.data, POINTER(c_char))[:self.size]
class MpvNode(Structure):
_fields_ = [('val', c_longlong),
('format', MpvFormat)]
def node_value(self, decoder=identity_decoder):
return MpvNode.node_cast_value(byref(c_void_p(self.val)), self.format.value, decoder)
return MpvNode.node_cast_value(self.val, self.format.value, decoder)
@staticmethod
def node_cast_value(v, fmt=MpvFormat.NODE, decoder=identity_decoder):
return {
MpvFormat.NONE: lambda v: None,
MpvFormat.STRING: lambda v: decoder(cast(v, POINTER(c_char_p)).contents.value),
MpvFormat.OSD_STRING: lambda v: cast(v, POINTER(c_char_p)).contents.value.decode('utf-8'),
MpvFormat.FLAG: lambda v: bool(cast(v, POINTER(c_int)).contents.value),
MpvFormat.INT64: lambda v: cast(v, POINTER(c_longlong)).contents.value,
MpvFormat.DOUBLE: lambda v: cast(v, POINTER(c_double)).contents.value,
MpvFormat.NODE: lambda v: cast(v, POINTER(MpvNode)).contents.node_value(decoder),
MpvFormat.NODE_ARRAY: lambda v: cast(v, POINTER(POINTER(MpvNodeList))).contents.contents.array_value(decoder),
MpvFormat.NODE_MAP: lambda v: cast(v, POINTER(POINTER(MpvNodeList))).contents.contents.dict_value(decoder),
MpvFormat.BYTE_ARRAY: lambda v: cast(v, POINTER(POINTER(MpvByteArray))).contents.contents.bytes_value(),
MpvFormat.STRING: lambda v: decoder(v.string),
MpvFormat.OSD_STRING: lambda v: v.string.decode('utf-8'),
MpvFormat.FLAG: lambda v: bool(v.flag),
MpvFormat.INT64: lambda v: v.int64,
MpvFormat.DOUBLE: lambda v: v.double,
MpvFormat.NODE: lambda v: v.node.contents.node_value(decoder) if v.node else None,
MpvFormat.NODE_ARRAY: lambda v: v.list.contents.array_value(decoder) if v.list else None,
MpvFormat.NODE_MAP: lambda v: v.map.contents.dict_value(decoder) if v.map else None,
MpvFormat.BYTE_ARRAY: lambda v: v.byte_array.contents.bytes_value() if v.byte_array else None,
}[fmt](v)
class MpvNodeUnion(Union):
_fields_ = [('string', c_char_p),
('flag', c_int),
('int64', c_int64),
('double', c_double),
('node', POINTER(MpvNode)),
('list', POINTER(MpvNodeList)),
('map', POINTER(MpvNodeList)),
('byte_array', POINTER(MpvByteArray))]
MpvNode._fields_ = [('val', MpvNodeUnion),
('format', MpvFormat)]
MpvNodeList._fields_ = [('num', c_int),
('values', POINTER(MpvNode)),
('keys', POINTER(c_char_p))]
@ -241,7 +251,7 @@ class MpvEvent(Structure):
class MpvEventProperty(Structure):
_fields_ = [('name', c_char_p),
('format', MpvFormat),
('data', c_void_p)]
('data', MpvNodeUnion)]
def as_dict(self, decoder=identity_decoder):
value = MpvNode.node_cast_value(self.data, self.format.value, decoder)
return {'name': self.name.decode('utf-8'),
@ -409,11 +419,11 @@ def _make_node_str_list(l):
keys=None,
values=( MpvNode * len(l))( *[ MpvNode(
format=MpvFormat.STRING,
val=cast(pointer(p), POINTER(c_longlong)).contents) # NOTE: ctypes is kinda crappy here
val=MpvNodeUnion(string=p))
for p in char_ps ]))
node = MpvNode(
format=MpvFormat.NODE_ARRAY,
val=addressof(node_list))
val=MpvNodeUnion(list=pointer(node_list)))
return char_ps, node_list, node, cast(pointer(node), c_void_p)
@ -607,11 +617,10 @@ class MPV(object):
def node_command(self, name, *args, decoder=strict_decoder):
_1, _2, _3, pointer = _make_node_str_list([name, *args])
out = cast(create_string_buffer(sizeof(MpvNode)), POINTER(MpvNode))
outptr = out #byref(out)
ppointer = cast(pointer, POINTER(MpvNode))
_mpv_command_node(self.handle, ppointer, outptr)
rv = MpvNode.node_cast_value(outptr, MpvFormat.NODE, decoder)
_mpv_free_node_contents(outptr)
_mpv_command_node(self.handle, ppointer, out)
rv = out.contents.node_value(decoder=decoder)
_mpv_free_node_contents(out)
return rv
def seek(self, amount, reference="relative", precision="default-precise"):
@ -1023,14 +1032,18 @@ class MPV(object):
# Property accessors
def _get_property(self, name, decoder=strict_decoder, fmt=MpvFormat.NODE):
out = cast(create_string_buffer(sizeof(c_void_p)), c_void_p)
outptr = byref(out)
out = create_string_buffer(sizeof(MpvNode))
try:
cval = _mpv_get_property(self.handle, name.encode('utf-8'), fmt, outptr)
rv = MpvNode.node_cast_value(outptr, fmt, decoder)
if fmt is MpvFormat.NODE:
_mpv_free_node_contents(outptr)
return rv
cval = _mpv_get_property(self.handle, name.encode('utf-8'), fmt, out)
if fmt is MpvFormat.OSD_STRING:
return cast(out, POINTER(c_char_p)).contents.value.decode('utf-8')
elif fmt is MpvFormat.NODE:
rv = cast(out, POINTER(MpvNode)).contents.node_value(decoder=decoder)
_mpv_free_node_contents(out)
return rv
else:
raise TypeError('_get_property only supports NODE and OSD_STRING formats.')
except PropertyUnavailableError as ex:
return None