Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def sequence_of_sequences(min=None, max=None):
return All(
Any(
None,
[Any(list, tuple)],
tuple([Any(list, tuple)]),
),
Any(
None,
[Length(min=min, max=max)],
tuple([Length(min=min, max=max)]),
),
"usage_gb": ["Usage", GIGABYTES, "mdi:download"],
"limit": ["Data limit", GIGABYTES, "mdi:download"],
"used_download": ["Used Download", GIGABYTES, "mdi:download"],
"used_upload": ["Used Upload", GIGABYTES, "mdi:upload"],
"used_total": ["Used Total", GIGABYTES, "mdi:download"],
"grace_download": ["Grace Download", GIGABYTES, "mdi:download"],
"grace_upload": ["Grace Upload", GIGABYTES, "mdi:upload"],
"grace_total": ["Grace Total", GIGABYTES, "mdi:download"],
"total_download": ["Total Download", GIGABYTES, "mdi:download"],
"total_upload": ["Total Upload", GIGABYTES, "mdi:download"],
"used_remaining": ["Remaining", GIGABYTES, "mdi:download"],
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_MONITORED_VARIABLES): vol.All(
cv.ensure_list, [vol.In(SENSOR_TYPES)]
),
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_TOTAL_BANDWIDTH): cv.positive_int,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
}
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the sensor platform."""
websession = async_get_clientsession(hass)
apikey = config.get(CONF_API_KEY)
bandwidthcap = config.get(CONF_TOTAL_BANDWIDTH)
ts_data = StartcaData(hass.loop, websession, apikey, bandwidthcap)
}
),
cv.has_at_least_one_key(CONF_PIN, CONF_ZONE),
)
# pylint: disable=no-value-for-parameter
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_ACCESS_TOKEN): cv.string,
vol.Optional(CONF_API_HOST): vol.Url(),
vol.Required(CONF_DEVICES): [
{
vol.Required(CONF_ID): cv.matches_regex("[0-9a-f]{12}"),
vol.Optional(CONF_BINARY_SENSORS): vol.All(
cv.ensure_list, [_BINARY_SENSOR_SCHEMA]
),
vol.Optional(CONF_SENSORS): vol.All(
cv.ensure_list, [_SENSOR_SCHEMA]
),
vol.Optional(CONF_SWITCHES): vol.All(
cv.ensure_list, [_SWITCH_SCHEMA]
),
vol.Optional(CONF_HOST): cv.string,
vol.Optional(CONF_PORT): cv.port,
vol.Optional(CONF_BLINK, default=True): cv.boolean,
vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
}
],
}
)
SENSOR_SCHEMA = vol.Schema(
{
vol.Optional(CONF_ENABLED, default=True): cv.boolean,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_DEVICE, default="homeassistant"): cv.string,
}
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_SENSOR, default=[{}]): vol.All(cv.ensure_list, [SENSOR_SCHEMA]),
}
)
},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass, config):
"""Set up this component."""
from mygpoclient import api
# Print startup message
startup = STARTUP.format(name=DOMAIN, version=VERSION, issueurl=ISSUE_URL)
_LOGGER.info(startup)
# Check that all required files are present
file_check = await check_files(hass)
CONST_HUISNUMMER = "huisnummer"
# Predefined types and id's
TRASH_TYPES = {
'gft': [5, 'Groene Bak GFT', 'mdi:delete-empty'],
'plastic': [6, 'Plastic en Verpakking', 'mdi:delete-empty'],
'papier': [3, 'Blauwe Bak Papier', 'mdi:delete-empty'],
'restafval': [2, 'Grijze Bak Restafval', 'mdi:delete-empty'],
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONST_POSTCODE): cv.string,
vol.Required(CONST_HUISNUMMER): cv.string,
vol.Required(CONF_RESOURCES, default=[]):
vol.All(cv.ensure_list, [vol.In(TRASH_TYPES)]),
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Setup the HVCGroep sensors."""
scan_interval = config.get(CONF_SCAN_INTERVAL)
postcode = config.get(CONST_POSTCODE)
huisnummer = config.get(CONST_HUISNUMMER)
name = config.get(CONF_NAME)
try:
data = TrashData(postcode, huisnummer)
except requests.exceptions.HTTPError as error:
_LOGGER.error(error)
return False
CONF_API_SECRET = "api_secret"
CONF_ACCOUNT_CURRENCIES = "account_balance_currencies"
CONF_EXCHANGE_CURRENCIES = "exchange_rate_currencies"
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1)
DATA_COINBASE = "coinbase_cache"
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_API_SECRET): cv.string,
vol.Optional(CONF_ACCOUNT_CURRENCIES): vol.All(
cv.ensure_list, [cv.string]
),
vol.Optional(CONF_EXCHANGE_CURRENCIES, default=[]): vol.All(
cv.ensure_list, [cv.string]
),
}
)
},
extra=vol.ALLOW_EXTRA,
)
def setup(hass, config):
"""Set up the Coinbase component.
Will automatically setup sensors to support
from . import ATTRIBUTION, DATA_SMARTWEATHER
DEPENDENCIES = ['smartweather']
_LOGGER = logging.getLogger(__name__)
SENSOR_TYPES = {
'raining': ['Raining', None, 'mdi:water', 'mdi:water-off'],
'freezing': ['Freezing', None, 'mdi:fridge', 'mdi:fridge-outline'],
'lightning': ['Lightning', None, 'mdi:weather-lightning', 'mdi:flash-off']
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_MONITORED_CONDITIONS, default=list(SENSOR_TYPES)):
vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)]),
vol.Optional(CONF_NAME, default=DATA_SMARTWEATHER): cv.string
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the SmartWeather binary sensor platform."""
name = config.get(CONF_NAME)
data = hass.data[DATA_SMARTWEATHER]
if data.data.timestamp is None:
return
sensors = []
for variable in config[CONF_MONITORED_CONDITIONS]:
sensors.append(SmartWeatherBinarySensor(hass, data, variable, name))
_LOGGER.debug("Binary ensor added: %s", variable)
def validate_ok_action_list(v):
validate_action_list(v, 'OK')
def validate_alarm_action_list(v):
validate_action_list(v, 'ALARM')
def validate_undetermined_action_list(v):
validate_action_list(v, 'UNDETERMINED')
alarm_definition_schema = {
Required('name'): All(Any(str, six.text_type), Length(max=255)),
Required('expression'): All(Any(str, six.text_type)),
Marker('description'): All(Any(str, six.text_type), Length(max=255)),
Marker('severity'): All(Upper, Any('LOW', 'MEDIUM', 'HIGH', 'CRITICAL')),
Marker('match_by'): Any([six.text_type], [str]),
Marker('ok_actions'): validate_ok_action_list,
Marker('alarm_actions'): validate_alarm_action_list,
Marker('undetermined_actions'): validate_undetermined_action_list,
Marker('actions_enabled'): bool}
def validate(msg, require_all=False):
try:
request_body_schema = Schema(alarm_definition_schema,
required=require_all,
extra=True)
request_body_schema(msg)
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=5)
# Sensor types: Name, device_class, event
SENSOR_TYPES = {
"button": ["Button", "occupancy", "device:sensor:button"],
"motion": ["Motion", "motion", "device:sensor:motion"],
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(
CONF_ENTITY_NAMESPACE, default=DEFAULT_ENTITY_NAMESPACE
): cv.string,
vol.Required(CONF_MONITORED_CONDITIONS, default=[]): vol.All(
cv.ensure_list, [vol.In(SENSOR_TYPES)]
),
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the platform for a Skybell device."""
skybell = hass.data.get(SKYBELL_DOMAIN)
sensors = []
for sensor_type in config.get(CONF_MONITORED_CONDITIONS):
for device in skybell.get_devices():
sensors.append(SkybellBinarySensor(device, sensor_type))
add_entities(sensors, True)
SIGNAL_TELLCORE_CALLBACK = "tellstick_callback"
# Use a global tellstick domain lock to avoid getting Tellcore errors when
# calling concurrently.
TELLSTICK_LOCK = threading.RLock()
# A TellstickRegistry that keeps a map from tellcore_id to the corresponding
# tellcore_device and HA device (entity).
TELLCORE_REGISTRY = None
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Inclusive(CONF_HOST, "tellcore-net"): cv.string,
vol.Inclusive(CONF_PORT, "tellcore-net"): vol.All(
cv.ensure_list, [cv.port], vol.Length(min=2, max=2)
),
vol.Optional(
CONF_SIGNAL_REPETITIONS, default=DEFAULT_SIGNAL_REPETITIONS
): vol.Coerce(int),
}
)
},
extra=vol.ALLOW_EXTRA,
)
def _discover(hass, config, component_name, found_tellcore_devices):
"""Set up and send the discovery event."""
if not found_tellcore_devices:
return