Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from ledfx.effects.audio import AudioReactiveEffect, MIN_MIDI, MAX_MIDI
from ledfx.effects.gradient import GradientEffect
from ledfx.effects import mix_colors
from ledfx.color import COLORS
import voluptuous as vol
import numpy as np
import aubio
class PitchSpectrumAudioEffect(AudioReactiveEffect, GradientEffect):
NAME = "PitchSpectrum"
CONFIG_SCHEMA = vol.Schema({
vol.Optional('blur', description='Amount to blur the effect', default = 1.0): vol.Coerce(float),
vol.Optional('mirror', description='Mirror the effect', default = True): bool,
vol.Optional('fade_rate', description='Rate at which notes fade', default = 0.15): vol.All(vol.Coerce(float), vol.Range(min=0.0, max=1.0)),
vol.Optional('responsiveness', description='Responsiveness of the note changes', default = 0.15): vol.All(vol.Coerce(float), vol.Range(min=0.0, max=1.0)),
})
def config_updated(self, config):
win_s = 1024
hop_s = 48000 // 60
tolerance = 0.8
# TODO: Move into the base audio effect class
self.pitch_o = aubio.pitch("schmitt", win_s, hop_s, 48000)
self.pitch_o.set_unit("midi")
self.pitch_o.set_tolerance(tolerance)
vol.All(vol.Coerce(float), vol.Range(min=0, max=100)),
)
),
vol.Coerce(tuple),
),
vol.Exclusive(ATTR_COLOR_TEMP, COLOR_GROUP): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
vol.Exclusive(ATTR_KELVIN, COLOR_GROUP): vol.All(vol.Coerce(int), vol.Range(min=0)),
ATTR_WHITE_VALUE: vol.All(vol.Coerce(int), vol.Range(min=0, max=255)),
ATTR_FLASH: vol.In([FLASH_SHORT, FLASH_LONG]),
ATTR_EFFECT: cv.string,
}
PROFILE_SCHEMA = vol.Schema(
vol.ExactSequence((str, cv.small_float, cv.small_float, cv.byte))
)
_LOGGER = logging.getLogger(__name__)
@bind_hass
def is_on(hass, entity_id=None):
"""Return if the lights are on based on the statemachine."""
entity_id = entity_id or ENTITY_ID_ALL_LIGHTS
return hass.states.is_state(entity_id, STATE_ON)
def preprocess_turn_on_alternatives(params):
"""Process extra data for turn light on request."""
profile = Profiles.get(params.pop(ATTR_PROFILE, None))
def load_ip_bans_config(path: str):
"""Loading list of banned IPs from config file."""
ip_list = []
ip_schema = vol.Schema({
vol.Optional('banned_at'): vol.Any(None, cv.datetime)
})
try:
try:
list_ = load_yaml_config_file(path)
except HomeAssistantError as err:
_LOGGER.error('Unable to load %s: %s', path, str(err))
return []
for ip_ban, ip_info in list_.items():
try:
ip_info = ip_schema(ip_info)
ip_info['ip_ban'] = ip_address(ip_ban)
ip_list.append(IpBan(**ip_info))
except vol.Invalid:
return self.async_abort(reason="already_setup")
if not flows:
return self.async_abort(reason="no_flows")
if len(flows) == 1:
self.flow_impl = list(flows)[0]
return await self.async_step_auth()
if user_input is not None:
self.flow_impl = user_input["flow_impl"]
return await self.async_step_auth()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required("flow_impl"): vol.In(list(flows))}),
)
async def _show_config_form(self, name=None, latitude=None, longitude=None):
"""Show the configuration form to edit location data."""
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_NAME, default=name): str,
vol.Required(CONF_LATITUDE, default=latitude): cv.latitude,
vol.Required(CONF_LONGITUDE, default=longitude): cv.longitude,
}
),
errors=self._errors,
)
max_ = data_type(max_)
range_ = vol.Range(min=min_, max=max_)
validators.append(range_)
# construct key
key = vol.Required('value')
if 'default_value' in type_info:
default_value = type_info['default_value']
if data_type == bool:
default_value = default_value == '1'
else:
default_value = data_type(default_value)
key.default = default_value
return vol.Schema({key: vol.All(*validators)})
DEFAULT_SIGNAL_REPETITIONS = 10
SWITCH_SCHEMA = vol.Schema({
vol.Required(CONF_CODE_OFF):
vol.All(cv.ensure_list_csv, [cv.positive_int]),
vol.Required(CONF_CODE_ON):
vol.All(cv.ensure_list_csv, [cv.positive_int]),
vol.Optional(CONF_PULSELENGTH): cv.positive_int,
vol.Optional(CONF_SIGNAL_REPETITIONS,
default=DEFAULT_SIGNAL_REPETITIONS): cv.positive_int,
vol.Optional(CONF_PROTOCOL, default=DEFAULT_PROTOCOL): cv.positive_int,
})
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_GPIO): cv.positive_int,
vol.Required(CONF_SWITCHES): vol.Schema({cv.string: SWITCH_SCHEMA}),
})
# pylint: disable=no-member
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Find and return switches controlled by a generic RF device via GPIO."""
import rpi_rf
from threading import RLock
gpio = config.get(CONF_GPIO)
rfdevice = rpi_rf.RFDevice(gpio)
rfdevice_lock = RLock()
switches = config.get(CONF_SWITCHES)
devices = []
for dev_name, properties in switches.items():
from . import (
CONF_BOUNCETIME,
CONF_INVERT_LOGIC,
CONF_PULL_MODE,
DEFAULT_BOUNCETIME,
DEFAULT_INVERT_LOGIC,
DEFAULT_PULL_MODE,
)
from .. import remote_rpi_gpio
_LOGGER = logging.getLogger(__name__)
CONF_PORTS = "ports"
_SENSORS_SCHEMA = vol.Schema({cv.positive_int: cv.string})
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PORTS): _SENSORS_SCHEMA,
vol.Optional(CONF_INVERT_LOGIC, default=DEFAULT_INVERT_LOGIC): cv.boolean,
vol.Optional(CONF_BOUNCETIME, default=DEFAULT_BOUNCETIME): cv.positive_int,
vol.Optional(CONF_PULL_MODE, default=DEFAULT_PULL_MODE): cv.string,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Raspberry PI GPIO devices."""
address = config["host"]
invert_logic = config[CONF_INVERT_LOGIC]
def _validate_scenario(scenario):
any_str = Any(str, six.text_type)
schema = Schema({
Required(TemplateFields.CONDITION, msg=83): any_str,
Required(TemplateFields.ACTIONS, msg=84): list
})
result = _validate_dict_schema(schema, scenario)
if result.is_valid_config:
return _validate_actions_schema(scenario[TemplateFields.ACTIONS])
return result
CONF_ALLOW_HUE_GROUPS = "allow_hue_groups"
DEFAULT_ALLOW_HUE_GROUPS = True
BRIDGE_CONFIG_SCHEMA = vol.Schema([{
vol.Optional(CONF_HOST): cv.string,
vol.Optional(CONF_FILENAME, default=PHUE_CONFIG_FILE): cv.string,
vol.Optional(CONF_ALLOW_UNREACHABLE,
default=DEFAULT_ALLOW_UNREACHABLE): cv.boolean,
vol.Optional(CONF_ALLOW_IN_EMULATED_HUE,
default=DEFAULT_ALLOW_IN_EMULATED_HUE): cv.boolean,
vol.Optional(CONF_ALLOW_HUE_GROUPS,
default=DEFAULT_ALLOW_HUE_GROUPS): cv.boolean,
}])
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_BRIDGES, default=[]): BRIDGE_CONFIG_SCHEMA,
}),
}, extra=vol.ALLOW_EXTRA)
ATTR_GROUP_NAME = "group_name"
ATTR_SCENE_NAME = "scene_name"
SCENE_SCHEMA = vol.Schema({
vol.Required(ATTR_GROUP_NAME): cv.string,
vol.Required(ATTR_SCENE_NAME): cv.string,
})
CONFIG_INSTRUCTIONS = """
Press the button on the bridge to register Philips Hue with Home Assistant.
![Location of button on bridge](/static/images/config_philips_hue.jpg)