Initial version

.github/FUNDING.yml

@@ -0,0 +1 @@
+custom: ['https://buymeacoffee.com/seseschneider']

.github/workflows/hacs.yml

@@ -0,0 +1,17 @@
+name: HACS
+
+on:
+  workflow_dispatch:
+  pull_request:
+  schedule:
+    - cron: "0 0 * * *"
+
+jobs:
+  hacs:
+    name: HACS Action
+    runs-on: "ubuntu-latest"
+    steps:
+      - uses: actions/checkout@v3
+      - uses: hacs/action@main
+        with:
+          category: integration

.github/workflows/hassfest.yml

@@ -0,0 +1,14 @@
+name: hassfest
+
+on:
+  workflow_dispatch:
+  pull_request:
+  schedule:
+    - cron:  '0 0 * * *'
+
+jobs:
+  validate:
+    runs-on: "ubuntu-latest"
+    steps:
+      - uses: "actions/checkout@v2"
+      - uses: home-assistant/actions/hassfest@master

.gitignore

@@ -0,0 +1,8 @@
+/node_modules/
+/.rpt2_cache/
+/.idea/
+/dist/
+
+__pycache__/
+
+*.iml

CHANGELOG.md

@@ -0,0 +1,5 @@
+## 1.0.0 (2023-02-07)
+
+### Features
+
+* Initial Release

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2019 Custom cards for Home Assistant
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,2 +1,43 @@
-# ha-personio
-Integration with the Personio API for Home Assistant.
+# Personio integration by [@Sese-Schneider](https://www.github.com/Sese-Schneider)
+
+A Home Assistant integration with the Personio API.
+
+
+[![hacs_badge](https://img.shields.io/badge/HACS-Custom-41BDF5.svg?style=for-the-badge)](https://github.com/hacs/integration)
+[![GitHub Release][releases-shield]][releases]
+![GitHub Downloads][downloads-shield]
+
+[![License][license-shield]](LICENSE)
+![Project Maintenance][maintenance-shield]
+[![GitHub Activity][commits-shield]][commits]
+
+
+[!["Buy Me A Coffee"](https://buymeacoffee.com/assets/img/custom_images/orange_img.png)](https://buymeacoffee.com/seseschneider)
+
+**Features:**
+
+- Automatically adding attendances through an added `switch`.
+
+![](.github/assets/attendance.png)
+
+## Install
+
+### HACS
+
+*This repo can be installed as a custom repository in HACS.*
+
+* Go to HACS → Integration
+* Click on the three-dot-menu → Custom repositories
+* Add `Sese-Schneider/ha-personio` as Integration.
+* Use the FAB "Explore and download repositories" to search "Personio".
+* Restart Home Assistant
+* Install "Personio" as an integration in your settings.
+
+
+[commits-shield]: https://img.shields.io/github/commit-activity/y/Sese-Schneider/ha-personio.svg?style=for-the-badge
+[commits]: https://github.com/Sese-Schneider/ha-personio/commits/master
+[downloads-shield]: https://img.shields.io/github/downloads/Sese-Schneider/ha-personio/total.svg?style=for-the-badge
+[license-shield]: https://img.shields.io/github/license/Sese-Schneider/ha-personio.svg?style=for-the-badge
+[maintenance-shield]: https://img.shields.io/maintenance/yes/2023.svg?style=for-the-badge
+[releases-shield]: https://img.shields.io/github/release/Sese-Schneider/ha-personio.svg?style=for-the-badge
+[releases]: https://github.com/Sese-Schneider/ha-personio/releases

custom_components/personio/__init__.py

@@ -0,0 +1,96 @@
+"""Personio integration."""
+
+from __future__ import annotations
+import logging
+
+from requests import HTTPError
+
+from homeassistant.helpers.template import now
+
+from .const import CONF_USER, COORDINATOR, DOMAIN
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .api.authentication import Authentication
+from .api.attendances import Attendances
+from .api.employees import Employees
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS = [Platform.SWITCH]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
+    """Set up entry."""
+
+    config = entry.as_dict()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
+        COORDINATOR: None,
+    }
+
+    authentication = Authentication()
+    coordinator = PersonioUpdateCoordinator(
+        hass, authentication, config["data"][CONF_USER]
+    )
+
+    hass.data[DOMAIN][entry.entry_id][COORDINATOR] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    await hass.async_add_executor_job(authentication.set_config, config)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+
+class PersonioUpdateCoordinator(DataUpdateCoordinator[None]):
+    """Class to manage personio API calls single endpoint."""
+
+    def __init__(
+        self, hass: HomeAssistant, authentication: Authentication, user_email: str
+    ) -> None:
+        """Initialize global Personio API coordiantor."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=None,
+        )
+        self._hass = hass
+        self._authentication = authentication
+        self._employees_api = Employees(authentication)
+        self._attendance_api = Attendances(authentication)
+
+        self._user_email = user_email
+        self._user_id = None
+
+    async def get_user_id(self) -> int:
+        """Get a users ID through their email."""
+        if self._user_id:
+            return self._user_id
+
+        self._user_id = await self._hass.async_add_executor_job(
+            self._employees_api.get_employee_id_by_mail,
+            self._user_email,
+        )
+
+        return self._user_id
+
+    async def add_attendance(self, start_time: float, end_time: float):
+        """Add users attendance data."""
+        uid = await self.get_user_id()
+
+        self._hass.async_add_executor_job(
+            self._attendance_api.add_attendance,
+            uid,
+            start_time,
+            end_time,
+            now(self._hass)
+        )

custom_components/personio/api/attendances.py

@@ -0,0 +1,71 @@
+"""Attendances for the Persionio API."""
+
+from datetime import datetime, timedelta
+import logging
+import requests
+from config.custom_components.personio.api.authentication import Authentication
+from config.custom_components.personio.api.base import BASE_URL
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class Attendances:
+    """Attendances for the Persionio API."""
+
+    def __init__(self, authentication: Authentication) -> None:
+        self._authentication = authentication
+
+    def add_attendance(
+        self,
+        employee_id: int,
+        start_time: float,
+        end_time: float,
+        now: datetime,
+    ):
+        """Add attendances to the Personio API."""
+
+        attendances = []
+
+        start_date = datetime.fromtimestamp(start_time, tz=now.tzinfo)
+        end_date = datetime.fromtimestamp(end_time, tz=now.tzinfo)
+
+        for single_date in _daterange(start_date, end_date):
+            attendances.append(
+                {
+                    "employee": employee_id,
+                    "date": single_date.strftime("%Y-%m-%d"),
+                    "start_time": start_date.strftime("%H:%M")
+                    if _is_on_date(start_date, single_date)
+                    else "00:00",
+                    "end_time": end_date.strftime("%H:%M")
+                    if _is_on_date(end_date, single_date)
+                    else "23:59",
+                    "break": 0,
+                    "project_id": None,
+                    "comment": "Generated by Sese-Schneider/ha-personio",
+                }
+            )
+
+        result = requests.post(
+            BASE_URL + "/company/attendances",
+            headers=self._authentication.get_headers(),
+            timeout=10000,
+            json={"attendances": attendances},
+        )
+        result.raise_for_status()
+        self._authentication.set_response(result)
+
+        _LOGGER.info("Attendance for employee %s added successfully", employee_id)
+
+
+def _daterange(start_date: datetime, end_date: datetime):
+    start_date = start_date.replace(hour=0, minute=0, second=0, microsecond=0)
+    end_date = end_date.replace(hour=0, minute=0, second=0, microsecond=0)
+    days = int((end_date - start_date).days)
+
+    for n in range(days + 1):  # pylint: disable=invalid-name
+        yield start_date + timedelta(days=n)
+
+
+def _is_on_date(to_check: datetime, on_date: datetime) -> bool:
+    return on_date < to_check < on_date + timedelta(days=1)

custom_components/personio/api/authentication.py

@@ -0,0 +1,104 @@
+"""Authentication for the Personio API"""
+
+import logging
+import time
+import jwt
+import requests
+from requests import Response
+
+from config.custom_components.personio.api.base import BASE_URL
+from homeassistant.exceptions import HomeAssistantError
+
+_LOGGER = logging.getLogger(__name__)
+
+class Authentication:
+    """Authentication for the Personio API."""
+
+    _current_config: dict = None
+    _current_token: str = None
+
+    def set_config(self, config: dict):
+        """Set the current API configuration."""
+        self._current_config = config["data"]
+
+        # self test current auth config, fetch initial token for usage
+        self.get_bearer(invalidate=False)
+
+    def get_bearer(self, invalidate: bool = True):
+        """Get a currently valid authentication bearer."""
+
+        if not self._current_config:
+            raise HomeAssistantError("Config not defined.")
+
+        if self._current_token:
+            jwt_token = jwt.decode(
+                self._current_token,
+                algorithms=["HS256"],
+                options={"verify_signature": False},
+            )
+            if jwt_token["exp"] > time.time():
+                _LOGGER.debug("Reusing existing JWT token")
+                bearer = "Bearer " + self._current_token
+                if invalidate:
+                    # bearers are one-time use only
+                    _LOGGER.debug("Invalidating JWT token")
+                    self._current_token = None
+
+                return bearer
+
+        _LOGGER.debug("Requesting new JWT token")
+        authentication = authenticate(
+            self._current_config["client_id"],
+            self._current_config["client_secret"],
+            self._current_config["partner_id"],
+            self._current_config["app_id"],
+        )
+
+        self._current_token = authentication.json()["data"]["token"]
+        return self.get_bearer(invalidate=invalidate)
+
+
+    def get_headers(self):
+        """Returns all headers required for a successful Personio API request."""
+        partner_id = self._current_config["partner_id"]
+        app_id = self._current_config["app_id"]
+
+        headers = {
+            "Authorization": self.get_bearer()
+        }
+
+        if partner_id:
+            headers["X-Personio-Partner-ID"] = partner_id
+        if app_id:
+            headers["X-Personio-App-ID"] = app_id
+        return headers
+
+
+    def set_response(self, response: Response):
+        """Callback after receiving a new response.
+        Call this to set new authorization headers after each successful request."""
+        self._current_token = response.headers["authorization"].removeprefix("Bearer ")
+        _LOGGER.debug("New JWT token received")
+
+
+def authenticate(
+    client_id: str,
+    client_secret: str,
+    partner_id: str = None,
+    app_id: str = None,
+):
+    """Authenticate agains the Personio API."""
+    headers = {}
+    if partner_id:
+        headers["X-Personio-Partner-ID"] = partner_id
+    if app_id:
+        headers["X-Personio-App-ID"] = app_id
+
+    result = requests.post(
+        BASE_URL + "/auth",
+        params={"client_id": client_id, "client_secret": client_secret},
+        headers=headers,
+        timeout=10000,
+    )
+    result.raise_for_status()
+    return result

custom_components/personio/api/base.py

@@ -0,0 +1 @@
+BASE_URL = "https://api.personio.de/v1"