Skip to content

Pull request plugin #1

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

Open
adelaisbruel wants to merge 2 commits into
base: 5.0.x
Choose a base branch
from
Open

Pull request plugin #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8962563
using quantities
adelaisbruel Aug 21, 2026
e51372d
Correcting dwell time
adelaisbruel Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ package-url = 'https://github.com/PyMoDAQ/pymodaq_plugins_ramping'
name = "pymodaq_plugins_ramping"
description = 'PyMoDAQ plugin implementing an extension to perform acquisition in the form of a ramp'
dependencies = [
"pymodaq>=5.3.0",
"pymodaq>=5.2.0",
"pymodaq_utils>=5.2.0",
"pymodaq_gui>=5.2.0",
"pymodaq_data>=5.2.0",
Expand Down
68 changes: 45 additions & 23 deletions src/pymodaq_plugins_ramping/extensions/ramping.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
from pathlib import Path
from time import perf_counter
from typing import Iterable, TYPE_CHECKING, Union, Mapping

import numpy as np
from qtpy import QtWidgets, QtCore

from pymodaq_data import Q_, DataDim, DataSource
Expand Down Expand Up @@ -78,9 +78,10 @@ class RampExtension(CustomExt):
{'title': 'Ramp:', 'name': 'ramp', 'type': 'group', 'children': [
{'title': 'Start:', 'name': 'start', 'type': 'float', 'value': 500.},
{'title': 'Stop:', 'name': 'stop', 'type': 'float', 'value': 560.},
{'title': 'Duration:', 'name': 'duration', 'type': 'float', 'value': 40, 'suffix': 's', 'siPrefix': True},
{'title': 'Velocity:', 'name': 'velocity', 'type': 'float', 'value': 0, 'suffix': '', 'siPrefix': True,
'readonly': True},
{'title': 'Duration:', 'name': 'duration', 'type': 'float', 'value': 40, 'suffix': 'min', 'siPrefix': True,
'readonly' : True},
{'title': 'Velocity:', 'name': 'velocity', 'type': 'float', 'value': 0, 'siPrefix': True,
'readonly': False},
]},
{'title': 'Use Steps:', 'name': 'use_steps', 'type': 'bool', 'value': True},
{'title': 'Steps:', 'name': 'steps', 'type': 'group', 'children': [
Expand All @@ -101,15 +102,19 @@ def __init__(self, parent: gutils.DockArea, dashboard):

super().__init__(parent, dashboard, add_toolbar_break=False)

self._start_time: float = None
self._paused_time: float = None
self._start_time: Q_ = None
self._paused_time: Q_ = None
self.ramp: RampGenerator = None

self._n_emitted = 0

self.ramp_timer = QtCore.QTimer()
self.ramp_timer.timeout.connect(self.update_ramp)

self.wait_after_stop_timer = QtCore.QTimer()
self.wait_after_stop_timer.setSingleShot(True)
self.wait_after_stop_timer.timeout.connect(self.go_home)

self.total_ramp_timer = QtCore.QTimer()
self.total_ramp_timer.timeout.connect(self.stop_ramp)

Expand All @@ -123,7 +128,7 @@ def __init__(self, parent: gutils.DockArea, dashboard):
self.setup_ui()

self.update_n_steps()
self.update_velocity()
self.update_duration()

self.enable_runflow_actions(False)

Expand Down Expand Up @@ -245,6 +250,7 @@ def update_histogramer(self):
self.histogramer.compute_plot_histogram(self.settings['actuator'])

def start_ramp(self):
self.wait_after_stop_timer.stop()
try:
self.actuator.move_done_signal.disconnect(self.start_ramp)
except TypeError:
Expand All @@ -256,9 +262,13 @@ def start_ramp(self):
pass

if self.settings['use_steps']:
self.ramp_timer.setInterval(int(self.settings['steps', 'time_step']))
self.ramp_timer.setInterval(
int(Q_(self.settings['steps', 'time_step'],
self.settings.child('steps', 'time_step').opts['suffix']).to('ms').magnitude))

self.total_ramp_timer.setInterval(int(self.settings['ramp', 'duration'] * 1000))
self.total_ramp_timer.setInterval(
int(Q_(self.settings['ramp', 'duration'],
self.settings.child('ramp', 'duration').opts['suffix']).to('ms').magnitude))
self.total_ramp_timer.setSingleShot(True)

if self.is_action_checked('save'):
Expand Down Expand Up @@ -348,7 +358,8 @@ def stop_ramp(self):
self.ramp_timer.stop()
self.total_ramp_timer.stop()
#self.histogramer_timer.stop()

self.wait_after_stop_timer.setInterval(3600000)
self.wait_after_stop_timer.start()
for detector in self.detectors:
try:
detector.grab_done_signal.disconnect(self.send_data)
Expand Down Expand Up @@ -376,6 +387,9 @@ def stop_ramp(self):
else:
self._worker_done.connect(self.terminate_worker)

def go_home(self):
self.actuator.move_home()

def terminate_worker(self):
self.exit_runner_thread()
self.h5saver.flush()
Expand All @@ -389,14 +403,14 @@ def pause_ramp(self, do_pause=True):
if do_pause:
self.ramp_timer.stop()
#self.histogramer_timer.stop()
self._paused_time = perf_counter()
self._paused_time = Q_(perf_counter(), 's')
for detector in self.detectors:
detector.grab_done_signal.disconnect(self.send_data)
for actuator in self.actuators:
actuator.current_value_signal.disconnect(self.send_data)
self.actuator.current_value_signal.disconnect(self.send_data)
else:
self._start_time = perf_counter() - (self._paused_time - self._start_time)
self._start_time = Q_(perf_counter(),'s') - (self._paused_time - self._start_time)

for detector in self.detectors:
detector.grab_done_signal.connect(self.send_data)
Expand All @@ -410,8 +424,8 @@ def pause_ramp(self, do_pause=True):

def update_ramp(self):
if self._start_time is None:
self._start_time = perf_counter()
elapsed_time = perf_counter() - self._start_time
self._start_time = Q_(perf_counter(), 's')
elapsed_time = Q_(perf_counter(), 's') - self._start_time

step = self.ramp(elapsed_time)
self.settings['steps', 'step'] = step
Expand Down Expand Up @@ -465,23 +479,31 @@ def value_changed(self, param):
self.update_ramp_settings()
elif param.name() == 'use_steps':
self.settings.child('steps').show(param.value())
if param.name() in ('duration', 'start', 'stop'):
self.update_velocity()
if param.name() in ('velocity', 'start', 'stop'):
self.update_duration()

def update_n_steps(self):
self.settings['steps', 'nsteps'] = (Q_(self.settings['ramp', 'duration'], 's') /
Q_(self.settings['steps', 'time_step'], 'ms').to('s')).magnitude
self.settings['steps', 'nsteps'] = (Q_(self.settings['ramp', 'duration'],
self.settings.child('ramp', 'duration').opts['suffix']) /
Q_(self.settings['steps', 'time_step'],
self.settings.child('steps', 'time_step').opts['suffix'])).to_reduced_units().magnitude

def update_velocity(self):
def update_duration(self):
if self.actuator is not None:
self.settings['ramp', 'velocity'] = (
(self.settings['ramp', 'stop'] - self.settings['ramp', 'start']) / self.settings['ramp', 'duration'])
self.settings.child('ramp', 'velocity').setOpts(suffix=f'{self.actuator.units}/s')
if not np.allclose(self.settings['ramp','velocity'], 0):
self.settings.child('ramp', 'velocity').setOpts(suffix=f'{self.actuator.units}/min')
self.settings['ramp', 'duration'] = (
(self.settings['ramp', 'stop'] - self.settings['ramp', 'start']) / self.settings['ramp', 'velocity'])
self.settings.child('ramp', 'duration').setOpts(suffix='min')
else :
self.settings['ramp', 'duration'] = 0


def get_ramp(self) -> RampGenerator:
return RampGenerator(self.settings['ramp', 'start'],
self.settings['ramp', 'stop'],
self.settings['ramp', 'duration'],)
Q_(self.settings['ramp', 'duration'],
self.settings.child('ramp', 'duration').opts['suffix']))

def send_data(self, dte: DataToExport | DataActuator):
if self.is_action_checked('save'):
Expand Down
15 changes: 9 additions & 6 deletions src/pymodaq_plugins_ramping/utilities/ramp_generator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,22 +1,25 @@
import numpy as np

from pymodaq_data import Q_

class RampGenerator:
def __init__(self, start=0.0, end=10.0, duration=5.0):
def __init__(self,
start: float,
end: float,
duration=Q_(5.0,'s')):
self.start = start
self.end = end
self.duration = duration

def __call__(self, elapsed_time: float) -> float:
def __call__(self, elapsed_time: Q_) -> float:
return self.update(elapsed_time)

def update(self, elapsed_time):
def update(self, elapsed_time: Q_):
# Clamp elapsed time to the valid duration window
t_clamped = np.clip(elapsed_time, 0.0, self.duration)

# Linear interpolation for the ramp output
if self.duration > 0:
fraction = t_clamped / self.duration
if self.duration.to_reduced_units().magnitude > 0:
fraction = (t_clamped / self.duration).to_reduced_units().magnitude
value = self.start + (self.end - self.start) * fraction
else:
value = self.end
Expand Down