from __future__ import annotations
import asyncio
import logging
from typing import Any
from pyobs.events import RoofClosingEvent, RoofOpenedEvent
from pyobs.interfaces import IRoof
from pyobs.modules import timeout
from pyobs.utils import exceptions as exc
from pyobs.utils.enums import MotionStatus
from pyobs.utils.threads import LockWithAbort
from .baseroof import BaseRoof
log = logging.getLogger(__name__)
class DummyRoof(BaseRoof, IRoof):
"""A dummy camera for testing."""
__module__ = "pyobs.modules.roof"
_ROOF_CLOSED_PERCENTAGE = 0
_ROOF_OPEN_PERCENTAGE = 100
def __init__(self, **kwargs: Any):
"""Creates a new dummy root."""
BaseRoof.__init__(self, **kwargs)
# dummy state
self._open_percentage: int = self._ROOF_CLOSED_PERCENTAGE
self._lock_motion = asyncio.Lock()
self._abort_motion = asyncio.Event()
[docs]
async def open(self) -> None:
"""Open module."""
await BaseRoof.open(self)
await self.comm.register_event(RoofOpenedEvent)
await self.comm.register_event(RoofClosingEvent)
# publish initial status — roof starts closed/parked
await self._change_motion_status(MotionStatus.PARKED)
[docs]
@timeout(15)
async def init(self, **kwargs: Any) -> None:
"""Open the roof.
Raises:
InitError: If the roof could not be initialized (e.g. another motion is in progress).
"""
if self._is_open():
return
try:
async with LockWithAbort(self._lock_motion, self._abort_motion):
await self._change_motion_status(MotionStatus.INITIALIZING)
await self._move_roof(self._ROOF_OPEN_PERCENTAGE)
await self._change_motion_status(MotionStatus.IDLE)
await self.comm.send_event(RoofOpenedEvent())
except Exception as e:
raise exc.InitError(str(e)) from e
def _is_open(self) -> bool:
return self._open_percentage == self._ROOF_OPEN_PERCENTAGE
[docs]
@timeout(15)
async def park(self, **kwargs: Any) -> None:
"""Close the roof.
Raises:
ParkError: If the roof could not be parked (e.g. another motion is in progress).
"""
if self._is_closed():
return
try:
async with LockWithAbort(self._lock_motion, self._abort_motion):
await self._change_motion_status(MotionStatus.PARKING)
await self.comm.send_event(RoofClosingEvent())
await self._move_roof(self._ROOF_CLOSED_PERCENTAGE)
await self._change_motion_status(MotionStatus.PARKED)
except Exception as e:
raise exc.ParkError(str(e)) from e
def _is_closed(self) -> bool:
return self._open_percentage == self._ROOF_CLOSED_PERCENTAGE
async def _move_roof(self, target_pos: int) -> None:
step = 1 if target_pos > self._open_percentage else -1
while self._open_percentage != target_pos:
if self._abort_motion.is_set():
await self._change_motion_status(MotionStatus.IDLE)
return
self._open_percentage += step
await asyncio.sleep(0.1)
[docs]
def get_percent_open(self) -> float:
"""Get the percentage the roof is open."""
return self._open_percentage
[docs]
async def stop_motion(self, device: str | None = None, **kwargs: Any) -> None:
"""Stop the motion.
Args:
device: Name of device to stop, or None for all.
Raises:
DeviceBusyError: If current motion could not be aborted.
"""
await self._change_motion_status(MotionStatus.ABORTING)
async with LockWithAbort(self._lock_motion, self._abort_motion):
await self._change_motion_status(MotionStatus.IDLE)
__all__ = ["DummyRoof"]