164 lines
5.3 KiB
Python
164 lines
5.3 KiB
Python
"""This component provides HA sensor support for Travis CI framework."""
|
|
from datetime import timedelta
|
|
import logging
|
|
|
|
from travispy import TravisPy
|
|
from travispy.errors import TravisError
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
|
|
from homeassistant.const import (
|
|
ATTR_ATTRIBUTION,
|
|
CONF_API_KEY,
|
|
CONF_MONITORED_CONDITIONS,
|
|
CONF_SCAN_INTERVAL,
|
|
TIME_SECONDS,
|
|
)
|
|
import homeassistant.helpers.config_validation as cv
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
ATTRIBUTION = "Information provided by https://travis-ci.org/"
|
|
|
|
CONF_BRANCH = "branch"
|
|
CONF_REPOSITORY = "repository"
|
|
|
|
DEFAULT_BRANCH_NAME = "master"
|
|
|
|
SCAN_INTERVAL = timedelta(seconds=30)
|
|
|
|
# sensor_type [ description, unit, icon ]
|
|
SENSOR_TYPES = {
|
|
"last_build_id": ["Last Build ID", "", "mdi:card-account-details"],
|
|
"last_build_duration": ["Last Build Duration", TIME_SECONDS, "mdi:timelapse"],
|
|
"last_build_finished_at": ["Last Build Finished At", "", "mdi:timetable"],
|
|
"last_build_started_at": ["Last Build Started At", "", "mdi:timetable"],
|
|
"last_build_state": ["Last Build State", "", "mdi:github"],
|
|
"state": ["State", "", "mdi:github"],
|
|
}
|
|
|
|
NOTIFICATION_ID = "travisci"
|
|
NOTIFICATION_TITLE = "Travis CI Sensor Setup"
|
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|
{
|
|
vol.Required(CONF_API_KEY): cv.string,
|
|
vol.Required(CONF_MONITORED_CONDITIONS, default=list(SENSOR_TYPES)): vol.All(
|
|
cv.ensure_list, [vol.In(SENSOR_TYPES)]
|
|
),
|
|
vol.Required(CONF_BRANCH, default=DEFAULT_BRANCH_NAME): cv.string,
|
|
vol.Optional(CONF_REPOSITORY, default=[]): vol.All(cv.ensure_list, [cv.string]),
|
|
vol.Optional(CONF_SCAN_INTERVAL, default=SCAN_INTERVAL): cv.time_period,
|
|
}
|
|
)
|
|
|
|
|
|
def setup_platform(hass, config, add_entities, discovery_info=None):
|
|
"""Set up the Travis CI sensor."""
|
|
|
|
token = config.get(CONF_API_KEY)
|
|
repositories = config.get(CONF_REPOSITORY)
|
|
branch = config.get(CONF_BRANCH)
|
|
|
|
try:
|
|
travis = TravisPy.github_auth(token)
|
|
user = travis.user()
|
|
|
|
except TravisError as ex:
|
|
_LOGGER.error("Unable to connect to Travis CI service: %s", str(ex))
|
|
hass.components.persistent_notification.create(
|
|
"Error: {}<br />"
|
|
"You will need to restart hass after fixing."
|
|
"".format(ex),
|
|
title=NOTIFICATION_TITLE,
|
|
notification_id=NOTIFICATION_ID,
|
|
)
|
|
return False
|
|
|
|
sensors = []
|
|
|
|
# non specific repository selected, then show all associated
|
|
if not repositories:
|
|
all_repos = travis.repos(member=user.login)
|
|
repositories = [repo.slug for repo in all_repos]
|
|
|
|
for repo in repositories:
|
|
if "/" not in repo:
|
|
repo = f"{user.login}/{repo}"
|
|
|
|
for sensor_type in config.get(CONF_MONITORED_CONDITIONS):
|
|
sensors.append(TravisCISensor(travis, repo, user, branch, sensor_type))
|
|
|
|
add_entities(sensors, True)
|
|
return True
|
|
|
|
|
|
class TravisCISensor(SensorEntity):
|
|
"""Representation of a Travis CI sensor."""
|
|
|
|
def __init__(self, data, repo_name, user, branch, sensor_type):
|
|
"""Initialize the sensor."""
|
|
self._build = None
|
|
self._sensor_type = sensor_type
|
|
self._data = data
|
|
self._repo_name = repo_name
|
|
self._user = user
|
|
self._branch = branch
|
|
self._state = None
|
|
self._name = "{} {}".format(self._repo_name, SENSOR_TYPES[self._sensor_type][0])
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the sensor."""
|
|
return self._name
|
|
|
|
@property
|
|
def unit_of_measurement(self):
|
|
"""Return the unit of measurement of this entity, if any."""
|
|
return SENSOR_TYPES[self._sensor_type][1]
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state of the sensor."""
|
|
return self._state
|
|
|
|
@property
|
|
def extra_state_attributes(self):
|
|
"""Return the state attributes."""
|
|
attrs = {}
|
|
attrs[ATTR_ATTRIBUTION] = ATTRIBUTION
|
|
|
|
if self._build and self._state is not None:
|
|
if self._user and self._sensor_type == "state":
|
|
attrs["Owner Name"] = self._user.name
|
|
attrs["Owner Email"] = self._user.email
|
|
else:
|
|
attrs["Committer Name"] = self._build.commit.committer_name
|
|
attrs["Committer Email"] = self._build.commit.committer_email
|
|
attrs["Commit Branch"] = self._build.commit.branch
|
|
attrs["Committed Date"] = self._build.commit.committed_at
|
|
attrs["Commit SHA"] = self._build.commit.sha
|
|
|
|
return attrs
|
|
|
|
@property
|
|
def icon(self):
|
|
"""Return the icon to use in the frontend, if any."""
|
|
return SENSOR_TYPES[self._sensor_type][2]
|
|
|
|
def update(self):
|
|
"""Get the latest data and updates the states."""
|
|
_LOGGER.debug("Updating sensor %s", self._name)
|
|
|
|
repo = self._data.repo(self._repo_name)
|
|
self._build = self._data.build(repo.last_build_id)
|
|
|
|
if self._build:
|
|
if self._sensor_type == "state":
|
|
branch_stats = self._data.branch(self._branch, self._repo_name)
|
|
self._state = branch_stats.state
|
|
|
|
else:
|
|
param = self._sensor_type.replace("last_build_", "")
|
|
self._state = getattr(self._build, param)
|