Stream pipeline example
This commit is contained in:
root
@@ -0,0 +1,101 @@
|
||||
|
||||
from logging import getLogger
|
||||
import epics
|
||||
from threading import Thread, RLock, Event
|
||||
from collections import deque
|
||||
import time
|
||||
from cam_server.pipeline.data_processing import functions
|
||||
from cam_server.utils import create_thread_pvs
|
||||
|
||||
import numpy as np
|
||||
|
||||
_logger = getLogger(__name__)
|
||||
|
||||
OUTPUT_CHANNEL_NAME = "SLAAR11-GEN:LAS-XRAY"
|
||||
PROC_CHANNEL_NAME = "SLAAR11-GEN:LAS-EVR"
|
||||
SIZE_BUFFER = 5
|
||||
|
||||
output_pv = None
|
||||
buffer = None
|
||||
buffer_lock = None
|
||||
initialized = False
|
||||
processing_thread = None
|
||||
|
||||
def initialize(parameters):
|
||||
global output_pv, buffer, buffer_lock, initialized, processing_thread
|
||||
epics.ca.clear_cache()
|
||||
[output_pv] = create_thread_pvs([OUTPUT_CHANNEL_NAME])
|
||||
output_pv.wait_for_connection()
|
||||
#If raising this exception then the pipeline won't start
|
||||
if not output_pv.connected:
|
||||
raise ("Cannot connect to " + OUTPUT_CHANNEL_NAME)
|
||||
buffer = deque(maxlen=SIZE_BUFFER)
|
||||
buffer_lock = RLock()
|
||||
processing_thread = Thread(target=process_thread_task, args=(buffer, buffer_lock))
|
||||
processing_thread.start()
|
||||
initialized = True
|
||||
|
||||
|
||||
#Processing the buffer every second and setting result to EPICS channel
|
||||
def process_thread_task(buffer, buffer_lock):
|
||||
_logger.info("Start buffer processing thread")
|
||||
try:
|
||||
[proc_pv] = create_thread_pvs([PROC_CHANNEL_NAME])
|
||||
proc_pv.wait_for_connection()
|
||||
if not proc_pv.connected:
|
||||
raise ("Cannot connect to " + PROC_CHANNEL_NAME)
|
||||
while True:
|
||||
if proc_pv.connected:
|
||||
b = None
|
||||
with buffer_lock:
|
||||
if (len(buffer) >= SIZE_BUFFER):
|
||||
b= buffer.copy()
|
||||
if b is not None:
|
||||
calc = sum(b)/len(b)
|
||||
proc_pv.put(calc)
|
||||
time.sleep(1.0)
|
||||
|
||||
except Exception as e:
|
||||
_logger.error("Error on buffer processing thread %s" % (str(e)))
|
||||
finally:
|
||||
_logger.info("Exit buffer processing thread")
|
||||
|
||||
|
||||
def process(data, pulse_id, timestamp, parameters):
|
||||
global output_pv, buffer, buffer_lock, initialized, processing_thread
|
||||
|
||||
#Initialize on first run
|
||||
if not initialized:
|
||||
initialize(parameters)
|
||||
|
||||
#Read stream inputs
|
||||
ch12 = data["SARFE10-CVME-PHO6212:Lnk9Ch12-DATA-SUM"]
|
||||
ch13 = data["SARFE10-CVME-PHO6212:Lnk9Ch13-DATA-SUM"]
|
||||
ch14 = data["SARFE10-CVME-PHO6212:Lnk9Ch14-DATA-SUM"]
|
||||
ch15 = data["SARFE10-CVME-PHO6212:Lnk9Ch15-DATA-SUM"]
|
||||
|
||||
#Calculations
|
||||
try:
|
||||
sum = ch12 + ch13 + ch14 + ch15
|
||||
except:
|
||||
sum = float("nan")
|
||||
|
||||
#Handle thread buffer
|
||||
with buffer_lock:
|
||||
buffer.append(sum)
|
||||
|
||||
#Update EPICS channels
|
||||
if output_pv and output_pv.connected:
|
||||
last = float(output_pv.value)
|
||||
output_pv.put(sum)
|
||||
else:
|
||||
last = float('nan')
|
||||
|
||||
#Set outputs
|
||||
data["sum"] = sum
|
||||
data["last"] = last
|
||||
data["alive"] = 1 if processing_thread.is_alive() else 0
|
||||
|
||||
return data
|
||||
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
from collections import OrderedDict
|
||||
from cam_server.pipeline.data_processing import functions, processor
|
||||
|
||||
|
||||
def process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, bsdata):
|
||||
r = processor.process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, bsdata)
|
||||
ret = OrderedDict()
|
||||
channels = ["intensity","x_center_of_mass","x_fwhm","x_rms","x_fit_amplitude", "x_fit_mean","x_fit_offset","x_fit_standard_deviation","x_profile"]
|
||||
prefix = parameters["camera_name"]
|
||||
for c in channels:
|
||||
ret[prefix+":"+c] = r[c]
|
||||
return ret
|
||||
|
||||
@@ -16,9 +16,18 @@ def get_roi_x_profile(image, roi):
|
||||
|
||||
return roi_image.sum(0)
|
||||
|
||||
|
||||
#_logger.warning("----- START ---- ")
|
||||
#pid = None
|
||||
#sent=None
|
||||
def process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, image_background_array=None):
|
||||
|
||||
#global pid, sent
|
||||
#if pid is not None:
|
||||
# if pid != sent:
|
||||
# _logger.warning("ERROR sending %s PID: %d" % (parameters["camera_name"], pid,))
|
||||
# if (pid+1) != pulse_id:
|
||||
# _logger.warning("ERROR %s PID: waiting %d - received %d" % (parameters["camera_name"], pid+1,pulse_id))
|
||||
#pid = pulse_id
|
||||
|
||||
processed_data = dict()
|
||||
|
||||
image_property_name = parameters["camera_name"]
|
||||
@@ -34,6 +43,6 @@ def process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, image_
|
||||
|
||||
if roi_background:
|
||||
processed_data[image_property_name + ".roi_background_x_profile"] = get_roi_x_profile(image, roi_background)
|
||||
|
||||
#sent = pid
|
||||
return processed_data
|
||||
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
from logging import getLogger
|
||||
|
||||
from cam_server.pipeline.data_processing import functions
|
||||
from cam_server.pipeline.data_processing.functions import chunk_copy
|
||||
from cam_server.utils import create_thread_pvs, epics_lock
|
||||
|
||||
|
||||
import json
|
||||
|
||||
@@ -9,28 +10,29 @@ import numpy
|
||||
import scipy.signal
|
||||
import scipy.optimize
|
||||
import numba
|
||||
numba.set_num_threads(4)
|
||||
|
||||
import epics
|
||||
numba.set_num_threads(4)
|
||||
|
||||
_logger = getLogger(__name__)
|
||||
|
||||
output_pv, center_pv, fwhm_pv, ymin_pv, ymax_pv, axis_pv = None, None, None, None, None, None
|
||||
channel_names = None
|
||||
output_pv, center_pv, fwhm_pv, ymin_pv, ymax_pv, axis_pv = None, None, None, None, None, None
|
||||
roi = [0, 0]
|
||||
initialized = False
|
||||
sent_pid = -1
|
||||
|
||||
|
||||
@numba.njit(parallel=True)
|
||||
def get_spectrum (image, background):
|
||||
def get_spectrum(image, background):
|
||||
y = image.shape[0]
|
||||
x = image.shape[1]
|
||||
|
||||
profile = numpy.zeros(x, dtype=numpy.uint32)
|
||||
b = 0
|
||||
|
||||
for i in numba.prange(y):
|
||||
for j in range(x):
|
||||
v = image[i,j]
|
||||
b = background[i,j]
|
||||
v = image[i, j]
|
||||
b = background[i, j]
|
||||
if v > b:
|
||||
v -= b
|
||||
else:
|
||||
@@ -41,9 +43,7 @@ def get_spectrum (image, background):
|
||||
|
||||
def initialize(parameters):
|
||||
global ymin_pv, ymax_pv, axis_pv, output_pv, center_pv, fwhm_pv
|
||||
|
||||
|
||||
|
||||
global channel_names
|
||||
epics_pv_name_prefix = parameters["camera_name"]
|
||||
output_pv_name = epics_pv_name_prefix + ":SPECTRUM_Y"
|
||||
center_pv_name = epics_pv_name_prefix + ":SPECTRUM_CENTER"
|
||||
@@ -51,30 +51,18 @@ def initialize(parameters):
|
||||
ymin_pv_name = epics_pv_name_prefix + ":SPC_ROI_YMIN"
|
||||
ymax_pv_name = epics_pv_name_prefix + ":SPC_ROI_YMAX"
|
||||
axis_pv_name = epics_pv_name_prefix + ":SPECTRUM_X"
|
||||
epics.ca.clear_cache()
|
||||
|
||||
output_pv = epics.PV(output_pv_name)
|
||||
center_pv = epics.PV(center_pv_name)
|
||||
fwhm_pv = epics.PV(fwhm_pv_name)
|
||||
|
||||
ymin_pv = epics.PV(ymin_pv_name)
|
||||
ymax_pv = epics.PV(ymax_pv_name)
|
||||
axis_pv = epics.PV(axis_pv_name)
|
||||
ymin_pv.wait_for_connection()
|
||||
ymax_pv.wait_for_connection()
|
||||
axis_pv.wait_for_connection()
|
||||
channel_names = [output_pv_name, center_pv_name, fwhm_pv_name, ymin_pv_name, ymax_pv_name, axis_pv_name]
|
||||
|
||||
|
||||
def process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, bsdata=None, background=None):
|
||||
global roi, initialized
|
||||
global ymin_pv, ymax_pv, axis_pv, output_pv, center_pv, fwhm_pv
|
||||
|
||||
global roi, initialized, sent_pid
|
||||
global channel_names
|
||||
|
||||
if not initialized:
|
||||
initialize(parameters)
|
||||
initialized = True
|
||||
|
||||
[output_pv, center_pv, fwhm_pv, ymin_pv, ymax_pv, axis_pv] = create_thread_pvs(channel_names)
|
||||
processed_data = dict()
|
||||
|
||||
epics_pv_name_prefix = parameters["camera_name"]
|
||||
|
||||
if ymin_pv and ymin_pv.connected:
|
||||
@@ -100,13 +88,12 @@ def process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, bsdata
|
||||
processing_image = image
|
||||
nrows, ncols = processing_image.shape
|
||||
|
||||
# validate background data
|
||||
# validate background data if passive mode (background subtraction handled here)
|
||||
background_image = parameters.pop('background_data', None)
|
||||
|
||||
if isinstance(background_image, numpy.ndarray):
|
||||
#background_image = chunk_copy(background_image)
|
||||
if background_image.shape != processing_image.shape:
|
||||
_logger.info("Invalid background shape: %s instead of %s" % (str(background_image.shape), str(processing_image.shape)))
|
||||
_logger.info("Invalid background shape: %s instead of %s" % (
|
||||
str(background_image.shape), str(processing_image.shape)))
|
||||
background_image = None
|
||||
else:
|
||||
background_image = None
|
||||
@@ -126,7 +113,7 @@ def process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, bsdata
|
||||
if background_image is not None:
|
||||
spectrum = get_spectrum(processing_image, background_image)
|
||||
else:
|
||||
spectrum = processing_image.sum(0, 'uint32')
|
||||
spectrum = processing_image.sum(0, 'uint32')
|
||||
|
||||
# smooth the spectrum with savgol filter with 51 window size and 3rd order polynomial
|
||||
smoothed_spectrum = scipy.signal.savgol_filter(spectrum, 51, 3)
|
||||
@@ -140,23 +127,28 @@ def process_image(image, pulse_id, timestamp, x_axis, y_axis, parameters, bsdata
|
||||
skip = False
|
||||
# gaussian fitting
|
||||
offset, amplitude, center, sigma = functions.gauss_fit_psss(smoothed_spectrum[::2], axis[::2],
|
||||
offset=minimum, amplitude=amplitude, skip=skip)
|
||||
offset=minimum, amplitude=amplitude, skip=skip, maxfev=20)
|
||||
|
||||
# outputs
|
||||
processed_data[epics_pv_name_prefix + ":SPECTRUM_Y"] = spectrum
|
||||
processed_data[epics_pv_name_prefix + ":SPECTRUM_X"] = axis
|
||||
processed_data[epics_pv_name_prefix + ":SPECTRUM_CENTER"] = center
|
||||
processed_data[epics_pv_name_prefix + ":SPECTRUM_FWHM"] = 2.355 * sigma
|
||||
processed_data[epics_pv_name_prefix + ":SPECTRUM_CENTER"] = numpy.float64(center)
|
||||
processed_data[epics_pv_name_prefix + ":SPECTRUM_FWHM"] = numpy.float64(2.355 * sigma)
|
||||
if epics_lock.acquire(False):
|
||||
try:
|
||||
if pulse_id > sent_pid:
|
||||
sent_pid = pulse_id
|
||||
if output_pv and output_pv.connected:
|
||||
output_pv.put(processed_data[epics_pv_name_prefix + ":SPECTRUM_Y"])
|
||||
#_logger.debug("caput on %s for pulse_id %s", output_pv, pulse_id)
|
||||
|
||||
if output_pv and output_pv.connected:
|
||||
output_pv.put(processed_data[epics_pv_name_prefix + ":SPECTRUM_Y"])
|
||||
_logger.debug("caput on %s for pulse_id %s", output_pv, pulse_id)
|
||||
if center_pv and center_pv.connected:
|
||||
center_pv.put(processed_data[epics_pv_name_prefix + ":SPECTRUM_CENTER"])
|
||||
|
||||
if center_pv and center_pv.connected:
|
||||
center_pv.put(processed_data[epics_pv_name_prefix + ":SPECTRUM_CENTER"])
|
||||
|
||||
if fwhm_pv and fwhm_pv.connected:
|
||||
fwhm_pv.put(processed_data[epics_pv_name_prefix + ":SPECTRUM_FWHM"])
|
||||
if fwhm_pv and fwhm_pv.connected:
|
||||
fwhm_pv.put(processed_data[epics_pv_name_prefix + ":SPECTRUM_FWHM"])
|
||||
finally:
|
||||
epics_lock.release()
|
||||
|
||||
return processed_data
|
||||
|
||||
|
||||
Reference in New Issue
Block a user