jlengrand/bipbipzizik
1
0
Fork 0
mirror of https://github.com/acatoire/bipbipzizik.git synced 2026-03-10 08:01:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source 
# Conflicts:
#	bipbip/bipbip.py
#	bipbip/bipbip_soco.py
This commit is contained in:
axxcat
2023-01-28 22:05:50 +01:00
parent b4420c30a9 566017e670
commit ae9f2f25cd
3 changed files with 124 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
bipbip/bipbip.py
View File

@@ -12,7 +12,7 @@ class BipBip:
The generic bipbip class
"""
def __init__(self, parameters: dict):
def __init__(self, parameters: dict, multi_read_timeout: int or float = 3):
"""
:param parameters: Parameter dict, each BipBip can implement its own parameters
Generic parameters are:
@@ -23,12 +23,16 @@ class BipBip:
action: type of action to be executed
data: extra data related to the action
user: allowed user for the card
:param multi_read_timeout: (optional) Value to consider a multi read, default 3s
"""
self.parameters = parameters
self.player = None
self.parameters = parameters
self.multi_read_timeout = multi_read_timeout
self._execution_log = []
def _parameter(self, param_name:str) -> str or None:
self.locked = False
def _parameter(self, param_name: str) -> str or None:
try:
return self.parameters[param_name]
except KeyError:
@@ -112,4 +116,26 @@ class BipBip:
logger.info("Execute the bipbip: %s", self.name)
# TODO manage the multi-read
if len(self.execution_log) > 1:
last_time_period = (self.execution_log[-1] - self.execution_log[-2])
if last_time_period < self.multi_read_timeout:
self.locked = True
else:
self.locked = False
self.start()
def start(self):
"""
Action of the bipbip to be overwritten by the child class
:return:
"""
####################################################
# ## Cancel conditions sample
####################################################
if self.locked:
logger.warning("Action canceled because of the %ds multi read protection.", self.multi_read_timeout)
return
logger.info("ACTION")

24
bipbip/bipbip_soco.py
View File

@@ -4,6 +4,8 @@ Interacting with Sonos using SoCo library
"""
import logging
import time
import soco
from soco.plugins.sharelink import ShareLinkPlugin
from bipbip import BipBip
@@ -69,7 +71,7 @@ class BipBipSoCo(BipBip):
The SoCo bipbip class
"""
def __init__(self, parameters: dict):
def __init__(self, parameters: dict, multi_read_timeout: int or float = 3):
"""
:param parameters: Parameter dict, each BipBip can implement its own parameters
Generic parameters are:
@@ -80,8 +82,9 @@ class BipBipSoCo(BipBip):
action: type of action to be executed
data: extra data related to the action
user: allowed user for the card
:param multi_read_timeout: (optional) Value to consider a multi read, default 3s
"""
super().__init__(parameters)
super().__init__(parameters, multi_read_timeout)
self.soco_mode = SoCoMode(self.mode)
@@ -95,18 +98,27 @@ class BipBipSoCo(BipBip):
logger.critical("No player found with name: %s!", player_name)
return
def execute(self):
def start(self):
"""
Execute of the bipbip
:return:
"""
super().execute()
####################################################
# ## Cancel conditions
####################################################
if self.locked and not self.action == "sonos-cmd":
logger.warning("Action canceled because of the %ds multi read protection.", self.multi_read_timeout)
return
if not self.player:
logger.critical("No player valid player configured during init, execution aborted!")
return
####################################################
# ####### Action
####################################################
if self.player.is_playing_radio:
logger.debug("Stop radio before continue")
# Todo manage the radio kill if actually playing
@@ -143,10 +155,6 @@ class BipBipSoCo(BipBip):
# radio action execution
# TODO to be implemented
self.player.play()
# Start the queue play from beginning
# self.player.play_from_queue(0)
else:
logger.critical('Action "%s" is not supported by BipBipSoCo!', self.action)
return

78
bipbip/test_bipbip_execution.py
View File

@@ -3,7 +3,9 @@ Card execution tests
"""
import logging
from time import sleep
from unittest import TestCase
from unittest.mock import MagicMock
from bipbip_soco import BipBipSoCo
logging.basicConfig(level=logging.INFO)
@@ -11,12 +13,18 @@ logger = logging.getLogger("bipbip.soco")
logger.setLevel(level=logging.DEBUG)
# logging.config.fileConfig('logging_unittest.yaml') TODO understand logging config
REAL = False
class TestBipBip(TestCase):
"""
TestBipBip
"""
@classmethod
def setUpClass(cls):
pass
def test_execute_soco_play(self):
"""
test_execute_soco_play
@@ -29,8 +37,15 @@ class TestBipBip(TestCase):
)
bip_bip = BipBipSoCo(param_dict)
if not REAL:
bip_bip.player.play = MagicMock()
bip_bip.execute()
if not REAL:
bip_bip.player.play.assert_called()
def test_execute_soco_pause(self):
"""
test_execute_soco_pause
@@ -119,3 +134,66 @@ class TestBipBip(TestCase):
bip_bip = BipBipSoCo(param_dict)
bip_bip.execute()
def test_execute_soco_spotify_multi_read(self):
param_dict = dict(
name='Les petits poissons',
action='spotify:track',
data='35VKLRwEjuR5IuFyGqjMaf',
mode='ClearQueue',
)
timeout = 1
bip_bip = BipBipSoCo(param_dict, multi_read_timeout=timeout)
if not REAL:
bip_bip.player.clear_queue = MagicMock()
bip_bip.sharelink.add_share_link_to_queue = MagicMock()
bip_bip.player.play = MagicMock()
bip_bip.execute()
if not REAL:
bip_bip.player.clear_queue.assert_called()
bip_bip.sharelink.add_share_link_to_queue.assert_called_with("spotify:track:35VKLRwEjuR5IuFyGqjMaf")
bip_bip.player.play.assert_called()
sleep(0.1)
if not REAL:
bip_bip.player.clear_queue.reset_mock()
bip_bip.sharelink.add_share_link_to_queue.reset_mock()
bip_bip.player.play.reset_mock()
bip_bip.execute()
if not REAL:
bip_bip.player.clear_queue.assert_not_called()
bip_bip.sharelink.add_share_link_to_queue.assert_not_called()
bip_bip.player.play.assert_not_called()
sleep(0.1)
if not REAL:
bip_bip.player.clear_queue.reset_mock()
bip_bip.sharelink.add_share_link_to_queue.reset_mock()
bip_bip.player.play.reset_mock()
bip_bip.execute()
if not REAL:
bip_bip.player.clear_queue.assert_not_called()
bip_bip.sharelink.add_share_link_to_queue.assert_not_called()
bip_bip.player.play.assert_not_called()
sleep(timeout)
if not REAL:
bip_bip.player.clear_queue.reset_mock()
bip_bip.sharelink.add_share_link_to_queue.reset_mock()
bip_bip.player.play.reset_mock()
bip_bip.execute()
if not REAL:
bip_bip.player.clear_queue.assert_called()
bip_bip.sharelink.add_share_link_to_queue.assert_called_with("spotify:track:35VKLRwEjuR5IuFyGqjMaf")
bip_bip.player.play.assert_called()