Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
schema[vol.Optional(CONF_TRIGGER_TIME)] = vol.All(
cv.time_period, cv.positive_timedelta
)
if state in SUPPORTED_PENDING_STATES:
schema[vol.Optional(CONF_PENDING_TIME)] = vol.All(
cv.time_period, cv.positive_timedelta
)
return vol.Schema(schema)
PLATFORM_SCHEMA = vol.Schema(
vol.All(
{
vol.Required(CONF_PLATFORM): "manual",
vol.Optional(CONF_NAME, default=DEFAULT_ALARM_NAME): cv.string,
vol.Exclusive(CONF_CODE, "code validation"): cv.string,
vol.Exclusive(CONF_CODE_TEMPLATE, "code validation"): cv.template,
vol.Optional(CONF_CODE_ARM_REQUIRED, default=True): cv.boolean,
vol.Optional(CONF_DELAY_TIME, default=DEFAULT_DELAY_TIME): vol.All(
cv.time_period, cv.positive_timedelta
),
vol.Optional(CONF_PENDING_TIME, default=DEFAULT_PENDING_TIME): vol.All(
cv.time_period, cv.positive_timedelta
),
vol.Optional(CONF_TRIGGER_TIME, default=DEFAULT_TRIGGER_TIME): vol.All(
cv.time_period, cv.positive_timedelta
),
vol.Optional(
CONF_DISARM_AFTER_TRIGGER, default=DEFAULT_DISARM_AFTER_TRIGGER
): cv.boolean,
vol.Optional(STATE_ALARM_ARMED_AWAY, default={}): _state_schema(
STATE_ALARM_ARMED_AWAY
CONTENT_TYPE_EXTENSIONS = {"audio/mpeg": "mp3", "audio/ogg": "ogg", "audio/pcm": "pcm"}
DEFAULT_ENGINE = "standard"
DEFAULT_VOICE = "Joanna"
DEFAULT_OUTPUT_FORMAT = "mp3"
DEFAULT_TEXT_TYPE = "text"
DEFAULT_SAMPLE_RATES = {"mp3": "22050", "ogg_vorbis": "22050", "pcm": "16000"}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_REGION, default=DEFAULT_REGION): vol.In(SUPPORTED_REGIONS),
vol.Inclusive(CONF_ACCESS_KEY_ID, ATTR_CREDENTIALS): cv.string,
vol.Inclusive(CONF_SECRET_ACCESS_KEY, ATTR_CREDENTIALS): cv.string,
vol.Exclusive(CONF_PROFILE_NAME, ATTR_CREDENTIALS): cv.string,
vol.Optional(CONF_VOICE, default=DEFAULT_VOICE): vol.In(SUPPORTED_VOICES),
vol.Optional(CONF_ENGINE, default=DEFAULT_ENGINE): vol.In(SUPPORTED_ENGINES),
vol.Optional(CONF_OUTPUT_FORMAT, default=DEFAULT_OUTPUT_FORMAT): vol.In(
SUPPORTED_OUTPUT_FORMATS
),
vol.Optional(CONF_SAMPLE_RATE): vol.All(
cv.string, vol.In(SUPPORTED_SAMPLE_RATES)
),
vol.Optional(CONF_TEXT_TYPE, default=DEFAULT_TEXT_TYPE): vol.In(
SUPPORTED_TEXT_TYPES
),
}
)
def get_engine(hass, config, discovery_info=None):
vol.Schema(
{
vol.Required(CONF_COMPONENT_NAME): cv.string,
vol.Optional(CONF_SIDEBAR_TITLE): cv.string,
vol.Optional(CONF_SIDEBAR_ICON, default=DEFAULT_ICON): cv.icon,
vol.Optional(CONF_URL_PATH): cv.string,
vol.Optional(CONF_CONFIG): dict,
vol.Exclusive(
CONF_WEBCOMPONENT_PATH,
CONF_URL_EXCLUSIVE_GROUP,
msg=MSG_URL_CONFLICT,
): cv.string,
vol.Exclusive(
CONF_JS_URL, CONF_URL_EXCLUSIVE_GROUP, msg=MSG_URL_CONFLICT
): cv.string,
vol.Exclusive(
CONF_MODULE_URL,
CONF_URL_EXCLUSIVE_GROUP,
msg=MSG_URL_CONFLICT,
): cv.string,
vol.Optional(
CONF_EMBED_IFRAME, default=DEFAULT_EMBED_IFRAME
): cv.boolean,
vol.Optional(
CONF_TRUST_EXTERNAL_SCRIPT, default=DEFAULT_TRUST_EXTERNAL
): cv.boolean,
vol.Optional(CONF_REQUIRE_ADMIN, default=False): cv.boolean,
}
)
],
)
},
import aiohttp
from homeassistant.components.switch import (SwitchDevice, PLATFORM_SCHEMA)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, CONF_TOKEN
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
HOOK_ENDPOINT = 'https://api.gethook.io/v1/'
TIMEOUT = 10
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Exclusive(CONF_PASSWORD, 'hook_secret', msg='hook: provide ' +
'username/password OR token'): cv.string,
vol.Exclusive(CONF_TOKEN, 'hook_secret', msg='hook: provide ' +
'username/password OR token'): cv.string,
vol.Inclusive(CONF_USERNAME, 'hook_auth'): cv.string,
vol.Inclusive(CONF_PASSWORD, 'hook_auth'): cv.string,
})
async def async_setup_platform(hass, config, async_add_entities,
discovery_info=None):
"""Set up Hook by getting the access token and list of actions."""
username = config.get(CONF_USERNAME)
password = config.get(CONF_PASSWORD)
token = config.get(CONF_TOKEN)
websession = async_get_clientsession(hass)
# If password is set in config, prefer it over token
if username is not None and password is not None:
try:
YEELIGHT_SLEEP_TRANSACTION = "SleepTransition"
YEELIGHT_FLOW_TRANSITION_SCHEMA = {
vol.Optional(ATTR_COUNT, default=0): cv.positive_int,
vol.Optional(ATTR_ACTION, default=ACTION_RECOVER): vol.Any(
ACTION_RECOVER, ACTION_OFF, ACTION_STAY
),
vol.Required(ATTR_TRANSITIONS): [
{
vol.Exclusive(YEELIGHT_RGB_TRANSITION, CONF_TRANSITION): vol.All(
cv.ensure_list, [cv.positive_int]
),
vol.Exclusive(YEELIGHT_HSV_TRANSACTION, CONF_TRANSITION): vol.All(
cv.ensure_list, [cv.positive_int]
),
vol.Exclusive(YEELIGHT_TEMPERATURE_TRANSACTION, CONF_TRANSITION): vol.All(
cv.ensure_list, [cv.positive_int]
),
vol.Exclusive(YEELIGHT_SLEEP_TRANSACTION, CONF_TRANSITION): vol.All(
cv.ensure_list, [cv.positive_int]
),
}
],
}
DEVICE_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_TRANSITION, default=DEFAULT_TRANSITION): cv.positive_int,
vol.Optional(CONF_MODE_MUSIC, default=False): cv.boolean,
vol.Optional(CONF_SAVE_ON_CHANGE, default=False): cv.boolean,
vol.Optional(CONF_NIGHTLIGHT_SWITCH_TYPE): vol.Any(
vol.Exclusive(ATTR_RGB_COLOR, COLOR_GROUP): vol.All(
vol.ExactSequence((cv.byte, cv.byte, cv.byte)), vol.Coerce(tuple)
),
vol.Exclusive(ATTR_XY_COLOR, COLOR_GROUP): vol.All(
vol.ExactSequence((cv.small_float, cv.small_float)), vol.Coerce(tuple)
),
vol.Exclusive(ATTR_HS_COLOR, COLOR_GROUP): vol.All(
vol.ExactSequence(
(
vol.All(vol.Coerce(float), vol.Range(min=0, max=360)),
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_PERIOD: vol.All(vol.Coerce(float), vol.Range(min=0.05)),
ATTR_CYCLES: vol.All(vol.Coerce(float), vol.Range(min=1)),
ATTR_MODE: vol.In(PULSE_MODES),
}
)
LIFX_EFFECT_COLORLOOP_SCHEMA = LIFX_EFFECT_SCHEMA.extend(
{
ATTR_BRIGHTNESS: VALID_BRIGHTNESS,
ATTR_BRIGHTNESS_PCT: VALID_BRIGHTNESS_PCT,
ATTR_PERIOD: vol.All(vol.Coerce(float), vol.Clamp(min=0.05)),
# Switch type platforms publish to MQTT and may subscribe
MQTT_RW_PLATFORM_SCHEMA = MQTT_BASE_PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_COMMAND_TOPIC): valid_publish_topic,
vol.Optional(CONF_RETAIN, default=DEFAULT_RETAIN): cv.boolean,
vol.Optional(CONF_STATE_TOPIC): valid_subscribe_topic,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
}
)
# Service call validation schema
MQTT_PUBLISH_SCHEMA = vol.Schema(
{
vol.Required(ATTR_TOPIC): valid_publish_topic,
vol.Exclusive(ATTR_PAYLOAD, CONF_PAYLOAD): object,
vol.Exclusive(ATTR_PAYLOAD_TEMPLATE, CONF_PAYLOAD): cv.string,
vol.Optional(ATTR_QOS, default=DEFAULT_QOS): _VALID_QOS_SCHEMA,
vol.Optional(ATTR_RETAIN, default=DEFAULT_RETAIN): cv.boolean,
},
required=True,
)
SubscribePayloadType = Union[str, bytes] # Only bytes if encoding is None
def _build_publish_data(topic: Any, qos: int, retain: bool) -> ServiceDataType:
"""Build the arguments for the publish service without the payload."""
data = {ATTR_TOPIC: topic}
if qos is not None:
data[ATTR_QOS] = qos
from cycledash import db
from cycledash.helpers import get_id_where, get_where, abort_if_none_for
from cycledash.validations import Doc, expect_one_of, PathString
from common.helpers import tables
import workers.runner
from . import genotypes, bams, marshal_with, validate_with, projects, Resource
CreateRun = Schema({
Required('uri'): PathString,
# One of `project` is required, but not supported in voluptuous, so we
# enforce this in code. cf. https://github.com/alecthomas/voluptuous/issues/115
Exclusive('project_id', 'project'): Coerce(int),
Exclusive('project_name', 'project'): unicode,
Exclusive('normal_bam_id', 'normal_bam'): Coerce(int),
Exclusive('normal_bam_uri', 'normal_bam'): PathString,
Exclusive('tumor_bam_id', 'tumor_bam'): Coerce(int),
Exclusive('tumor_bam_uri', 'tumor_bam'): PathString,
'caller_name': unicode,
'project_id': Coerce(int),
'tumor_dataset_id': Coerce(int),
'normal_dataset_id': Coerce(int),
'truth_vcf_path': PathString,
'is_validation': bool,
'notes': unicode,
'dataset': unicode,
'project_name': unicode,
extra=vol.ALLOW_EXTRA,
)
_EVENT_IN_TYPES = vol.Schema(
{
vol.Exclusive(EVENT_IN_DAYS, EVENT_TYPES_CONF): cv.positive_int,
vol.Exclusive(EVENT_IN_WEEKS, EVENT_TYPES_CONF): cv.positive_int,
}
)
ADD_EVENT_SERVICE_SCHEMA = vol.Schema(
{
vol.Required(EVENT_CALENDAR_ID): cv.string,
vol.Required(EVENT_SUMMARY): cv.string,
vol.Optional(EVENT_DESCRIPTION, default=""): cv.string,
vol.Exclusive(EVENT_START_DATE, EVENT_START_CONF): cv.date,
vol.Exclusive(EVENT_END_DATE, EVENT_END_CONF): cv.date,
vol.Exclusive(EVENT_START_DATETIME, EVENT_START_CONF): cv.datetime,
vol.Exclusive(EVENT_END_DATETIME, EVENT_END_CONF): cv.datetime,
vol.Exclusive(EVENT_IN, EVENT_START_CONF, EVENT_END_CONF): _EVENT_IN_TYPES,
}
)
def do_authentication(hass, hass_config, config):
"""Notify user of actions and authenticate.
Notify user of user_code and verification_url then poll
until we have an access token.
"""
oauth = OAuth2WebServerFlow(
client_id=config[CONF_CLIENT_ID],
{
vol.Exclusive(EVENT_IN_DAYS, EVENT_TYPES_CONF): cv.positive_int,
vol.Exclusive(EVENT_IN_WEEKS, EVENT_TYPES_CONF): cv.positive_int,
}
)
ADD_EVENT_SERVICE_SCHEMA = vol.Schema(
{
vol.Required(EVENT_CALENDAR_ID): cv.string,
vol.Required(EVENT_SUMMARY): cv.string,
vol.Optional(EVENT_DESCRIPTION, default=""): cv.string,
vol.Exclusive(EVENT_START_DATE, EVENT_START_CONF): cv.date,
vol.Exclusive(EVENT_END_DATE, EVENT_END_CONF): cv.date,
vol.Exclusive(EVENT_START_DATETIME, EVENT_START_CONF): cv.datetime,
vol.Exclusive(EVENT_END_DATETIME, EVENT_END_CONF): cv.datetime,
vol.Exclusive(EVENT_IN, EVENT_START_CONF, EVENT_END_CONF): _EVENT_IN_TYPES,
}
)
def do_authentication(hass, hass_config, config):
"""Notify user of actions and authenticate.
Notify user of user_code and verification_url then poll
until we have an access token.
"""
oauth = OAuth2WebServerFlow(
client_id=config[CONF_CLIENT_ID],
client_secret=config[CONF_CLIENT_SECRET],
scope="https://www.googleapis.com/auth/calendar",
redirect_uri="Home-Assistant.io",
)