Clean up twitch (#67595)

This commit is contained in:
Robert Hillis 2022-03-15 13:33:16 -04:00 committed by GitHub
parent 668e0c19a2
commit 0c4efae31f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -62,13 +62,16 @@ def setup_platform(
client_id = config[CONF_CLIENT_ID] client_id = config[CONF_CLIENT_ID]
client_secret = config[CONF_CLIENT_SECRET] client_secret = config[CONF_CLIENT_SECRET]
oauth_token = config.get(CONF_TOKEN) oauth_token = config.get(CONF_TOKEN)
client = Twitch(app_id=client_id, app_secret=client_secret)
client.auto_refresh_auth = False
try: try:
client.authenticate_app(scope=OAUTH_SCOPES) client = Twitch(
app_id=client_id,
app_secret=client_secret,
target_app_auth_scope=OAUTH_SCOPES,
)
client.auto_refresh_auth = False
except TwitchAuthorizationException: except TwitchAuthorizationException:
_LOGGER.error("INvalid client ID or client secret") _LOGGER.error("Invalid client ID or client secret")
return return
if oauth_token: if oauth_token:
@ -86,7 +89,7 @@ def setup_platform(
channels = client.get_users(logins=channels) channels = client.get_users(logins=channels)
add_entities( add_entities(
[TwitchSensor(channel=channel, client=client) for channel in channels["data"]], [TwitchSensor(channel, client) for channel in channels["data"]],
True, True,
) )
@ -94,80 +97,38 @@ def setup_platform(
class TwitchSensor(SensorEntity): class TwitchSensor(SensorEntity):
"""Representation of an Twitch channel.""" """Representation of an Twitch channel."""
def __init__(self, channel, client: Twitch): _attr_icon = ICON
def __init__(self, channel: dict[str, str], client: Twitch) -> None:
"""Initialize the sensor.""" """Initialize the sensor."""
self._client = client self._client = client
self._channel = channel
self._enable_user_auth = client.has_required_auth(AuthType.USER, OAUTH_SCOPES) self._enable_user_auth = client.has_required_auth(AuthType.USER, OAUTH_SCOPES)
self._state = None self._attr_name = channel["display_name"]
self._preview = None self._attr_unique_id = channel["id"]
self._game = None
self._title = None
self._subscription = None
self._follow = None
self._statistics = None
@property def update(self) -> None:
def name(self):
"""Return the name of the sensor."""
return self._channel["display_name"]
@property
def native_value(self):
"""Return the state of the sensor."""
return self._state
@property
def entity_picture(self):
"""Return preview of current game."""
return self._preview
@property
def extra_state_attributes(self):
"""Return the state attributes."""
attr = dict(self._statistics)
if self._enable_user_auth:
attr.update(self._subscription)
attr.update(self._follow)
if self._state == STATE_STREAMING:
attr.update({ATTR_GAME: self._game, ATTR_TITLE: self._title})
return attr
@property
def unique_id(self):
"""Return unique ID for this sensor."""
return self._channel["id"]
@property
def icon(self):
"""Icon to use in the frontend, if any."""
return ICON
def update(self):
"""Update device state.""" """Update device state."""
followers = self._client.get_users_follows(to_id=self._channel["id"])["total"] followers = self._client.get_users_follows(to_id=self.unique_id)["total"]
channel = self._client.get_users(user_ids=[self._channel["id"]])["data"][0] channel = self._client.get_users(user_ids=[self.unique_id])["data"][0]
self._statistics = { self._attr_extra_state_attributes = {
ATTR_FOLLOWING: followers, ATTR_FOLLOWING: followers,
ATTR_VIEWS: channel["view_count"], ATTR_VIEWS: channel["view_count"],
} }
if self._enable_user_auth: if self._enable_user_auth:
user = self._client.get_users()["data"][0] user = self._client.get_users()["data"][0]["id"]
subs = self._client.check_user_subscription( subs = self._client.check_user_subscription(
user_id=user["id"], broadcaster_id=self._channel["id"] user_id=user, broadcaster_id=self.unique_id
) )
if "data" in subs: if "data" in subs:
self._subscription = { self._attr_extra_state_attributes[ATTR_SUBSCRIPTION] = True
ATTR_SUBSCRIPTION: True, self._attr_extra_state_attributes[ATTR_SUBSCRIPTION_GIFTED] = subs[
ATTR_SUBSCRIPTION_GIFTED: subs["data"][0]["is_gift"], "data"
} ][0]["is_gift"]
elif "status" in subs and subs["status"] == 404: elif "status" in subs and subs["status"] == 404:
self._subscription = {ATTR_SUBSCRIPTION: False} self._attr_extra_state_attributes[ATTR_SUBSCRIPTION] = False
elif "error" in subs: elif "error" in subs:
raise Exception( raise Exception(
f"Error response on check_user_subscription: {subs['error']}" f"Error response on check_user_subscription: {subs['error']}"
@ -176,23 +137,22 @@ class TwitchSensor(SensorEntity):
raise Exception("Unknown error response on check_user_subscription") raise Exception("Unknown error response on check_user_subscription")
follows = self._client.get_users_follows( follows = self._client.get_users_follows(
from_id=user["id"], to_id=self._channel["id"] from_id=user, to_id=self.unique_id
)["data"] )["data"]
if len(follows) > 0: self._attr_extra_state_attributes[ATTR_FOLLOW] = len(follows) > 0
self._follow = { if len(follows):
ATTR_FOLLOW: True, self._attr_extra_state_attributes[ATTR_FOLLOW_SINCE] = follows[0][
ATTR_FOLLOW_SINCE: follows[0]["followed_at"], "followed_at"
} ]
else:
self._follow = {ATTR_FOLLOW: False}
streams = self._client.get_streams(user_id=[self._channel["id"]])["data"] if streams := self._client.get_streams(user_id=[self.unique_id])["data"]:
if len(streams) > 0:
stream = streams[0] stream = streams[0]
self._game = stream["game_name"] self._attr_native_value = STATE_STREAMING
self._title = stream["title"] self._attr_extra_state_attributes[ATTR_GAME] = stream["game_name"]
self._preview = stream["thumbnail_url"] self._attr_extra_state_attributes[ATTR_TITLE] = stream["title"]
self._state = STATE_STREAMING self._attr_entity_picture = stream["thumbnail_url"]
else: else:
self._preview = channel["offline_image_url"] self._attr_native_value = STATE_OFFLINE
self._state = STATE_OFFLINE self._attr_extra_state_attributes[ATTR_GAME] = None
self._attr_extra_state_attributes[ATTR_TITLE] = None
self._attr_entity_picture = channel["offline_image_url"]