jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 22:27:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove global variable from zigbee (#33750)

Browse Source 
* Remove global variable from zigbee

* Pass device instead of hass into the constructor
This commit is contained in:
springstan 2020-04-07 17:41:23 +02:00 committed by GitHub
parent 894aac1b45
commit 325e5416ef
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 64 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
homeassistant/components/zigbee/__init__.py
View File

@ -33,20 +33,8 @@ DEFAULT_DEVICE = "/dev/ttyUSB0"
DEFAULT_BAUD = 9600 DEFAULT_BAUD = 9600
DEFAULT_ADC_MAX_VOLTS = 1.2 DEFAULT_ADC_MAX_VOLTS = 1.2
# Copied from xbee_helper during setup()
GPIO_DIGITAL_OUTPUT_LOW = None
GPIO_DIGITAL_OUTPUT_HIGH = None
ADC_PERCENTAGE = None
DIGITAL_PINS = None
ANALOG_PINS = None
CONVERT_ADC = None
ZIGBEE_EXCEPTION = None
ZIGBEE_TX_FAILURE = None
ATTR_FRAME = "frame" ATTR_FRAME = "frame"
DEVICE = None
CONFIG_SCHEMA = vol.Schema( CONFIG_SCHEMA = vol.Schema(
{ {
DOMAIN: vol.Schema( DOMAIN: vol.Schema(
@ -71,24 +59,6 @@ PLATFORM_SCHEMA = vol.Schema(
def setup(hass, config): def setup(hass, config):
"""Set up the connection to the Zigbee device.""" """Set up the connection to the Zigbee device."""
global DEVICE # pylint: disable=global-statement
global GPIO_DIGITAL_OUTPUT_LOW # pylint: disable=global-statement
global GPIO_DIGITAL_OUTPUT_HIGH # pylint: disable=global-statement
global ADC_PERCENTAGE # pylint: disable=global-statement
global DIGITAL_PINS # pylint: disable=global-statement
global ANALOG_PINS # pylint: disable=global-statement
global CONVERT_ADC # pylint: disable=global-statement
global ZIGBEE_EXCEPTION # pylint: disable=global-statement
global ZIGBEE_TX_FAILURE # pylint: disable=global-statement
GPIO_DIGITAL_OUTPUT_LOW = xb_const.GPIO_DIGITAL_OUTPUT_LOW
GPIO_DIGITAL_OUTPUT_HIGH = xb_const.GPIO_DIGITAL_OUTPUT_HIGH
ADC_PERCENTAGE = xb_const.ADC_PERCENTAGE
DIGITAL_PINS = xb_const.DIGITAL_PINS
ANALOG_PINS = xb_const.ANALOG_PINS
CONVERT_ADC = convert_adc
ZIGBEE_EXCEPTION = ZigBeeException
ZIGBEE_TX_FAILURE = ZigBeeTxFailure
usb_device = config[DOMAIN].get(CONF_DEVICE, DEFAULT_DEVICE) usb_device = config[DOMAIN].get(CONF_DEVICE, DEFAULT_DEVICE)
baud = int(config[DOMAIN].get(CONF_BAUD, DEFAULT_BAUD)) baud = int(config[DOMAIN].get(CONF_BAUD, DEFAULT_BAUD))
@ -97,8 +67,11 @@ def setup(hass, config):
except SerialException as exc: except SerialException as exc:
_LOGGER.exception("Unable to open serial port for Zigbee: %s", exc) _LOGGER.exception("Unable to open serial port for Zigbee: %s", exc)
return False return False
DEVICE = ZigBee(ser) zigbee_device = ZigBee(ser)
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, close_serial_port)
def close_serial_port(*args):
"""Close the serial port we're using to communicate with the Zigbee."""
zigbee_device.zb.serial.close()
def _frame_received(frame): def _frame_received(frame):
"""Run when a Zigbee frame is received. """Run when a Zigbee frame is received.
@ -108,16 +81,13 @@ def setup(hass, config):
""" """
dispatcher_send(hass, SIGNAL_ZIGBEE_FRAME_RECEIVED, frame) dispatcher_send(hass, SIGNAL_ZIGBEE_FRAME_RECEIVED, frame)
DEVICE.add_frame_rx_handler(_frame_received) hass.data[DOMAIN] = zigbee_device
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, close_serial_port)
zigbee_device.add_frame_rx_handler(_frame_received)
return True return True
def close_serial_port(*args):
"""Close the serial port we're using to communicate with the Zigbee."""
DEVICE.zb.serial.close()
def frame_is_relevant(entity, frame): def frame_is_relevant(entity, frame):
"""Test whether the frame is relevant to the entity.""" """Test whether the frame is relevant to the entity."""
if frame.get("source_addr_long") != entity.config.address: if frame.get("source_addr_long") != entity.config.address:
@ -229,13 +199,13 @@ class ZigBeeDigitalOutConfig(ZigBeePinConfig):
""" """
if self._config.get("on_state", "").lower() == "low": if self._config.get("on_state", "").lower() == "low":
bool2state = { bool2state = {
True: GPIO_DIGITAL_OUTPUT_LOW, True: xb_const.GPIO_DIGITAL_OUTPUT_LOW,
False: GPIO_DIGITAL_OUTPUT_HIGH, False: xb_const.GPIO_DIGITAL_OUTPUT_HIGH,
} }
else: else:
bool2state = { bool2state = {
True: GPIO_DIGITAL_OUTPUT_HIGH, True: xb_const.GPIO_DIGITAL_OUTPUT_HIGH,
False: GPIO_DIGITAL_OUTPUT_LOW, False: xb_const.GPIO_DIGITAL_OUTPUT_LOW,
} }
state2bool = {v: k for k, v in bool2state.items()} state2bool = {v: k for k, v in bool2state.items()}
return bool2state, state2bool return bool2state, state2bool
@ -269,9 +239,10 @@ class ZigBeeAnalogInConfig(ZigBeePinConfig):
class ZigBeeDigitalIn(Entity): class ZigBeeDigitalIn(Entity):
"""Representation of a GPIO pin configured as a digital input.""" """Representation of a GPIO pin configured as a digital input."""
def __init__(self, hass, config): def __init__(self, config, device):
"""Initialize the device.""" """Initialize the device."""
self._config = config self._config = config
self._device = device
self._state = False self._state = False
async def async_added_to_hass(self): async def async_added_to_hass(self):
@ -286,7 +257,7 @@ class ZigBeeDigitalIn(Entity):
if not frame_is_relevant(self, frame): if not frame_is_relevant(self, frame):
return return
sample = next(iter(frame["samples"])) sample = next(iter(frame["samples"]))
pin_name = DIGITAL_PINS[self._config.pin] pin_name = xb_const.DIGITAL_PINS[self._config.pin]
if pin_name not in sample: if pin_name not in sample:
# Doesn't contain information about our pin # Doesn't contain information about our pin
return return
@ -322,18 +293,18 @@ class ZigBeeDigitalIn(Entity):
def update(self): def update(self):
"""Ask the Zigbee device what state its input pin is in.""" """Ask the Zigbee device what state its input pin is in."""
try: try:
sample = DEVICE.get_sample(self._config.address) sample = self._device.get_sample(self._config.address)
except ZIGBEE_TX_FAILURE: except ZigBeeTxFailure:
_LOGGER.warning( _LOGGER.warning(
"Transmission failure when attempting to get sample from " "Transmission failure when attempting to get sample from "
"Zigbee device at address: %s", "Zigbee device at address: %s",
hexlify(self._config.address), hexlify(self._config.address),
) )
return return
except ZIGBEE_EXCEPTION as exc: except ZigBeeException as exc:
_LOGGER.exception("Unable to get sample from Zigbee device: %s", exc) _LOGGER.exception("Unable to get sample from Zigbee device: %s", exc)
return return
pin_name = DIGITAL_PINS[self._config.pin] pin_name = xb_const.DIGITAL_PINS[self._config.pin]
if pin_name not in sample: if pin_name not in sample:
_LOGGER.warning( _LOGGER.warning(
"Pin %s (%s) was not in the sample provided by Zigbee device %s.", "Pin %s (%s) was not in the sample provided by Zigbee device %s.",
@ -351,17 +322,17 @@ class ZigBeeDigitalOut(ZigBeeDigitalIn):
def _set_state(self, state): def _set_state(self, state):
"""Initialize the Zigbee digital out device.""" """Initialize the Zigbee digital out device."""
try: try:
DEVICE.set_gpio_pin( self._device.set_gpio_pin(
self._config.pin, self._config.bool2state[state], self._config.address self._config.pin, self._config.bool2state[state], self._config.address
) )
except ZIGBEE_TX_FAILURE: except ZigBeeTxFailure:
_LOGGER.warning( _LOGGER.warning(
"Transmission failure when attempting to set output pin on " "Transmission failure when attempting to set output pin on "
"Zigbee device at address: %s", "Zigbee device at address: %s",
hexlify(self._config.address), hexlify(self._config.address),
) )
return return
except ZIGBEE_EXCEPTION as exc: except ZigBeeException as exc:
_LOGGER.exception("Unable to set digital pin on Zigbee device: %s", exc) _LOGGER.exception("Unable to set digital pin on Zigbee device: %s", exc)
return return
self._state = state self._state = state
@ -379,15 +350,17 @@ class ZigBeeDigitalOut(ZigBeeDigitalIn):
def update(self): def update(self):
"""Ask the Zigbee device what its output is set to.""" """Ask the Zigbee device what its output is set to."""
try: try:
pin_state = DEVICE.get_gpio_pin(self._config.pin, self._config.address) pin_state = self._device.get_gpio_pin(
except ZIGBEE_TX_FAILURE: self._config.pin, self._config.address
)
except ZigBeeTxFailure:
_LOGGER.warning( _LOGGER.warning(
"Transmission failure when attempting to get output pin status" "Transmission failure when attempting to get output pin status"
" from Zigbee device at address: %s", " from Zigbee device at address: %s",
hexlify(self._config.address), hexlify(self._config.address),
) )
return return
except ZIGBEE_EXCEPTION as exc: except ZigBeeException as exc:
_LOGGER.exception( _LOGGER.exception(
"Unable to get output pin status from Zigbee device: %s", exc "Unable to get output pin status from Zigbee device: %s", exc
) )
@ -398,9 +371,10 @@ class ZigBeeDigitalOut(ZigBeeDigitalIn):
class ZigBeeAnalogIn(Entity): class ZigBeeAnalogIn(Entity):
"""Representation of a GPIO pin configured as an analog input.""" """Representation of a GPIO pin configured as an analog input."""
def __init__(self, hass, config): def __init__(self, config, device):
"""Initialize the ZigBee analog in device.""" """Initialize the ZigBee analog in device."""
self._config = config self._config = config
self._device = device
self._value = None self._value = None
async def async_added_to_hass(self): async def async_added_to_hass(self):
@ -415,12 +389,12 @@ class ZigBeeAnalogIn(Entity):
if not frame_is_relevant(self, frame): if not frame_is_relevant(self, frame):
return return
sample = frame["samples"].pop() sample = frame["samples"].pop()
pin_name = ANALOG_PINS[self._config.pin] pin_name = xb_const.ANALOG_PINS[self._config.pin]
if pin_name not in sample: if pin_name not in sample:
# Doesn't contain information about our pin # Doesn't contain information about our pin
return return
self._value = CONVERT_ADC( self._value = convert_adc(
sample[pin_name], ADC_PERCENTAGE, self._config.max_voltage sample[pin_name], xb_const.ADC_PERCENTAGE, self._config.max_voltage
) )
self.schedule_update_ha_state() self.schedule_update_ha_state()
@ -454,17 +428,17 @@ class ZigBeeAnalogIn(Entity):
def update(self): def update(self):
"""Get the latest reading from the ADC.""" """Get the latest reading from the ADC."""
try: try:
self._value = DEVICE.read_analog_pin( self._value = self._device.read_analog_pin(
self._config.pin, self._config.pin,
self._config.max_voltage, self._config.max_voltage,
self._config.address, self._config.address,
ADC_PERCENTAGE, xb_const.ADC_PERCENTAGE,
) )
except ZIGBEE_TX_FAILURE: except ZigBeeTxFailure:
_LOGGER.warning( _LOGGER.warning(
"Transmission failure when attempting to get sample from " "Transmission failure when attempting to get sample from "
"Zigbee device at address: %s", "Zigbee device at address: %s",
hexlify(self._config.address), hexlify(self._config.address),
) )
except ZIGBEE_EXCEPTION as exc: except ZigBeeException as exc:
_LOGGER.exception("Unable to get sample from Zigbee device: %s", exc) _LOGGER.exception("Unable to get sample from Zigbee device: %s", exc)

7
homeassistant/components/zigbee/binary_sensor.py
View File

@ -3,7 +3,7 @@ import voluptuous as vol
from homeassistant.components.binary_sensor import BinarySensorDevice from homeassistant.components.binary_sensor import BinarySensorDevice
from . import PLATFORM_SCHEMA, ZigBeeDigitalIn, ZigBeeDigitalInConfig from . import DOMAIN, PLATFORM_SCHEMA, ZigBeeDigitalIn, ZigBeeDigitalInConfig
CONF_ON_STATE = "on_state" CONF_ON_STATE = "on_state"
@ -15,7 +15,10 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({vol.Optional(CONF_ON_STATE): vol.In(ST
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Zigbee binary sensor platform.""" """Set up the Zigbee binary sensor platform."""
add_entities([ZigBeeBinarySensor(hass, ZigBeeDigitalInConfig(config))], True) zigbee_device = hass.data[DOMAIN]
add_entities(
[ZigBeeBinarySensor(ZigBeeDigitalInConfig(config), zigbee_device)], True
)
class ZigBeeBinarySensor(ZigBeeDigitalIn, BinarySensorDevice): class ZigBeeBinarySensor(ZigBeeDigitalIn, BinarySensorDevice):

5
homeassistant/components/zigbee/light.py
View File

@ -3,7 +3,7 @@ import voluptuous as vol
from homeassistant.components.light import Light from homeassistant.components.light import Light
from . import PLATFORM_SCHEMA, ZigBeeDigitalOut, ZigBeeDigitalOutConfig from . import DOMAIN, PLATFORM_SCHEMA, ZigBeeDigitalOut, ZigBeeDigitalOutConfig
CONF_ON_STATE = "on_state" CONF_ON_STATE = "on_state"
@ -17,7 +17,8 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Create and add an entity based on the configuration.""" """Create and add an entity based on the configuration."""
add_entities([ZigBeeLight(hass, ZigBeeDigitalOutConfig(config))]) zigbee_device = hass.data[DOMAIN]
add_entities([ZigBeeLight(ZigBeeDigitalOutConfig(config), zigbee_device)])
class ZigBeeLight(ZigBeeDigitalOut, Light): class ZigBeeLight(ZigBeeDigitalOut, Light):

26
homeassistant/components/zigbee/sensor.py
View File

@ -3,12 +3,18 @@ from binascii import hexlify
import logging import logging
import voluptuous as vol import voluptuous as vol
from xbee_helper.exceptions import ZigBeeException, ZigBeeTxFailure
from homeassistant.components import zigbee
from homeassistant.const import TEMP_CELSIUS from homeassistant.const import TEMP_CELSIUS
from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity import Entity
from . import PLATFORM_SCHEMA from . import (
DOMAIN,
PLATFORM_SCHEMA,
ZigBeeAnalogIn,
ZigBeeAnalogInConfig,
ZigBeeConfig,
)
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -32,6 +38,7 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
Uses the 'type' config value to work out which type of Zigbee sensor we're Uses the 'type' config value to work out which type of Zigbee sensor we're
dealing with and instantiates the relevant classes to handle it. dealing with and instantiates the relevant classes to handle it.
""" """
zigbee_device = hass.data[DOMAIN]
typ = config.get(CONF_TYPE) typ = config.get(CONF_TYPE)
try: try:
@ -40,15 +47,16 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
_LOGGER.exception("Unknown Zigbee sensor type: %s", typ) _LOGGER.exception("Unknown Zigbee sensor type: %s", typ)
return return
add_entities([sensor_class(hass, config_class(config))], True) add_entities([sensor_class(config_class(config), zigbee_device)], True)
class ZigBeeTemperatureSensor(Entity): class ZigBeeTemperatureSensor(Entity):
"""Representation of XBee Pro temperature sensor.""" """Representation of XBee Pro temperature sensor."""
def __init__(self, hass, config): def __init__(self, config, device):
"""Initialize the sensor.""" """Initialize the sensor."""
self._config = config self._config = config
self._device = device
self._temp = None self._temp = None
@property @property
@ -69,19 +77,19 @@ class ZigBeeTemperatureSensor(Entity):
def update(self): def update(self):
"""Get the latest data.""" """Get the latest data."""
try: try:
self._temp = zigbee.DEVICE.get_temperature(self._config.address) self._temp = self._device.get_temperature(self._config.address)
except zigbee.ZIGBEE_TX_FAILURE: except ZigBeeTxFailure:
_LOGGER.warning( _LOGGER.warning(
"Transmission failure when attempting to get sample from " "Transmission failure when attempting to get sample from "
"Zigbee device at address: %s", "Zigbee device at address: %s",
hexlify(self._config.address), hexlify(self._config.address),
) )
except zigbee.ZIGBEE_EXCEPTION as exc: except ZigBeeException as exc:
_LOGGER.exception("Unable to get sample from Zigbee device: %s", exc) _LOGGER.exception("Unable to get sample from Zigbee device: %s", exc)
# This must be below the classes to which it refers. # This must be below the classes to which it refers.
TYPE_CLASSES = { TYPE_CLASSES = {
"temperature": (ZigBeeTemperatureSensor, zigbee.ZigBeeConfig), "temperature": (ZigBeeTemperatureSensor, ZigBeeConfig),
"analog": (zigbee.ZigBeeAnalogIn, zigbee.ZigBeeAnalogInConfig), "analog": (ZigBeeAnalogIn, ZigBeeAnalogInConfig),
} }

5
homeassistant/components/zigbee/switch.py
View File

@ -3,7 +3,7 @@ import voluptuous as vol
from homeassistant.components.switch import SwitchDevice from homeassistant.components.switch import SwitchDevice
from . import PLATFORM_SCHEMA, ZigBeeDigitalOut, ZigBeeDigitalOutConfig from . import DOMAIN, PLATFORM_SCHEMA, ZigBeeDigitalOut, ZigBeeDigitalOutConfig
CONF_ON_STATE = "on_state" CONF_ON_STATE = "on_state"
@ -16,7 +16,8 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({vol.Optional(CONF_ON_STATE): vol.In(ST
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Zigbee switch platform.""" """Set up the Zigbee switch platform."""
add_entities([ZigBeeSwitch(hass, ZigBeeDigitalOutConfig(config))]) zigbee_device = hass.data[DOMAIN]
add_entities([ZigBeeSwitch(ZigBeeDigitalOutConfig(config), zigbee_device)])
class ZigBeeSwitch(ZigBeeDigitalOut, SwitchDevice): class ZigBeeSwitch(ZigBeeDigitalOut, SwitchDevice):