evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Cleanup GitHub sensor classes and descriptions (#64853)

Browse source
This commit is contained in:
Joakim Sørensen 2022-01-24 18:47:21 +01:00 • committed by GitHub
parent 24ee4256b9
commit b07f4ba398
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 94 additions and 232 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
homeassistant/components/github/coordinator.py
View file

@ -144,8 +144,8 @@ class RepositoryIssueDataUpdateCoordinator(
pull_last = self.data.pull_last
else:
self._pull_etag = pull_response.etag
pulls_count = pull_response.last_page_number or 0
pull_last = pull_response.data[0] if pulls_count != 0 else None
pulls_count = pull_response.last_page_number or len(pull_response.data)
pull_last = pull_response.data[0] if pull_response.data else None
try:
issue_response = await self._client.repos.issues.list(
@ -158,8 +158,10 @@ class RepositoryIssueDataUpdateCoordinator(
issue_last = self.data.issue_last
else:
self._issue_etag = issue_response.etag
issues_count = (issue_response.last_page_number or 0) - pulls_count
issue_last = issue_response.data[0] if issues_count != 0 else None
issues_count = (
issue_response.last_page_number or len(issue_response.data)
) - pulls_count
issue_last = issue_response.data[0] if issue_response.data else None
if issue_last is not None and issue_last.pull_request:
issue_response = await self._client.repos.issues.list(self.repository)

316
homeassistant/components/github/sensor.py
View file

@ -3,8 +3,7 @@ from __future__ import annotations
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from aiogithubapi import GitHubRepositoryModel
from typing import Any
from homeassistant.components.sensor import (
SensorEntity,
@ -19,69 +18,39 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, IssuesPulls
from .const import DOMAIN
from .coordinator import (
CoordinatorKeyType,
DataUpdateCoordinators,
GitHubBaseDataUpdateCoordinator,
RepositoryCommitDataUpdateCoordinator,
RepositoryIssueDataUpdateCoordinator,
RepositoryReleaseDataUpdateCoordinator,
)
@dataclass
class GitHubSensorBaseEntityDescriptionMixin:
class BaseEntityDescriptionMixin:
"""Mixin for required GitHub base description keys."""
coordinator_key: CoordinatorKeyType
value_fn: Callable[[Any], StateType]
@dataclass
class GitHubSensorInformationEntityDescriptionMixin(
GitHubSensorBaseEntityDescriptionMixin
):
"""Mixin for required GitHub information description keys."""
value_fn: Callable[[GitHubRepositoryModel], StateType]
@dataclass
class GitHubSensorIssueEntityDescriptionMixin(GitHubSensorBaseEntityDescriptionMixin):
"""Mixin for required GitHub information description keys."""
value_fn: Callable[[IssuesPulls], StateType]
@dataclass
class GitHubSensorBaseEntityDescription(SensorEntityDescription):
class BaseEntityDescription(SensorEntityDescription):
"""Describes GitHub sensor entity default overrides."""
icon: str = "mdi:github"
entity_registry_enabled_default: bool = False
attr_fn: Callable[[Any], Mapping[str, Any] | None] = lambda data: None
avabl_fn: Callable[[Any], bool] = lambda data: True
@dataclass
class GitHubSensorInformationEntityDescription(
GitHubSensorBaseEntityDescription,
GitHubSensorInformationEntityDescriptionMixin,
):
"""Describes GitHub information sensor entity."""
@dataclass
class GitHubSensorIssueEntityDescription(
GitHubSensorBaseEntityDescription,
GitHubSensorIssueEntityDescriptionMixin,
):
class GitHubSensorEntityDescription(BaseEntityDescription, BaseEntityDescriptionMixin):
"""Describes GitHub issue sensor entity."""
SENSOR_DESCRIPTIONS: tuple[
GitHubSensorInformationEntityDescription | GitHubSensorIssueEntityDescription,
...,
] = (
GitHubSensorInformationEntityDescription(
SENSOR_DESCRIPTIONS: tuple[GitHubSensorEntityDescription, ...] = (
GitHubSensorEntityDescription(
key="stargazers_count",
name="Stars",
icon="mdi:star",
@ -91,7 +60,7 @@ SENSOR_DESCRIPTIONS: tuple[
value_fn=lambda data: data.stargazers_count,
coordinator_key="information",
),
GitHubSensorInformationEntityDescription(
GitHubSensorEntityDescription(
key="subscribers_count",
name="Watchers",
icon="mdi:glasses",
@ -102,7 +71,7 @@ SENSOR_DESCRIPTIONS: tuple[
value_fn=lambda data: data.subscribers_count,
coordinator_key="information",
),
GitHubSensorInformationEntityDescription(
GitHubSensorEntityDescription(
key="forks_count",
name="Forks",
icon="mdi:source-fork",
@ -112,7 +81,7 @@ SENSOR_DESCRIPTIONS: tuple[
value_fn=lambda data: data.forks_count,
coordinator_key="information",
),
GitHubSensorIssueEntityDescription(
GitHubSensorEntityDescription(
key="issues_count",
name="Issues",
native_unit_of_measurement="Issues",
@ -121,7 +90,7 @@ SENSOR_DESCRIPTIONS: tuple[
value_fn=lambda data: data.issues_count,
coordinator_key="issue",
),
GitHubSensorIssueEntityDescription(
GitHubSensorEntityDescription(
key="pulls_count",
name="Pull Requests",
native_unit_of_measurement="Pull Requests",
@ -130,6 +99,49 @@ SENSOR_DESCRIPTIONS: tuple[
value_fn=lambda data: data.pulls_count,
coordinator_key="issue",
),
GitHubSensorEntityDescription(
coordinator_key="commit",
key="latest_commit",
name="Latest Commit",
value_fn=lambda data: data.commit.message.splitlines()[0][:255],
attr_fn=lambda data: {
"sha": data.sha,
"url": data.html_url,
},
),
GitHubSensorEntityDescription(
coordinator_key="release",
key="latest_release",
name="Latest Release",
entity_registry_enabled_default=True,
value_fn=lambda data: data.name[:255],
attr_fn=lambda data: {
"url": data.html_url,
"tag": data.tag_name,
},
),
GitHubSensorEntityDescription(
coordinator_key="issue",
key="latest_issue",
name="Latest Issue",
value_fn=lambda data: data.issue_last.title[:255],
avabl_fn=lambda data: data.issue_last is not None,
attr_fn=lambda data: {
"url": data.issue_last.html_url,
"number": data.issue_last.number,
},
),
GitHubSensorEntityDescription(
coordinator_key="issue",
key="latest_pull_request",
name="Latest Pull Request",
value_fn=lambda data: data.pull_last.title[:255],
avabl_fn=lambda data: data.pull_last is not None,
attr_fn=lambda data: {
"url": data.pull_last.html_url,
"number": data.pull_last.number,
},
),
)
@ -140,214 +152,62 @@ async def async_setup_entry(
) -> None:
"""Set up GitHub sensor based on a config entry."""
repositories: dict[str, DataUpdateCoordinators] = hass.data[DOMAIN]
entities: list[GitHubSensorBaseEntity] = []
for coordinators in repositories.values():
repository_information = coordinators["information"].data
entities.extend(
sensor(coordinators, repository_information)
for sensor in (
GitHubSensorLatestCommitEntity,
GitHubSensorLatestIssueEntity,
GitHubSensorLatestPullEntity,
GitHubSensorLatestReleaseEntity,
)
)
entities.extend(
GitHubSensorDescriptionEntity(
coordinators, description, repository_information
)
async_add_entities(
(
GitHubSensorEntity(coordinators, description)
for description in SENSOR_DESCRIPTIONS
)
async_add_entities(entities)
for coordinators in repositories.values()
),
update_before_add=True,
)
class GitHubSensorBaseEntity(CoordinatorEntity, SensorEntity):
"""Defines a base GitHub sensor entity."""
class GitHubSensorEntity(CoordinatorEntity, SensorEntity):
"""Defines a GitHub sensor entity."""
_attr_attribution = "Data provided by the GitHub API"
coordinator: GitHubBaseDataUpdateCoordinator
entity_description: GitHubSensorEntityDescription
def __init__(
self,
coordinator: GitHubBaseDataUpdateCoordinator,
repository_information: GitHubRepositoryModel,
coordinators: DataUpdateCoordinators,
entity_description: GitHubSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
coordinator = coordinators[entity_description.coordinator_key]
_information = coordinators["information"].data
super().__init__(coordinator=coordinator)
self.entity_description = entity_description
self._attr_name = f"{_information.full_name} {entity_description.name}"
self._attr_unique_id = f"{_information.id}_{entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self.coordinator.repository)},
name=repository_information.full_name,
identifiers={(DOMAIN, coordinator.repository)},
name=_information.full_name,
manufacturer="GitHub",
configuration_url=f"https://github.com/{self.coordinator.repository}",
configuration_url=f"https://github.com/{coordinator.repository}",
entry_type=DeviceEntryType.SERVICE,
)
@property
def available(self) -> bool:
"""Return if entity is available."""
return super().available and self.coordinator.data is not None
class GitHubSensorDescriptionEntity(GitHubSensorBaseEntity):
"""Defines a GitHub sensor entity based on entity descriptions."""
coordinator: GitHubBaseDataUpdateCoordinator
entity_description: GitHubSensorInformationEntityDescription | GitHubSensorIssueEntityDescription
def __init__(
self,
coordinators: DataUpdateCoordinators,
description: GitHubSensorInformationEntityDescription
| GitHubSensorIssueEntityDescription,
repository_information: GitHubRepositoryModel,
) -> None:
"""Initialize a GitHub sensor entity."""
super().__init__(
coordinator=coordinators[description.coordinator_key],
repository_information=repository_information,
"""Return True if entity is available."""
return (
super().available
and self.coordinator.data is not None
and self.entity_description.avabl_fn(self.coordinator.data)
)
self.entity_description = description
self._attr_name = f"{repository_information.full_name} {description.name}"
self._attr_unique_id = f"{repository_information.id}_{description.key}"
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)
class GitHubSensorLatestBaseEntity(GitHubSensorBaseEntity):
"""Defines a base GitHub latest sensor entity."""
_name: str = "Latest"
_coordinator_key: CoordinatorKeyType = "information"
_attr_entity_registry_enabled_default = False
_attr_icon = "mdi:github"
def __init__(
self,
coordinators: DataUpdateCoordinators,
repository_information: GitHubRepositoryModel,
) -> None:
"""Initialize a GitHub sensor entity."""
super().__init__(
coordinator=coordinators[self._coordinator_key],
repository_information=repository_information,
)
self._attr_name = f"{repository_information.full_name} {self._name}"
self._attr_unique_id = (
f"{repository_information.id}_{self._name.lower().replace(' ', '_')}"
)
class GitHubSensorLatestReleaseEntity(GitHubSensorLatestBaseEntity):
"""Defines a GitHub latest release sensor entity."""
_coordinator_key: CoordinatorKeyType = "release"
_name: str = "Latest Release"
_attr_entity_registry_enabled_default = True
coordinator: RepositoryReleaseDataUpdateCoordinator
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.coordinator.data.name[:255]
@property
def extra_state_attributes(self) -> Mapping[str, str | None]:
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return the extra state attributes."""
release = self.coordinator.data
return {
"url": release.html_url,
"tag": release.tag_name,
}
class GitHubSensorLatestIssueEntity(GitHubSensorLatestBaseEntity):
"""Defines a GitHub latest issue sensor entity."""
_name: str = "Latest Issue"
_coordinator_key: CoordinatorKeyType = "issue"
coordinator: RepositoryIssueDataUpdateCoordinator
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self.coordinator.data.issues_count != 0
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
if (issue := self.coordinator.data.issue_last) is None:
return None
return issue.title[:255]
@property
def extra_state_attributes(self) -> Mapping[str, str | int | None] | None:
"""Return the extra state attributes."""
if (issue := self.coordinator.data.issue_last) is None:
return None
return {
"url": issue.html_url,
"number": issue.number,
}
class GitHubSensorLatestPullEntity(GitHubSensorLatestBaseEntity):
"""Defines a GitHub latest pull sensor entity."""
_coordinator_key: CoordinatorKeyType = "issue"
_name: str = "Latest Pull Request"
coordinator: RepositoryIssueDataUpdateCoordinator
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self.coordinator.data.pulls_count != 0
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
if (pull := self.coordinator.data.pull_last) is None:
return None
return pull.title[:255]
@property
def extra_state_attributes(self) -> Mapping[str, str | int | None] | None:
"""Return the extra state attributes."""
if (pull := self.coordinator.data.pull_last) is None:
return None
return {
"url": pull.html_url,
"number": pull.number,
}
class GitHubSensorLatestCommitEntity(GitHubSensorLatestBaseEntity):
"""Defines a GitHub latest commit sensor entity."""
_coordinator_key: CoordinatorKeyType = "commit"
_name: str = "Latest Commit"
coordinator: RepositoryCommitDataUpdateCoordinator
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.coordinator.data.commit.message.splitlines()[0][:255]
@property
def extra_state_attributes(self) -> Mapping[str, str | int | None]:
"""Return the extra state attributes."""
return {
"sha": self.coordinator.data.sha,
"url": self.coordinator.data.html_url,
}
return self.entity_description.attr_fn(self.coordinator.data)