Skip to content

Commit

Permalink
unittest
Browse files Browse the repository at this point in the history
  • Loading branch information
isotuela committed Jul 10, 2023
1 parent a83ddcb commit c093cc8
Show file tree
Hide file tree
Showing 2 changed files with 177 additions and 28 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
__all__ = ["PowerOffATCalSys"]

import asyncio
import time as time

from lsst.ts.idl.enums import ATWhiteLight

Expand Down Expand Up @@ -50,7 +49,7 @@ def __init__(self, index, add_remotes: bool = True):
self.white_light_source = None

# White lamp config
self.timeout_lamp_cool_down = 60 * 15
self.timeout_lamp_cool_down = 60 * 16
self.cmd_timeout = 30

@classmethod
Expand All @@ -71,47 +70,58 @@ async def run(self):
await self.assert_components_enabled()

await self.checkpoint("Turning lamp off")
await self.white_light_source.cmd_turnLampOff.set_start(
timeout=self.timeout_lamp_cool_down
)
await self.switch_lamp_off()

await self.checkpoint("Closing the shutter")
await self.white_light_source.cmd_closeShutter.start()
await self.white_light_source.cmd_closeShutter.start(
timeout=self.timeout_close_shutter
)

await self.checkpoint("Waiting for lamp to cool down")
await self.wait_for_lamp_to_cool_down()

await self.checkpoint("Stopping chiller")
await self.white_light_source.cmd_stopChiller.set_start(
timeout=self.cmd_timeout
await self.white_light_source.cmd_stopChiller.start(timeout=self.cmd_timeout)

async def switch_lamp_off(self):
await self.white_light_source.evt_lampState.flush()

await self.white_light_source.cmd_turnLampOff.start(
timeout=self.timeout_lamp_cool_down
)

async def wait_for_lamp_to_cool_down(self):
self.white_light_source.evt_lampState.flush()
start_time_lamp_warm_up = time.time()
while time.time() - start_time_lamp_warm_up < self.timeout_lamp_warm_up:
lamp_state = await self.white_light_source.evt_lampState.next(
flush=False, timeout=self.timeout_lamp_warm_up
)
if lamp_state.basicState == ATWhiteLight.LampBasicState.WARMUP:
warmup_time_left = (
lamp_state.warmupEndTime - lamp_state.private_sndStamp
)
self.log.info(f"Time Left for lamp warmup: {warmup_time_left} min.")
await asyncio.sleep(self.track_lamp_warmup)
lamp_state = await self.white_light_source.evt_lampState.aget(
timeout=self.timeout_lamp_cool_down
)
self.log.info(
f"Lamp state: {ATWhiteLight.LampBasicState(lamp_state.basicState)!r}."
)

# Now enters the loop with the stop condition being
# that the lamp is OFF

elif lamp_state.basicState == ATWhiteLight.LampBasicState.ON:
while lamp_state.basicState != ATWhiteLight.LampBasicState.OFF:
try:
lamp_state = await self.white_light_source.evt_lampState.next(
flush=False, timeout=self.timeout_lamp_cool_down
)
cool_down_wait_time = (
lamp_state.cooldownEndTime - lamp_state.private_sndStamp
)
self.log.info(
f"White Light Lamp is on after a warm up of {time.time() - start_time_lamp_warm_up} s"
f"Lamp state: {ATWhiteLight.LampBasicState(lamp_state.basicState)!r}."
f"Need to wait {cool_down_wait_time/60} min"
)
asyncio.sleep(60)
except asyncio.TimeoutError:
raise RuntimeError(
f"White Light Lamp failed to turn off after {self.timeout_lamp_warm_up} s."
)
else:
raise TimeoutError(
f"White Light Lamp failed to turn on after {self.timeout_lamp_warm_up} s"
)

async def assert_components_enabled(self):
"""Check if ATWhiteLight and ATMonochromator are ENABLED"""
for comp in [self.white_light_source, self.monochromator]:
"""Check if ATWhiteLight is ENABLED"""
for comp in [self.white_light_source]:
summary_state = await comp.evt_summaryState.aget()
if salobj.State(summary_state.summaryState) != salobj.State(
salobj.State.ENABLED
Expand Down
139 changes: 139 additions & 0 deletions tests/test_auxtel_power_off_atcalsys.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
# This file is part of ts_standardscripts
#
# Developed for the LSST Telescope and Site Systems.
# This product includes software developed by the LSST Project
# (https://www.lsst.org).
# See the COPYRIGHT file at the top-level directory of this distribution
# for details of code ownership.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.

import logging
import random
import unittest
import types
import asyncio

from lsst.ts import salobj, utils

from lsst.ts import standardscripts
from lsst.ts.standardscripts.auxtel.calibrations import PowerOffATCalSys
from lsst.ts.idl.enums import ATWhiteLight


random.seed(47) # for set_random_lsst_dds_partition_prefix

logging.basicConfig()


class TestPowerOffATCalSys(
standardscripts.BaseScriptTestCase, unittest.IsolatedAsyncioTestCase
):
async def basic_make_script(self, index):
self.script = PowerOffATCalSys(index=index, add_remotes=False)

self.lamp_state = types.SimpleNamespace(
basicState=ATWhiteLight.LampBasicState.ON
)
self.shutter_status = types.SimpleNamespace(
shutterState=ATWhiteLight.ShutterState.OPEN
)

await self.configure_mocks()

return [
self.script,
]

async def configure_mocks(self):
self.script.white_light_source = unittest.mock.AsyncMock()
self.script.white_light_source.start_task = utils.make_done_future()

# Configure mocks

self.script.white_light_source.configure_mock(
**{
"evt_summaryState.aget.side_effect": self.mock_get_whitelightsource_summary_state,
"cmd_turnLampOff.start.side_effect": self.mock_get_lamp_status,
"cmd_closeShutter.start.side_effect": self.mock_close_shutter,
}
)

# Mock check methods
self.script.wait_for_lamp_to_cool_down = unittest.mock.AsyncMock(
side_effect=self.mock_lamp_temp
)

# Summary State

async def mock_get_whitelightsource_summary_state(self, **kwargs):
return types.SimpleNamespace(summaryState=salobj.State.ENABLED)

# Lamp

async def mock_get_lamp_status(self, **kwargs):
await asyncio.sleep(0.5)
return self.lamp_state

async def mock_lamp_temp(self, **kwargs):
self.lamp_state.basicState = ATWhiteLight.LampBasicState.TURNING_OFF
await asyncio.sleep(15.0)
self.lamp_state.basicState = ATWhiteLight.LampBasicState.OFF

# Shutter
async def mock_close_shutter(self, **kwargs):
types.SimpleNamespace(shutterState=ATWhiteLight.ShutterState.OPEN)
await asyncio.sleep(3)
self.shutter_status = types.SimpleNamespace(
shutterState=ATWhiteLight.ShutterState.CLOSED
)

async def test_run_without_without_failures(self):
async with self.make_script():
await self.configure_script()

await self.run_script()

# Summary State
self.script.white_light_source.evt_summaryState.aget.assert_awaited_once()

# White lamp
self.script.white_light_source.cmd_turnLampOff.start.assert_awaited_with(
timeout=self.script.timeout_lamp_cool_down,
)

self.script.wait_for_lamp_to_cool_down.assert_awaited_once()

# Shutter
self.script.white_light_source.cmd_openShutter.start.assert_awaited_with(
timeout=self.script.timeout_close_shutter,
)

# Chiller
self.script.white_light_source.cmd_stopChiller.start.assert_awaited_once_with(
timeout=self.script.script.cmd_timeout
)

# Check status
assert self.lamp_state.basicState == ATWhiteLight.LampBasicState.OFF
assert self.shutter_status.shutterState == ATWhiteLight.ShutterState.CLOSED

async def test_executable(self):
scripts_dir = standardscripts.get_scripts_dir()
script_path = scripts_dir / "auxtel" / "calibrations" / "power_off_atcalsys.py"
await self.check_executable(script_path)


if __name__ == "__main__":
unittest.main()

0 comments on commit c093cc8

Please sign in to comment.