You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
224 lines
6.6 KiB
224 lines
6.6 KiB
"""Platform for sensor integration.""" |
|
import logging |
|
import json |
|
import requests |
|
import voluptuous as vol |
|
import homeassistant.helpers.config_validation as cv |
|
from homeassistant.helpers.entity import Entity |
|
from homeassistant.components.switch import PLATFORM_SCHEMA |
|
from homeassistant.const import ( |
|
CONF_TOKEN, |
|
CONF_PORT, |
|
CONF_HOST, |
|
CONF_PATH, |
|
CONF_NAME, |
|
CONF_PROTOCOL, |
|
) |
|
|
|
_LOGGER = logging.getLogger(__name__) |
|
|
|
DEFAULT_NAME = "gitea" |
|
CONF_REPOS = "repositories" |
|
ATTR_REPO_NAME = "Repository" |
|
ATTR_REPO_ID = "ID" |
|
ATTR_DESCRIPTION = "Description" |
|
ATTR_OPEN_ISSUES = "Open issues" |
|
ATTR_DEFAULT_BR = "Branch" |
|
ATTR_OWNER = "Owner" |
|
ATTR_SIZE = "Size" |
|
ATTR_PRIVATE_REPO = "isPrivate" |
|
ATTR_FORK = "Forks" |
|
ATTR_MIRROR = "isMirror" |
|
ATTR_REPO_URL = "Repository Url" |
|
ATTR_STARS = "Stars" |
|
ATTR_WATCH = "Watchers" |
|
ATTR_UPDATED_AT = "Updated at" |
|
ATTR_LAST_ISSUE_LINK = "Last Issue Link" |
|
ATTR_LAST_ISSUE_STATE = "Last Issue Status" |
|
ATTR_LAST_ISSUE_TITLE = "Last Issue Title" |
|
ATTR_ALL_ISSUES = "All Issues" |
|
|
|
|
|
URL_ISSUE = "/issues?state=all" |
|
|
|
|
|
REPO_SCHEMA = vol.Schema( |
|
{vol.Required(CONF_PATH): cv.string, vol.Optional(CONF_NAME): cv.string} |
|
) |
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( |
|
{ |
|
vol.Required(CONF_TOKEN): cv.string, |
|
vol.Required(CONF_PORT): cv.string, |
|
vol.Required(CONF_HOST): cv.string, |
|
vol.Required(CONF_PROTOCOL): cv.string, |
|
vol.Required(CONF_REPOS): vol.All(cv.ensure_list, [REPO_SCHEMA]), |
|
} |
|
) |
|
|
|
|
|
def setup_platform(hass, config, add_entities, discovery_info=None): |
|
"""Setup Platform.""" |
|
for repo in config[CONF_REPOS]: |
|
add_entities( |
|
[ |
|
GiteaSensor( |
|
config.get(CONF_TOKEN), |
|
config.get(CONF_PROTOCOL), |
|
config.get(CONF_HOST), |
|
config.get(CONF_PORT), |
|
repo, |
|
) |
|
] |
|
) |
|
|
|
|
|
class GiteaSensor(Entity): |
|
"""Representation of a Sensor.""" |
|
|
|
def __init__( |
|
self, |
|
token=None, |
|
proto=None, |
|
api_url=None, |
|
api_port=None, |
|
repo=None, |
|
id=None, |
|
description=None, |
|
open_issues_count=None, |
|
default_branch=None, |
|
size=None, |
|
owner_name=None, |
|
private=None, |
|
stars=None, |
|
fork=None, |
|
mirror=None, |
|
url=None, |
|
updated_at=None, |
|
issue_title=None, |
|
issue_link=None, |
|
issue_state=None, |
|
all_issues=None, |
|
): |
|
self._state = None |
|
self.token = token |
|
self.proto = proto |
|
self.api_url = api_url |
|
self.api_port = api_port |
|
self.repo = repo[CONF_PATH] |
|
self.id_repo = id |
|
self.description = description |
|
self.open_issues_count = open_issues_count |
|
self.default_branch = default_branch |
|
self.size = size |
|
self.owner_name = owner_name |
|
self.private = private |
|
self.mirror = mirror |
|
self.fork = fork |
|
self.stars = stars |
|
self.url = url |
|
self.watcher = None |
|
self.updated_at = updated_at |
|
self.issue_title = issue_title |
|
self.issue_link = issue_link |
|
self.issue_state = issue_state |
|
self.all_issues = all_issues |
|
|
|
@property |
|
def name(self): |
|
"""Return the name of the sensor.""" |
|
return DEFAULT_NAME + "_" + self.repo.split("/")[1] |
|
|
|
@property |
|
def state(self): |
|
"""Return the state of the sensor.""" |
|
return self._state |
|
|
|
@property |
|
def icon(self): |
|
"""Return the icon to use in the frontend.""" |
|
return "mdi:tea" |
|
|
|
@property |
|
def extra_state_attributes(self): |
|
"""Return the state attributes.""" |
|
attrs = { |
|
ATTR_REPO_ID: self.id_repo, |
|
ATTR_REPO_NAME: self.repo, |
|
ATTR_OWNER: self.owner_name, |
|
ATTR_PRIVATE_REPO: self.private, |
|
ATTR_FORK: self.fork, |
|
ATTR_MIRROR: self.mirror, |
|
ATTR_STARS: self.stars, |
|
ATTR_DESCRIPTION: self.description, |
|
ATTR_OPEN_ISSUES: self.open_issues_count, |
|
ATTR_DEFAULT_BR: self.default_branch, |
|
ATTR_REPO_URL: self.url, |
|
ATTR_SIZE: self.size, |
|
ATTR_WATCH: self.watcher, |
|
ATTR_UPDATED_AT: self.updated_at, |
|
ATTR_LAST_ISSUE_LINK: self.issue_link, |
|
ATTR_LAST_ISSUE_STATE: self.issue_state, |
|
ATTR_LAST_ISSUE_TITLE: self.issue_title, |
|
ATTR_ALL_ISSUES: self.all_issues, |
|
} |
|
return attrs |
|
|
|
def update(self): |
|
"""Update all sensor attributes.""" |
|
infos = self.api_call(self.generate_url()) |
|
self.id_repo = infos["id"] |
|
self.description = infos["description"] |
|
self.open_issues_count = infos["description"] |
|
self.open_issues_count = infos["open_issues_count"] |
|
self.default_branch = infos["default_branch"] |
|
self.size = str(infos["size"]) + " Mo" |
|
self.owner_name = infos["owner"]["login"] |
|
self.private = infos["private"] |
|
self.mirror = infos["mirror"] |
|
self.stars = infos["stars_count"] |
|
self.fork = infos["forks_count"] |
|
self.url = infos["html_url"] |
|
self._state = infos["default_branch"] |
|
self.watcher = infos["watchers_count"] |
|
self.updated_at = infos["updated_at"] |
|
|
|
if infos["open_issues_count"] != 0: |
|
issues_tab = [] |
|
issues = self.api_call(self.generate_url(URL_ISSUE)) |
|
self.issue_link = issues[0]["html_url"] |
|
self.issue_state = issues[0]["state"] |
|
self.issue_title = issues[0]["title"] |
|
|
|
for iss in issues: |
|
card_items = {} |
|
card_items["id"] = iss["id"] |
|
card_items["state"] = iss["state"] |
|
card_items["title"] = iss["title"] |
|
card_items["url"] = iss["html_url"] |
|
issues_tab.append(card_items) |
|
|
|
self.all_issues = json.dumps(issues_tab) |
|
|
|
def generate_url(self, path=""): |
|
"""Return api url.""" |
|
return "{0}://{1}:{2}/api/v1/repos/{3}/{4}{5}".format( |
|
self.proto, |
|
self.api_url, |
|
self.api_port, |
|
self.repo.split("/")[0], |
|
self.repo.split("/")[1], |
|
path, |
|
) |
|
|
|
def get_header(self): |
|
"""Return headers for api request.""" |
|
headers = { |
|
"Content-Type": "application/json", |
|
"Authorization": "Bearer {0}".format(self.token), |
|
} |
|
return headers |
|
|
|
def api_call(self, url): |
|
"""Return result of api request.""" |
|
return requests.request(method="GET", url=url, headers=self.get_header()).json() |