TriggeredSeverities#

class lsst.ts.watcher.rules.test.TriggeredSeverities(config, log=None)#

Bases: BaseRule

A test rule that transitions through a specified list of severities, repeatedly, when manually triggered by test code.

This is only intended for unit tests, since it will not transition between severities on its own. It gives unit tests complete control over when to report the next severity.

Parameters:
Raises:

RuntimeError – If compute_alarm_severity is called. When used as a normal rule this method should never be called because the rule specifies topics to call it.

run_task#

The task used to run the run method. Once started, you may check if this task is done to determine that all repeats have run.

Type:

asyncio.Future | asyncio.Task

trigger_next_severity_event#

An event the user can set to trigger the next severity.

Type:

asyncio.Event

Notes

The alarm name is f"test.TriggeredSeverities.{config.name}"

Methods Summary

compute_alarm_severity()

Compute and set alarm severity and reason.

get_schema()

Return a jsonschema as a dict, to validate configuration.

run()

Run through the configured severities, repeatedly, forever.

start()

Start any background tasks, such as a polling loop.

stop()

Stop all background tasks.

Methods Documentation

compute_alarm_severity()#

Compute and set alarm severity and reason.

Parameters:

**kwargs (dict [str, typing.Any]) –

Keyword arguments. If triggered by TopicCallback calling update_alarm_severity, the arguments will be as follows:

  • data : salobj.BaseMsgType Message from the topic described by topic_callback.

  • topic_callback : TopicCallback Topic callback wrapper.

Return type:

None

Returns:

  • None, if no change or unknown, or a tuple of two values

  • severity (lsst.ts.xml.enums.Watcher.AlarmSeverity) – The new alarm severity.

  • reason (str) – Detailed reason for the severity, e.g. a string describing what value is out of range, and what the range is. If severity is NONE then this value is ignored (but still required) and the old reason is retained until the alarm is reset to nominal state.

Notes

You may return NoneNoReason if the alarm state is NONE.

classmethod get_schema()#

Return a jsonschema as a dict, to validate configuration.

Notes

Please provide default values for all fields for which defaults make sense. This makes watcher configuration files easier to write.

If your rule has no configuration then return None.

We recommend that you write the schema as yaml, for compactness, then use yaml.safe_load to convert it to a dict. For example:

schema_yaml = """
    $schema: http://json-schema.org/draft-07/schema#
    description: Configuration for MyRule
    type: object
    properties:
    ...
    required: [...]
    additionalProperties: false
"""
return yaml.safe_load(schema_yaml)
async run()#

Run through the configured severities, repeatedly, forever.

start()#

Start any background tasks, such as a polling loop.

This is called when the watcher goes into the enabled state.

Notes

Do not assume that start is called before stop; the order depends on the initial state of the Watcher.

Immediate subclasses need not call super().start()

stop()#

Stop all background tasks.

This is called when the watcher goes out of the enabled state, and must stop any tasks that might trigger an alarm state change.

Notes

Do not assume that start is called before stop; the order depends on the initial state of the Watcher.

This base implementation does nothing, so immediate subclasses need not call super().stop().