150 lines
6.6 KiB
Python
150 lines
6.6 KiB
Python
import numpy
|
|
import sys
|
|
import zmq
|
|
import time
|
|
|
|
from cam_server.camera.source.camera import *
|
|
|
|
from logging import getLogger
|
|
from cam_server import config
|
|
from cam_server.camera.sender import *
|
|
from cam_server.camera.source.common import transform_image
|
|
from copy import copy
|
|
from cam_server.utils import set_statistics, on_message_sent, init_statistics, MaxLenDict
|
|
|
|
import jungfrau_utils as ju
|
|
|
|
|
|
_logger = getLogger(__name__)
|
|
|
|
|
|
class jungfrau(Camera):
|
|
def __init__(self, camera_config):
|
|
Camera.__init__(self, camera_config, check_sender_data=False)
|
|
self.camera_config = camera_config
|
|
self.backend = self.camera_config.get_source()
|
|
self.backend = "tcp://192.168.30.8:9102"
|
|
self.timeout = self.camera_config.parameters.get("timeout")
|
|
if self.timeout is None:
|
|
self.timeout = 5.0
|
|
self.ju_stream_adapter = ju.StreamAdapter()
|
|
|
|
self.zmq_context = zmq.Context(io_threads=4)
|
|
self.poller = zmq.Poller()
|
|
|
|
# Width and height of the raw image
|
|
self.width_raw = 0
|
|
self.height_raw = 0
|
|
self.pid = None
|
|
self.metadata = {}
|
|
|
|
self.init=True
|
|
|
|
def get_raw_geometry(self):
|
|
return self.width_raw, self.height_raw
|
|
|
|
def verify_camera_online(self):
|
|
return
|
|
|
|
def connect(self):
|
|
"""
|
|
flags=0
|
|
zmq_context = zmq.Context(io_threads=4)
|
|
poller = zmq.Poller()
|
|
backend_socket = zmq_context.socket(zmq.PULL)
|
|
backend_socket.connect(self.backend)
|
|
poller.register(backend_socket, zmq.POLLIN)
|
|
for i in range(5):
|
|
events = dict(poller.poll(2000))
|
|
if backend_socket in events:
|
|
metadata = backend_socket.recv_json(flags)
|
|
image = backend_socket.recv(flags, copy=False, track=False)
|
|
_logger.info(f"got it {metadata}")
|
|
else:
|
|
_logger.info("nothing")
|
|
"""
|
|
try:
|
|
_logger.info("Connecting to Jungfrau: " + str(self.backend))
|
|
self.verify_camera_online()
|
|
self.backend_socket = self.zmq_context.socket(zmq.PULL)
|
|
self.backend_socket.connect(self.backend)
|
|
self.poller.register(self.backend_socket, zmq.POLLIN)
|
|
image = self.get_image()
|
|
self.height_raw, self.width_raw = image.shape
|
|
#metadata, image = self._read()
|
|
#self.width_raw, self.height_raw = metadata['shape'][1], metadata['shape'][0]
|
|
_logger.info("Connected to Jungfrau - shape: " + str(self.width_raw) + "x" + str(self.height_raw))
|
|
except:
|
|
_logger.warning("Error connecting to Jungfrau: " + str(sys.exc_info()[1]))
|
|
raise
|
|
|
|
def disconnect(self):
|
|
try:
|
|
self.backend_socket.disconnect()
|
|
except:
|
|
pass
|
|
|
|
def _read(self):
|
|
flags=0
|
|
start = time.time()
|
|
try:
|
|
while True:
|
|
events = dict(self.poller.poll(20)) # check every 2 seconds in each worker
|
|
if self.backend_socket in events:
|
|
metadata = self.backend_socket.recv_json(flags)
|
|
data = self.backend_socket.recv(flags, copy=False, track=False)
|
|
image = numpy.frombuffer(data, dtype=metadata['type']).reshape(metadata['shape'])
|
|
results = copy(metadata)
|
|
# we may not send each frame(real image) every time, to save bandwidth. If image is not sent, we send "2x2 dummy image"
|
|
if results['shape'][0] == 2 and results['shape'][1] == 2:
|
|
#_logger.info("Dummy")
|
|
pass
|
|
else:
|
|
return results, image
|
|
if self.timeout>0:
|
|
if (time.time() - start) > self.timeout:
|
|
raise Exception("Timeout")
|
|
except:
|
|
raise Exception("Cannot get data from detector: " + str(sys.exc_info()[1]))
|
|
|
|
def read(self):
|
|
try:
|
|
#width, height = self.get_raw_geometry()
|
|
#return numpy.random.randint(1, 101, width * height, "uint16").reshape((height, width))
|
|
self.metadata, image = self._read()
|
|
#double_pixel_action = results.get('double_pixels_action', "mask")
|
|
#image = self.ju_stream_adapter.process(image, results, double_pixels=double_pixel_action)
|
|
#image=image.astype(dtype="uint16", copy=False)
|
|
#self.pid = self.metadata.get('pulse_id', None)
|
|
return image
|
|
except:
|
|
_logger.warning("Error reading from Jungfrau: " + str(sys.exc_info()[1]))
|
|
raise
|
|
|
|
def get_data(self):
|
|
image = self.get_image()
|
|
timestamp=time.time()
|
|
return image, time.time(), self.metadata.get('pulse_id', None)
|
|
|
|
def register_channels(self, sender):
|
|
Camera.register_channels(self, sender) #default channels
|
|
sender.add_channel("frame",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "int64"})
|
|
sender.add_channel("is_good_frame",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "int64"})
|
|
sender.add_channel("daq_rec",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "int64"})
|
|
sender.add_channel("number_frames_expected",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "int64"})
|
|
sender.add_channel("daq_rec",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "int64"})
|
|
sender.add_channel("pedestal_file",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "string"})
|
|
sender.add_channel("gain_file",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "string"})
|
|
sender.add_channel("run_name",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "string"})
|
|
sender.add_channel("detector_name",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "string"})
|
|
sender.add_channel("htype",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "string"})
|
|
sender.add_channel("type",metadata={"compression": config.CAMERA_BSREAD_SCALAR_COMPRESSION, "type": "string"})
|
|
|
|
def get_send_channels(self, default_channels):
|
|
for channel in ("frame", "is_good_frame","daq_rec","number_frames_expected","daq_rec","pedestal_file","gain_file", "run_name", "detector_name", "htype","type"):
|
|
default_channels[channel] = self.metadata.get(channel, None)
|
|
return default_channels
|
|
|
|
def process(self, stop_event, statistics, parameter_queue, port):
|
|
return Camera.process(self, stop_event, statistics, parameter_queue, port)
|