raoul/openlp
1
0
Fork 0
forked from openlp/openlp
Code Issues Pull Requests Projects Releases Wiki Activity

- Added PJLINK_DEFAULT_CODES to pjink1 imports

Browse Source 
- Refactor command class checks and methods
- Update PJLink1.get_data for UTF-8
- Added method to clear busy flags and send received data signals
- Added class check on command reply data v. stated projector class compatibility
- Added test for PJLink1.socket_abort
- Added test for PJLink1.poll_loop
- Fix regression in test_projector_process_power_on

--------------------------------
lp:~alisonken1/openlp/pjlink2-b (revision 2735)
[SUCCESS] https...

bzr-revno: 2737
This commit is contained in:
Ken Roberts 2017-05-17 21:35:43 +01:00 committed by Tim Bentley
commit 81be8f5093
3 changed files with 143 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
openlp/core/lib/projector/constants.py
View File

@ -57,20 +57,35 @@ LF = chr(0x0A) # \n
PJLINK_PORT = 4352
TIMEOUT = 30.0
PJLINK_MAX_PACKET = 136
PJLINK_VALID_CMD = {'1': ['PJLINK', # Initial connection
'POWR', # Power option
'INPT', # Video sources option
'AVMT', # Shutter option
'ERST', # Error status option
'LAMP', # Lamp(s) query (Includes fans)
'INST', # Input sources available query
'NAME', # Projector name query
'INF1', # Manufacturer name query
'INF2', # Product name query
'INFO', # Other information query
'CLSS' # PJLink class support query
]}
# NOTE: Change format to account for some commands are both class 1 and 2
PJLINK_VALID_CMD = {
'ACKN': ['2', ], # UDP Reply to 'SRCH'
'AVMT': ['1', ], # Shutter option
'CLSS': ['1', ], # PJLink class support query
'ERST': ['1', '2'], # Error status option
'FILT': ['2', ], # Get current filter usage time
'FREZ': ['2', ], # Set freeze/unfreeze picture being projected
'INF1': ['1', ], # Manufacturer name query
'INF2': ['1', ], # Product name query
'INFO': ['1', ], # Other information query
'INNM': ['2', ], # Get Video source input terminal name
'INPT': ['1', ], # Video sources option
'INST': ['1', ], # Input sources available query
'IRES': ['2', ], # Get Video source resolution
'LAMP': ['1', ], # Lamp(s) query (Includes fans)
'LKUP': ['2', ], # UPD Linkup status notification
'MVOL': ['2', ], # Set microphone volume
'NAME': ['1', ], # Projector name query
'PJLINK': ['1', ], # Initial connection
'POWR': ['1', ], # Power option
'RFIL': ['2', ], # Get replacement air filter model number
'RLMP': ['2', ], # Get lamp replacement model number
'RRES': ['2', ], # Get projector recommended video resolution
'SNUM': ['2', ], # Get projector serial number
'SRCH': ['2', ], # UDP broadcast search for available projectors on local network
'SVER': ['2', ], # Get projector software version
'SVOL': ['2', ] # Set speaker volume
}
# Error and status codes
S_OK = E_OK = 0 # E_OK included since I sometimes forget
# Error codes. Start at 200 so we don't duplicate system error codes.

58
openlp/core/lib/projector/pjlink1.py
View File

@ -53,16 +53,18 @@ from openlp.core.lib.projector.constants import CONNECTION_ERRORS, CR, ERROR_MSG
E_AUTHENTICATION, E_CONNECTION_REFUSED, E_GENERAL, E_INVALID_DATA, E_NETWORK, E_NOT_CONNECTED, \
E_PARAMETER, E_PROJECTOR, E_SOCKET_TIMEOUT, E_UNAVAILABLE, E_UNDEFINED, PJLINK_ERRORS, \
PJLINK_ERST_STATUS, PJLINK_MAX_PACKET, PJLINK_PORT, PJLINK_POWR_STATUS, PJLINK_VALID_CMD, \
STATUS_STRING, S_CONNECTED, S_CONNECTING, S_NETWORK_RECEIVED, S_NETWORK_SENDING, S_NOT_CONNECTED, \
S_OFF, S_OK, S_ON, S_STATUS
PJLINK_DEFAULT_CODES, STATUS_STRING, S_CONNECTED, S_CONNECTING, S_NETWORK_RECEIVED, S_NETWORK_SENDING, \
S_NOT_CONNECTED, S_OFF, S_OK, S_ON, S_STATUS
# Shortcuts
SocketError = QtNetwork.QAbstractSocket.SocketError
SocketSTate = QtNetwork.QAbstractSocket.SocketState
PJLINK_PREFIX = '%'
PJLINK_CLASS = '1'
PJLINK_HEADER = '{prefix}{linkclass}'.format(prefix=PJLINK_PREFIX, linkclass=PJLINK_CLASS)
PJLINK_CLASS = '1' # Default to class 1 until we query the projector
# Add prefix here, but defer linkclass expansion until later when we have the actual
# PJLink class for the command
PJLINK_HEADER = '{prefix}{{linkclass}}'.format(prefix=PJLINK_PREFIX)
PJLINK_SUFFIX = CR
@ -271,8 +273,6 @@ class PJLink1(QtNetwork.QTcpSocket):
self.send_command('INF2', queue=True)
if self.pjlink_name is None:
self.send_command('NAME', queue=True)
if self.power == S_ON and self.source_available is None:
self.send_command('INST', queue=True)
def _get_status(self, status):
"""
@ -349,7 +349,7 @@ class PJLink1(QtNetwork.QTcpSocket):
elif len(read) < 8:
log.warning('({ip}) Not enough data read)'.format(ip=self.ip))
return
data = decode(read, 'ascii')
data = decode(read, 'utf-8')
# Possibility of extraneous data on input when reading.
# Clean out extraneous characters in buffer.
dontcare = self.readLine(self.max_size)
@ -436,20 +436,18 @@ class PJLink1(QtNetwork.QTcpSocket):
if len(data) < 7:
# Not enough data for a packet
log.debug('({ip}) get_data(): Packet length < 7: "{data}"'.format(ip=self.ip, data=data))
self.send_busy = False
self.projectorReceivedData.emit()
return
log.debug('({ip}) get_data(): Checking new data "{data}"'.format(ip=self.ip, data=data))
if data.upper().startswith('PJLINK'):
# Reconnected from remote host disconnect ?
self.check_login(data)
self.send_busy = False
self.projectorReceivedData.emit()
self.receive_data_signal()
return
elif '=' not in data:
log.warning('({ip}) get_data(): Invalid packet received'.format(ip=self.ip))
self.send_busy = False
self.projectorReceivedData.emit()
self.receive_data_signal()
return
log.debug('({ip}) get_data(): Checking new data "{data}"'.format(ip=self.ip, data=data))
# At this point, we should have something to work with
if data.upper().startswith('PJLINK'):
# Reconnected from remote host disconnect ?
self.check_login(data)
self.receive_data_signal()
return
data_split = data.split('=')
try:
@ -458,15 +456,15 @@ class PJLink1(QtNetwork.QTcpSocket):
log.warning('({ip}) get_data(): Invalid packet - expected header + command + data'.format(ip=self.ip))
log.warning('({ip}) get_data(): Received data: "{data}"'.format(ip=self.ip, data=data_in.strip()))
self.change_status(E_INVALID_DATA)
self.send_busy = False
self.projectorReceivedData.emit()
self.receive_data_signal()
return
if not (self.pjlink_class in PJLINK_VALID_CMD and cmd in PJLINK_VALID_CMD[self.pjlink_class]):
if not (cmd in PJLINK_VALID_CMD and class_ in PJLINK_VALID_CMD[cmd]):
log.warning('({ip}) get_data(): Invalid packet - unknown command "{data}"'.format(ip=self.ip, data=cmd))
self.send_busy = False
self.projectorReceivedData.emit()
self.receive_data_signal()
return
if int(self.pjlink_class) < int(class_):
log.warn('({ip}) get_data(): Projector returned class reply higher '
'than projector stated class'.format(ip=self.ip))
return self.process_command(cmd, data)
@QtCore.pyqtSlot(QtNetwork.QAbstractSocket.SocketError)
@ -515,8 +513,10 @@ class PJLink1(QtNetwork.QTcpSocket):
data=opts,
salt='' if salt is None
else ' with hash'))
# TODO: Check for class of command rather than default to projector PJLink class
header = PJLINK_HEADER.format(linkclass=self.pjlink_class)
out = '{salt}{header}{command} {options}{suffix}'.format(salt="" if salt is None else salt,
header=PJLINK_HEADER,
header=header,
command=cmd,
options=opts,
suffix=CR)
@ -997,6 +997,14 @@ class PJLink1(QtNetwork.QTcpSocket):
self.send_command(cmd='AVMT', opts='10')
self.poll_loop()
def receive_data_signal(self):
"""
Clear any busy flags and send data received signal
"""
self.send_busy = False
self.projectorReceivedData.emit()
return
def _not_implemented(self, cmd):
"""
Log when a future PJLink command has not been implemented yet.

83
tests/functional/openlp_core_lib/test_projector_pjlink1.py
View File

@ -23,10 +23,11 @@
Package to test the openlp.core.lib.projector.pjlink1 package.
"""
from unittest import TestCase
from unittest.mock import patch, MagicMock
from unittest.mock import call, patch, MagicMock
from openlp.core.lib.projector.pjlink1 import PJLink1
from openlp.core.lib.projector.constants import E_PARAMETER, ERROR_STRING, S_OFF, S_STANDBY, S_ON, PJLINK_POWR_STATUS
from openlp.core.lib.projector.constants import E_PARAMETER, ERROR_STRING, S_OFF, S_STANDBY, S_ON, \
PJLINK_POWR_STATUS, S_CONNECTED
from tests.resources.projector.data import TEST_PIN, TEST_SALT, TEST_CONNECT_AUTHENTICATE, TEST_HASH
@ -170,6 +171,7 @@ class TestPJLink(TestCase):
# GIVEN: Test object and preset
pjlink = pjlink_test
pjlink.power = S_STANDBY
pjlink.socket_timer = MagicMock()
# WHEN: Call process_command with turn power on command
pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_ON])
@ -381,3 +383,80 @@ class TestPJLink(TestCase):
# THEN: pjlink1.__not_implemented should have been called with test_cmd
mock_not_implemented.assert_called_with(test_cmd)
@patch.object(pjlink_test, 'disconnect_from_host')
def socket_abort_test(self, mock_disconnect):
"""
Test PJLink1.socket_abort calls disconnect_from_host
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Calling socket_abort
pjlink.socket_abort()
# THEN: disconnect_from_host should be called
self.assertTrue(mock_disconnect.called, 'Should have called disconnect_from_host')
def poll_loop_not_connected_test(self):
"""
Test PJLink1.poll_loop not connected return
"""
# GIVEN: Test object and mocks
pjlink = pjlink_test
pjlink.state = MagicMock()
pjlink.timer = MagicMock()
pjlink.state.return_value = False
pjlink.ConnectedState = True
# WHEN: PJLink1.poll_loop called
pjlink.poll_loop()
# THEN: poll_loop should exit without calling any other method
self.assertFalse(pjlink.timer.called, 'Should have returned without calling any other method')
@patch.object(pjlink_test, 'send_command')
def poll_loop_start_test(self, mock_send_command):
"""
Test PJLink1.poll_loop makes correct calls
"""
# GIVEN: test object and test data
pjlink = pjlink_test
pjlink.state = MagicMock()
pjlink.timer = MagicMock()
pjlink.timer.interval = MagicMock()
pjlink.timer.setInterval = MagicMock()
pjlink.timer.start = MagicMock()
pjlink.poll_time = 20
pjlink.power = S_ON
pjlink.source_available = None
pjlink.other_info = None
pjlink.manufacturer = None
pjlink.model = None
pjlink.pjlink_name = None
pjlink.ConnectedState = S_CONNECTED
pjlink.timer.interval.return_value = 10
pjlink.state.return_value = S_CONNECTED
call_list = [
call('POWR', queue=True),
call('ERST', queue=True),
call('LAMP', queue=True),
call('AVMT', queue=True),
call('INPT', queue=True),
call('INST', queue=True),
call('INFO', queue=True),
call('INF1', queue=True),
call('INF2', queue=True),
call('NAME', queue=True),
]
# WHEN: PJLink1.poll_loop is called
pjlink.poll_loop()
# THEN: proper calls were made to retrieve projector data
# First, call to update the timer with the next interval
self.assertTrue(pjlink.timer.setInterval.called, 'Should have updated the timer')
# Next, should have called the timer to start
self.assertTrue(pjlink.timer.start.called, 'Should have started the timer')
# Finally, should have called send_command with a list of projetctor status checks
mock_send_command.assert_has_calls(call_list, 'Should have queued projector queries')