jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove 1-Wire SysBus (ADR-0019) (#71232)

Browse Source
This commit is contained in:
epenet 2022-05-09 13:16:23 +02:00 committed by GitHub
parent 30fdfc454f
commit 08856cfab0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 142 additions and 909 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
homeassistant/components/onewire/binary_sensor.py
View File

@ -3,7 +3,6 @@ from __future__ import annotations
from dataclasses import dataclass
import os
from typing import TYPE_CHECKING
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -11,21 +10,18 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
CONF_TYPE_OWSERVER,
DEVICE_KEYS_0_3,
DEVICE_KEYS_0_7,
DEVICE_KEYS_A_B,
DOMAIN,
READ_MODE_BOOL,
)
from .model import OWServerDeviceDescription
from .onewire_entities import OneWireEntityDescription, OneWireProxyEntity
from .onewire_entities import OneWireEntity, OneWireEntityDescription
from .onewirehub import OneWireHub
@ -98,23 +94,19 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up 1-Wire platform."""
# Only OWServer implementation works with binary sensors
if config_entry.data[CONF_TYPE] == CONF_TYPE_OWSERVER:
onewirehub = hass.data[DOMAIN][config_entry.entry_id]
onewirehub = hass.data[DOMAIN][config_entry.entry_id]
entities = await hass.async_add_executor_job(get_entities, onewirehub)
async_add_entities(entities, True)
entities = await hass.async_add_executor_job(get_entities, onewirehub)
async_add_entities(entities, True)
def get_entities(onewirehub: OneWireHub) -> list[BinarySensorEntity]:
def get_entities(onewirehub: OneWireHub) -> list[OneWireBinarySensor]:
"""Get a list of entities."""
if not onewirehub.devices:
return []
entities: list[BinarySensorEntity] = []
entities: list[OneWireBinarySensor] = []
for device in onewirehub.devices:
if TYPE_CHECKING:
assert isinstance(device, OWServerDeviceDescription)
family = device.family
device_id = device.id
device_type = device.type
@ -130,7 +122,7 @@ def get_entities(onewirehub: OneWireHub) -> list[BinarySensorEntity]:
device_file = os.path.join(os.path.split(device.path)[0], description.key)
name = f"{device_id} {description.name}"
entities.append(
OneWireProxyBinarySensor(
OneWireBinarySensor(
description=description,
device_id=device_id,
device_file=device_file,
@ -143,7 +135,7 @@ def get_entities(onewirehub: OneWireHub) -> list[BinarySensorEntity]:
return entities
class OneWireProxyBinarySensor(OneWireProxyEntity, BinarySensorEntity):
class OneWireBinarySensor(OneWireEntity, BinarySensorEntity):
"""Implementation of a 1-Wire binary sensor."""
entity_description: OneWireBinarySensorEntityDescription

117
homeassistant/components/onewire/config_flow.py
View File

@ -6,19 +6,15 @@ from typing import Any
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.device_registry import DeviceRegistry
from .const import (
CONF_MOUNT_DIR,
CONF_TYPE_OWSERVER,
CONF_TYPE_SYSBUS,
DEFAULT_OWSERVER_HOST,
DEFAULT_OWSERVER_PORT,
DEFAULT_SYSBUS_MOUNT_DIR,
DEFAULT_HOST,
DEFAULT_PORT,
DEVICE_SUPPORT_OPTIONS,
DOMAIN,
INPUT_ENTRY_CLEAR_OPTIONS,
@ -27,31 +23,21 @@ from .const import (
OPTION_ENTRY_SENSOR_PRECISION,
PRECISION_MAPPING_FAMILY_28,
)
from .model import OWServerDeviceDescription
from .onewirehub import CannotConnect, InvalidPath, OneWireHub
from .model import OWDeviceDescription
from .onewirehub import CannotConnect, OneWireHub
DATA_SCHEMA_USER = vol.Schema(
{vol.Required(CONF_TYPE): vol.In([CONF_TYPE_OWSERVER, CONF_TYPE_SYSBUS])}
)
DATA_SCHEMA_OWSERVER = vol.Schema(
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST, default=DEFAULT_OWSERVER_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_OWSERVER_PORT): int,
}
)
DATA_SCHEMA_MOUNTDIR = vol.Schema(
{
vol.Required(CONF_MOUNT_DIR, default=DEFAULT_SYSBUS_MOUNT_DIR): str,
vol.Required(CONF_HOST, default=DEFAULT_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
}
)
async def validate_input_owserver(
hass: HomeAssistant, data: dict[str, Any]
) -> dict[str, str]:
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, str]:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA_OWSERVER with values provided by the user.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
hub = OneWireHub(hass)
@ -65,24 +51,6 @@ async def validate_input_owserver(
return {"title": host}
async def validate_input_mount_dir(
hass: HomeAssistant, data: dict[str, Any]
) -> dict[str, str]:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA_MOUNTDIR with values provided by the user.
"""
hub = OneWireHub(hass)
mount_dir = data[CONF_MOUNT_DIR]
# Raises InvalidDir exception on failure
await hub.check_mount_dir(mount_dir)
# Return info that you want to store in the config entry.
return {"title": mount_dir}
class OneWireFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle 1-Wire config flow."""
@ -100,29 +68,10 @@ class OneWireFlowHandler(ConfigFlow, domain=DOMAIN):
Let user manually input configuration.
"""
errors: dict[str, str] = {}
if user_input is not None:
self.onewire_config.update(user_input)
if CONF_TYPE_OWSERVER == user_input[CONF_TYPE]:
return await self.async_step_owserver()
if CONF_TYPE_SYSBUS == user_input[CONF_TYPE]:
return await self.async_step_mount_dir()
return self.async_show_form(
step_id="user",
data_schema=DATA_SCHEMA_USER,
errors=errors,
)
async def async_step_owserver(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle OWServer configuration."""
errors = {}
if user_input:
# Prevent duplicate entries
self._async_abort_entries_match(
{
CONF_TYPE: CONF_TYPE_OWSERVER,
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
}
@ -131,7 +80,7 @@ class OneWireFlowHandler(ConfigFlow, domain=DOMAIN):
self.onewire_config.update(user_input)
try:
info = await validate_input_owserver(self.hass, user_input)
info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
else:
@ -140,37 +89,8 @@ class OneWireFlowHandler(ConfigFlow, domain=DOMAIN):
)
return self.async_show_form(
step_id="owserver",
data_schema=DATA_SCHEMA_OWSERVER,
errors=errors,
)
async def async_step_mount_dir(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle SysBus configuration."""
errors = {}
if user_input:
# Prevent duplicate entries
await self.async_set_unique_id(
f"{CONF_TYPE_SYSBUS}:{user_input[CONF_MOUNT_DIR]}"
)
self._abort_if_unique_id_configured()
self.onewire_config.update(user_input)
try:
info = await validate_input_mount_dir(self.hass, user_input)
except InvalidPath:
errors["base"] = "invalid_path"
else:
return self.async_create_entry(
title=info["title"], data=self.onewire_config
)
return self.async_show_form(
step_id="mount_dir",
data_schema=DATA_SCHEMA_MOUNTDIR,
step_id="user",
data_schema=DATA_SCHEMA,
errors=errors,
)
@ -188,8 +108,8 @@ class OnewireOptionsFlowHandler(OptionsFlow):
"""Initialize OneWire Network options flow."""
self.entry_id = config_entry.entry_id
self.options = dict(config_entry.options)
self.configurable_devices: dict[str, OWServerDeviceDescription] = {}
self.devices_to_configure: dict[str, OWServerDeviceDescription] = {}
self.configurable_devices: dict[str, OWDeviceDescription] = {}
self.devices_to_configure: dict[str, OWDeviceDescription] = {}
self.current_device: str = ""
async def async_step_init(
@ -197,12 +117,7 @@ class OnewireOptionsFlowHandler(OptionsFlow):
) -> FlowResult:
"""Manage the options."""
controller: OneWireHub = self.hass.data[DOMAIN][self.entry_id]
if controller.type == CONF_TYPE_SYSBUS:
return self.async_abort(
reason="SysBus setup does not have any config options."
)
all_devices: list[OWServerDeviceDescription] = controller.devices # type: ignore[assignment]
all_devices: list[OWDeviceDescription] = controller.devices # type: ignore[assignment]
if not all_devices:
return self.async_abort(reason="No configurable devices found.")

13
homeassistant/components/onewire/const.py
View File

@ -3,14 +3,8 @@ from __future__ import annotations
from homeassistant.const import Platform
CONF_MOUNT_DIR = "mount_dir"
CONF_TYPE_OWSERVER = "OWServer"
CONF_TYPE_SYSBUS = "SysBus"
DEFAULT_OWSERVER_HOST = "localhost"
DEFAULT_OWSERVER_PORT = 4304
DEFAULT_SYSBUS_MOUNT_DIR = "/sys/bus/w1/devices/"
DEFAULT_HOST = "localhost"
DEFAULT_PORT = 4304
DOMAIN = "onewire"
@ -18,7 +12,7 @@ DEVICE_KEYS_0_3 = range(4)
DEVICE_KEYS_0_7 = range(8)
DEVICE_KEYS_A_B = ("A", "B")
DEVICE_SUPPORT_OWSERVER = {
DEVICE_SUPPORT = {
"05": (),
"10": (),
"12": (),
@ -35,7 +29,6 @@ DEVICE_SUPPORT_OWSERVER = {
"7E": ("EDS0066", "EDS0068"),
"EF": ("HB_HUB", "HB_MOISTURE_METER", "HobbyBoards_EF"),
}
DEVICE_SUPPORT_SYSBUS = ["10", "22", "28", "3B", "42"]
DEVICE_SUPPORT_OPTIONS = ["28"]

4
homeassistant/components/onewire/manifest.json
View File

@ -3,8 +3,8 @@
"name": "1-Wire",
"documentation": "https://www.home-assistant.io/integrations/onewire",
"config_flow": true,
"requirements": ["pyownet==0.10.0.post1", "pi1wire==0.1.0"],
"requirements": ["pyownet==0.10.0.post1"],
"codeowners": ["@garbled1", "@epenet"],
"iot_class": "local_polling",
"loggers": ["pi1wire", "pyownet"]
"loggers": ["pyownet"]
}

16
homeassistant/components/onewire/model.py
View File

@ -3,29 +3,15 @@ from __future__ import annotations
from dataclasses import dataclass
from pi1wire import OneWireInterface
from homeassistant.helpers.entity import DeviceInfo
@dataclass
class OWDeviceDescription:
"""OWDeviceDescription device description class."""
"""1-Wire device description class."""
device_info: DeviceInfo
@dataclass
class OWDirectDeviceDescription(OWDeviceDescription):
"""SysBus device description class."""
interface: OneWireInterface
@dataclass
class OWServerDeviceDescription(OWDeviceDescription):
"""OWServer device description class."""
family: str
id: str
path: str

39
homeassistant/components/onewire/onewire_entities.py
View File

@ -23,7 +23,7 @@ class OneWireEntityDescription(EntityDescription):
_LOGGER = logging.getLogger(__name__)
class OneWireBaseEntity(Entity):
class OneWireEntity(Entity):
"""Implementation of a 1-Wire entity."""
entity_description: OneWireEntityDescription
@ -35,6 +35,7 @@ class OneWireBaseEntity(Entity):
device_info: DeviceInfo,
device_file: str,
name: str,
owproxy: protocol._Proxy,
) -> None:
"""Initialize the entity."""
self.entity_description = description
@ -44,6 +45,7 @@ class OneWireBaseEntity(Entity):
self._device_file = device_file
self._state: StateType = None
self._value_raw: float | None = None
self._owproxy = owproxy
@property
def extra_state_attributes(self) -> dict[str, Any] | None:
@ -53,44 +55,21 @@ class OneWireBaseEntity(Entity):
"raw_value": self._value_raw,
}
class OneWireProxyEntity(OneWireBaseEntity):
"""Implementation of a 1-Wire entity connected through owserver."""
def __init__(
self,
description: OneWireEntityDescription,
device_id: str,
device_info: DeviceInfo,
device_file: str,
name: str,
owproxy: protocol._Proxy,
) -> None:
"""Initialize the sensor."""
super().__init__(
description=description,
device_id=device_id,
device_info=device_info,
device_file=device_file,
name=name,
)
self._owproxy = owproxy
def _read_value_ownet(self) -> str:
"""Read a value from the owserver."""
def _read_value(self) -> str:
"""Read a value from the server."""
read_bytes: bytes = self._owproxy.read(self._device_file)
return read_bytes.decode().lstrip()
def _write_value_ownet(self, value: bytes) -> None:
"""Write a value to the owserver."""
def _write_value(self, value: bytes) -> None:
"""Write a value to the server."""
self._owproxy.write(self._device_file, value)
def update(self) -> None:
"""Get the latest data from the device."""
try:
self._value_raw = float(self._read_value_ownet())
self._value_raw = float(self._read_value())
except protocol.Error as exc:
_LOGGER.error("Owserver failure in read(), got: %s", exc)
_LOGGER.error("Failure to read server value, got: %s", exc)
self._state = None
else:
if self.entity_description.read_mode == READ_MODE_INT:

113
homeassistant/components/onewire/onewirehub.py
View File

@ -5,7 +5,6 @@ import logging
import os
from typing import TYPE_CHECKING
from pi1wire import Pi1Wire
from pyownet import protocol
from homeassistant.config_entries import ConfigEntry
@ -17,7 +16,6 @@ from homeassistant.const import (
ATTR_VIA_DEVICE,
CONF_HOST,
CONF_PORT,
CONF_TYPE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
@ -25,21 +23,13 @@ from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo
from .const import (
CONF_MOUNT_DIR,
CONF_TYPE_OWSERVER,
CONF_TYPE_SYSBUS,
DEVICE_SUPPORT_OWSERVER,
DEVICE_SUPPORT_SYSBUS,
DEVICE_SUPPORT,
DOMAIN,
MANUFACTURER_EDS,
MANUFACTURER_HOBBYBOARDS,
MANUFACTURER_MAXIM,
)
from .model import (
OWDeviceDescription,
OWDirectDeviceDescription,
OWServerDeviceDescription,
)
from .model import OWDeviceDescription
DEVICE_COUPLERS = {
# Family : [branches]
@ -54,26 +44,24 @@ DEVICE_MANUFACTURER = {
_LOGGER = logging.getLogger(__name__)
def _is_known_owserver_device(device_family: str, device_type: str) -> bool:
def _is_known_device(device_family: str, device_type: str) -> bool:
"""Check if device family/type is known to the library."""
if device_family in ("7E", "EF"): # EDS or HobbyBoard
return device_type in DEVICE_SUPPORT_OWSERVER[device_family]
return device_family in DEVICE_SUPPORT_OWSERVER
return device_type in DEVICE_SUPPORT[device_family]
return device_family in DEVICE_SUPPORT
class OneWireHub:
"""Hub to communicate with SysBus or OWServer."""
"""Hub to communicate with server."""
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize."""
self.hass = hass
self.type: str | None = None
self.pi1proxy: Pi1Wire | None = None
self.owproxy: protocol._Proxy | None = None
self.devices: list[OWDeviceDescription] | None = None
async def connect(self, host: str, port: int) -> None:
"""Connect to the owserver host."""
"""Connect to the server."""
try:
self.owproxy = await self.hass.async_add_executor_job(
protocol.proxy, host, port
@ -81,32 +69,12 @@ class OneWireHub:
except protocol.ConnError as exc:
raise CannotConnect from exc
async def check_mount_dir(self, mount_dir: str) -> None:
"""Test that the mount_dir is a valid path."""
if not await self.hass.async_add_executor_job(os.path.isdir, mount_dir):
raise InvalidPath
self.pi1proxy = Pi1Wire(mount_dir)
async def initialize(self, config_entry: ConfigEntry) -> None:
"""Initialize a config entry."""
self.type = config_entry.data[CONF_TYPE]
if self.type == CONF_TYPE_SYSBUS:
mount_dir = config_entry.data[CONF_MOUNT_DIR]
_LOGGER.debug("Initializing using SysBus %s", mount_dir)
_LOGGER.warning(
"Using the 1-Wire integration via SysBus is deprecated and will be removed "
"in Home Assistant Core 2022.6; this integration is being adjusted to comply "
"with Architectural Decision Record 0019, more information can be found here: "
"https://github.com/home-assistant/architecture/blob/master/adr/0019-GPIO.md "
"Access via OWServer is still supported"
)
await self.check_mount_dir(mount_dir)
elif self.type == CONF_TYPE_OWSERVER:
host = config_entry.data[CONF_HOST]
port = config_entry.data[CONF_PORT]
_LOGGER.debug("Initializing using OWServer %s:%s", host, port)
await self.connect(host, port)
host = config_entry.data[CONF_HOST]
port = config_entry.data[CONF_PORT]
_LOGGER.debug("Initializing connection to %s:%s", host, port)
await self.connect(host, port)
await self.discover_devices()
if TYPE_CHECKING:
assert self.devices
@ -126,63 +94,22 @@ class OneWireHub:
async def discover_devices(self) -> None:
"""Discover all devices."""
if self.devices is None:
if self.type == CONF_TYPE_SYSBUS:
self.devices = await self.hass.async_add_executor_job(
self._discover_devices_sysbus
)
if self.type == CONF_TYPE_OWSERVER:
self.devices = await self.hass.async_add_executor_job(
self._discover_devices_owserver
)
def _discover_devices_sysbus(self) -> list[OWDeviceDescription]:
"""Discover all sysbus devices."""
devices: list[OWDeviceDescription] = []
assert self.pi1proxy
all_sensors = self.pi1proxy.find_all_sensors()
if not all_sensors:
_LOGGER.error(
"No onewire sensor found. Check if dtoverlay=w1-gpio "
"is in your /boot/config.txt. "
"Check the mount_dir parameter if it's defined"
self.devices = await self.hass.async_add_executor_job(
self._discover_devices
)
for interface in all_sensors:
device_family = interface.mac_address[:2]
device_id = f"{device_family}-{interface.mac_address[2:]}"
if device_family not in DEVICE_SUPPORT_SYSBUS:
_LOGGER.warning(
"Ignoring unknown device family (%s) found for device %s",
device_family,
device_id,
)
continue
device_info: DeviceInfo = {
ATTR_IDENTIFIERS: {(DOMAIN, device_id)},
ATTR_MANUFACTURER: DEVICE_MANUFACTURER.get(
device_family, MANUFACTURER_MAXIM
),
ATTR_MODEL: device_family,
ATTR_NAME: device_id,
}
device = OWDirectDeviceDescription(
device_info=device_info,
interface=interface,
)
devices.append(device)
return devices
def _discover_devices_owserver(
def _discover_devices(
self, path: str = "/", parent_id: str | None = None
) -> list[OWDeviceDescription]:
"""Discover all owserver devices."""
"""Discover all server devices."""
devices: list[OWDeviceDescription] = []
assert self.owproxy
for device_path in self.owproxy.dir(path):
device_id = os.path.split(os.path.split(device_path)[0])[1]
device_family = self.owproxy.read(f"{device_path}family").decode()
_LOGGER.debug("read `%sfamily`: %s", device_path, device_family)
device_type = self._get_device_type_owserver(device_path)
if not _is_known_owserver_device(device_family, device_type):
device_type = self._get_device_type(device_path)
if not _is_known_device(device_family, device_type):
_LOGGER.warning(
"Ignoring unknown device family/type (%s/%s) found for device %s",
device_family,
@ -200,7 +127,7 @@ class OneWireHub:
}
if parent_id:
device_info[ATTR_VIA_DEVICE] = (DOMAIN, parent_id)
device = OWServerDeviceDescription(
device = OWDeviceDescription(
device_info=device_info,
id=device_id,
family=device_family,
@ -210,13 +137,13 @@ class OneWireHub:
devices.append(device)
if device_branches := DEVICE_COUPLERS.get(device_family):
for branch in device_branches:
devices += self._discover_devices_owserver(
devices += self._discover_devices(
f"{device_path}{branch}", device_id
)
return devices
def _get_device_type_owserver(self, device_path: str) -> str:
def _get_device_type(self, device_path: str) -> str:
"""Get device model."""
if TYPE_CHECKING:
assert self.owproxy

207
homeassistant/components/onewire/sensor.py
View File

@ -1,16 +1,13 @@
"""Support for 1-Wire environment sensors."""
from __future__ import annotations
import asyncio
from collections.abc import Callable, Mapping
import copy
from dataclasses import dataclass
import logging
import os
from types import MappingProxyType
from typing import TYPE_CHECKING, Any
from pi1wire import InvalidCRCException, OneWireInterface, UnsupportResponseException
from typing import Any
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -20,7 +17,6 @@ from homeassistant.components.sensor import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_TYPE,
ELECTRIC_POTENTIAL_VOLT,
LIGHT_LUX,
PERCENTAGE,
@ -29,13 +25,10 @@ from homeassistant.const import (
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import (
CONF_TYPE_OWSERVER,
CONF_TYPE_SYSBUS,
DEVICE_KEYS_0_3,
DEVICE_KEYS_A_B,
DOMAIN,
@ -45,12 +38,7 @@ from .const import (
READ_MODE_FLOAT,
READ_MODE_INT,
)
from .model import OWDirectDeviceDescription, OWServerDeviceDescription
from .onewire_entities import (
OneWireBaseEntity,
OneWireEntityDescription,
OneWireProxyEntity,
)
from .onewire_entities import OneWireEntity, OneWireEntityDescription
from .onewirehub import OneWireHub
@ -263,8 +251,6 @@ DEVICE_SENSORS: dict[str, tuple[OneWireSensorEntityDescription, ...]] = {
}
# EF sensors are usually hobbyboards specialized sensors.
# These can only be read by OWFS. Currently this driver only supports them
# via owserver (network protocol)
HOBBYBOARD_EF: dict[str, tuple[OneWireSensorEntityDescription, ...]] = {
"HobbyBoards_EF": (
@ -383,109 +369,72 @@ async def async_setup_entry(
"""Set up 1-Wire platform."""
onewirehub = hass.data[DOMAIN][config_entry.entry_id]
entities = await hass.async_add_executor_job(
get_entities, onewirehub, config_entry.data, config_entry.options
get_entities, onewirehub, config_entry.options
)
async_add_entities(entities, True)
def get_entities(
onewirehub: OneWireHub,
config: MappingProxyType[str, Any],
options: MappingProxyType[str, Any],
) -> list[SensorEntity]:
onewirehub: OneWireHub, options: MappingProxyType[str, Any]
) -> list[OneWireSensor]:
"""Get a list of entities."""
if not onewirehub.devices:
return []
entities: list[SensorEntity] = []
conf_type = config[CONF_TYPE]
# We have an owserver on a remote(or local) host/port
if conf_type == CONF_TYPE_OWSERVER:
assert onewirehub.owproxy
for device in onewirehub.devices:
if TYPE_CHECKING:
assert isinstance(device, OWServerDeviceDescription)
family = device.family
device_type = device.type
device_id = device.id
device_info = device.device_info
device_sub_type = "std"
device_path = device.path
if "EF" in family:
device_sub_type = "HobbyBoard"
family = device_type
elif "7E" in family:
device_sub_type = "EDS"
family = device_type
entities: list[OneWireSensor] = []
assert onewirehub.owproxy
for device in onewirehub.devices:
family = device.family
device_type = device.type
device_id = device.id
device_info = device.device_info
device_sub_type = "std"
device_path = device.path
if "EF" in family:
device_sub_type = "HobbyBoard"
family = device_type
elif "7E" in family:
device_sub_type = "EDS"
family = device_type
if family not in get_sensor_types(device_sub_type):
continue
for description in get_sensor_types(device_sub_type)[family]:
if description.key.startswith("moisture/"):
s_id = description.key.split(".")[1]
is_leaf = int(
onewirehub.owproxy.read(
f"{device_path}moisture/is_leaf.{s_id}"
).decode()
)
if is_leaf:
description = copy.deepcopy(description)
description.device_class = SensorDeviceClass.HUMIDITY
description.native_unit_of_measurement = PERCENTAGE
description.name = f"Wetness {s_id}"
override_key = None
if description.override_key:
override_key = description.override_key(device_id, options)
device_file = os.path.join(
os.path.split(device.path)[0],
override_key or description.key,
if family not in get_sensor_types(device_sub_type):
continue
for description in get_sensor_types(device_sub_type)[family]:
if description.key.startswith("moisture/"):
s_id = description.key.split(".")[1]
is_leaf = int(
onewirehub.owproxy.read(
f"{device_path}moisture/is_leaf.{s_id}"
).decode()
)
name = f"{device_id} {description.name}"
entities.append(
OneWireProxySensor(
description=description,
device_id=device_id,
device_file=device_file,
device_info=device_info,
name=name,
owproxy=onewirehub.owproxy,
)
)
# We have a raw GPIO ow sensor on a Pi
elif conf_type == CONF_TYPE_SYSBUS:
for device in onewirehub.devices:
if TYPE_CHECKING:
assert isinstance(device, OWDirectDeviceDescription)
p1sensor: OneWireInterface = device.interface
family = p1sensor.mac_address[:2]
device_id = f"{family}-{p1sensor.mac_address[2:]}"
device_info = device.device_info
description = SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION
device_file = f"/sys/bus/w1/devices/{device_id}/w1_slave"
if is_leaf:
description = copy.deepcopy(description)
description.device_class = SensorDeviceClass.HUMIDITY
description.native_unit_of_measurement = PERCENTAGE
description.name = f"Wetness {s_id}"
override_key = None
if description.override_key:
override_key = description.override_key(device_id, options)
device_file = os.path.join(
os.path.split(device.path)[0],
override_key or description.key,
)
name = f"{device_id} {description.name}"
entities.append(
OneWireDirectSensor(
OneWireSensor(
description=description,
device_id=device_id,
device_file=device_file,
device_info=device_info,
name=name,
owsensor=p1sensor,
owproxy=onewirehub.owproxy,
)
)
return entities
class OneWireSensor(OneWireBaseEntity, SensorEntity):
"""Mixin for sensor specific attributes."""
entity_description: OneWireSensorEntityDescription
class OneWireProxySensor(OneWireProxyEntity, OneWireSensor):
"""Implementation of a 1-Wire sensor connected through owserver."""
class OneWireSensor(OneWireEntity, SensorEntity):
"""Implementation of a 1-Wire sensor."""
entity_description: OneWireSensorEntityDescription
@ -493,69 +442,3 @@ class OneWireProxySensor(OneWireProxyEntity, OneWireSensor):
def native_value(self) -> StateType:
"""Return the state of the entity."""
return self._state
class OneWireDirectSensor(OneWireSensor):
"""Implementation of a 1-Wire sensor directly connected to RPI GPIO."""
def __init__(
self,
description: OneWireSensorEntityDescription,
device_id: str,
device_info: DeviceInfo,
device_file: str,
name: str,
owsensor: OneWireInterface,
) -> None:
"""Initialize the sensor."""
super().__init__(
description=description,
device_id=device_id,
device_info=device_info,
device_file=device_file,
name=name,
)
self._attr_unique_id = device_file
self._owsensor = owsensor
@property
def native_value(self) -> StateType:
"""Return the state of the entity."""
return self._state
async def get_temperature(self) -> float:
"""Get the latest data from the device."""
attempts = 1
while True:
try:
return await self.hass.async_add_executor_job(
self._owsensor.get_temperature
)
except UnsupportResponseException as ex:
_LOGGER.debug(
"Cannot read from sensor %s (retry attempt %s): %s",
self._device_file,
attempts,
ex,
)
await asyncio.sleep(0.2)
attempts += 1
if attempts > 10:
raise
async def async_update(self) -> None:
"""Get the latest data from the device."""
try:
self._value_raw = await self.get_temperature()
self._state = round(self._value_raw, 1)
except (
FileNotFoundError,
InvalidCRCException,
UnsupportResponseException,
) as ex:
_LOGGER.warning(
"Cannot read from sensor %s: %s",
self._device_file,
ex,
)
self._state = None

18
homeassistant/components/onewire/strings.json
View File

@ -4,22 +4,15 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_path": "Directory not found."
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"step": {
"owserver": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]"
},
"title": "Set owserver details"
},
"user": {
"data": {
"type": "Connection type"
},
"title": "Set up 1-Wire"
"title": "Set server details"
}
}
},
@ -28,11 +21,6 @@
"device_not_selected": "Select devices to configure"
},
"step": {
"ack_no_options": {
"data": {},
"description": "There are no options for the SysBus implementation",
"title": "OneWire SysBus Options"
},
"device_selection": {
"data": {
"clear_device_options": "Clear all device configurations",

29
homeassistant/components/onewire/switch.py
View File

@ -3,25 +3,22 @@ from __future__ import annotations
from dataclasses import dataclass
import os
from typing import TYPE_CHECKING, Any
from typing import Any
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
CONF_TYPE_OWSERVER,
DEVICE_KEYS_0_3,
DEVICE_KEYS_0_7,
DEVICE_KEYS_A_B,
DOMAIN,
READ_MODE_BOOL,
)
from .model import OWServerDeviceDescription
from .onewire_entities import OneWireEntityDescription, OneWireProxyEntity
from .onewire_entities import OneWireEntity, OneWireEntityDescription
from .onewirehub import OneWireHub
@ -153,24 +150,20 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up 1-Wire platform."""
# Only OWServer implementation works with switches
if config_entry.data[CONF_TYPE] == CONF_TYPE_OWSERVER:
onewirehub = hass.data[DOMAIN][config_entry.entry_id]
onewirehub = hass.data[DOMAIN][config_entry.entry_id]
entities = await hass.async_add_executor_job(get_entities, onewirehub)
async_add_entities(entities, True)
entities = await hass.async_add_executor_job(get_entities, onewirehub)
async_add_entities(entities, True)
def get_entities(onewirehub: OneWireHub) -> list[SwitchEntity]:
def get_entities(onewirehub: OneWireHub) -> list[OneWireSwitch]:
"""Get a list of entities."""
if not onewirehub.devices:
return []
entities: list[SwitchEntity] = []
entities: list[OneWireSwitch] = []
for device in onewirehub.devices:
if TYPE_CHECKING:
assert isinstance(device, OWServerDeviceDescription)
family = device.family
device_type = device.type
device_id = device.id
@ -186,7 +179,7 @@ def get_entities(onewirehub: OneWireHub) -> list[SwitchEntity]:
device_file = os.path.join(os.path.split(device.path)[0], description.key)
name = f"{device_id} {description.name}"
entities.append(
OneWireProxySwitch(
OneWireSwitch(
description=description,
device_id=device_id,
device_file=device_file,
@ -199,7 +192,7 @@ def get_entities(onewirehub: OneWireHub) -> list[SwitchEntity]:
return entities
class OneWireProxySwitch(OneWireProxyEntity, SwitchEntity):
class OneWireSwitch(OneWireEntity, SwitchEntity):
"""Implementation of a 1-Wire switch."""
entity_description: OneWireSwitchEntityDescription
@ -211,8 +204,8 @@ class OneWireProxySwitch(OneWireProxyEntity, SwitchEntity):
def turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
self._write_value_ownet(b"1")
self._write_value(b"1")
def turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
self._write_value_ownet(b"0")
self._write_value(b"0")

3
requirements_all.txt
View File

@ -1210,9 +1210,6 @@ pexpect==4.6.0
# homeassistant.components.modem_callerid
phone_modem==0.1.1
# homeassistant.components.onewire
pi1wire==0.1.0
# homeassistant.components.remote_rpi_gpio
pigpio==1.78

3
requirements_test_all.txt
View File

@ -812,9 +812,6 @@ pexpect==4.6.0
# homeassistant.components.modem_callerid
phone_modem==0.1.1
# homeassistant.components.onewire
pi1wire==0.1.0
# homeassistant.components.pilight
pilight==0.1.1

29
tests/components/onewire/__init__.py
View File

@ -7,7 +7,6 @@ from unittest.mock import MagicMock
from pyownet.protocol import ProtocolError
from homeassistant.components.onewire.const import DEFAULT_SYSBUS_MOUNT_DIR
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_IDENTIFIERS,
@ -29,7 +28,6 @@ from .const import (
ATTR_UNIQUE_ID,
FIXED_ATTRIBUTES,
MOCK_OWPROXY_DEVICES,
MOCK_SYSBUS_DEVICES,
)
@ -181,30 +179,3 @@ def _setup_owproxy_mock_device_reads(
device_sensors = mock_device.get(platform, [])
for expected_sensor in device_sensors:
sub_read_side_effect.append(expected_sensor[ATTR_INJECT_READS])
def setup_sysbus_mock_devices(
platform: str, device_ids: list[str]
) -> tuple[list[str], list[Any]]:
"""Set up mock for sysbus."""
glob_result = []
read_side_effect = []
for device_id in device_ids:
mock_device = MOCK_SYSBUS_DEVICES[device_id]
# Setup directory listing
glob_result += [f"/{DEFAULT_SYSBUS_MOUNT_DIR}/{device_id}"]
# Setup sub-device reads
device_sensors = mock_device.get(platform, [])
for expected_sensor in device_sensors:
if isinstance(expected_sensor[ATTR_INJECT_READS], list):
read_side_effect += expected_sensor[ATTR_INJECT_READS]
else:
read_side_effect.append(expected_sensor[ATTR_INJECT_READS])
# Ensure enough read side effect
read_side_effect.extend([FileNotFoundError("Missing injected value")] * 20)
return (glob_result, read_side_effect)

38
tests/components/onewire/conftest.py
View File

@ -4,15 +4,9 @@ from unittest.mock import MagicMock, patch
from pyownet.protocol import ConnError
import pytest
from homeassistant.components.onewire.const import (
CONF_MOUNT_DIR,
CONF_TYPE_OWSERVER,
CONF_TYPE_SYSBUS,
DEFAULT_SYSBUS_MOUNT_DIR,
DOMAIN,
)
from homeassistant.components.onewire.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
from .const import MOCK_OWPROXY_DEVICES
@ -33,7 +27,6 @@ def get_config_entry(hass: HomeAssistant) -> ConfigEntry:
domain=DOMAIN,
source=SOURCE_USER,
data={
CONF_TYPE: CONF_TYPE_OWSERVER,
CONF_HOST: "1.2.3.4",
CONF_PORT: 1234,
},
@ -49,24 +42,6 @@ def get_config_entry(hass: HomeAssistant) -> ConfigEntry:
return config_entry
@pytest.fixture(name="sysbus_config_entry")
def get_sysbus_config_entry(hass: HomeAssistant) -> ConfigEntry:
"""Create and register mock config entry."""
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
data={
CONF_TYPE: CONF_TYPE_SYSBUS,
CONF_MOUNT_DIR: DEFAULT_SYSBUS_MOUNT_DIR,
},
unique_id=f"{CONF_TYPE_SYSBUS}:{DEFAULT_SYSBUS_MOUNT_DIR}",
options={},
entry_id="3",
)
config_entry.add_to_hass(hass)
return config_entry
@pytest.fixture(name="owproxy")
def get_owproxy() -> MagicMock:
"""Mock owproxy."""
@ -82,12 +57,3 @@ def get_owproxy_with_connerror() -> MagicMock:
side_effect=ConnError,
) as owproxy:
yield owproxy
@pytest.fixture(name="sysbus")
def get_sysbus() -> MagicMock:
"""Mock sysbus."""
with patch(
"homeassistant.components.onewire.onewirehub.os.path.isdir", return_value=True
):
yield

140
tests/components/onewire/const.py
View File

@ -1,5 +1,4 @@
"""Constants for 1-Wire integration."""
from pi1wire import InvalidCRCException, UnsupportResponseException
from pyownet.protocol import Error as ProtocolError
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
@ -1131,142 +1130,3 @@ MOCK_OWPROXY_DEVICES = {
],
},
}
MOCK_SYSBUS_DEVICES = {
"00-111111111111": {
ATTR_UNKNOWN_DEVICE: True,
},
"10-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "10-111111111111")},
ATTR_MANUFACTURER: MANUFACTURER_MAXIM,
ATTR_MODEL: "10",
ATTR_NAME: "10-111111111111",
},
Platform.SENSOR: [
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ENTITY_ID: "sensor.10_111111111111_temperature",
ATTR_INJECT_READS: 25.123,
ATTR_STATE: "25.1",
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
ATTR_UNIQUE_ID: "/sys/bus/w1/devices/10-111111111111/w1_slave",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
},
],
},
"22-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "22-111111111111")},
ATTR_MANUFACTURER: MANUFACTURER_MAXIM,
ATTR_MODEL: "22",
ATTR_NAME: "22-111111111111",
},
Platform.SENSOR: [
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ENTITY_ID: "sensor.22_111111111111_temperature",
ATTR_INJECT_READS: FileNotFoundError,
ATTR_STATE: STATE_UNKNOWN,
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
ATTR_UNIQUE_ID: "/sys/bus/w1/devices/22-111111111111/w1_slave",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
},
],
},
"28-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "28-111111111111")},
ATTR_MANUFACTURER: MANUFACTURER_MAXIM,
ATTR_MODEL: "28",
ATTR_NAME: "28-111111111111",
},
Platform.SENSOR: [
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ENTITY_ID: "sensor.28_111111111111_temperature",
ATTR_INJECT_READS: InvalidCRCException,
ATTR_STATE: STATE_UNKNOWN,
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
ATTR_UNIQUE_ID: "/sys/bus/w1/devices/28-111111111111/w1_slave",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
},
],
},
"3B-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "3B-111111111111")},
ATTR_MANUFACTURER: MANUFACTURER_MAXIM,
ATTR_MODEL: "3B",
ATTR_NAME: "3B-111111111111",
},
Platform.SENSOR: [
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ENTITY_ID: "sensor.3b_111111111111_temperature",
ATTR_INJECT_READS: 29.993,
ATTR_STATE: "30.0",
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
ATTR_UNIQUE_ID: "/sys/bus/w1/devices/3B-111111111111/w1_slave",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
},
],
},
"42-111111111111": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "42-111111111111")},
ATTR_MANUFACTURER: MANUFACTURER_MAXIM,
ATTR_MODEL: "42",
ATTR_NAME: "42-111111111111",
},
Platform.SENSOR: [
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ENTITY_ID: "sensor.42_111111111111_temperature",
ATTR_INJECT_READS: UnsupportResponseException,
ATTR_STATE: STATE_UNKNOWN,
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
ATTR_UNIQUE_ID: "/sys/bus/w1/devices/42-111111111111/w1_slave",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
},
],
},
"42-111111111112": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "42-111111111112")},
ATTR_MANUFACTURER: MANUFACTURER_MAXIM,
ATTR_MODEL: "42",
ATTR_NAME: "42-111111111112",
},
Platform.SENSOR: [
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ENTITY_ID: "sensor.42_111111111112_temperature",
ATTR_INJECT_READS: [UnsupportResponseException] * 9 + [27.993],
ATTR_STATE: "28.0",
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
ATTR_UNIQUE_ID: "/sys/bus/w1/devices/42-111111111112/w1_slave",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
},
],
},
"42-111111111113": {
ATTR_DEVICE_INFO: {
ATTR_IDENTIFIERS: {(DOMAIN, "42-111111111113")},
ATTR_MANUFACTURER: MANUFACTURER_MAXIM,
ATTR_MODEL: "42",
ATTR_NAME: "42-111111111113",
},
Platform.SENSOR: [
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ENTITY_ID: "sensor.42_111111111113_temperature",
ATTR_INJECT_READS: [UnsupportResponseException] * 10 + [27.993],
ATTR_STATE: STATE_UNKNOWN,
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
ATTR_UNIQUE_ID: "/sys/bus/w1/devices/42-111111111113/w1_slave",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
},
],
},
}

4
tests/components/onewire/test_binary_sensor.py
View File

@ -1,4 +1,4 @@
"""Tests for 1-Wire devices connected on OWServer."""
"""Tests for 1-Wire binary sensors."""
import logging
from unittest.mock import MagicMock, patch
@ -27,7 +27,7 @@ def override_platforms():
yield
async def test_owserver_binary_sensor(
async def test_binary_sensors(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,

130
tests/components/onewire/test_config_flow.py
View File

@ -4,15 +4,9 @@ from unittest.mock import AsyncMock, patch
from pyownet import protocol
import pytest
from homeassistant.components.onewire.const import (
CONF_MOUNT_DIR,
CONF_TYPE_OWSERVER,
CONF_TYPE_SYSBUS,
DEFAULT_SYSBUS_MOUNT_DIR,
DOMAIN,
)
from homeassistant.components.onewire.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
@ -30,23 +24,14 @@ def override_async_setup_entry() -> AsyncMock:
yield mock_setup_entry
async def test_user_owserver(hass: HomeAssistant, mock_setup_entry: AsyncMock):
"""Test OWServer user flow."""
async def test_user_flow(hass: HomeAssistant, mock_setup_entry: AsyncMock):
"""Test user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_TYPE: CONF_TYPE_OWSERVER},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "owserver"
assert not result["errors"]
# Invalid server
with patch(
"homeassistant.components.onewire.onewirehub.protocol.proxy",
@ -58,7 +43,7 @@ async def test_user_owserver(hass: HomeAssistant, mock_setup_entry: AsyncMock):
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "owserver"
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
# Valid server
@ -73,7 +58,6 @@ async def test_user_owserver(hass: HomeAssistant, mock_setup_entry: AsyncMock):
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "1.2.3.4"
assert result["data"] == {
CONF_TYPE: CONF_TYPE_OWSERVER,
CONF_HOST: "1.2.3.4",
CONF_PORT: 1234,
}
@ -81,10 +65,10 @@ async def test_user_owserver(hass: HomeAssistant, mock_setup_entry: AsyncMock):
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_owserver_duplicate(
async def test_user_duplicate(
hass: HomeAssistant, config_entry: ConfigEntry, mock_setup_entry: AsyncMock
):
"""Test OWServer flow."""
"""Test user duplicate flow."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
@ -93,15 +77,7 @@ async def test_user_owserver_duplicate(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_TYPE: CONF_TYPE_OWSERVER},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "owserver"
assert result["step_id"] == "user"
assert not result["errors"]
# Duplicate server
@ -113,93 +89,3 @@ async def test_user_owserver_duplicate(
assert result["reason"] == "already_configured"
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_sysbus(hass: HomeAssistant, mock_setup_entry: AsyncMock):
"""Test SysBus flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_TYPE: CONF_TYPE_SYSBUS},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "mount_dir"
assert not result["errors"]
# Invalid path
with patch(
"homeassistant.components.onewire.onewirehub.os.path.isdir",
return_value=False,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_MOUNT_DIR: "/sys/bus/invalid_directory"},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "mount_dir"
assert result["errors"] == {"base": "invalid_path"}
# Valid path
with patch(
"homeassistant.components.onewire.onewirehub.os.path.isdir",
return_value=True,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_MOUNT_DIR: "/sys/bus/directory"},
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "/sys/bus/directory"
assert result["data"] == {
CONF_TYPE: CONF_TYPE_SYSBUS,
CONF_MOUNT_DIR: "/sys/bus/directory",
}
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_sysbus_duplicate(
hass: HomeAssistant, sysbus_config_entry: ConfigEntry, mock_setup_entry: AsyncMock
):
"""Test SysBus duplicate flow."""
await hass.config_entries.async_setup(sysbus_config_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_TYPE: CONF_TYPE_SYSBUS},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "mount_dir"
assert not result["errors"]
# Valid path
with patch(
"homeassistant.components.onewire.onewirehub.os.path.isdir",
return_value=True,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_MOUNT_DIR: DEFAULT_SYSBUS_MOUNT_DIR},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1

1
tests/components/onewire/test_diagnostics.py
View File

@ -52,7 +52,6 @@ async def test_entry_diagnostics(
"data": {
"host": REDACTED,
"port": 1234,
"type": "OWServer",
},
"options": {
"device_options": {

36
tests/components/onewire/test_init.py
View File

@ -1,5 +1,4 @@
"""Tests for 1-Wire config flow."""
import logging
from unittest.mock import MagicMock
from pyownet import protocol
@ -11,7 +10,7 @@ from homeassistant.core import HomeAssistant
@pytest.mark.usefixtures("owproxy_with_connerror")
async def test_owserver_connect_failure(hass: HomeAssistant, config_entry: ConfigEntry):
async def test_connect_failure(hass: HomeAssistant, config_entry: ConfigEntry):
"""Test connection failure raises ConfigEntryNotReady."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@ -21,7 +20,7 @@ async def test_owserver_connect_failure(hass: HomeAssistant, config_entry: Confi
assert not hass.data.get(DOMAIN)
async def test_owserver_listing_failure(
async def test_listing_failure(
hass: HomeAssistant, config_entry: ConfigEntry, owproxy: MagicMock
):
"""Test listing failure raises ConfigEntryNotReady."""
@ -49,34 +48,3 @@ async def test_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
assert config_entry.state is ConfigEntryState.NOT_LOADED
assert not hass.data.get(DOMAIN)
@pytest.mark.usefixtures("sysbus")
async def test_warning_no_devices(
hass: HomeAssistant,
sysbus_config_entry: ConfigEntry,
caplog: pytest.LogCaptureFixture,
):
"""Test warning is generated when no sysbus devices found."""
with caplog.at_level(logging.WARNING, logger="homeassistant.components.onewire"):
await hass.config_entries.async_setup(sysbus_config_entry.entry_id)
await hass.async_block_till_done()
assert "No onewire sensor found. Check if dtoverlay=w1-gpio" in caplog.text
@pytest.mark.usefixtures("sysbus")
async def test_unload_sysbus_entry(
hass: HomeAssistant, sysbus_config_entry: ConfigEntry
):
"""Test being able to unload an entry."""
await hass.config_entries.async_setup(sysbus_config_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert sysbus_config_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(sysbus_config_entry.entry_id)
await hass.async_block_till_done()
assert sysbus_config_entry.state is ConfigEntryState.NOT_LOADED
assert not hass.data.get(DOMAIN)

30
tests/components/onewire/test_options_flow.py
View File

@ -2,8 +2,6 @@
from unittest.mock import MagicMock, patch
from homeassistant.components.onewire.const import (
CONF_TYPE_SYSBUS,
DOMAIN,
INPUT_ENTRY_CLEAR_OPTIONS,
INPUT_ENTRY_DEVICE_SELECTION,
)
@ -26,13 +24,7 @@ class FakeDevice:
name_by_user = "Given Name"
class FakeOWHubSysBus:
"""Mock Class for mocking onewire hub."""
type = CONF_TYPE_SYSBUS
async def test_user_owserver_options_clear(
async def test_user_options_clear(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
@ -61,7 +53,7 @@ async def test_user_owserver_options_clear(
assert result["data"] == {}
async def test_user_owserver_options_empty_selection(
async def test_user_options_empty_selection(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
@ -91,7 +83,7 @@ async def test_user_owserver_options_empty_selection(
assert result["errors"] == {"base": "device_not_selected"}
async def test_user_owserver_options_set_single(
async def test_user_options_set_single(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
@ -134,7 +126,7 @@ async def test_user_owserver_options_set_single(
)
async def test_user_owserver_options_set_multiple(
async def test_user_options_set_multiple(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
@ -208,7 +200,7 @@ async def test_user_owserver_options_set_multiple(
)
async def test_user_owserver_options_no_devices(
async def test_user_options_no_devices(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
@ -223,15 +215,3 @@ async def test_user_owserver_options_no_devices(
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "No configurable devices found."
async def test_user_sysbus_options(
hass: HomeAssistant,
config_entry: ConfigEntry,
):
"""Test that SysBus options flow aborts on init."""
hass.data[DOMAIN] = {config_entry.entry_id: FakeOWHubSysBus()}
result = await hass.config_entries.options.async_init(config_entry.entry_id)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "SysBus setup does not have any config options."

53
tests/components/onewire/test_sensor.py
View File

@ -1,4 +1,4 @@
"""Tests for 1-Wire sensor platform."""
"""Tests for 1-Wire sensors."""
import logging
from unittest.mock import MagicMock, patch
@ -14,14 +14,8 @@ from . import (
check_device_registry,
check_entities,
setup_owproxy_mock_devices,
setup_sysbus_mock_devices,
)
from .const import (
ATTR_DEVICE_INFO,
ATTR_UNKNOWN_DEVICE,
MOCK_OWPROXY_DEVICES,
MOCK_SYSBUS_DEVICES,
)
from .const import ATTR_DEVICE_INFO, ATTR_UNKNOWN_DEVICE, MOCK_OWPROXY_DEVICES
from tests.common import mock_device_registry, mock_registry
@ -33,7 +27,7 @@ def override_platforms():
yield
async def test_owserver_sensor(
async def test_sensors(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,
@ -73,44 +67,3 @@ async def test_owserver_sensor(
await hass.async_block_till_done()
check_entities(hass, entity_registry, expected_entities)
@pytest.mark.usefixtures("sysbus")
@pytest.mark.parametrize("device_id", MOCK_SYSBUS_DEVICES.keys(), indirect=True)
async def test_onewiredirect_setup_valid_device(
hass: HomeAssistant,
sysbus_config_entry: ConfigEntry,
device_id: str,
caplog: pytest.LogCaptureFixture,
):
"""Test that sysbus config entry works correctly."""
device_registry = mock_device_registry(hass)
entity_registry = mock_registry(hass)
glob_result, read_side_effect = setup_sysbus_mock_devices(
Platform.SENSOR, [device_id]
)
mock_device = MOCK_SYSBUS_DEVICES[device_id]
expected_entities = mock_device.get(Platform.SENSOR, [])
expected_devices = ensure_list(mock_device.get(ATTR_DEVICE_INFO))
with patch("pi1wire._finder.glob.glob", return_value=glob_result,), patch(
"pi1wire.OneWire.get_temperature",
side_effect=read_side_effect,
), caplog.at_level(
logging.WARNING, logger="homeassistant.components.onewire"
), patch(
"homeassistant.components.onewire.sensor.asyncio.sleep"
):
await hass.config_entries.async_setup(sysbus_config_entry.entry_id)
await hass.async_block_till_done()
assert "No onewire sensor found. Check if dtoverlay=w1-gpio" not in caplog.text
if mock_device.get(ATTR_UNKNOWN_DEVICE):
assert "Ignoring unknown device family" in caplog.text
else:
assert "Ignoring unknown device family" not in caplog.text
check_device_registry(device_registry, expected_devices)
assert len(entity_registry.entities) == len(expected_entities)
check_entities(hass, entity_registry, expected_entities)

4
tests/components/onewire/test_switch.py
View File

@ -1,4 +1,4 @@
"""Tests for 1-Wire devices connected on OWServer."""
"""Tests for 1-Wire switches."""
import logging
from unittest.mock import MagicMock, patch
@ -35,7 +35,7 @@ def override_platforms():
yield
async def test_owserver_switch(
async def test_switches(
hass: HomeAssistant,
config_entry: ConfigEntry,
owproxy: MagicMock,