Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
DEFAULT_TIMEOUT = 10
DEFAULT_CONFIDENCE = 80
SOURCE_SCHEMA = vol.Schema(
{
vol.Required(CONF_ENTITY_ID): cv.entity_domain("camera"),
vol.Optional(CONF_NAME): cv.string,
}
)
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_SOURCE): vol.All(cv.ensure_list, [SOURCE_SCHEMA]),
vol.Optional(CONF_CONFIDENCE, default=DEFAULT_CONFIDENCE): vol.All(
vol.Coerce(float), vol.Range(min=0, max=100)
),
}
)
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE.extend(PLATFORM_SCHEMA.schema)
async def async_setup(hass, config):
"""Set up the image processing."""
component = EntityComponent(_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
await component.async_setup(config)
async def async_scan_service(service):
"""Service handler for scan."""
image_entities = await component.async_extract_from_service(service)
),
[SUPPORT_VOLUME_SET],
)
component.async_register_entity_service(
SERVICE_VOLUME_MUTE,
{vol.Required(ATTR_MEDIA_VOLUME_MUTED): cv.boolean},
lambda entity, call: entity.async_mute_volume(
mute=call.data[ATTR_MEDIA_VOLUME_MUTED]
),
[SUPPORT_VOLUME_MUTE],
)
component.async_register_entity_service(
SERVICE_MEDIA_SEEK,
{
vol.Required(ATTR_MEDIA_SEEK_POSITION): vol.All(
vol.Coerce(float), vol.Range(min=0)
)
},
lambda entity, call: entity.async_media_seek(
position=call.data[ATTR_MEDIA_SEEK_POSITION]
),
[SUPPORT_SEEK],
)
component.async_register_entity_service(
SERVICE_SELECT_SOURCE,
{vol.Required(ATTR_INPUT_SOURCE): cv.string},
"async_select_source",
[SUPPORT_SELECT_SOURCE],
)
component.async_register_entity_service(
SERVICE_SELECT_SOUND_MODE,
{vol.Required(ATTR_SOUND_MODE): cv.string},
MIN_TIME_BETWEEN_FORCED_SCANS = timedelta(seconds=1)
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
SUPPORTED_COMMANDS = SUPPORT_TURN_ON | SUPPORT_TURN_OFF \
| SUPPORT_SELECT_SOURCE \
| SUPPORT_NEXT_TRACK | SUPPORT_PREVIOUS_TRACK \
| SUPPORT_VOLUME_MUTE | SUPPORT_VOLUME_STEP \
| SUPPORT_VOLUME_SET
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_ACCESS_TOKEN): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_SUPPRESS_WARNING, default=False): cv.boolean,
vol.Optional(CONF_VOLUME_STEP, default=DEFAULT_VOLUME_STEP):
vol.All(vol.Coerce(int), vol.Range(min=1, max=10)),
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the VizioTV media player platform."""
host = config.get(CONF_HOST)
token = config.get(CONF_ACCESS_TOKEN)
name = config.get(CONF_NAME)
volume_step = config.get(CONF_VOLUME_STEP)
device = VizioDevice(host, token, name, volume_step)
if device.validate_setup() is False:
_LOGGER.error("Failed to set up Vizio TV platform, "
"please check if host and API key are correct")
return
INTENT_SET = "HassLightSet"
async def async_setup_intents(hass: HomeAssistant) -> None:
"""Set up the light intents."""
hass.helpers.intent.async_register(SetIntentHandler())
class SetIntentHandler(intent.IntentHandler):
"""Handle set color intents."""
intent_type = INTENT_SET
slot_schema = {
vol.Required("name"): cv.string,
vol.Optional("color"): color_util.color_name_to_rgb,
vol.Optional("brightness"): vol.All(vol.Coerce(int), vol.Range(0, 100)),
}
async def async_handle(self, intent_obj: intent.Intent) -> intent.IntentResponse:
"""Handle the hass intent."""
hass = intent_obj.hass
slots = self.async_validate_slots(intent_obj.slots)
state = hass.helpers.intent.async_match_state(
slots["name"]["value"],
[state for state in hass.states.async_all() if state.domain == DOMAIN],
)
service_data = {ATTR_ENTITY_ID: state.entity_id}
speech_parts = []
if "color" in slots:
intent.async_test_feature(state, SUPPORT_COLOR, "changing colors")
Coerce(float), Range(0.3, 1), msg="wavelength must be float between .3 - 1"
),
"wiener": Any(-1, All(Coerce(float), Range(0, 50))),
"background": All(
Coerce(int),
Range(0, 65535),
msg="background must be int between 0 - 65,535",
),
"napodize": All(
Coerce(int), Range(0, 400), msg="napodize must be int between 0 - 400"
),
"nzblend": All(
Coerce(int), Range(0, 100), msg="nzblend must be int between 0 - 100"
),
"NA": All(
Coerce(float), Range(0.2, 1.33), msg="NA must be float between 0.2 - 1.33"
),
Exclusive("RL", "iterations"): All(
Coerce(int), Range(0, 30), msg="RL (nIters) must be int between 0 - 30"
),
Exclusive("nIters", "iterations"): All(
Coerce(int), Range(0, 30), msg="RL (nIters) must be int between 0 - 30"
),
"deskew": All(
Coerce(float),
Range(-180, 180),
msg="deskew angle must be float between -180 and 180",
),
"width": All(
Coerce(int), Range(0, 2000), msg="width must be int between 0 - 2000"
),
"shift": All(
SUPPORT_TURN_ON = 1
SUPPORT_TURN_OFF = 2
SUPPORT_LOAD_URL = 4
SUPPORT_SET_BRIGHTNESS = 8
DISPLAY_DEVICE_SCHEMA = vol.Schema({
ATTR_ENTITY_ID: cv.entity_ids,
})
DISPLAY_DEVICE_LOAD_URL_SCHEMA = DISPLAY_DEVICE_SCHEMA.extend({
vol.Required(ATTR_URL): cv.string,
})
DISPLAY_DEVICE_SET_BRIGHTNESS_SCHEMA = DISPLAY_DEVICE_SCHEMA.extend({
vol.Optional(ATTR_BRIGHTNESS, default=None):
vol.Any(
vol.All(str, vol.Length(min=0, max=3)),
vol.All(int, vol.Range(min=0, max=255))
)
})
@bind_hass
def is_on(hass, entity_id=None):
entity_id = entity_id or ENTITY_ID_ALL_DISPLAYS
return hass.states.is_state(entity_id, STATE_ON)
async def async_setup(hass, config):
component = EntityComponent(
_LOGGER, DOMAIN, hass, SCAN_INTERVAL, GROUP_NAME_ALL_DISPLAYS)
await component.async_setup(config)
CONF_WEEKDAY_ORDER_NUMBER: {
"step": 4,
"valid_for": lambda f: f in MONTHLY_FREQUENCY,
"method": vol.Optional,
"type": [int],
"validator": vol.All(
cv.ensure_list, [vol.All(vol.Coerce(int), vol.Range(min=1, max=5))]
),
},
CONF_WEEK_ORDER_NUMBER: {
"step": 4,
"valid_for": lambda f: f in MONTHLY_FREQUENCY,
"method": vol.Optional,
"type": [int],
"validator": vol.All(
cv.ensure_list, [vol.All(vol.Coerce(int), vol.Range(min=1, max=5))]
),
},
CONF_FIRST_MONTH: {
"step": 4,
"valid_for": lambda f: f in EXCEPT_ANNUAL_GROUP,
"method": vol.Optional,
"default": DEFAULT_FIRST_MONTH,
"type": vol.In(MONTH_OPTIONS),
},
CONF_LAST_MONTH: {
"step": 4,
"valid_for": lambda f: f in EXCEPT_ANNUAL_GROUP,
"method": vol.Optional,
"default": DEFAULT_LAST_MONTH,
"type": vol.In(MONTH_OPTIONS),
},
vol.Optional(CONF_BUS_VOLTAGE): cv.nameable(sensor.SENSOR_SCHEMA.extend({
cv.GenerateID(): cv.declare_variable_id(INA219VoltageSensor),
})),
vol.Optional(CONF_SHUNT_VOLTAGE): cv.nameable(sensor.SENSOR_SCHEMA.extend({
cv.GenerateID(): cv.declare_variable_id(INA219VoltageSensor),
})),
vol.Optional(CONF_CURRENT): cv.nameable(sensor.SENSOR_SCHEMA.extend({
cv.GenerateID(): cv.declare_variable_id(INA219CurrentSensor),
})),
vol.Optional(CONF_POWER): cv.nameable(sensor.SENSOR_SCHEMA.extend({
cv.GenerateID(): cv.declare_variable_id(INA219PowerSensor),
})),
vol.Optional(CONF_SHUNT_RESISTANCE, default=0.1): vol.All(cv.resistance,
vol.Range(min=0.0, max=32.0)),
vol.Optional(CONF_MAX_VOLTAGE, default=32.0): vol.All(cv.voltage, vol.Range(min=0.0, max=32.0)),
vol.Optional(CONF_MAX_CURRENT, default=3.2): vol.All(cv.current, vol.Range(min=0.0)),
vol.Optional(CONF_UPDATE_INTERVAL): cv.update_interval,
}).extend(cv.COMPONENT_SCHEMA.schema), cv.has_at_least_one_key(*SENSOR_KEYS))
def to_code(config):
rhs = App.make_ina219(config[CONF_SHUNT_RESISTANCE],
config[CONF_MAX_CURRENT], config[CONF_MAX_VOLTAGE],
config[CONF_ADDRESS], config.get(CONF_UPDATE_INTERVAL))
ina = Pvariable(config[CONF_ID], rhs)
if CONF_BUS_VOLTAGE in config:
conf = config[CONF_BUS_VOLTAGE]
sensor.register_sensor(ina.Pmake_bus_voltage_sensor(conf[CONF_NAME]), conf)
if CONF_SHUNT_VOLTAGE in config:
conf = config[CONF_SHUNT_VOLTAGE]
sensor.register_sensor(ina.Pmake_shunt_voltage_sensor(conf[CONF_NAME]), conf)
if CONF_CURRENT in config:
"tram": "mdi:tram",
"water": "mdi:ferry",
}
SCAN_INTERVAL = timedelta(seconds=45)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_STOP_IDS): vol.All(cv.ensure_list, [cv.string]),
vol.Optional(CONF_EXPAND_PLATFORMS, default=True): cv.boolean,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_SHOW_ON_MAP, default=False): cv.boolean,
vol.Optional(CONF_WHITELIST_LINES, default=[]): cv.ensure_list,
vol.Optional(CONF_OMIT_NON_BOARDING, default=True): cv.boolean,
vol.Optional(CONF_NUMBER_OF_DEPARTURES, default=2): vol.All(
cv.positive_int, vol.Range(min=2, max=10)
),
}
)
ATTR_STOP_ID = "stop_id"
ATTR_ROUTE = "route"
ATTR_ROUTE_ID = "route_id"
ATTR_EXPECTED_AT = "due_at"
ATTR_DELAY = "delay"
ATTR_REALTIME = "real_time"
ATTR_NEXT_UP_IN = "next_due_in"
ATTR_NEXT_UP_ROUTE = "next_route"
ATTR_NEXT_UP_ROUTE_ID = "next_route_id"
from homeassistant.core import valid_entity_id, split_entity_id
from homeassistant.exceptions import TemplateError
from homeassistant.helpers.logging import KeywordStyleAdapter
from homeassistant.util import slugify as util_slugify
# mypy: allow-untyped-calls, allow-untyped-defs
# mypy: no-check-untyped-defs, no-warn-return-any
# pylint: disable=invalid-name
TIME_PERIOD_ERROR = "offset {} should be format 'HH:MM' or 'HH:MM:SS'"
# Home Assistant types
byte = vol.All(vol.Coerce(int), vol.Range(min=0, max=255))
small_float = vol.All(vol.Coerce(float), vol.Range(min=0, max=1))
positive_int = vol.All(vol.Coerce(int), vol.Range(min=0))
latitude = vol.All(
vol.Coerce(float), vol.Range(min=-90, max=90), msg="invalid latitude"
)
longitude = vol.All(
vol.Coerce(float), vol.Range(min=-180, max=180), msg="invalid longitude"
)
gps = vol.ExactSequence([latitude, longitude])
sun_event = vol.All(vol.Lower, vol.Any(SUN_EVENT_SUNSET, SUN_EVENT_SUNRISE))
port = vol.All(vol.Coerce(int), vol.Range(min=1, max=65535))
# typing typevar
T = TypeVar("T")
# Adapted from: