__init__.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402
  1. """Support to check for available updates."""
  2. from __future__ import annotations
  3. import asyncio, logging, async_timeout, os, voluptuous as vol
  4. from awesomeversion import AwesomeVersion
  5. from datetime import timedelta
  6. from homeassistant.const import __version__ as current_version
  7. from homeassistant.helpers import discovery, update_coordinator
  8. from homeassistant.helpers.aiohttp_client import async_get_clientsession
  9. import homeassistant.helpers.config_validation as cv
  10. _LOGGER = logging.getLogger(__name__)
  11. _log_wmsg = """
  12. NOTE! Using a replacement (custom) 'updater' component for the snap package.
  13. Do NOT report bugs of any kind related to 'updater' to the Home Assistant core project.
  14. Report any issues at https://github.com/home-assistant-snap/home-assistant-snap/issues"""
  15. ATTR_RELEASE_NOTES = "release_notes"
  16. ATTR_UPDATE_NOTES = "update_notes"
  17. ATTR_NEWEST_VERSION = "newest_version"
  18. # Keeping for consistency (we're overriding)
  19. CONF_REPORTING = "reporting"
  20. CONF_COMPONENT_REPORTING = "include_used_components"
  21. DOMAIN = "updater"
  22. UPDATER_URL = "https://api.snapcraft.io/v2/snaps/info/home-assistant-snap?architecture=%s&fields=channel-map,revision,version"
  23. RESPONSE_SCHEMA = vol.Schema(
  24. {
  25. vol.Required("channel-map"): cv.ensure_list,
  26. vol.Required("default-track"): cv.string,
  27. },
  28. extra=vol.REMOVE_EXTRA,
  29. )
  30. class Channel:
  31. def __init__(self, track: Track, channel: dict, revision: int, version: str):
  32. self.__arch = channel['architecture']
  33. self.__risk = 0
  34. self.__track = track
  35. self.__risks = {'stable': 4, 'candidate': 3, 'beta': 2, 'edge': 1}
  36. if channel['risk'] in self.__risks:
  37. self.__risk = self.__risks[channel['risk']]
  38. self.__revision = revision
  39. self.__version = AwesomeVersion(version)
  40. def __str__(self) -> str:
  41. return f"{str(self.__track)}/{self.get_risk()}"
  42. def __repr__(self) -> str:
  43. return f"{self.__version}, revision: {self.__revision}, channel: {str(self)}"
  44. def get_track(self) -> Track:
  45. return self.__track
  46. def get_risk(self, as_str: bool = True) -> str|int:
  47. if not as_str:
  48. return self.__risk
  49. for v,k in self.__risks.items():
  50. if k == self.__risk:
  51. return v
  52. return self.__risk
  53. def get_revision(self) -> int:
  54. return self.__revision
  55. def get_version(self) -> AwesomeVersion:
  56. return self.__version
  57. def __gt__(self, other: Channel) -> bool:
  58. if self.__track == 'latest':
  59. return False
  60. elif other.get_track() == 'latest':
  61. return False
  62. return self.__revision > other.get_revision() and self.__risk >= other.get_risk(False)
  63. class Track:
  64. def __init__(self, track: str):
  65. self.__track = AwesomeVersion(track)
  66. self.__channels = []
  67. def get_track(self) -> str:
  68. return self.__track
  69. def __eq__(self, other: Track|str) -> bool:
  70. if isinstance(other, Track):
  71. return self.get_track() == other.get_track()
  72. return self.get_track() == other
  73. def add_channel(self, channel) -> Track:
  74. self.__channels.append(Channel(self, channel['channel'], channel['revision'], channel['version']))
  75. self.__channels.sort(key=lambda x: x.get_risk(False))
  76. return self
  77. def get_channels(self) -> list:
  78. return self.__channels
  79. def channel_with_revision(self, revision: int) -> Channel|None:
  80. for channel in self.__channels:
  81. if channel.get_revision() == revision:
  82. return channel
  83. return None
  84. def channel_with_higher_revision(self, channel: Channel) -> Channel|None:
  85. newest = channel
  86. for channel in self.__channels:
  87. if channel > newest:
  88. newest = channel
  89. return newest
  90. def get_latest(self) -> Channel|None:
  91. if len(self.__channels) == 0:
  92. return None
  93. return self.__channels[len(self.__channels)-1]
  94. def __repr__(self) -> str:
  95. risks = []
  96. for channel in self.__channels:
  97. risks.append(f"{channel.get_risk()}/{channel.get_revision()}")
  98. return f"{self.__track}({', '.join(risks)})"
  99. def __str__(self) -> str:
  100. return str(self.__track)
  101. class Tracks:
  102. def __init__(self, channel_map: list) -> None:
  103. self.__tracks = []
  104. for channel in channel_map:
  105. track = self.get_track(channel['channel']['track'])
  106. if track is not None:
  107. track.add_channel(channel)
  108. else:
  109. track = Track(channel['channel']['track'])
  110. track.add_channel(channel)
  111. self.__tracks.append(track)
  112. self.__tracks.sort(key=lambda x: x.get_track())
  113. def get_latest(self) -> Track|None:
  114. self.__tracks.reverse()
  115. latest = None
  116. for track in self.__tracks:
  117. if len(track.get_channels()) != 0 and track.get_latest().get_risk() == 'latest':
  118. latest = track.get_latest()
  119. break
  120. self.__tracks.reverse()
  121. return latest
  122. def get_track(self, track: str) -> Track|None:
  123. for t in self.__tracks:
  124. if t == track:
  125. return t
  126. return None
  127. def find_for_revision(self, revision: int) -> Channel|None:
  128. for track in self.__tracks:
  129. channel = track.channel_with_revision(revision)
  130. if channel is not None:
  131. return channel
  132. return None
  133. def channel_with_higher_revision(self, channel: Channel) -> Channel|None:
  134. for track in self.__tracks:
  135. for chan in track.get_channels():
  136. if chan > channel:
  137. return chan
  138. return None
  139. def track_with_lower_revision(self, revision: int) -> Channel|None:
  140. closest = None
  141. for track in self.__tracks:
  142. for chan in track.get_channels():
  143. if chan.get_revision() < revision:
  144. if closest is not None and closest < chan:
  145. closest = chan
  146. else:
  147. closest = chan
  148. return closest
  149. def __str__(self) -> str:
  150. tracks = []
  151. for track in self.__tracks:
  152. tracks.append(str(track))
  153. return ", ".join(tracks)
  154. def __repr__(self) -> str:
  155. return self.__str__()
  156. class Updater:
  157. """ Updater class for data exchange."""
  158. def __init__(self, update_available: bool, default: Channel, current: Channel|None, newer: Channel|None, update_notes: str) -> None:
  159. self.update_available = update_available
  160. self.newest_version = str(newer.get_version()) if newer is not None else str(default.get_version())
  161. self.update_notes = update_notes
  162. if isinstance(newer, Channel) and default > newer:
  163. self.update_notes += f"\n\nLatest channel is: _«{repr(default)}»_."
  164. elif newer is None and isinstance(current, Channel) and default > current:
  165. self.update_notes += f"\n\nLatest channel is: _«{repr(default)}»_. Upgrade with: `snap switch home-assistant-snap --channel={default}`"
  166. elif current is None and newer is None:
  167. self.update_notes += f"\n\nLatest channel is: _«{repr(default)}»_."
  168. if update_available:
  169. _LOGGER.info("UPDATE AVAILABLE: %s, newer: %s, current: %s, default: %s, notes: %s", update_available, newer, current, default, self.update_notes)
  170. self.release_notes = "https://www.home-assistant.io/blog/categories/core/"
  171. async def async_setup(hass, config):
  172. conf = config.get(DOMAIN, {})
  173. _LOGGER.warning(_log_wmsg)
  174. # Keeping for consistency (we're overriding)
  175. for option in (CONF_COMPONENT_REPORTING, CONF_REPORTING):
  176. if option in conf:
  177. _LOGGER.warning(
  178. "Analytics reporting with the option '%s' "
  179. "is deprecated and you should remove that from your configuration. "
  180. "The analytics part of this integration has moved to the new 'analytics' integration",
  181. option,
  182. )
  183. async def check_new_version() -> Updater:
  184. _LOGGER.warning(_log_wmsg)
  185. snap_rev = os.getenv('SNAP_REVISION')
  186. tracks, default_track = await get_versions(hass)
  187. if snap_rev is None:
  188. if AwesomeVersion(current_version).dev:
  189. snap_rev = 327
  190. _LOGGER.warning(f"Development, using SNAP_REVISION: {snap_rev}")
  191. else:
  192. raise update_coordinator.UpdateFailed(Exception("Missing SNAP_REVISION environment variable."))
  193. if f"{snap_rev[0] if type(snap_rev) is str else 'y'}" == 'x':
  194. c_v = AwesomeVersion(current_version)
  195. track = tracks.get_track(f"{c_v.section(0)}.{c_v.section(1)}")
  196. if track is not None and len(track.get_channels()) != 0:
  197. xsnap_rev = track.get_latest().get_revision()
  198. else:
  199. xsnap_rev = default_track.get_latest().get_revision()
  200. _LOGGER.warning(f"Locally built ({snap_rev}), using SNAP_REVISION: {xsnap_rev}")
  201. snap_rev = xsnap_rev
  202. snap_rev = int(snap_rev)
  203. current_channel = tracks.find_for_revision(snap_rev)
  204. """
  205. NOTE: This is just predictions - as a revision of a snap might be in several channels,
  206. and always in latest. Therefore you can be on latest and reciving notification on new
  207. releases in another channel, if they have the same revision available
  208. """
  209. if current_channel.get_track() == "latest":
  210. _LOGGER.warning(
  211. f"You're on the channel «{current_channel}», please consider switch to «{default_track.get_latest()}». "
  212. f"Switch with: sudo snap switch --channel={default_track.get_latest()}"
  213. f"Staying on {current_channel} will auto-upgrade your Home Assistant instance, which "
  214. f"can cause your Home Assistant instance to stop working as of breaking changes."
  215. )
  216. return Updater(False, default_track.get_latest(), current_channel, None,
  217. f"You're on the channel «{current_channel}», please consider switch to «{default_track.get_latest()}». "
  218. f"Switch with: sudo snap switch --channel={default_track.get_latest()}"
  219. f"Staying on {current_channel} will auto-upgrade your Home Assistant instance, which "
  220. f"can cause your Home Assistant instance to stop working as of breaking changes."
  221. )
  222. if current_channel is not None:
  223. newer_channel = current_channel.get_track().channel_with_higher_revision(current_channel)
  224. if newer_channel is not None and newer_channel > current_channel:
  225. return Updater(True, default_track.get_latest(), current_channel, newer_channel,
  226. f"You're currently on _«{repr(current_channel)}»_ and can upgrade to _«{repr(newer_channel)}»_. "
  227. f"Update with `sudo snap switch home-assistant-snap --channel={newer_channel}`."
  228. )
  229. newer_channel = tracks.channel_with_higher_revision(current_channel)
  230. if newer_channel is not None and newer_channel > current_channel:
  231. return Updater(True, default_track.get_latest(), current_channel, newer_channel,
  232. f"You're currently on _«{repr(current_channel)}»_ and can upgrade to _«{repr(newer_channel)}»_. "
  233. f"Update with `sudo snap switch home-assistant-snap --channel={newer_channel}`."
  234. )
  235. return Updater(False, default_track.get_latest(), None, current_channel, f"You're on _«{repr(current_channel)}»_!")
  236. else:
  237. c_v = AwesomeVersion(current_version)
  238. track = tracks.get_track(f"{c_v.section(0)}.{c_v.section(1)}")
  239. if track is not None and track.get_latest() is not None:
  240. current_channel = track.get_latest()
  241. newer_channel = tracks.channel_with_higher_revision(current_channel)
  242. if newer_channel is not None and newer_channel > current_channel:
  243. return Updater(True, default_track.get_latest(), current_channel, newer_channel,
  244. f"Unknown revision «{snap_rev}», assuming on any channel for {current_channel.get_track()}. The snap package "
  245. f"should automatically update, but you can also upgrade to _«{repr(newer_channel)}»_ with: "
  246. f"`sudo snap switch home-assistant-snap --channel={newer_channel}`."
  247. )
  248. return Updater(True, default_track.get_latest(), current_channel, None,
  249. f"Unknown revision «{snap_rev}», assuming on any channel for track {current_channel.get_track()}. The snap package "
  250. f"should automatically update, but double check that the channel is not closed. You can force the update with: "
  251. f"`sudo snap refresh home-assistant-snap` and find channels with: `sudo info home-assistant-snap`."
  252. )
  253. older_track = tracks.track_with_lower_revision(snap_rev)
  254. if older_track is not None:
  255. newer_channel = tracks.channel_with_higher_revision(older_track)
  256. if newer_channel is not None:
  257. newer_channel = newer_channel.get_track().get_latest()
  258. return Updater(True, default_track.get_latest(), None, newer_channel,
  259. f"No channel found for {c_v.section(0)}.{c_v.section(1)}, it might have been deleted - and you will not receive updates. "
  260. f"A newer channel _«{repr(newer_channel)}»_ is available! "
  261. f"You can switch with: `sudo snap refresh home-assistant-snap --channel={newer_channel}`."
  262. )
  263. return Updater(True, default_track.get_latest(), None, None,
  264. f"No channel found for «{snap_rev}» ({c_v.section(0)}.{c_v.section(1)}). "
  265. f"Please consult `snap info home-assistant-snap` to find a suitable track to upgrade to, "
  266. f"and switch channel with: `snap switch home-assistant-snap --channel=<channel>`."
  267. )
  268. coordinator = hass.data[DOMAIN] = update_coordinator.DataUpdateCoordinator[Updater](
  269. hass,
  270. _LOGGER,
  271. name="Home Assistant Snap update",
  272. update_method=check_new_version,
  273. update_interval=timedelta(days=1)
  274. )
  275. asyncio.create_task(coordinator.async_refresh())
  276. hass.async_create_task(
  277. discovery.async_load_platform(hass, 'binary_sensor', DOMAIN, {}, config)
  278. )
  279. return True
  280. from urllib.parse import urlparse
  281. async def get_versions(hass):
  282. session = async_get_clientsession(hass)
  283. snap_arch = os.getenv('SNAP_ARCH')
  284. if snap_arch is None:
  285. if AwesomeVersion(current_version).dev:
  286. snap_arch = "amd64"
  287. else:
  288. raise update_coordinator.UpdateFailed(Exception("Missing SNAP_ARCH environment variable."))
  289. with async_timeout.timeout(45):
  290. req = await session.get(UPDATER_URL % snap_arch, headers={
  291. 'Snap-Device-Series': '16'
  292. })
  293. try:
  294. res = await req.json()
  295. except ValueError as err:
  296. raise update_coordinator.UpdateFailed(
  297. f"Received invalid JSON from {urlparse(UPDATER_URL).netloc}"
  298. ) from err
  299. try:
  300. res = RESPONSE_SCHEMA(res)
  301. tracks = Tracks(res['channel-map'])
  302. default_track = res['default-track'] if 'default-track' in res else None
  303. if default_track is None:
  304. default_track = tracks.get_latest()
  305. else:
  306. default_track = tracks.get_track(default_track)
  307. return [tracks, default_track]
  308. except vol.Invalid as err:
  309. raise update_coordinator.UpdateFailed(
  310. f"Got unexepected response: {err}"
  311. ) from err