Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add power command #169

Merged
merged 5 commits into from
Dec 5, 2021
Merged

Add power command #169

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a7a08ec
Always output status of IEB in status command
albireox Dec 5, 2021
cf1b07c
Move the ieb power and ieb switch command to power
albireox Dec 5, 2021
3c13c73
Update changelog
albireox Dec 5, 2021
08d82f7
Do not send command in status if no positioners
albireox Dec 5, 2021
46449bd
Fix outputing category
albireox Dec 5, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

* [#163](https://github.com/sdss/jaeger/issues/163) The to and from destination trajectories are saved when `BaseConfiguration.decollide_and_get_paths()` is called. The reverse path can be sent from the actor using `configuration reverse`. The paths can be generated in advance when loading the design. An ``--epoch-delay`` parameter can be passed when loading the design to create a configuration for some time in the future.
* [#167](https://github.com/sdss/jaeger/issues/167) Add the ability of loading a configuration from the current positions of the robots.
* [#169](https://github.com/sdss/jaeger/issues/169) Move `ieb power` and `ieb switch` to simply `power`.

### ✨ Improved

Expand Down
1 change: 1 addition & 0 deletions python/jaeger/actor/commands/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
from .ieb import *
from .pollers import *
from .positioner import *
from .power import *
from .snapshot import *
from .talk import *
from .unwind import *
327 changes: 7 additions & 320 deletions python/jaeger/actor/commands/ieb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,55 +6,27 @@
# @Filename: ieb.py
# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)

import asyncio
import math
from __future__ import annotations

from typing import Tuple
from typing import TYPE_CHECKING, Tuple

import click

from clu.command import Command
from clu.parsers.click import pass_args

from jaeger import config
from jaeger.fps import FPS
from jaeger.ieb import IEB
from jaeger.testing import VirtualFPS
from jaeger.ieb import IEB, _get_category_data

from .. import JaegerActor
from . import jaeger_parser


__all__ = ["ieb"]


async def _get_category_data(command, category) -> list:

ieb = command.actor.fps.ieb
schema = command.actor.model.schema
if TYPE_CHECKING:
from clu.command import Command

items = schema["properties"][category]["items"]
measured = []
from .. import JaegerActor

async with ieb:
for item in items:
name = item["title"]
type_ = item["type"]
device = ieb.get_device(name)
value = (await device.read(connect=False))[0]
if type_ == "boolean" and device.__type__ == "relay":
value = True if value == "closed" else False
elif type_ == "integer":
value = int(value)
elif type_ == "number":
if "multipleOf" in item:
precision = int(-math.log10(item["multipleOf"]))
else:
precision = 3
value = round(value, precision)
measured.append(value)

return measured
__all__ = ["ieb"]


@jaeger_parser.group()
Expand Down Expand Up @@ -98,291 +70,6 @@ async def status(command, fps):
return command.finish(message=status_data, concatenate=True)


@ieb.command()
@click.argument("DEVICES", type=str, nargs=-1)
@click.option(
"--on/--off",
default=None,
help="the desired state of the device. If not passed, switches the current state.",
)
@click.option(
"--cycle",
is_flag=True,
help="power cycles a relay. The final status is on.",
)
@click.option(
"--delay",
type=float,
default=1,
help="When powering multiple devices, the delay to wait between them.",
)
@click.option(
"--force",
is_flag=True,
help="Forces a device to turn on/off even if disabled..",
)
async def switch(command, fps, devices=(), on=None, cycle=False, delay=1, force=False):
"""Switches the status of an on/off device."""

ieb = fps.ieb

if len(devices) == 0:
return command.fail(error="No devices specified.")

for idev, device in enumerate(devices):

if device.upper() in config["ieb"]["disabled_devices"]:
if force is False:
command.warning(text=f"{device} is disabled. Skipping.")
continue
else:
command.warning(text=f"{device} is disabled but overriding.")

if len(devices) > 1 and idev != 0:
await asyncio.sleep(delay)

if cycle:
on = False

try:
device_obj = ieb.get_device(device)
dev_name = device_obj.name
category = device_obj.category.lower()
except ValueError:
return command.fail(error=f"cannot find device {device!r}.")

if device_obj.module.mode not in ["holding_register", "coil"]:
return command.fail(error=f"{dev_name!r} is not an output.")

if on is None: # The --on/--off was not passed
current_status = (await device_obj.read())[0]
if current_status == "closed":
on = False
elif current_status == "open":
on = True
else:
return command.fail(
error=f"invalid status for device {dev_name!r}: {current_status!r}."
)

try:
if on is True:
await device_obj.close()
elif on is False:
await device_obj.open()
except Exception:
return command.fail(error=f"failed to set status of device {dev_name!r}.")

if cycle:
command.write("d", text="waiting 1 second before powering up.")
await asyncio.sleep(1)
try:
await device_obj.close()
except Exception:
return command.fail(
error=f"failed to power device {dev_name!r} back on."
)

# If we read the status immediately sometimes we still get the old one.
# Sleep a bit to avoid that.
await asyncio.sleep(0.2)

status = "on" if (await device_obj.read())[0] == "closed" else "off"

command.info(
message={
"text": f"device {dev_name!r} is now {status!r}.",
category: await _get_category_data(command, category),
}
)

return command.finish()


async def _power_sequence(command, ieb, seq, mode="on", delay=3) -> bool:
"""Applies the power on/off sequence."""

# To speed up tests
if isinstance(command.actor.fps, VirtualFPS):
delay = 0.01

relay_result = "closed" if mode == "on" else "open"

command.info(text=f"Running power {mode} sequence")

# First check that the SYNC line is open.
sync = ieb.get_device("SYNC")
if (await sync.read())[0] == "closed":
command.debug(text="SYNC line is high. Opening it.")
await sync.open()

if (await sync.read())[0] != "open":
command.fail(error="Failed opening SYNC line.")
return False

command.debug(power_sync=[False])

disabled = config["ieb"]["disabled_devices"]

for devname in seq:

do_delay = False
if isinstance(devname, str):
if devname.upper() in config["ieb"]["disabled_devices"]:
command.warning(text=f"{devname} is disabled. Skipping.")
continue

dev = ieb.get_device(devname)
category = dev.category.lower()

if (await dev.read())[0] == relay_result:
command.debug(text=f"{devname} already powered {mode}.")
else:
command.debug(text=f"Powering {mode} {devname}.")
await dev.close() if mode == "on" else await dev.open()
await asyncio.sleep(0.2)
if (await dev.read())[0] != relay_result:
command.fail(error=f"Failed powering {mode} {devname}.")
return False

do_delay = True

command.debug({category: await _get_category_data(command, category)})

elif isinstance(devname, (tuple, list)):
devname = list([dn for dn in devname if dn not in disabled])
devs = [ieb.get_device(dn) for dn in devname]
category = devs[0].category.lower()

status = list(await asyncio.gather(*[dev.read() for dev in devs]))

keep = []
for ii, res in enumerate(status):
if res[0] == relay_result:
command.debug({"text": f"{devname[ii]} already powered {mode}."})
continue
keep.append(ii)

devs = [devs[ii] for ii in keep]
devname = [devname[ii] for ii in keep]

if len(devs) > 0:
command.debug(text=f"Powering {mode} {', '.join(devname)}")
await asyncio.gather(
*[dev.close() if mode == "on" else dev.open() for dev in devs]
)
do_delay = True

status = list(await asyncio.gather(*[dev.read() for dev in devs]))
for ii, res in enumerate(status):
if res[0] != relay_result:
command.fail(error=f"Failed powering {mode} {devname[ii]}.")
return False

command.debug({category: await _get_category_data(command, category)})

else:
command.fail(error=f"Invalid relay {devname!r}.")
return False

if do_delay:
await asyncio.sleep(delay)

return True


@ieb.group()
@pass_args()
def power(command, fps):
"""Runs the power on/off sequences."""

pass


@power.command()
@click.option("--no-gfas", is_flag=True, help="Do not power the GFAs.")
async def on(command, fps: FPS, no_gfas: bool = False):
"""Powers on all the FPS IEB components."""

command.info(text="Turning pollers off.")
await fps.pollers.stop()

ieb = fps.ieb
if not isinstance(ieb, IEB) or ieb.disabled:
return command.fail(error="IEB is not conencted or is disabled.")

# Sequence of relays to power on. Tuples indicate relays that can be powered
# on concurrently.
on_seq = [
("CM1", "CM2", "CM3", "CM4", "CM5", "CM6"),
"PS1",
"PS2",
"PS3",
"PS4",
"PS5",
"PS6",
"NUC1",
"NUC2",
"NUC3",
"NUC4",
"NUC5",
"NUC6",
]

if no_gfas is False:
on_seq += [
"GFA1",
"GFA2",
"GFA3",
"GFA4",
"GFA5",
"GFA6",
]

if not (await _power_sequence(command, ieb, on_seq, mode="on")):
return

return command.finish(text="Power on sequence complete.")


@power.command()
@click.option("--nucs", is_flag=True, help="Also power down NUCs")
async def off(command, fps, nucs):
"""Powers off all the FPS IEB components."""

command.info(text="Turning pollers off.")
await fps.pollers.stop()

ieb = fps.ieb
if not isinstance(ieb, IEB) or ieb.disabled:
return command.fail(error="IEB is not conencted or is disabled.")

# Sequence of relays to power off. Tuples indicate relays that can be powered
# off concurrently.
off_seq = [
"PS1",
"PS2",
"PS3",
"PS4",
"PS5",
"PS6",
"GFA1",
"GFA2",
"GFA3",
"GFA4",
"GFA5",
"GFA6",
]

if nucs:
off_seq += ["NUC1", "NUC2", "NUC3", "NUC4", "NUC5", "NUC6"]

if not (await _power_sequence(command, ieb, off_seq, mode="off")):
return

return command.finish(text="Power off sequence complete.")


@ieb.command()
@click.argument("device_names", metavar="DEVICES", type=str, nargs=-1)
@click.argument("VALUE", type=click.FloatRange(0, 100))
Expand Down