Mutiple update towards cover state

Development
1

Hello,

I need some assistance on the below. I have a domotic system “nikobus” for which I’ve written a custom integration fdebrus/Nikobus-HA (github.com)

Complexity come from managing covers entities.

The domotic system does not have a feedback on cover position so I simulate the position based on execution time of a close or open command knowing from the configuration file how long a full close or open takes in seconds.

Nikobus supports 0x01 open / 0x02 close / 0x00 stop

Command sent from HA async_open_cover / async_close_cover / async_stop_cover

I’m adding an async_set_cover_position based on time calculation as not nativly supported by Nikobus as explained above.

and a _handle_coordinator_update that trigger at integration refresh time AND when a physical button of Nikobus would be pressed.

Does anyone as an example, code advice; good practice. because I’m strungling to make it work.

Current issue arise as a cover would be in movement and a scheduled refresh would trigger, calling _handle_coordinator_update and messing up with the state and position calculator while the cover is still moving towards desired position.

Any guidance would be greatly appreciated. direct link to the code Nikobus-HA/custom_components/nikobus/cover.py at main · fdebrus/Nikobus-HA (github.com)

Thank you

2

A few ways to tackle this (and there may be others).

  1. You could use an asyncio.lock and wait for it in your update method. Then aquire the lock when you are setting the position and release when finished. This way if an update comes in during a position movement, the update will wait until finished for that cover.

  2. Ignore updates if in_motion is true and then call an update when finished doing your position move. This will then ignore any updates for that cover and get the final state from an update when finished.

Sorry i dont have any code examples but the asyncio docs on sync primitives (which lock is one) is easy to follow.

3

Thanks for the shift reply, problem is an update could be a stop command or reverting up vs down, etc… so I still need to process the update “physical button press” but not the refresh from the coordinator.

So current way of thinking, if cover is moving and the update is from the refresh, ignore, else process.

How could I identify the source of the call to the update method? is it from a refresh of data ‘api call’ or button being pressed ?

I call a self.async_update_listeners() as a button is pressed

5

So I added the below to the coordinator

    def set_update_source(self, source: str):
        self._update_source = source

    def get_update_source(self) -> str:
        return self._update_source or "unknown"

then based on the update source, I set it

    async def initial_update_data(self):
        """Fetch the latest data from Nikobus initially"""
        try:
            _LOGGER.debug("Performing initial data refresh for Nikobus")
            self.set_update_source("api_call")
            return await self.api.refresh_nikobus_data()
        except NikobusDataError as e:
            _LOGGER.error("Error fetching Nikobus data: %s", e)
            raise UpdateFailed(f"Error fetching Nikobus data: {e}")

and finally use it in the cover.py

    @callback
    def _handle_coordinator_update(self) -> None:
        """Handle updated data from the coordinator."""
        _LOGGER.debug("Handling coordinator update.")
        current_state = self._dataservice.api.get_cover_state(self._address, self._channel)

        source = self._dataservice.get_update_source()

Based on the source, I then execute command on the cover or not.

Could you find anything better so far.