custom_components/nea_sg_weather/camera.py
async def async_camera_image(
        self, width: int | None = None, height: int | None = None
    ) -> bytes | None:
        """Return a still image response from the camera."""

        async def get_image(current_image_time: int) -> bytes | None:
            url = RAIN_MAP_URL_PREFIX + str(current_image_time) + RAIN_MAP_URL_SUFFIX
            _LOGGER.debug("Getting rain map image from %s", url)
            try:
                async_client = get_async_client(self.hass, verify_ssl=self.verify_ssl)
                response = await async_client.get(url, headers=RAIN_MAP_HEADERS)
                response.raise_for_status()
                self._last_image = response.content
                self._last_image_time = current_image_time
                self._last_image_time_pretty = datetime.strptime(
                    str(current_image_time), "%Y%m%d%H%M"
                ).isoformat()
                self._last_url = url
                # Update timestamp from external coordinator entity
                self._last_state = self.hass.states.get(self.entity_id).state
                self._last_attributes = self.hass.states.get(self.entity_id).attributes
                self._updated_attributes = dict(self._last_attributes)
                self._updated_attributes["Updated at"] = self._last_image_time_pretty
                self._updated_attributes["URL"] = url
                self.hass.states.async_set(
                    self.entity_id, self._last_state, self._updated_attributes
                )
                _LOGGER.debug(
                    "Rain map image successfully updated at %s, new URL is %s",
                    datetime.strptime(
                        str(current_image_time), "%Y%m%d%H%M"
                    ).isoformat(),
                    url,
                )
                return self._last_image

            except httpx.TimeoutException:
                _LOGGER.warning(
                    "Timeout getting camera image for %s from %s", self._name, url
                )
                return self._last_image

            except (httpx.HTTPStatusError, httpx.RequestError) as err:
                if response.status_code == 404:
                    # Image not ready, check older image urls
                    _LOGGER.debug(
                        "%s rain map image not ready, trying previous images",
                        current_image_time,
                    )

                    # minor fix for whole hours
                    if str(current_image_time)[-2:] == "00":
                        _skip_minutes = 45
                    else:
                        _skip_minutes = 5

                    if current_image_time - _skip_minutes == self._last_image_time:
                        return self._last_image
                    else:
                        return await get_image(current_image_time - _skip_minutes)
                else:
                    _LOGGER.warning(
                        "Error getting new camera image for %s from %s: %s",
                        self._name,
                        url,
                        err,
                    )
                    return self._last_image

        _current_query_time = int(
            datetime.strftime(datetime.now(timezone(timedelta(hours=8))), "%Y%m%d%H%M")
        )

        if _current_query_time != self._last_query_time:
            self._last_query_time = _current_query_time
            _current_image_time = math.floor(_current_query_time / 5) * 5
            if _current_image_time != self._last_image_time:
                return await get_image(_current_image_time)

        return self._last_image
Return a still image response from the camera.
Return a still image response from the camera.
Return a still image response from the camera .
async def async_camera_image(
        self, width: int | None = None, height: int | None = None
    ) -> bytes | None:
        async def get_image(current_image_time: int) -> bytes | None:
            url = RAIN_MAP_URL_PREFIX + str(current_image_time) + RAIN_MAP_URL_SUFFIX
            _LOGGER.debug("Getting rain map image from %s", url)
            try:
                async_client = get_async_client(self.hass, verify_ssl=self.verify_ssl)
                response = await async_client.get(url, headers=RAIN_MAP_HEADERS)
                response.raise_for_status()
                self._last_image = response.content
                self._last_image_time = current_image_time
                self._last_image_time_pretty = datetime.strptime(
                    str(current_image_time), "%Y%m%d%H%M"
                ).isoformat()
                self._last_url = url
                self._last_state = self.hass.states.get(self.entity_id).state
                self._last_attributes = self.hass.states.get(self.entity_id).attributes
                self._updated_attributes = dict(self._last_attributes)
                self._updated_attributes["Updated at"] = self._last_image_time_pretty
                self._updated_attributes["URL"] = url
                self.hass.states.async_set(
                    self.entity_id, self._last_state, self._updated_attributes
                )
                _LOGGER.debug(
                    "Rain map image successfully updated at %s, new URL is %s",
                    datetime.strptime(
                        str(current_image_time), "%Y%m%d%H%M"
                    ).isoformat(),
                    url,
                )
                return self._last_image
            except httpx.TimeoutException:
                _LOGGER.warning(
                    "Timeout getting camera image for %s from %s", self._name, url
                )
                return self._last_image
            except (httpx.HTTPStatusError, httpx.RequestError) as err:
                if response.status_code == 404:
                    _LOGGER.debug(
                        "%s rain map image not ready, trying previous images",
                        current_image_time,
                    )
                    if str(current_image_time)[-2:] == "00":
                        _skip_minutes = 45
                    else:
                        _skip_minutes = 5
                    if current_image_time - _skip_minutes == self._last_image_time:
                        return self._last_image
                    else:
                        return await get_image(current_image_time - _skip_minutes)
                else:
                    _LOGGER.warning(
                        "Error getting new camera image for %s from %s: %s",
                        self._name,
                        url,
                        err,
                    )
                    return self._last_image
        _current_query_time = int(
            datetime.strftime(datetime.now(timezone(timedelta(hours=8))), "%Y%m%d%H%M")
        )
        if _current_query_time != self._last_query_time:
            self._last_query_time = _current_query_time
            _current_image_time = math.floor(_current_query_time / 5) * 5
            if _current_image_time != self._last_image_time:
                return await get_image(_current_image_time)
        return self._last_image
async def async_camera_image ( self , width : int | None = None , height : int | None = None ) -> bytes | None : async def get_image ( current_image_time : int ) -> bytes | None : url = RAIN_MAP_URL_PREFIX + str ( current_image_time ) + RAIN_MAP_URL_SUFFIX _LOGGER . debug ( "Getting rain map image from %s" , url ) try : async_client = get_async_client ( self . hass , verify_ssl = self . verify_ssl ) response = await async_client . get ( url , headers = RAIN_MAP_HEADERS ) response . raise_for_status ( ) self . _last_image = response . content self . _last_image_time = current_image_time self . _last_image_time_pretty = datetime . strptime ( str ( current_image_time ) , "%Y%m%d%H%M" ) . isoformat ( ) self . _last_url = url self . _last_state = self . hass . states . get ( self . entity_id ) . state self . _last_attributes = self . hass . states . get ( self . entity_id ) . attributes self . _updated_attributes = dict ( self . _last_attributes ) self . _updated_attributes [ "Updated at" ] = self . _last_image_time_pretty self . _updated_attributes [ "URL" ] = url self . hass . states . async_set ( self . entity_id , self . _last_state , self . _updated_attributes ) _LOGGER . debug ( "Rain map image successfully updated at %s, new URL is %s" , datetime . strptime ( str ( current_image_time ) , "%Y%m%d%H%M" ) . isoformat ( ) , url , ) return self . _last_image except httpx . TimeoutException : _LOGGER . warning ( "Timeout getting camera image for %s from %s" , self . _name , url ) return self . _last_image except ( httpx . HTTPStatusError , httpx . RequestError ) as err : if response . status_code == 404 : _LOGGER . debug ( "%s rain map image not ready, trying previous images" , current_image_time , ) if str ( current_image_time ) [ - 2 : ] == "00" : _skip_minutes = 45 else : _skip_minutes = 5 if current_image_time - _skip_minutes == self . _last_image_time : return self . _last_image else : return await get_image ( current_image_time - _skip_minutes ) else : _LOGGER . warning ( "Error getting new camera image for %s from %s: %s" , self . _name , url , err , ) return self . _last_image _current_query_time = int ( datetime . strftime ( datetime . now ( timezone ( timedelta ( hours = 8 ) ) ) , "%Y%m%d%H%M" ) ) if _current_query_time != self . _last_query_time : self . _last_query_time = _current_query_time _current_image_time = math . floor ( _current_query_time / 5 ) * 5 if _current_image_time != self . _last_image_time : return await get_image ( _current_image_time ) return self . _last_image
Return a still image response from the camera.
Return a still image response from the camera .
"""Return a still image response from the camera.""" # Update timestamp from external coordinator entity # Image not ready, check older image urls # minor fix for whole hours
self
width
int | None
height
int | None