Joren/dispatcharr_sensor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add files via upload

Browse Source
This commit is contained in:
Lyfesaver
2025-10-02 15:58:48 -05:00
committed by GitHub
parent c2b31d9eaf
commit c52a79d81f
8 changed files with 500 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
custom_components/dispatcharr_sensor/LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Lyfesaver
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.

17
custom_components/dispatcharr_sensor/__init__.py Normal file
View File

@@ -0,0 +1,17 @@
"""The Dispatcharr Sensor integration."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import PLATFORMS
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Dispatcharr Sensor from a config entry."""
# This is the correct method name: async_forward_entry_setups
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
# The unload method name is different, which is confusing. This one is correct.
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

36
custom_components/dispatcharr_sensor/config_flow.py Normal file
View File

@@ -0,0 +1,36 @@
"""Config flow for Dispatcharr Sensor integration."""
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
# Ask for username and password instead of API token
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required("host"): str,
vol.Required("port", default=9191): int,
vol.Required("username"): str,
vol.Required("password"): str,
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Dispatcharr Sensor."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
return self.async_create_entry(title="Dispatcharr", data=user_input)

6
custom_components/dispatcharr_sensor/const.py Normal file
View File

@@ -0,0 +1,6 @@
"""Constants for the Dispatcharr Sensor integration."""
DOMAIN = "dispatcharr_sensor"
# List of platforms to support. In this case, just the "sensor" platform.
PLATFORMS = ["sensor"]

7
custom_components/dispatcharr_sensor/hacs.json Normal file
View File

@@ -0,0 +1,7 @@
{
"name": "Dispatcharr Sessions Sensor",
"content_in_root": false,
"domains": ["sensor"],
"country": "us",
"homeassistant": "2022.6.0"
}

11
custom_components/dispatcharr_sensor/manifest.json Normal file
View File

@@ -0,0 +1,11 @@
{
"domain": "dispatcharr_sensor",
"name": "Dispatcharr Sessions Sensor",
"version": "2.0.0",
"documentation": "https://github.com/lyfesaver74/ha-dispatcharr",
"requirements": [],
"dependencies": [],
"codeowners": ["@lyfesaver74"],
"iot_class": "local_polling",
"config_flow": true
}

119
custom_components/dispatcharr_sensor/readme.md
View File

@@ -1 +1,120 @@
# Dispatcharr Integration for Home Assistant
[![hacs_badge](https://img.shields.io/badge/HACS-Custom-orange.svg)](https://github.com/custom-components/hacs)
This is a custom integration for [Home Assistant](https://www.home-assistant.io/) that connects to your [Dispatcharr](https://github.com/Dispatcharr/Dispatcharr) instance. It provides real-time monitoring of active streams, creating dynamic sensors for each stream and a summary sensor for the total count.
## Features
* **Total Stream Count:** A dedicated sensor (`sensor.dispatcharr_total_active_streams`) that shows the total number of currently active streams.
* **Dynamic Stream Sensors:** Creates a new sensor entity for each active stream automatically. These sensors are removed automatically when the stream stops, keeping your Home Assistant instance clean.
* **Rich Stream Data:** Each stream sensor provides detailed attributes, including:
* Channel Name and Number
* Channel Logo URL
* Client Count
* Video Resolution, FPS, and Codec
* Audio Codec
* Average Bitrate
* **Live Program Information:** The integration parses your EPG data to display the currently airing program's title, description, start time, and stop time for each active stream.
## Prerequisites
* A running instance of [Dispatcharr](https://github.com/Dispatcharr/Dispatcharr).
* [Home Assistant Community Store (HACS)](https://hacs.xyz/) installed on your Home Assistant instance.
* The username and password for your Dispatcharr user account.
## Installation via HACS
This integration is not yet in the default HACS repository. You can add it as a custom repository.
1. In Home Assistant, go to **HACS** > **Integrations**.
2. Click the three dots in the top-right corner and select **"Custom repositories"**.
3. In the "Repository" field, enter the URL to this GitHub repository: `https://github.com/lyfesaver74/ha-dispatcharr`
4. For the "Category" dropdown, select **"Integration"**.
5. Click **"Add"**.
6. You should now see the "Dispatcharr Sessions Sensor" in your HACS integrations list. Click **"Install"** and proceed with the installation.
7. Restart Home Assistant when prompted.
## Configuration
Once the integration is installed, you can add it to Home Assistant via the UI.
1. Go to **Settings** > **Devices & Services**.
2. Click the **"+ Add Integration"** button in the bottom right.
3. Search for **"Dispatcharr"** and select it.
4. A configuration dialog will appear. Enter the following information:
* **Host:** The IP address of your Dispatcharr server (e.g., `192.168.0.121`).
* **Port:** The port your Dispatcharr server is running on (default is `9191`).
* **Username:** Your Dispatcharr username.
* **Password:** Your Dispatcharr password.
5. Click **"Submit"**. The integration will be set up and your sensors will be created.
## Provided Sensors
The integration will create the following entities:
* **`sensor.dispatcharr_total_active_streams`**: A sensor that always exists and shows the total number of active streams. Its state is a number (e.g., `0`, `1`, `2`).
* **`sensor.dispatcharr_stream_<channel_name>`** (Dynamic): A new sensor will be created for each active stream. The entity ID is based on the channel name. For example, a stream of "BBC America" might create `sensor.dispatcharr_stream_us_bbc_america_hd`.
* The state of these sensors is either `Streaming` (when active) or `Idle` (if it persists briefly after stopping).
* These sensors will be removed automatically when no longer active.
### Stream Sensor Attributes
Each dynamic stream sensor will have the following attributes:
| Attribute | Description | Example |
|---|---|---|
| `channel_number` | The stream's channel number or ID. | `98209` |
| `channel_name` | The friendly name of the channel. | `US| BBC America ᴴᴰ` |
| `logo_url` | A direct URL to the channel's logo image. | `http://.../api/channels/logos/262/cache/` |
| `clients` | The number of clients watching this stream. | `1` |
| `resolution` | The current video resolution. | `1280x720` |
| `fps` | The current frames per second. | `59.94` |
| `video_codec` | The video codec being used. | `h264` |
| `audio_codec` | The audio codec being used. | `aac` |
| `avg_bitrate` | The average bitrate of the stream. | `4.11 Mbps` |
| `program_title` | The title of the currently airing program. | `Doctor Who` |
| `program_description` | A description of the current program. | `The Doctor travels through time...` |
| `program_start` | The start time of the current program. | `2025-10-02T14:00:00-05:00` |
| `program_stop` | The end time of the current program. | `2025-10-02T15:00:00-05:00` |
## Troubleshooting
* **Program Data is `null`:** If your stream sensors are created but the `program_title`, `program_description`, etc., are `null`, it means the integration was unable to find matching guide data for that specific channel in your Dispatcharr EPG file. Please ensure that the channel has EPG data assigned within the Dispatcharr UI and that your EPG source has been recently refreshed.
* **Authentication Errors:** If you receive errors after setup, double-check that your Dispatcharr username and password are correct.
## License
This project is licensed under the MIT License - see the `LICENSE` file for details.
```

282
custom_components/dispatcharr_sensor/sensor.py Normal file
View File

@@ -0,0 +1,282 @@
"""Sensor platform for Dispatcharr."""
import logging
from datetime import timedelta, datetime, timezone
from urllib.parse import urlencode
import xml.etree.ElementTree as ET
import aiohttp
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.core import HomeAssistant, callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.components.sensor import SensorEntity, SensorStateClass
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.util import slugify
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the sensor platform."""
coordinator = DispatcharrDataUpdateCoordinator(hass, config_entry)
try:
await coordinator.async_populate_channel_details()
await coordinator.async_config_entry_first_refresh()
except ConfigEntryNotReady:
raise
# This manager will add and remove stream sensors dynamically
DispatcharrStreamManager(coordinator, async_add_entities)
# Add the static "total" sensor
async_add_entities([DispatcharrTotalStreamSensor(coordinator)])
class DispatcharrDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching data from the API."""
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry):
"""Initialize."""
self.config_entry = config_entry
self.websession = aiohttp.ClientSession()
self._access_token: str | None = None
self.channel_details: dict = {}
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=30),
)
@property
def base_url(self) -> str:
"""Get the base URL for API calls."""
data = self.config_entry.data
protocol = "https" if data.get("ssl", False) else "http"
return f"{protocol}://{data['host']}:{data['port']}"
async def _api_request(self, method: str, url: str, is_json: bool = True, **kwargs):
"""Make an authenticated API request, refreshing token if necessary."""
if not self._access_token:
await self._get_new_token()
headers = {"Authorization": f"Bearer {self._access_token}"}
try:
response = await self.websession.request(method, url, headers=headers, **kwargs)
if response.status == 401:
_LOGGER.info("Access token expired, requesting a new one")
await self._get_new_token()
headers["Authorization"] = f"Bearer {self._access_token}"
response = await self.websession.request(method, url, headers=headers, **kwargs)
response.raise_for_status()
return await response.json() if is_json else await response.text()
except aiohttp.ClientResponseError as err:
if "epg/grid" in url and err.status == 404:
_LOGGER.warning("EPG Grid returned a 404 for the requested channels, treating as no program data.")
return {}
raise UpdateFailed(f"API request failed for {url}: {err.status} {err.message}") from err
except aiohttp.ClientError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
async def _get_new_token(self) -> str:
"""Get a new access token using username and password."""
_LOGGER.debug("Requesting new access token from Dispatcharr")
url = f"{self.base_url}/api/accounts/token/"
auth_data = {
"username": self.config_entry.data["username"],
"password": self.config_entry.data["password"],
}
try:
async with self.websession.post(url, json=auth_data) as response:
response.raise_for_status()
tokens = await response.json()
self._access_token = tokens.get("access")
if self._access_token:
_LOGGER.info("Successfully authenticated with Dispatcharr")
return self._access_token
raise ConfigEntryNotReady("Authentication successful, but no access token was provided")
except aiohttp.ClientError as err:
_LOGGER.error("Authentication failed: %s", err)
raise ConfigEntryNotReady(f"Authentication failed: {err}") from err
async def async_populate_channel_details(self):
"""Fetch all channel details to build a lookup map."""
_LOGGER.info("Populating Dispatcharr channel details")
all_channels = await self._api_request("GET", f"{self.base_url}/api/channels/channels/")
if isinstance(all_channels, list):
self.channel_details = {
channel['uuid']: channel for channel in all_channels if 'uuid' in channel
}
else:
_LOGGER.warning("Expected a list of channels but received: %s", type(all_channels))
self.channel_details = {}
_LOGGER.debug("Found %d channels", len(self.channel_details))
async def _get_current_programs_from_xml(self, epg_ids: list[str]) -> dict:
"""Get current program for EPG IDs by parsing the raw XMLTV file."""
if not epg_ids:
return {}
now = datetime.now(timezone.utc)
try:
xml_string = await self._api_request("GET", f"{self.base_url}/output/epg", is_json=False)
root = ET.fromstring(xml_string)
current_programs = {}
for program in root.findall(".//programme"):
channel_id = program.get("channel")
if channel_id in epg_ids and channel_id not in current_programs:
start_str = program.get("start")
stop_str = program.get("stop")
if start_str and stop_str:
try:
start_time = datetime.strptime(start_str, "%Y%m%d%H%M%S %z")
stop_time = datetime.strptime(stop_str, "%Y%m%d%H%M%S %z")
if start_time <= now < stop_time:
current_programs[channel_id] = {
"title": program.findtext("title"),
"description": program.findtext("desc"),
"start_time": start_time.isoformat(),
"end_time": stop_time.isoformat(),
}
except (ValueError, TypeError):
_LOGGER.debug("Could not parse timestamp for program: %s", program.findtext("title"))
return current_programs
except Exception as e:
_LOGGER.error("Failed to parse EPG XML file: %s", e)
return {}
async def _async_update_data(self):
"""Update data via library, enriching with logo and EPG info."""
status_data = await self._api_request("GET", f"{self.base_url}/proxy/ts/status")
active_streams = status_data.get("channels", [])
if not active_streams:
return {}
active_epg_ids = list(set([
details['tvg_id']
for stream in active_streams
if (details := self.channel_details.get(stream['channel_id'])) and details.get('tvg_id')
]))
current_programs_map = await self._get_current_programs_from_xml(active_epg_ids)
enriched_streams = {}
for stream in active_streams:
stream_id = stream['channel_id']
enriched_stream = stream.copy()
details = self.channel_details.get(stream_id)
if details:
if logo_id := details.get("logo_id"):
enriched_stream["logo_url"] = f"{self.base_url}/api/channels/logos/{logo_id}/cache/"
if epg_id := details.get("tvg_id"):
enriched_stream["program"] = current_programs_map.get(epg_id)
enriched_streams[stream_id] = enriched_stream
return enriched_streams
class DispatcharrStreamManager:
"""Manages the creation and removal of stream sensors."""
def __init__(self, coordinator: DispatcharrDataUpdateCoordinator, async_add_entities: AddEntitiesCallback):
self._coordinator = coordinator
self._async_add_entities = async_add_entities
self._known_stream_ids = set()
self._coordinator.async_add_listener(self._update_sensors)
self._update_sensors()
@callback
def _update_sensors(self) -> None:
"""Update, add, or remove sensors based on coordinator data."""
if not isinstance(self._coordinator.data, dict):
current_stream_ids = set()
else:
current_stream_ids = set(self._coordinator.data.keys())
new_stream_ids = current_stream_ids - self._known_stream_ids
if new_stream_ids:
new_sensors = [DispatcharrStreamSensor(self._coordinator, stream_id) for stream_id in new_stream_ids]
self._async_add_entities(new_sensors)
self._known_stream_ids.update(new_stream_ids)
class DispatcharrTotalStreamSensor(CoordinatorEntity, SensorEntity):
"""A sensor to show the total number of active Dispatcharr streams."""
_attr_state_class = SensorStateClass.MEASUREMENT
_attr_has_entity_name = True
def __init__(self, coordinator: DispatcharrDataUpdateCoordinator):
super().__init__(coordinator)
self._attr_name = "Total Active Streams"
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_total_streams"
self._attr_icon = "mdi:play-network"
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, coordinator.config_entry.entry_id)}, name="Dispatcharr")
# --- FIX: Removed premature call to _handle_coordinator_update ---
@callback
def _handle_coordinator_update(self) -> None:
self._attr_native_value = len(self.coordinator.data or {})
self.async_write_ha_state()
class DispatcharrStreamSensor(CoordinatorEntity, SensorEntity):
"""Representation of a single, dynamic Dispatcharr stream sensor."""
_attr_should_poll = False
_attr_has_entity_name = True
def __init__(self, coordinator: DispatcharrDataUpdateCoordinator, stream_id: str):
super().__init__(coordinator)
self._stream_id = stream_id
stream_data = self.coordinator.data.get(self._stream_id, {})
name = stream_data.get("stream_name", f"Stream {self._stream_id[-6:]}")
self._attr_name = name
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{self._stream_id}"
self._attr_icon = "mdi:television-stream"
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, coordinator.config_entry.entry_id)}, name="Dispatcharr")
@property
def available(self) -> bool:
"""Return True if the stream is still in the coordinator's data."""
return super().available and self._stream_id in self.coordinator.data
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if not self.available:
self.async_write_ha_state()
return
stream_data = self.coordinator.data[self._stream_id]
program_data = stream_data.get("program") or {}
self._attr_native_value = "Streaming"
self._attr_entity_picture = stream_data.get("logo_url")
self._attr_extra_state_attributes = {
"channel_number": stream_data.get("stream_id"),
"channel_name": stream_data.get("stream_name"),
"logo_url": stream_data.get("logo_url"),
"clients": stream_data.get("client_count"),
"resolution": stream_data.get("resolution"),
"fps": stream_data.get("source_fps"),
"video_codec": stream_data.get("video_codec"),
"audio_codec": stream_data.get("audio_codec"),
"avg_bitrate": stream_data.get("avg_bitrate"),
"program_title": program_data.get("title"),
"program_description": program_data.get("description"),
"program_start": program_data.get("start_time"),
"program_stop": program_data.get("end_time"),
}
self.async_write_ha_state()