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

PJLink2 update G - restructuring

This commit is contained in:
Ken Roberts 2017-08-06 00:23:26 -07:00
parent f6edbf952d
commit 864fd291a0
9 changed files with 1003 additions and 892 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
openlp/core/lib/__init__.py
View File

@ -621,5 +621,5 @@ from .imagemanager import ImageManager
from .renderer import Renderer
from .mediamanageritem import MediaManagerItem
from .projector.db import ProjectorDB, Projector
from .projector.pjlink1 import PJLink
from .projector.pjlink import PJLink
from .projector.constants import PJLINK_PORT, ERROR_MSG, ERROR_STRING

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

@ -154,7 +154,7 @@ PJLINK_VALID_CMD = {
},
'SRCH': {'version': ['2', ],
'description': translate('OpenLP.PJLinkConstants',
'UDP broadcast search request for available projectors.')
'UDP broadcast search request for available projectors. Reply is ACKN.')
},
'SVER': {'version': ['2', ],
'description': translate('OpenLP.PJLinkConstants',

885
openlp/core/lib/projector/pjlink1.py → openlp/core/lib/projector/pjlink.py
View File

@ -20,14 +20,17 @@
# Temple Place, Suite 330, Boston, MA 02111-1307 USA #
###############################################################################
"""
:mod:`openlp.core.lib.projector.pjlink1` module
:mod:`openlp.core.lib.projector.pjlink` module
Provides the necessary functions for connecting to a PJLink-capable projector.
See PJLink Class 1 Specifications for details.
http://pjlink.jbmia.or.jp/english/dl.html
PJLink Class 1 Specifications.
http://pjlink.jbmia.or.jp/english/dl_class1.html
Section 5-1 PJLink Specifications
Section 5-5 Guidelines for Input Terminals
PJLink Class 2 Specifications.
http://pjlink.jbmia.or.jp/english/dl_class2.html
Section 5-1 PJLink Specifications
Section 5-5 Guidelines for Input Terminals
NOTE:
@ -40,7 +43,7 @@
import logging
log = logging.getLogger(__name__)
log.debug('pjlink1 loaded')
log.debug('pjlink loaded')
__all__ = ['PJLink']
@ -69,88 +72,17 @@ PJLINK_HEADER = '{prefix}{{linkclass}}'.format(prefix=PJLINK_PREFIX)
PJLINK_SUFFIX = CR
class PJLink(QtNetwork.QTcpSocket):
class PJLinkCommands(object):
"""
Socket service for connecting to a PJLink-capable projector.
Process replies from PJLink projector.
"""
# Signals sent by this module
changeStatus = QtCore.pyqtSignal(str, int, str)
projectorNetwork = QtCore.pyqtSignal(int) # Projector network activity
projectorStatus = QtCore.pyqtSignal(int) # Status update
projectorAuthentication = QtCore.pyqtSignal(str) # Authentication error
projectorNoAuthentication = QtCore.pyqtSignal(str) # PIN set and no authentication needed
projectorReceivedData = QtCore.pyqtSignal() # Notify when received data finished processing
projectorUpdateIcons = QtCore.pyqtSignal() # Update the status icons on toolbar
# New commands available in PJLink Class 2
pjlink_udp_commands = [
'ACKN',
'ERST', # Class 1 or 2
'INPT', # Class 1 or 2
'LKUP',
'POWR', # Class 1 or 2
'SRCH'
]
def __init__(self, name=None, ip=None, port=PJLINK_PORT, pin=None, *args, **kwargs):
def __init__(self, *args, **kwargs):
"""
Setup for instance.
:param name: Display name
:param ip: IP address to connect to
:param port: Port to use. Default to PJLINK_PORT
:param pin: Access pin (if needed)
Optional parameters
:param dbid: Database ID number
:param location: Location where projector is physically located
:param notes: Extra notes about the projector
:param poll_time: Time (in seconds) to poll connected projector
:param socket_timeout: Time (in seconds) to abort the connection if no response
Setup for the process commands
"""
log.debug('PJlink(args={args} kwargs={kwargs})'.format(args=args, kwargs=kwargs))
self.name = name
self.ip = ip
self.port = port
self.pin = pin
log.debug('PJlinkCommands(args={args} kwargs={kwargs})'.format(args=args, kwargs=kwargs))
super().__init__()
self.model_lamp = None
self.model_filter = None
self.mac_adx = kwargs.get('mac_adx')
self.serial_no = None
self.serial_no_received = None # Used only if saved serial number is different than received serial number
self.dbid = None
self.db_update = False # Use to check if db needs to be updated prior to exiting
self.location = None
self.notes = None
self.dbid = kwargs.get('dbid')
self.location = kwargs.get('location')
self.notes = kwargs.get('notes')
# Poll time 20 seconds unless called with something else
self.poll_time = 20000 if 'poll_time' not in kwargs else kwargs['poll_time'] * 1000
# Timeout 5 seconds unless called with something else
self.socket_timeout = 5000 if 'socket_timeout' not in kwargs else kwargs['socket_timeout'] * 1000
# In case we're called from somewhere that only wants information
self.no_poll = 'no_poll' in kwargs
self.i_am_running = False
self.status_connect = S_NOT_CONNECTED
self.last_command = ''
self.projector_status = S_NOT_CONNECTED
self.error_status = S_OK
# Socket information
# Add enough space to input buffer for extraneous \n \r
self.max_size = PJLINK_MAX_PACKET + 2
self.setReadBufferSize(self.max_size)
# PJLink information
self.pjlink_class = '1' # Default class
self.reset_information()
# Set from ProjectorManager.add_projector()
self.widget = None # QListBox entry
self.timer = None # Timer that calls the poll_loop
self.send_queue = []
self.send_busy = False
# Socket timer for some possible brain-dead projectors or network cable pulled
self.socket_timer = None
# Map command to function
self.pjlink_functions = {
'AVMT': self.process_avmt,
@ -173,29 +105,30 @@ class PJLink(QtNetwork.QTcpSocket):
def reset_information(self):
"""
Reset projector-specific information to default
Initialize instance variables. Also used to reset projector-specific information to default.
"""
log.debug('({ip}) reset_information() connect status is {state}'.format(ip=self.ip, state=self.state()))
self.send_queue = []
self.power = S_OFF
self.pjlink_name = None
self.manufacturer = None
self.model = None
self.serial_no = None
self.fan = None # ERST
self.filter_time = None # FILT
self.lamp = None # LAMP
self.mac_adx_received = None # ACKN
self.manufacturer = None # INF1
self.model = None # INF2
self.model_filter = None # RFIL
self.model_lamp = None # RLMP
self.mute = None # AVMT
self.other_info = None # INFO
self.pjlink_class = PJLINK_CLASS # Default class
self.pjlink_name = None # NAME
self.power = S_OFF # POWR
self.serial_no = None # SNUM
self.serial_no_received = None
self.sw_version = None
self.sw_version = None # SVER
self.sw_version_received = None
self.mac_adx = None
self.shutter = None
self.mute = None
self.lamp = None
self.model_lamp = None
self.fan = None
self.filter_time = None
self.model_filter = None
self.source_available = None
self.source = None
self.other_info = None
self.shutter = None # AVMT
self.source_available = None # INST
self.source = None # INPT
# These should be part of PJLink() class, but set here for convenience
if hasattr(self, 'timer'):
log.debug('({ip}): Calling timer.stop()'.format(ip=self.ip))
self.timer.stop()
@ -203,6 +136,416 @@ class PJLink(QtNetwork.QTcpSocket):
log.debug('({ip}): Calling socket_timer.stop()'.format(ip=self.ip))
self.socket_timer.stop()
self.send_busy = False
self.send_queue = []
def process_command(self, cmd, data):
"""
Verifies any return error code. Calls the appropriate command handler.
:param cmd: Command to process
:param data: Data being processed
"""
log.debug('({ip}) Processing command "{cmd}" with data "{data}"'.format(ip=self.ip,
cmd=cmd,
data=data))
# Check if we have a future command not available yet
if cmd not in PJLINK_VALID_CMD:
log.error("({ip}) Ignoring command='{cmd}' (Invalid/Unknown)".format(ip=self.ip, cmd=cmd))
return
elif cmd not in self.pjlink_functions:
log.warn("({ip}) Unable to process command='{cmd}' (Future option)".format(ip=self.ip, cmd=cmd))
return
elif data in PJLINK_ERRORS:
# Oops - projector error
log.error('({ip}) Projector returned error "{data}"'.format(ip=self.ip, data=data))
if data.upper() == 'ERRA':
# Authentication error
self.disconnect_from_host()
self.change_status(E_AUTHENTICATION)
log.debug('({ip}) emitting projectorAuthentication() signal'.format(ip=self.ip))
self.projectorAuthentication.emit(self.name)
elif data.upper() == 'ERR1':
# Undefined command
self.change_status(E_UNDEFINED, '{error}: "{data}"'.format(error=ERROR_MSG[E_UNDEFINED],
data=cmd))
elif data.upper() == 'ERR2':
# Invalid parameter
self.change_status(E_PARAMETER)
elif data.upper() == 'ERR3':
# Projector busy
self.change_status(E_UNAVAILABLE)
elif data.upper() == 'ERR4':
# Projector/display error
self.change_status(E_PROJECTOR)
self.receive_data_signal()
return
# Command succeeded - no extra information
elif data.upper() == 'OK':
log.debug('({ip}) Command returned OK'.format(ip=self.ip))
# A command returned successfully
self.receive_data_signal()
return
# Command checks already passed
log.debug('({ip}) Calling function for {cmd}'.format(ip=self.ip, cmd=cmd))
self.receive_data_signal()
self.pjlink_functions[cmd](data)
def process_avmt(self, data):
"""
Process shutter and speaker status. See PJLink specification for format.
Update self.mute (audio) and self.shutter (video shutter).
:param data: Shutter and audio status
"""
shutter = self.shutter
mute = self.mute
if data == '11':
shutter = True
mute = False
elif data == '21':
shutter = False
mute = True
elif data == '30':
shutter = False
mute = False
elif data == '31':
shutter = True
mute = True
else:
log.warning('({ip}) Unknown shutter response: {data}'.format(ip=self.ip, data=data))
update_icons = shutter != self.shutter
update_icons = update_icons or mute != self.mute
self.shutter = shutter
self.mute = mute
if update_icons:
self.projectorUpdateIcons.emit()
return
def process_clss(self, data):
"""
PJLink class that this projector supports. See PJLink specification for format.
Updates self.class.
:param data: Class that projector supports.
"""
# bug 1550891: Projector returns non-standard class response:
# : Expected: '%1CLSS=1'
# : Received: '%1CLSS=Class 1' (Optoma)
# : Received: '%1CLSS=Version1' (BenQ)
if len(data) > 1:
log.warn("({ip}) Non-standard CLSS reply: '{data}'".format(ip=self.ip, data=data))
# Due to stupid projectors not following standards (Optoma, BenQ comes to mind),
# AND the different responses that can be received, the semi-permanent way to
# fix the class reply is to just remove all non-digit characters.
try:
clss = re.findall('\d', data)[0] # Should only be the first match
except IndexError:
log.error("({ip}) No numbers found in class version reply - defaulting to class '1'".format(ip=self.ip))
clss = '1'
elif not data.isdigit():
log.error("({ip}) NAN class version reply - defaulting to class '1'".format(ip=self.ip))
clss = '1'
else:
clss = data
self.pjlink_class = clss
log.debug('({ip}) Setting pjlink_class for this projector to "{data}"'.format(ip=self.ip,
data=self.pjlink_class))
return
def process_erst(self, data):
"""
Error status. See PJLink Specifications for format.
Updates self.projector_errors
\ :param data: Error status
"""
try:
datacheck = int(data)
except ValueError:
# Bad data - ignore
return
if datacheck == 0:
self.projector_errors = None
else:
self.projector_errors = {}
# Fan
if data[0] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Fan')] = \
PJLINK_ERST_STATUS[data[0]]
# Lamp
if data[1] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Lamp')] = \
PJLINK_ERST_STATUS[data[1]]
# Temp
if data[2] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Temperature')] = \
PJLINK_ERST_STATUS[data[2]]
# Cover
if data[3] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Cover')] = \
PJLINK_ERST_STATUS[data[3]]
# Filter
if data[4] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Filter')] = \
PJLINK_ERST_STATUS[data[4]]
# Other
if data[5] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Other')] = \
PJLINK_ERST_STATUS[data[5]]
return
def process_inf1(self, data):
"""
Manufacturer name set in projector.
Updates self.manufacturer
:param data: Projector manufacturer
"""
self.manufacturer = data
log.debug('({ip}) Setting projector manufacturer data to "{data}"'.format(ip=self.ip, data=self.manufacturer))
return
def process_inf2(self, data):
"""
Projector Model set in projector.
Updates self.model.
:param data: Model name
"""
self.model = data
log.debug('({ip}) Setting projector model to "{data}"'.format(ip=self.ip, data=self.model))
return
def process_info(self, data):
"""
Any extra info set in projector.
Updates self.other_info.
:param data: Projector other info
"""
self.other_info = data
log.debug('({ip}) Setting projector other_info to "{data}"'.format(ip=self.ip, data=self.other_info))
return
def process_inpt(self, data):
"""
Current source input selected. See PJLink specification for format.
Update self.source
:param data: Currently selected source
"""
self.source = data
log.info('({ip}) Setting data source to "{data}"'.format(ip=self.ip, data=self.source))
return
def process_inst(self, data):
"""
Available source inputs. See PJLink specification for format.
Updates self.source_available
:param data: Sources list
"""
sources = []
check = data.split()
for source in check:
sources.append(source)
sources.sort()
self.source_available = sources
self.projectorUpdateIcons.emit()
log.debug('({ip}) Setting projector sources_available to "{data}"'.format(ip=self.ip,
data=self.source_available))
return
def process_lamp(self, data):
"""
Lamp(s) status. See PJLink Specifications for format.
Data may have more than 1 lamp to process.
Update self.lamp dictionary with lamp status.
:param data: Lamp(s) status.
"""
lamps = []
data_dict = data.split()
while data_dict:
try:
fill = {'Hours': int(data_dict[0]), 'On': False if data_dict[1] == '0' else True}
except ValueError:
# In case of invalid entry
log.warning('({ip}) process_lamp(): Invalid data "{data}"'.format(ip=self.ip, data=data))
return
lamps.append(fill)
data_dict.pop(0) # Remove lamp hours
data_dict.pop(0) # Remove lamp on/off
self.lamp = lamps
return
def process_name(self, data):
"""
Projector name set in projector.
Updates self.pjlink_name
:param data: Projector name
"""
self.pjlink_name = data
log.debug('({ip}) Setting projector PJLink name to "{data}"'.format(ip=self.ip, data=self.pjlink_name))
return
def process_powr(self, data):
"""
Power status. See PJLink specification for format.
Update self.power with status. Update icons if change from previous setting.
:param data: Power status
"""
log.debug('({ip}: Processing POWR command'.format(ip=self.ip))
if data in PJLINK_POWR_STATUS:
power = PJLINK_POWR_STATUS[data]
update_icons = self.power != power
self.power = power
self.change_status(PJLINK_POWR_STATUS[data])
if update_icons:
self.projectorUpdateIcons.emit()
# Update the input sources available
if power == S_ON:
self.send_command('INST')
else:
# Log unknown status response
log.warning('({ip}) Unknown power response: {data}'.format(ip=self.ip, data=data))
return
def process_rfil(self, data):
"""
Process replacement filter type
"""
if self.model_filter is None:
self.model_filter = data
else:
log.warn("({ip}) Filter model already set".format(ip=self.ip))
log.warn("({ip}) Saved model: '{old}'".format(ip=self.ip, old=self.model_filter))
log.warn("({ip}) New model: '{new}'".format(ip=self.ip, new=data))
def process_rlmp(self, data):
"""
Process replacement lamp type
"""
if self.model_lamp is None:
self.model_lamp = data
else:
log.warn("({ip}) Lamp model already set".format(ip=self.ip))
log.warn("({ip}) Saved lamp: '{old}'".format(ip=self.ip, old=self.model_lamp))
log.warn("({ip}) New lamp: '{new}'".format(ip=self.ip, new=data))
def process_snum(self, data):
"""
Serial number of projector.
:param data: Serial number from projector.
"""
if self.serial_no is None:
log.debug("({ip}) Setting projector serial number to '{data}'".format(ip=self.ip, data=data))
self.serial_no = data
self.db_update = False
else:
# Compare serial numbers and see if we got the same projector
if self.serial_no != data:
log.warn("({ip}) Projector serial number does not match saved serial number".format(ip=self.ip))
log.warn("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.serial_no))
log.warn("({ip}) Received: '{new}'".format(ip=self.ip, new=data))
log.warn("({ip}) NOT saving serial number".format(ip=self.ip))
self.serial_no_received = data
def process_sver(self, data):
"""
Software version of projector
"""
if self.sw_version is None:
log.debug("({ip}) Setting projector software version to '{data}'".format(ip=self.ip, data=data))
self.sw_version = data
self.db_update = True
else:
# Compare software version and see if we got the same projector
if self.serial_no != data:
log.warn("({ip}) Projector software version does not match saved software version".format(ip=self.ip))
log.warn("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.sw_version))
log.warn("({ip}) Received: '{new}'".format(ip=self.ip, new=data))
log.warn("({ip}) NOT saving serial number".format(ip=self.ip))
self.sw_version_received = data
class PJLink(PJLinkCommands, QtNetwork.QTcpSocket):
"""
Socket service for connecting to a PJLink-capable projector.
"""
# Signals sent by this module
changeStatus = QtCore.pyqtSignal(str, int, str)
projectorNetwork = QtCore.pyqtSignal(int) # Projector network activity
projectorStatus = QtCore.pyqtSignal(int) # Status update
projectorAuthentication = QtCore.pyqtSignal(str) # Authentication error
projectorNoAuthentication = QtCore.pyqtSignal(str) # PIN set and no authentication needed
projectorReceivedData = QtCore.pyqtSignal() # Notify when received data finished processing
projectorUpdateIcons = QtCore.pyqtSignal() # Update the status icons on toolbar
# New commands available in PJLink Class 2
pjlink_udp_commands = [
'ACKN', # Class 2
'ERST', # Class 1 or 2
'INPT', # Class 1 or 2
'LKUP', # Class 2
'POWR', # Class 1 or 2
'SRCH' # Class 2
]
def __init__(self, port=PJLINK_PORT, *args, **kwargs):
"""
Setup for instance.
Options should be in kwargs except for port which does have a default.
:param name: Display name
:param ip: IP address to connect to
:param port: Port to use. Default to PJLINK_PORT
:param pin: Access pin (if needed)
Optional parameters
:param dbid: Database ID number
:param location: Location where projector is physically located
:param notes: Extra notes about the projector
:param poll_time: Time (in seconds) to poll connected projector
:param socket_timeout: Time (in seconds) to abort the connection if no response
"""
log.debug('PJlink(args={args} kwargs={kwargs})'.format(args=args, kwargs=kwargs))
super().__init__()
self.dbid = kwargs.get('dbid')
self.ip = kwargs.get('ip')
self.location = kwargs.get('location')
self.mac_adx = kwargs.get('mac_adx')
self.name = kwargs.get('name')
self.notes = kwargs.get('notes')
self.pin = kwargs.get('pin')
self.port = port
self.db_update = False # Use to check if db needs to be updated prior to exiting
# Poll time 20 seconds unless called with something else
self.poll_time = 20000 if 'poll_time' not in kwargs else kwargs['poll_time'] * 1000
# Timeout 5 seconds unless called with something else
self.socket_timeout = 5000 if 'socket_timeout' not in kwargs else kwargs['socket_timeout'] * 1000
# In case we're called from somewhere that only wants information
self.no_poll = 'no_poll' in kwargs
self.i_am_running = False
self.status_connect = S_NOT_CONNECTED
self.last_command = ''
self.projector_status = S_NOT_CONNECTED
self.error_status = S_OK
# Socket information
# Add enough space to input buffer for extraneous \n \r
self.max_size = PJLINK_MAX_PACKET + 2
self.setReadBufferSize(self.max_size)
self.reset_information()
# Set from ProjectorManager.add_projector()
self.widget = None # QListBox entry
self.timer = None # Timer that calls the poll_loop
self.send_queue = []
self.send_busy = False
# Socket timer for some possible brain-dead projectors or network cable pulled
self.socket_timer = None
def thread_started(self):
"""
@ -290,28 +633,6 @@ class PJLink(QtNetwork.QTcpSocket):
if self.model_lamp is None:
self.send_command('RLMP', queue=True)
def process_rfil(self, data):
"""
Process replacement filter type
"""
if self.model_filter is None:
self.model_filter = data
else:
log.warn("({ip}) Filter model already set".format(ip=self.ip))
log.warn("({ip}) Saved model: '{old}'".format(ip=self.ip, old=self.model_filter))
log.warn("({ip}) New model: '{new}'".format(ip=self.ip, new=data))
def process_rlmp(self, data):
"""
Process replacement lamp type
"""
if self.model_lamp is None:
self.model_lamp = data
else:
log.warn("({ip}) Lamp model already set".format(ip=self.ip))
log.warn("({ip}) Saved lamp: '{old}'".format(ip=self.ip, old=self.model_lamp))
log.warn("({ip}) New lamp: '{new}'".format(ip=self.ip, new=data))
def _get_status(self, status):
"""
Helper to retrieve status/error codes and convert to strings.
@ -474,6 +795,7 @@ class PJLink(QtNetwork.QTcpSocket):
self.send_busy = False
return
read = self.readLine(self.max_size)
log.debug("({ip}) get_data(): '{buff}'".format(ip=self.ip, buff=read))
if read == -1:
# No data available
log.debug('({ip}) get_data(): No data available (-1)'.format(ip=self.ip))
@ -626,317 +948,6 @@ class PJLink(QtNetwork.QTcpSocket):
self.change_status(E_NETWORK,
translate('OpenLP.PJLink', 'Error while sending data to projector'))
def process_command(self, cmd, data):
"""
Verifies any return error code. Calls the appropriate command handler.
:param cmd: Command to process
:param data: Data being processed
"""
log.debug('({ip}) Processing command "{cmd}" with data "{data}"'.format(ip=self.ip,
cmd=cmd,
data=data))
# Check if we have a future command not available yet
if cmd not in PJLINK_VALID_CMD:
log.error('({ip}) Unknown command received - ignoring'.format(ip=self.ip))
return
elif cmd not in self.pjlink_functions:
log.warn('({ip}) Future command received - unable to process yet'.format(ip=self.ip))
return
elif data in PJLINK_ERRORS:
# Oops - projector error
log.error('({ip}) Projector returned error "{data}"'.format(ip=self.ip, data=data))
if data.upper() == 'ERRA':
# Authentication error
self.disconnect_from_host()
self.change_status(E_AUTHENTICATION)
log.debug('({ip}) emitting projectorAuthentication() signal'.format(ip=self.ip))
self.projectorAuthentication.emit(self.name)
elif data.upper() == 'ERR1':
# Undefined command
self.change_status(E_UNDEFINED, '{error}: "{data}"'.format(error=ERROR_MSG[E_UNDEFINED],
data=cmd))
elif data.upper() == 'ERR2':
# Invalid parameter
self.change_status(E_PARAMETER)
elif data.upper() == 'ERR3':
# Projector busy
self.change_status(E_UNAVAILABLE)
elif data.upper() == 'ERR4':
# Projector/display error
self.change_status(E_PROJECTOR)
self.receive_data_signal()
return
# Command succeeded - no extra information
elif data.upper() == 'OK':
log.debug('({ip}) Command returned OK'.format(ip=self.ip))
# A command returned successfully
self.receive_data_signal()
return
# Command checks already passed
log.debug('({ip}) Calling function for {cmd}'.format(ip=self.ip, cmd=cmd))
self.receive_data_signal()
self.pjlink_functions[cmd](data)
def process_lamp(self, data):
"""
Lamp(s) status. See PJLink Specifications for format.
Data may have more than 1 lamp to process.
Update self.lamp dictionary with lamp status.
:param data: Lamp(s) status.
"""
lamps = []
data_dict = data.split()
while data_dict:
try:
fill = {'Hours': int(data_dict[0]), 'On': False if data_dict[1] == '0' else True}
except ValueError:
# In case of invalid entry
log.warning('({ip}) process_lamp(): Invalid data "{data}"'.format(ip=self.ip, data=data))
return
lamps.append(fill)
data_dict.pop(0) # Remove lamp hours
data_dict.pop(0) # Remove lamp on/off
self.lamp = lamps
return
def process_powr(self, data):
"""
Power status. See PJLink specification for format.
Update self.power with status. Update icons if change from previous setting.
:param data: Power status
"""
log.debug('({ip}: Processing POWR command'.format(ip=self.ip))
if data in PJLINK_POWR_STATUS:
power = PJLINK_POWR_STATUS[data]
update_icons = self.power != power
self.power = power
self.change_status(PJLINK_POWR_STATUS[data])
if update_icons:
self.projectorUpdateIcons.emit()
# Update the input sources available
if power == S_ON:
self.send_command('INST')
else:
# Log unknown status response
log.warning('({ip}) Unknown power response: {data}'.format(ip=self.ip, data=data))
return
def process_avmt(self, data):
"""
Process shutter and speaker status. See PJLink specification for format.
Update self.mute (audio) and self.shutter (video shutter).
:param data: Shutter and audio status
"""
shutter = self.shutter
mute = self.mute
if data == '11':
shutter = True
mute = False
elif data == '21':
shutter = False
mute = True
elif data == '30':
shutter = False
mute = False
elif data == '31':
shutter = True
mute = True
else:
log.warning('({ip}) Unknown shutter response: {data}'.format(ip=self.ip, data=data))
update_icons = shutter != self.shutter
update_icons = update_icons or mute != self.mute
self.shutter = shutter
self.mute = mute
if update_icons:
self.projectorUpdateIcons.emit()
return
def process_inpt(self, data):
"""
Current source input selected. See PJLink specification for format.
Update self.source
:param data: Currently selected source
"""
self.source = data
log.info('({ip}) Setting data source to "{data}"'.format(ip=self.ip, data=self.source))
return
def process_clss(self, data):
"""
PJLink class that this projector supports. See PJLink specification for format.
Updates self.class.
:param data: Class that projector supports.
"""
# bug 1550891: Projector returns non-standard class response:
# : Expected: '%1CLSS=1'
# : Received: '%1CLSS=Class 1' (Optoma)
# : Received: '%1CLSS=Version1' (BenQ)
if len(data) > 1:
log.warn("({ip}) Non-standard CLSS reply: '{data}'".format(ip=self.ip, data=data))
# Due to stupid projectors not following standards (Optoma, BenQ comes to mind),
# AND the different responses that can be received, the semi-permanent way to
# fix the class reply is to just remove all non-digit characters.
try:
clss = re.findall('\d', data)[0] # Should only be the first match
except IndexError:
log.error("({ip}) No numbers found in class version reply - defaulting to class '1'".format(ip=self.ip))
clss = '1'
elif not data.isdigit():
log.error("({ip}) NAN class version reply - defaulting to class '1'".format(ip=self.ip))
clss = '1'
else:
clss = data
self.pjlink_class = clss
log.debug('({ip}) Setting pjlink_class for this projector to "{data}"'.format(ip=self.ip,
data=self.pjlink_class))
return
def process_name(self, data):
"""
Projector name set in projector.
Updates self.pjlink_name
:param data: Projector name
"""
self.pjlink_name = data
log.debug('({ip}) Setting projector PJLink name to "{data}"'.format(ip=self.ip, data=self.pjlink_name))
return
def process_inf1(self, data):
"""
Manufacturer name set in projector.
Updates self.manufacturer
:param data: Projector manufacturer
"""
self.manufacturer = data
log.debug('({ip}) Setting projector manufacturer data to "{data}"'.format(ip=self.ip, data=self.manufacturer))
return
def process_inf2(self, data):
"""
Projector Model set in projector.
Updates self.model.
:param data: Model name
"""
self.model = data
log.debug('({ip}) Setting projector model to "{data}"'.format(ip=self.ip, data=self.model))
return
def process_info(self, data):
"""
Any extra info set in projector.
Updates self.other_info.
:param data: Projector other info
"""
self.other_info = data
log.debug('({ip}) Setting projector other_info to "{data}"'.format(ip=self.ip, data=self.other_info))
return
def process_inst(self, data):
"""
Available source inputs. See PJLink specification for format.
Updates self.source_available
:param data: Sources list
"""
sources = []
check = data.split()
for source in check:
sources.append(source)
sources.sort()
self.source_available = sources
self.projectorUpdateIcons.emit()
log.debug('({ip}) Setting projector sources_available to "{data}"'.format(ip=self.ip,
data=self.source_available))
return
def process_erst(self, data):
"""
Error status. See PJLink Specifications for format.
Updates self.projector_errors
:param data: Error status
"""
try:
datacheck = int(data)
except ValueError:
# Bad data - ignore
return
if datacheck == 0:
self.projector_errors = None
else:
self.projector_errors = {}
# Fan
if data[0] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Fan')] = \
PJLINK_ERST_STATUS[data[0]]
# Lamp
if data[1] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Lamp')] = \
PJLINK_ERST_STATUS[data[1]]
# Temp
if data[2] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Temperature')] = \
PJLINK_ERST_STATUS[data[2]]
# Cover
if data[3] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Cover')] = \
PJLINK_ERST_STATUS[data[3]]
# Filter
if data[4] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Filter')] = \
PJLINK_ERST_STATUS[data[4]]
# Other
if data[5] != '0':
self.projector_errors[translate('OpenLP.ProjectorPJLink', 'Other')] = \
PJLINK_ERST_STATUS[data[5]]
return
def process_snum(self, data):
"""
Serial number of projector.
:param data: Serial number from projector.
"""
if self.serial_no is None:
log.debug("({ip}) Setting projector serial number to '{data}'".format(ip=self.ip, data=data))
self.serial_no = data
self.db_update = False
else:
# Compare serial numbers and see if we got the same projector
if self.serial_no != data:
log.warn("({ip}) Projector serial number does not match saved serial number".format(ip=self.ip))
log.warn("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.serial_no))
log.warn("({ip}) Received: '{new}'".format(ip=self.ip, new=data))
log.warn("({ip}) NOT saving serial number".format(ip=self.ip))
self.serial_no_received = data
def process_sver(self, data):
"""
Software version of projector
"""
if self.sw_version is None:
log.debug("({ip}) Setting projector software version to '{data}'".format(ip=self.ip, data=data))
self.sw_version = data
self.db_update = True
else:
# Compare software version and see if we got the same projector
if self.serial_no != data:
log.warn("({ip}) Projector software version does not match saved software version".format(ip=self.ip))
log.warn("({ip}) Saved: '{old}'".format(ip=self.ip, old=self.sw_version))
log.warn("({ip}) Received: '{new}'".format(ip=self.ip, new=data))
log.warn("({ip}) NOT saving serial number".format(ip=self.ip))
self.sw_version_received = data
def connect_to_host(self):
"""
Initiate connection to projector.
@ -1098,11 +1109,3 @@ class PJLink(QtNetwork.QTcpSocket):
self.send_busy = False
self.projectorReceivedData.emit()
return
def _not_implemented(self, cmd):
"""
Log when a future PJLink command has not been implemented yet.
"""
log.warn("({ip}) Future command '{cmd}' has not been implemented yet".format(ip=self.ip,
cmd=cmd))
return

10
openlp/core/lib/projector/upgrade.py
View File

@ -42,7 +42,7 @@ def upgrade_1(session, metadata):
"""
Version 1 upgrade - old db might/might not be versioned.
"""
log.debug('Skipping upgrade_1 of projector DB - not used')
log.debug('Skipping projector DB upgrade to version 1 - not used')
def upgrade_2(session, metadata):
@ -60,14 +60,14 @@ def upgrade_2(session, metadata):
:param session: DB session instance
:param metadata: Metadata of current DB
"""
log.debug('Checking projector DB upgrade to version 2')
projector_table = Table('projector', metadata, autoload=True)
if 'mac_adx' not in [col.name for col in projector_table.c.values()]:
log.debug("Upgrading projector DB to version '2'")
upgrade_db = 'mac_adx' not in [col.name for col in projector_table.c.values()]
if upgrade_db:
new_op = get_upgrade_op(session)
new_op.add_column('projector', Column('mac_adx', types.String(18), server_default=null()))
new_op.add_column('projector', Column('serial_no', types.String(30), server_default=null()))
new_op.add_column('projector', Column('sw_version', types.String(30), server_default=null()))
new_op.add_column('projector', Column('model_filter', types.String(30), server_default=null()))
new_op.add_column('projector', Column('model_lamp', types.String(30), server_default=null()))
else:
log.warn("Skipping upgrade_2 of projector DB")
log.debug('{status} projector DB upgrade to version 2'.format(status='Updated' if upgrade_db else 'Skipping'))

2
openlp/core/ui/projector/manager.py
View File

@ -38,7 +38,7 @@ from openlp.core.lib.projector.constants import ERROR_MSG, ERROR_STRING, E_AUTHE
E_NETWORK, E_NOT_CONNECTED, E_UNKNOWN_SOCKET_ERROR, STATUS_STRING, S_CONNECTED, S_CONNECTING, S_COOLDOWN, \
S_INITIALIZE, S_NOT_CONNECTED, S_OFF, S_ON, S_STANDBY, S_WARMUP
from openlp.core.lib.projector.db import ProjectorDB
from openlp.core.lib.projector.pjlink1 import PJLink
from openlp.core.lib.projector.pjlink import PJLink
from openlp.core.lib.projector.pjlink2 import PJLinkUDP
from openlp.core.ui.projector.editform import ProjectorEditForm
from openlp.core.ui.projector.sourceselectform import SourceSelectTabs, SourceSelectSingle

4
tests/functional/openlp_core_lib/test_projector_constants.py
View File

@ -22,7 +22,7 @@
"""
Package to test the openlp.core.lib.projector.constants package.
"""
from unittest import TestCase, skip
from unittest import TestCase
class TestProjectorConstants(TestCase):
@ -40,4 +40,4 @@ class TestProjectorConstants(TestCase):
from openlp.core.lib.projector.constants import PJLINK_DEFAULT_CODES
# THEN: Verify dictionary was build correctly
self.assertEquals(PJLINK_DEFAULT_CODES, TEST_VIDEO_CODES, 'PJLink video strings should match')
self.assertEqual(PJLINK_DEFAULT_CODES, TEST_VIDEO_CODES, 'PJLink video strings should match')

8
tests/functional/openlp_core_lib/test_projector_db.py
View File

@ -29,8 +29,8 @@ import os
import shutil
from tempfile import mkdtemp
from unittest import TestCase, skip
from unittest.mock import MagicMock, patch
from unittest import TestCase
from unittest.mock import patch
from openlp.core.lib.projector import upgrade
from openlp.core.lib.db import upgrade_db
@ -413,7 +413,7 @@ class TestProjectorDB(TestCase):
Test add_projector() fail
"""
# GIVEN: Test entry in the database
ignore_result = self.projector.add_projector(Projector(**TEST1_DATA))
self.projector.add_projector(Projector(**TEST1_DATA))
# WHEN: Attempt to add same projector entry
results = self.projector.add_projector(Projector(**TEST1_DATA))
@ -439,7 +439,7 @@ class TestProjectorDB(TestCase):
Test update_projector() when entry not in database
"""
# GIVEN: Projector entry in database
ignore_result = self.projector.add_projector(Projector(**TEST1_DATA))
self.projector.add_projector(Projector(**TEST1_DATA))
projector = Projector(**TEST2_DATA)
# WHEN: Attempt to update data with a different ID

208
tests/functional/openlp_core_lib/test_projector_pjlink_base.py Normal file
View File

@ -0,0 +1,208 @@
# -*- coding: utf-8 -*-
# vim: autoindent shiftwidth=4 expandtab textwidth=120 tabstop=4 softtabstop=4
###############################################################################
# OpenLP - Open Source Lyrics Projection #
# --------------------------------------------------------------------------- #
# Copyright (c) 2008-2015 OpenLP Developers #
# --------------------------------------------------------------------------- #
# This program is free software; you can redistribute it and/or modify it #
# under the terms of the GNU General Public License as published by the Free #
# Software Foundation; version 2 of the License. #
# #
# This program is distributed in the hope that it will be useful, but WITHOUT #
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or #
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for #
# more details. #
# #
# You should have received a copy of the GNU General Public License along #
# with this program; if not, write to the Free Software Foundation, Inc., 59 #
# Temple Place, Suite 330, Boston, MA 02111-1307 USA #
###############################################################################
"""
Package to test the openlp.core.lib.projector.pjlink base package.
"""
from unittest import TestCase
from unittest.mock import call, patch, MagicMock
from openlp.core.lib.projector.pjlink import PJLink
from openlp.core.lib.projector.constants import E_PARAMETER, ERROR_STRING, S_ON, S_CONNECTED
from tests.resources.projector.data import TEST_PIN, TEST_SALT, TEST_CONNECT_AUTHENTICATE, TEST_HASH
pjlink_test = PJLink(name='test', ip='127.0.0.1', pin=TEST_PIN, no_poll=True)
class TestPJLinkBase(TestCase):
"""
Tests for the PJLink module
"""
@patch.object(pjlink_test, 'readyRead')
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'waitForReadyRead')
@patch('openlp.core.common.qmd5_hash')
def test_authenticated_connection_call(self,
mock_qmd5_hash,
mock_waitForReadyRead,
mock_send_command,
mock_readyRead):
"""
Ticket 92187: Fix for projector connect with PJLink authentication exception.
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Calling check_login with authentication request:
pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE)
# THEN: Should have called qmd5_hash
self.assertTrue(mock_qmd5_hash.called_with(TEST_SALT,
"Connection request should have been called with TEST_SALT"))
self.assertTrue(mock_qmd5_hash.called_with(TEST_PIN,
"Connection request should have been called with TEST_PIN"))
@patch.object(pjlink_test, 'change_status')
def test_status_change(self, mock_change_status):
"""
Test process_command call with ERR2 (Parameter) status
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: process_command is called with "ERR2" status from projector
pjlink.process_command('POWR', 'ERR2')
# THEN: change_status should have called change_status with E_UNDEFINED
# as first parameter
mock_change_status.called_with(E_PARAMETER,
'change_status should have been called with "{}"'.format(
ERROR_STRING[E_PARAMETER]))
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'waitForReadyRead')
@patch.object(pjlink_test, 'projectorAuthentication')
@patch.object(pjlink_test, 'timer')
@patch.object(pjlink_test, 'socket_timer')
def test_bug_1593882_no_pin_authenticated_connection(self,
mock_socket_timer,
mock_timer,
mock_authentication,
mock_ready_read,
mock_send_command):
"""
Test bug 1593882 no pin and authenticated request exception
"""
# GIVEN: Test object and mocks
pjlink = pjlink_test
pjlink.pin = None
mock_ready_read.return_value = True
# WHEN: call with authentication request and pin not set
pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE)
# THEN: 'No Authentication' signal should have been sent
mock_authentication.emit.assert_called_with(pjlink.ip)
@patch.object(pjlink_test, 'waitForReadyRead')
@patch.object(pjlink_test, 'state')
@patch.object(pjlink_test, '_send_command')
@patch.object(pjlink_test, 'timer')
@patch.object(pjlink_test, 'socket_timer')
def test_bug_1593883_pjlink_authentication(self,
mock_socket_timer,
mock_timer,
mock_send_command,
mock_state,
mock_waitForReadyRead):
"""
Test bugfix 1593883 pjlink authentication
"""
# GIVEN: Test object and data
pjlink = pjlink_test
pjlink.pin = TEST_PIN
mock_state.return_value = pjlink.ConnectedState
mock_waitForReadyRead.return_value = True
# WHEN: Athenticated connection is called
pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE)
# THEN: send_command should have the proper authentication
self.assertEqual("{test}".format(test=mock_send_command.call_args),
"call(data='{hash}%1CLSS ?\\r')".format(hash=TEST_HASH))
@patch.object(pjlink_test, 'disconnect_from_host')
def test_socket_abort(self, mock_disconnect):
"""
Test PJLink.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 test_poll_loop_not_connected(self):
"""
Test PJLink.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: PJLink.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 test_poll_loop_start(self, mock_send_command):
"""
Test PJLink.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: PJLink.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')

774
tests/functional/openlp_core_lib/test_projector_pjlink1.py → tests/functional/openlp_core_lib/test_projector_pjlink_commands.py
View File

@ -20,44 +20,360 @@
# Temple Place, Suite 330, Boston, MA 02111-1307 USA #
###############################################################################
"""
Package to test the openlp.core.lib.projector.pjlink1 package.
Package to test the openlp.core.lib.projector.pjlink commands package.
"""
from unittest import TestCase
from unittest.mock import call, patch, MagicMock
from unittest.mock import patch, MagicMock
from openlp.core.lib.projector.pjlink1 import PJLink
from openlp.core.lib.projector.constants import E_PARAMETER, ERROR_STRING, S_OFF, S_STANDBY, S_ON, \
PJLINK_POWR_STATUS, S_CONNECTED
from openlp.core.lib.projector.pjlink import PJLink
from openlp.core.lib.projector.constants import PJLINK_ERST_STATUS, PJLINK_POWR_STATUS, \
S_OFF, S_STANDBY, S_ON
from tests.resources.projector.data import TEST_PIN, TEST_SALT, TEST_CONNECT_AUTHENTICATE, TEST_HASH
from tests.resources.projector.data import TEST_PIN
pjlink_test = PJLink(name='test', ip='127.0.0.1', pin=TEST_PIN, no_poll=True)
# ERST status codes
ERST_OK, ERST_WARN, ERST_ERR = '0', '1', '2'
class TestPJLink(TestCase):
class TestPJLinkCommands(TestCase):
"""
Tests for the PJLink module
"""
@patch.object(pjlink_test, 'readyRead')
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'waitForReadyRead')
@patch('openlp.core.common.qmd5_hash')
def test_authenticated_connection_call(self, mock_qmd5_hash, mock_waitForReadyRead, mock_send_command,
mock_readyRead):
def test_projector_reset_information(self):
"""
Ticket 92187: Fix for projector connect with PJLink authentication exception.
Test reset_information() resets all information and stops timers
"""
# GIVEN: Test object and test data
pjlink = pjlink_test
pjlink.power = S_ON
pjlink.pjlink_name = 'OPENLPTEST'
pjlink.manufacturer = 'PJLINK'
pjlink.model = '1'
pjlink.shutter = True
pjlink.mute = True
pjlink.lamp = True
pjlink.fan = True
pjlink.source_available = True
pjlink.other_info = 'ANOTHER TEST'
pjlink.send_queue = True
pjlink.send_busy = True
pjlink.timer = MagicMock()
pjlink.socket_timer = MagicMock()
# WHEN: reset_information() is called
with patch.object(pjlink.timer, 'stop') as mock_timer:
with patch.object(pjlink.socket_timer, 'stop') as mock_socket_timer:
pjlink.reset_information()
# THEN: All information should be reset and timers stopped
self.assertEqual(pjlink.power, S_OFF, 'Projector power should be OFF')
self.assertIsNone(pjlink.pjlink_name, 'Projector pjlink_name should be None')
self.assertIsNone(pjlink.manufacturer, 'Projector manufacturer should be None')
self.assertIsNone(pjlink.model, 'Projector model should be None')
self.assertIsNone(pjlink.shutter, 'Projector shutter should be None')
self.assertIsNone(pjlink.mute, 'Projector shuttter should be None')
self.assertIsNone(pjlink.lamp, 'Projector lamp should be None')
self.assertIsNone(pjlink.fan, 'Projector fan should be None')
self.assertIsNone(pjlink.source_available, 'Projector source_available should be None')
self.assertIsNone(pjlink.source, 'Projector source should be None')
self.assertIsNone(pjlink.other_info, 'Projector other_info should be None')
self.assertEqual(pjlink.send_queue, [], 'Projector send_queue should be an empty list')
self.assertFalse(pjlink.send_busy, 'Projector send_busy should be False')
self.assertTrue(mock_timer.called, 'Projector timer.stop() should have been called')
self.assertTrue(mock_socket_timer.called, 'Projector socket_timer.stop() should have been called')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_closed_muted(self, mock_projectorReceivedData):
"""
Test avmt status shutter closed and mute off
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = False
pjlink.mute = False
# WHEN: Called with setting shutter to closed and mute on
pjlink.process_avmt('31')
# THEN: Shutter should be closed and mute should be True
self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed')
self.assertTrue(pjlink.mute, 'Audio should be on')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_closed_unmuted(self, mock_projectorReceivedData):
"""
Test avmt status shutter closed and audio muted
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = False
pjlink.mute = True
# WHEN: Called with setting shutter closed and mute off
pjlink.process_avmt('11')
# THEN: Shutter should be True and mute should be False
self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed')
self.assertFalse(pjlink.mute, 'Audio should be off')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_open_muted(self, mock_projectorReceivedData):
"""
Test avmt status shutter open and mute on
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = True
pjlink.mute = False
# WHEN: Called with setting shutter closed and mute on
pjlink.process_avmt('21')
# THEN: Shutter should be closed and mute should be True
self.assertFalse(pjlink.shutter, 'Shutter should have been set to closed')
self.assertTrue(pjlink.mute, 'Audio should be off')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_open_unmuted(self, mock_projectorReceivedData):
"""
Test avmt status shutter open and mute off off
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = True
pjlink.mute = True
# WHEN: Called with setting shutter to closed and mute on
pjlink.process_avmt('30')
# THEN: Shutter should be closed and mute should be True
self.assertFalse(pjlink.shutter, 'Shutter should have been set to open')
self.assertFalse(pjlink.mute, 'Audio should be on')
def test_projector_process_clss_one(self):
"""
Test class 1 sent from projector
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Calling check_login with authentication request:
pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE)
# WHEN: Process class response
pjlink.process_clss('1')
# THEN: Should have called qmd5_hash
self.assertTrue(mock_qmd5_hash.called_with(TEST_SALT,
"Connection request should have been called with TEST_SALT"))
self.assertTrue(mock_qmd5_hash.called_with(TEST_PIN,
"Connection request should have been called with TEST_PIN"))
# THEN: Projector class should be set to 1
self.assertEqual(pjlink.pjlink_class, '1',
'Projector should have returned class=1')
def test_projector_process_clss_two(self):
"""
Test class 2 sent from projector
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Process class response
pjlink.process_clss('2')
# THEN: Projector class should be set to 1
self.assertEqual(pjlink.pjlink_class, '2',
'Projector should have returned class=2')
def test_projector_process_clss_nonstandard_reply(self):
"""
Bugfix 1550891: CLSS request returns non-standard reply
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Process non-standard reply
pjlink.process_clss('Class 1')
# THEN: Projector class should be set with proper value
self.assertEqual(pjlink.pjlink_class, '1',
'Non-standard class reply should have set class=1')
def test_projector_process_erst_all_ok(self):
"""
Test test_projector_process_erst_all_ok
"""
# GIVEN: Test object
pjlink = pjlink_test
chk_test = ERST_OK
# WHEN: process_erst with no errors
pjlink.process_erst('{fan}{lamp}{temp}{cover}{filter}{other}'.format(fan=chk_test,
lamp=chk_test,
temp=chk_test,
cover=chk_test,
filter=chk_test,
other=chk_test))
# PJLink instance errors should be None
self.assertIsNone(pjlink.projector_errors, 'projector_errors should have been set to None')
def test_projector_process_erst_all_warn(self):
"""
Test test_projector_process_erst_all_warn
"""
# GIVEN: Test object
pjlink = pjlink_test
chk_test = ERST_WARN
chk_code = PJLINK_ERST_STATUS[chk_test]
chk_value = {'Fan': chk_code,
'Lamp': chk_code,
'Temperature': chk_code,
'Cover': chk_code,
'Filter': chk_code,
'Other': chk_code
}
# WHEN: process_erst with status set to WARN
pjlink.process_erst('{fan}{lamp}{temp}{cover}{filter}{other}'.format(fan=chk_test,
lamp=chk_test,
temp=chk_test,
cover=chk_test,
filter=chk_test,
other=chk_test))
# PJLink instance errors should match chk_value
self.assertEqual(pjlink.projector_errors, chk_value,
'projector_errors should have been set to all {err}'.format(err=chk_code))
def test_projector_process_erst_all_error(self):
"""
Test test_projector_process_erst_all_error
"""
# GIVEN: Test object
pjlink = pjlink_test
chk_test = ERST_ERR
chk_code = PJLINK_ERST_STATUS[chk_test]
chk_value = {'Fan': chk_code,
'Lamp': chk_code,
'Temperature': chk_code,
'Cover': chk_code,
'Filter': chk_code,
'Other': chk_code
}
# WHEN: process_erst with status set to ERROR
pjlink.process_erst('{fan}{lamp}{temp}{cover}{filter}{other}'.format(fan=chk_test,
lamp=chk_test,
temp=chk_test,
cover=chk_test,
filter=chk_test,
other=chk_test))
# PJLink instance errors should be set to chk_value
self.assertEqual(pjlink.projector_errors, chk_value,
'projector_errors should have been set to all {err}'.format(err=chk_code))
def test_projector_process_inpt(self):
"""
Test input source status shows current input
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.source = '0'
# WHEN: Called with input source
pjlink.process_inpt('1')
# THEN: Input selected should reflect current input
self.assertEqual(pjlink.source, '1', 'Input source should be set to "1"')
@patch.object(pjlink_test, 'projectorReceivedData')
def test_projector_process_lamp_single(self, mock_projectorReceivedData):
"""
Test status lamp on/off and hours
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Call process_command with lamp data
pjlink.process_command('LAMP', '22222 1')
# THEN: Lamp should have been set with status=ON and hours=22222
self.assertEqual(pjlink.lamp[0]['On'], True,
'Lamp power status should have been set to TRUE')
self.assertEqual(pjlink.lamp[0]['Hours'], 22222,
'Lamp hours should have been set to 22222')
@patch.object(pjlink_test, 'projectorReceivedData')
def test_projector_process_lamp_multiple(self, mock_projectorReceivedData):
"""
Test status multiple lamp on/off and hours
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Call process_command with lamp data
pjlink.process_command('LAMP', '11111 1 22222 0 33333 1')
# THEN: Lamp should have been set with proper lamp status
self.assertEqual(len(pjlink.lamp), 3,
'Projector should have 3 lamps specified')
self.assertEqual(pjlink.lamp[0]['On'], True,
'Lamp 1 power status should have been set to TRUE')
self.assertEqual(pjlink.lamp[0]['Hours'], 11111,
'Lamp 1 hours should have been set to 11111')
self.assertEqual(pjlink.lamp[1]['On'], False,
'Lamp 2 power status should have been set to FALSE')
self.assertEqual(pjlink.lamp[1]['Hours'], 22222,
'Lamp 2 hours should have been set to 22222')
self.assertEqual(pjlink.lamp[2]['On'], True,
'Lamp 3 power status should have been set to TRUE')
self.assertEqual(pjlink.lamp[2]['Hours'], 33333,
'Lamp 3 hours should have been set to 33333')
@patch.object(pjlink_test, 'projectorReceivedData')
@patch.object(pjlink_test, 'projectorUpdateIcons')
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'change_status')
def test_projector_process_powr_on(self,
mock_change_status,
mock_send_command,
mock_UpdateIcons,
mock_ReceivedData):
"""
Test status power to ON
"""
# GIVEN: Test object and preset
pjlink = pjlink_test
pjlink.power = S_STANDBY
# WHEN: Call process_command with turn power on command
pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_ON])
# THEN: Power should be set to ON
self.assertEqual(pjlink.power, S_ON, 'Power should have been set to ON')
mock_send_command.assert_called_once_with('INST')
self.assertEqual(mock_UpdateIcons.emit.called, True, 'projectorUpdateIcons should have been called')
@patch.object(pjlink_test, 'projectorReceivedData')
@patch.object(pjlink_test, 'projectorUpdateIcons')
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'change_status')
def test_projector_process_powr_off(self,
mock_change_status,
mock_send_command,
mock_UpdateIcons,
mock_ReceivedData):
"""
Test status power to STANDBY
"""
# GIVEN: Test object and preset
pjlink = pjlink_test
pjlink.power = S_ON
# WHEN: Call process_command with turn power on command
pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_STANDBY])
# THEN: Power should be set to STANDBY
self.assertEqual(pjlink.power, S_STANDBY, 'Power should have been set to STANDBY')
self.assertEqual(mock_send_command.called, False, 'send_command should not have been called')
self.assertEqual(mock_UpdateIcons.emit.called, True, 'projectorUpdateIcons should have been called')
def test_projector_process_rfil_save(self):
"""
@ -150,419 +466,3 @@ class TestPJLink(TestCase):
# THEN: Serial number should be set
self.assertNotEquals(pjlink.serial_no, test_number,
'Projector serial number should NOT have been set')
def test_projector_clss_one(self):
"""
Test class 1 sent from projector
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Process class response
pjlink.process_clss('1')
# THEN: Projector class should be set to 1
self.assertEqual(pjlink.pjlink_class, '1',
'Projector should have returned class=1')
def test_projector_clss_two(self):
"""
Test class 2 sent from projector
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Process class response
pjlink.process_clss('2')
# THEN: Projector class should be set to 1
self.assertEqual(pjlink.pjlink_class, '2',
'Projector should have returned class=2')
def test_bug_1550891_non_standard_class_reply(self):
"""
Bugfix 1550891: CLSS request returns non-standard reply
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Process non-standard reply
pjlink.process_clss('Class 1')
# THEN: Projector class should be set with proper value
self.assertEqual(pjlink.pjlink_class, '1',
'Non-standard class reply should have set class=1')
@patch.object(pjlink_test, 'change_status')
def test_status_change(self, mock_change_status):
"""
Test process_command call with ERR2 (Parameter) status
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: process_command is called with "ERR2" status from projector
pjlink.process_command('POWR', 'ERR2')
# THEN: change_status should have called change_status with E_UNDEFINED
# as first parameter
mock_change_status.called_with(E_PARAMETER,
'change_status should have been called with "{}"'.format(
ERROR_STRING[E_PARAMETER]))
@patch.object(pjlink_test, 'process_inpt')
def test_projector_return_ok(self, mock_process_inpt):
"""
Test projector calls process_inpt command when process_command is called with INPT option
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: process_command is called with INST command and 31 input:
pjlink.process_command('INPT', '31')
# THEN: process_inpt method should have been called with 31
mock_process_inpt.called_with('31',
"process_inpt should have been called with 31")
@patch.object(pjlink_test, 'projectorReceivedData')
def test_projector_process_lamp(self, mock_projectorReceivedData):
"""
Test status lamp on/off and hours
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Call process_command with lamp data
pjlink.process_command('LAMP', '22222 1')
# THEN: Lamp should have been set with status=ON and hours=22222
self.assertEqual(pjlink.lamp[0]['On'], True,
'Lamp power status should have been set to TRUE')
self.assertEqual(pjlink.lamp[0]['Hours'], 22222,
'Lamp hours should have been set to 22222')
@patch.object(pjlink_test, 'projectorReceivedData')
def test_projector_process_multiple_lamp(self, mock_projectorReceivedData):
"""
Test status multiple lamp on/off and hours
"""
# GIVEN: Test object
pjlink = pjlink_test
# WHEN: Call process_command with lamp data
pjlink.process_command('LAMP', '11111 1 22222 0 33333 1')
# THEN: Lamp should have been set with proper lamp status
self.assertEqual(len(pjlink.lamp), 3,
'Projector should have 3 lamps specified')
self.assertEqual(pjlink.lamp[0]['On'], True,
'Lamp 1 power status should have been set to TRUE')
self.assertEqual(pjlink.lamp[0]['Hours'], 11111,
'Lamp 1 hours should have been set to 11111')
self.assertEqual(pjlink.lamp[1]['On'], False,
'Lamp 2 power status should have been set to FALSE')
self.assertEqual(pjlink.lamp[1]['Hours'], 22222,
'Lamp 2 hours should have been set to 22222')
self.assertEqual(pjlink.lamp[2]['On'], True,
'Lamp 3 power status should have been set to TRUE')
self.assertEqual(pjlink.lamp[2]['Hours'], 33333,
'Lamp 3 hours should have been set to 33333')
@patch.object(pjlink_test, 'projectorReceivedData')
@patch.object(pjlink_test, 'projectorUpdateIcons')
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'change_status')
def test_projector_process_power_on(self, mock_change_status,
mock_send_command,
mock_UpdateIcons,
mock_ReceivedData):
"""
Test status power to ON
"""
# GIVEN: Test object and preset
pjlink = pjlink_test
pjlink.power = S_STANDBY
# WHEN: Call process_command with turn power on command
pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_ON])
# THEN: Power should be set to ON
self.assertEqual(pjlink.power, S_ON, 'Power should have been set to ON')
mock_send_command.assert_called_once_with('INST')
self.assertEqual(mock_UpdateIcons.emit.called, True, 'projectorUpdateIcons should have been called')
@patch.object(pjlink_test, 'projectorReceivedData')
@patch.object(pjlink_test, 'projectorUpdateIcons')
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'change_status')
def test_projector_process_power_off(self, mock_change_status,
mock_send_command,
mock_UpdateIcons,
mock_ReceivedData):
"""
Test status power to STANDBY
"""
# GIVEN: Test object and preset
pjlink = pjlink_test
pjlink.power = S_ON
# WHEN: Call process_command with turn power on command
pjlink.process_command('POWR', PJLINK_POWR_STATUS[S_STANDBY])
# THEN: Power should be set to STANDBY
self.assertEqual(pjlink.power, S_STANDBY, 'Power should have been set to STANDBY')
self.assertEqual(mock_send_command.called, False, 'send_command should not have been called')
self.assertEqual(mock_UpdateIcons.emit.called, True, 'projectorUpdateIcons should have been called')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_closed_unmuted(self, mock_projectorReceivedData):
"""
Test avmt status shutter closed and audio muted
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = False
pjlink.mute = True
# WHEN: Called with setting shutter closed and mute off
pjlink.process_avmt('11')
# THEN: Shutter should be True and mute should be False
self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed')
self.assertFalse(pjlink.mute, 'Audio should be off')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_open_muted(self, mock_projectorReceivedData):
"""
Test avmt status shutter open and mute on
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = True
pjlink.mute = False
# WHEN: Called with setting shutter closed and mute on
pjlink.process_avmt('21')
# THEN: Shutter should be closed and mute should be True
self.assertFalse(pjlink.shutter, 'Shutter should have been set to closed')
self.assertTrue(pjlink.mute, 'Audio should be off')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_open_unmuted(self, mock_projectorReceivedData):
"""
Test avmt status shutter open and mute off off
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = True
pjlink.mute = True
# WHEN: Called with setting shutter to closed and mute on
pjlink.process_avmt('30')
# THEN: Shutter should be closed and mute should be True
self.assertFalse(pjlink.shutter, 'Shutter should have been set to open')
self.assertFalse(pjlink.mute, 'Audio should be on')
@patch.object(pjlink_test, 'projectorUpdateIcons')
def test_projector_process_avmt_closed_muted(self, mock_projectorReceivedData):
"""
Test avmt status shutter closed and mute off
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.shutter = False
pjlink.mute = False
# WHEN: Called with setting shutter to closed and mute on
pjlink.process_avmt('31')
# THEN: Shutter should be closed and mute should be True
self.assertTrue(pjlink.shutter, 'Shutter should have been set to closed')
self.assertTrue(pjlink.mute, 'Audio should be on')
def test_projector_process_input(self):
"""
Test input source status shows current input
"""
# GIVEN: Test object
pjlink = pjlink_test
pjlink.source = '0'
# WHEN: Called with input source
pjlink.process_inpt('1')
# THEN: Input selected should reflect current input
self.assertEqual(pjlink.source, '1', 'Input source should be set to "1"')
def test_projector_reset_information(self):
"""
Test reset_information() resets all information and stops timers
"""
# GIVEN: Test object and test data
pjlink = pjlink_test
pjlink.power = S_ON
pjlink.pjlink_name = 'OPENLPTEST'
pjlink.manufacturer = 'PJLINK'
pjlink.model = '1'
pjlink.shutter = True
pjlink.mute = True
pjlink.lamp = True
pjlink.fan = True
pjlink.source_available = True
pjlink.other_info = 'ANOTHER TEST'
pjlink.send_queue = True
pjlink.send_busy = True
pjlink.timer = MagicMock()
pjlink.socket_timer = MagicMock()
# WHEN: reset_information() is called
with patch.object(pjlink.timer, 'stop') as mock_timer:
with patch.object(pjlink.socket_timer, 'stop') as mock_socket_timer:
pjlink.reset_information()
# THEN: All information should be reset and timers stopped
self.assertEqual(pjlink.power, S_OFF, 'Projector power should be OFF')
self.assertIsNone(pjlink.pjlink_name, 'Projector pjlink_name should be None')
self.assertIsNone(pjlink.manufacturer, 'Projector manufacturer should be None')
self.assertIsNone(pjlink.model, 'Projector model should be None')
self.assertIsNone(pjlink.shutter, 'Projector shutter should be None')
self.assertIsNone(pjlink.mute, 'Projector shuttter should be None')
self.assertIsNone(pjlink.lamp, 'Projector lamp should be None')
self.assertIsNone(pjlink.fan, 'Projector fan should be None')
self.assertIsNone(pjlink.source_available, 'Projector source_available should be None')
self.assertIsNone(pjlink.source, 'Projector source should be None')
self.assertIsNone(pjlink.other_info, 'Projector other_info should be None')
self.assertEqual(pjlink.send_queue, [], 'Projector send_queue should be an empty list')
self.assertFalse(pjlink.send_busy, 'Projector send_busy should be False')
self.assertTrue(mock_timer.called, 'Projector timer.stop() should have been called')
self.assertTrue(mock_socket_timer.called, 'Projector socket_timer.stop() should have been called')
@patch.object(pjlink_test, 'send_command')
@patch.object(pjlink_test, 'waitForReadyRead')
@patch.object(pjlink_test, 'projectorAuthentication')
@patch.object(pjlink_test, 'timer')
@patch.object(pjlink_test, 'socket_timer')
def test_bug_1593882_no_pin_authenticated_connection(self, mock_socket_timer,
mock_timer,
mock_authentication,
mock_ready_read,
mock_send_command):
"""
Test bug 1593882 no pin and authenticated request exception
"""
# GIVEN: Test object and mocks
pjlink = pjlink_test
pjlink.pin = None
mock_ready_read.return_value = True
# WHEN: call with authentication request and pin not set
pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE)
# THEN: 'No Authentication' signal should have been sent
mock_authentication.emit.assert_called_with(pjlink.ip)
@patch.object(pjlink_test, 'waitForReadyRead')
@patch.object(pjlink_test, 'state')
@patch.object(pjlink_test, '_send_command')
@patch.object(pjlink_test, 'timer')
@patch.object(pjlink_test, 'socket_timer')
def test_bug_1593883_pjlink_authentication(self, mock_socket_timer,
mock_timer,
mock_send_command,
mock_state,
mock_waitForReadyRead):
"""
Test bugfix 1593883 pjlink authentication
"""
# GIVEN: Test object and data
pjlink = pjlink_test
pjlink.pin = TEST_PIN
mock_state.return_value = pjlink.ConnectedState
mock_waitForReadyRead.return_value = True
# WHEN: Athenticated connection is called
pjlink.check_login(data=TEST_CONNECT_AUTHENTICATE)
# THEN: send_command should have the proper authentication
self.assertEqual("{test}".format(test=mock_send_command.call_args),
"call(data='{hash}%1CLSS ?\\r')".format(hash=TEST_HASH))
@patch.object(pjlink_test, 'disconnect_from_host')
def socket_abort_test(self, mock_disconnect):
"""
Test PJLink.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 PJLink.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: PJLink.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 PJLink.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: PJLink.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')