script

frog.gui.measure_script.script

An interface for using the YAML-formatted measure scripts.

This includes code for parsing and running the scripts.

Attributes

CURRENT_SCRIPT_VERSION = 1 module-attribute

The current version of the measure script format.

Classes

Measurement(angle, measurements) dataclass

Represents a single step (i.e. angle + number of measurements).

Attributes

angle instance-attribute

Either an angle in degrees or the name of a preset angle.

measurements instance-attribute

The number of times to record a measurement at this position.

ParseError(_)

Bases: Exception

An error occurred while parsing a measure script.

Create a new ParseError.

Source code in frog/gui/measure_script/script.py

def __init__(_) -> None:
    """Create a new ParseError."""
    super().__init__("Error parsing measure script")

Functions

Script(path, repeats, sequence)

Represents a measure script, including its file path and data.

Create a new Script.

Parameters:

Name	Type	Description	Default
path	Path	The file path to this measure script	required
repeats	int	The number of times to repeat the sequence of measurements	required
sequence	Sequence[dict[str, Any]]	Different measurements (i.e. angle + num measurements) to record	required

Source code in frog/gui/measure_script/script.py

def __init__(
    self, path: Path, repeats: int, sequence: Sequence[dict[str, Any]]
) -> None:
    """Create a new Script.

    Args:
        path: The file path to this measure script
        repeats: The number of times to repeat the sequence of measurements
        sequence: Different measurements (i.e. angle + num measurements) to record
    """
    self.path = path
    self.repeats = repeats
    self.sequence = [Measurement(**val) for val in sequence]
    self.runner: ScriptRunner | None = None

Functions

__iter__()

Get an iterator for the measurements.

Source code in frog/gui/measure_script/script.py

def __iter__(self) -> ScriptIterator:
    """Get an iterator for the measurements."""
    return ScriptIterator(self)

run(parent=None)

Run this measure script.

Source code in frog/gui/measure_script/script.py

def run(self, parent: QWidget | None = None) -> None:
    """Run this measure script."""
    logging.info(f"Running {self.path}")
    self.runner = ScriptRunner(self, parent=parent)
    self.runner.start_moving()

try_load(parent, file_path) classmethod

Try to load a measure script at the specified path.

Parameters:

Name	Type	Description	Default
parent	QWidget	The parent widget (for error messages shown)	required
file_path	Path	The path to the script to be loaded	required

Returns: A Script if successful, else None

Source code in frog/gui/measure_script/script.py

@classmethod
def try_load(cls, parent: QWidget, file_path: Path) -> Script | None:
    """Try to load a measure script at the specified path.

    Args:
        parent: The parent widget (for error messages shown)
        file_path: The path to the script to be loaded
    Returns:
        A Script if successful, else None
    """
    try:
        with open(file_path) as f:
            return cls(file_path, **parse_script(f))
    except OSError as e:
        show_error_message(parent, f"Error: Could not read {file_path}: {e!s}")
    except ParseError:
        show_error_message(parent, f"Error: {file_path} is in an invalid format")
    return None

ScriptIterator(script)

Allows for iterating through a Script with the required number of repeats.

Create a new ScriptIterator.

Parameters:

Name	Type	Description	Default
script	Script	The Script from which to create this iterator.	required

Source code in frog/gui/measure_script/script.py

def __init__(self, script: Script) -> None:
    """Create a new ScriptIterator.

    Args:
        script: The Script from which to create this iterator.
    """
    self._sequence_iter = iter(script.sequence)
    self.script = script
    self.current_repeat = 0

Functions

__iter__()

Return self.

Source code in frog/gui/measure_script/script.py

def __iter__(self) -> ScriptIterator:
    """Return self."""
    return self

__next__()

Return the next Measurement in the sequence.

Source code in frog/gui/measure_script/script.py

def __next__(self) -> Measurement:
    """Return the next Measurement in the sequence."""
    try:
        return next(self._sequence_iter)
    except StopIteration:
        self.current_repeat = min(self.script.repeats, self.current_repeat + 1)
        if self.current_repeat == self.script.repeats:
            raise

        self._sequence_iter = iter(self.script.sequence)
        return next(self)

ScriptRunner(script, parent=None)

Bases: StateMachine

A class for running measure scripts.

The ScriptRunner is a finite state machine. Besides the one initial state, the runner can either be in a "moving" state (i.e. the motor is moving) or a "measuring" state (i.e. the motor is stationary and the EM27 is recording a measurement).

The state diagram looks like this:

Create a new ScriptRunner.

Parameters:

Name	Type	Description	Default
script	Script	The script to run	required
parent	QWidget | None	The parent widget	None

Source code in frog/gui/measure_script/script.py

def __init__(
    self,
    script: Script,
    parent: QWidget | None = None,
) -> None:
    """Create a new ScriptRunner.

    Args:
        script: The script to run
        parent: The parent widget
    """
    self.script = script
    """The running script."""
    self.measurement_iter = iter(self.script)
    """An iterator yielding the required sequence of measurements."""
    self.parent = parent
    """The parent widget."""
    self.paused = False
    """Whether the script is paused."""

    self.current_measurement: Measurement
    """The current measurement to acquire."""
    self.current_measurement_count: int
    """How many times a measurement has been recorded at the current angle."""

    # Actions to control the script
    pub.subscribe(self.abort, "measure_script.abort")
    pub.subscribe(self.pause, "measure_script.pause")
    pub.subscribe(self.unpause, "measure_script.unpause")

    super().__init__()

Attributes

cancel_measuring = measuring.to(not_running) class-attribute instance-attribute

Cancel the current measurement.

cancel_move = moving.to(not_running) class-attribute instance-attribute

Cancel the current movement.

cancel_waiting_to_measure = waiting_to_measure.to(not_running) class-attribute instance-attribute

Cancel the script from a waiting to measure state.

cancel_waiting_to_move = waiting_to_move.to(not_running) class-attribute instance-attribute

Cancel the script from a waiting to move state.

current_measurement instance-attribute

The current measurement to acquire.

current_measurement_count instance-attribute

How many times a measurement has been recorded at the current angle.

finish = moving.to(not_running) class-attribute instance-attribute

To be called when all measurements are complete.

finish_moving = moving.to(waiting_to_measure) class-attribute instance-attribute

Finish the moving stage.

finish_waiting_for_move = waiting_to_move.to(moving) class-attribute instance-attribute

Stop waiting and start the next move.

measurement_iter = iter(self.script) instance-attribute

An iterator yielding the required sequence of measurements.

measuring = State('Measuring') class-attribute instance-attribute

State indicating that a measurement is taking place.

moving = State('Moving') class-attribute instance-attribute

State indicating that the motor is moving.

not_running = State('Not running', initial=True) class-attribute instance-attribute

State indicating that the script is not yet running or has finished.

parent = parent instance-attribute

The parent widget.

paused = False instance-attribute

Whether the script is paused.

repeat_measuring = measuring.to(waiting_to_measure) class-attribute instance-attribute

Record another measurement at the same angle.

script = script instance-attribute

The running script.

start_measuring = waiting_to_measure.to(measuring) class-attribute instance-attribute

Start recording for the current measurement.

start_moving = not_running.to(moving) class-attribute instance-attribute

Start moving the motor to the required angle for the current measurement.

start_next_move = measuring.to(waiting_to_move) class-attribute instance-attribute

Trigger a move to the angle for the next measurement.

waiting_to_measure = State('Waiting to measure') class-attribute instance-attribute

State indicating that measurement will start when the script is unpaused.

waiting_to_move = State('Waiting to move') class-attribute instance-attribute

State indicating that the motor will move when the script is unpaused.

Functions

abort()

Abort the current measure script run.

Source code in frog/gui/measure_script/script.py

def abort(self) -> None:
    """Abort the current measure script run."""
    # For some reason mypy seems not to be able to deduce the type of current_state
    match self.current_state:  # type: ignore[has-type]
        case self.moving:
            self.cancel_move()
        case self.measuring:
            self.cancel_measuring()
        case self.waiting_to_measure:
            self.cancel_waiting_to_measure()
        case self.waiting_to_move:
            self.cancel_waiting_to_move()

    logging.info("Aborting measure script")

before_start_moving()

Send a pubsub message to indicate that the script is running.

Source code in frog/gui/measure_script/script.py

def before_start_moving(self) -> None:
    """Send a pubsub message to indicate that the script is running."""
    pub.sendMessage("measure_script.begin", script_runner=self)

on_enter_moving()

Try to load the next measurement and start the next movement.

If there are no more measurements, the ScriptRunner will return to a not_running state.

Source code in frog/gui/measure_script/script.py

def on_enter_moving(self) -> None:
    """Try to load the next measurement and start the next movement.

    If there are no more measurements, the ScriptRunner will return to a not_running
    state.
    """
    if not self._load_next_measurement():
        self.finish()
        return

    pub.sendMessage("measure_script.start_moving", script_runner=self)

    # Start moving the stepper motor
    pub.sendMessage(
        f"device.{STEPPER_MOTOR_TOPIC}.move.begin",
        target=self.current_measurement.angle,
    )

on_enter_not_running(event)

If finished, unsubscribe from pubsub messages and send message.

Source code in frog/gui/measure_script/script.py

def on_enter_not_running(self, event: str) -> None:
    """If finished, unsubscribe from pubsub messages and send message."""
    if event == "__initial__":
        # If this is the first state, do nothing
        return

    # Stepper motor messages
    pub.unsubscribe(self.finish_moving, f"device.{STEPPER_MOTOR_TOPIC}.move.end")
    pub.unsubscribe(
        self._on_stepper_motor_error, f"device.error.{STEPPER_MOTOR_TOPIC}"
    )

    # EM27 messages
    pub.unsubscribe(
        self._on_spectrometer_error, f"device.error.{SPECTROMETER_TOPIC}"
    )

    # Reset mirror to point downwards on measure script end
    pub.sendMessage(f"device.{STEPPER_MOTOR_TOPIC}.move.begin", target="nadir")

    # Send message signalling that the measure script is no longer running
    pub.sendMessage("measure_script.end")

on_enter_state(target, event)

Log the state every time it changes.

Source code in frog/gui/measure_script/script.py

def on_enter_state(self, target: State, event: str) -> None:
    """Log the state every time it changes."""
    logging.info(f"Measure script: Entering state {target.name} (event: {event})")

on_enter_waiting_to_measure()

Move onto the next measurement unless the script is paused.

Source code in frog/gui/measure_script/script.py

def on_enter_waiting_to_measure(self) -> None:
    """Move onto the next measurement unless the script is paused."""
    pub.subscribe(
        self._measuring_start, f"device.{SPECTROMETER_TOPIC}.status.measuring"
    )

    if not self.paused:
        self._request_measurement()

on_enter_waiting_to_move()

Move onto the next move unless the script is paused.

Source code in frog/gui/measure_script/script.py

def on_enter_waiting_to_move(self) -> None:
    """Move onto the next move unless the script is paused."""
    if not self.paused:
        self.finish_waiting_for_move()

on_exit_measuring()

Unsubscribe from pubsub topics.

Source code in frog/gui/measure_script/script.py

def on_exit_measuring(self) -> None:
    """Unsubscribe from pubsub topics."""
    pub.unsubscribe(
        self._measuring_end,
        f"device.{SPECTROMETER_TOPIC}.status.connected",
    )

on_exit_not_running()

Subscribe to pubsub messages for the stepper motor and spectrometer.

Source code in frog/gui/measure_script/script.py

def on_exit_not_running(self) -> None:
    """Subscribe to pubsub messages for the stepper motor and spectrometer."""
    # Listen for stepper motor messages
    pub.subscribe(self.finish_moving, f"device.{STEPPER_MOTOR_TOPIC}.move.end")
    pub.subscribe(
        self._on_stepper_motor_error, f"device.error.{STEPPER_MOTOR_TOPIC}"
    )

    # Listen for spectrometer messages
    pub.subscribe(self._on_spectrometer_error, f"device.error.{SPECTROMETER_TOPIC}")

on_exit_waiting_to_measure()

Unsubscribe from pubsub topics.

Source code in frog/gui/measure_script/script.py

def on_exit_waiting_to_measure(self) -> None:
    """Unsubscribe from pubsub topics."""
    pub.unsubscribe(
        self._measuring_start, f"device.{SPECTROMETER_TOPIC}.status.measuring"
    )

pause()

Pause the current measure script run.

Source code in frog/gui/measure_script/script.py

def pause(self) -> None:
    """Pause the current measure script run."""
    self.paused = True

unpause()

Unpause the current measure script run.

Source code in frog/gui/measure_script/script.py

def unpause(self) -> None:
    """Unpause the current measure script run."""
    self.paused = False

    # Move onto the next step if needed
    match self.current_state:
        case self.waiting_to_move:
            self.finish_waiting_for_move()
        case self.waiting_to_measure:
            self._request_measurement()

Functions

parse_script(script)

Parse a measure script.

Parameters:

Name	Type	Description	Default
script	str | TextIOBase	The contents of the script as YAML or a stream containing YAML	required

Raises: ParseError: The script's contents were invalid

Source code in frog/gui/measure_script/script.py

def parse_script(script: str | TextIOBase) -> dict[str, Any]:
    """Parse a measure script.

    Args:
        script: The contents of the script as YAML or a stream containing YAML
    Raises:
        ParseError: The script's contents were invalid
    """
    valid_float = And(float, lambda f: 0.0 <= f < 360.0)
    valid_preset = And(str, lambda s: s in ANGLE_PRESETS)
    measurements_type = And(int, lambda x: x > 0)
    nonempty_list = And(list, lambda x: x)

    schema = Schema(
        {
            "version": Const(
                CURRENT_SCRIPT_VERSION,
                f"Current script version number must be {CURRENT_SCRIPT_VERSION}",
            ),
            "repeats": measurements_type,
            "sequence": And(
                nonempty_list,
                [
                    {
                        "angle": Or(valid_float, valid_preset),
                        "measurements": measurements_type,
                    }
                ],
            ),
        }
    )

    try:
        output = yaml.safe_load(script)

        # v2.0.0 and older didn't have a version field, but the file formats are
        # otherwise identical
        if "version" not in output:
            output["version"] = 1

        output = schema.validate(output)
        output.pop("version")
        return output
    except (yaml.YAMLError, SchemaError) as e:
        raise ParseError() from e