Commit 1734e887 authored by Alejandro Homs Puron's avatar Alejandro Homs Puron Committed by operator for beamline
Browse files

Add Stream NoCompression support

parent 5537b9b7
...@@ -39,6 +39,8 @@ ...@@ -39,6 +39,8 @@
#include <ostream> #include <ostream>
DEB_GLOBAL_NAMESPC(DebModCamera, "Eiger");
namespace eigerapi namespace eigerapi
{ {
class Requests; class Requests;
...@@ -65,7 +67,7 @@ namespace lima ...@@ -65,7 +67,7 @@ namespace lima
enum ApiGeneration { Eiger1, Eiger2 }; enum ApiGeneration { Eiger1, Eiger2 };
enum Status { Ready, Initialising, Exposure, Readout, Fault }; enum Status { Ready, Initialising, Exposure, Readout, Fault };
enum CompressionType {LZ4,BSLZ4}; enum CompressionType {NoCompression,LZ4,BSLZ4};
Camera(const std::string& detector_ip, ApiGeneration api = Eiger1); Camera(const std::string& detector_ip, ApiGeneration api = Eiger1);
~Camera(); ~Camera();
......
...@@ -71,6 +71,8 @@ namespace eigerapi ...@@ -71,6 +71,8 @@ namespace eigerapi
bool lock = true); bool lock = true);
Value get_max(double timeout = CurlLoop::FutureRequest::TIMEOUT, Value get_max(double timeout = CurlLoop::FutureRequest::TIMEOUT,
bool lock = true); bool lock = true);
Value get_allowed_values(double timeout = CurlLoop::FutureRequest::TIMEOUT,
bool lock = true);
private: private:
void _fill_get_request(); void _fill_get_request();
template <class T> template <class T>
......
...@@ -496,7 +496,7 @@ Requests::Param::Value Requests::Param::_get(double timeout,bool lock, ...@@ -496,7 +496,7 @@ Requests::Param::Value Requests::Param::_get(double timeout,bool lock,
THROW_EIGER_EXCEPTION(eigerapi::JSON_PARSE_FAILED,""); THROW_EIGER_EXCEPTION(eigerapi::JSON_PARSE_FAILED,"");
Value value; Value value;
std::string json_type; std::string json_type;
bool is_list = root.isArray(); bool is_list = root.isArray() || root.get(param_name, "no_value").isArray();
if (!is_list) { if (!is_list) {
json_type = root.get("value_type", "dummy").asString(); json_type = root.get("value_type", "dummy").asString();
is_list = (json_type == "list"); is_list = (json_type == "list");
...@@ -504,7 +504,7 @@ Requests::Param::Value Requests::Param::_get(double timeout,bool lock, ...@@ -504,7 +504,7 @@ Requests::Param::Value Requests::Param::_get(double timeout,bool lock,
if(is_list) if(is_list)
{ {
value.type = Requests::Param::STRING_ARRAY; value.type = Requests::Param::STRING_ARRAY;
Json::Value& array = root.isArray() ? root : root["value"]; Json::Value& array = root.isArray() ? root : root[param_name];
int array_size = array.size(); int array_size = array.size();
for(int i = 0;i < array_size;++i) for(int i = 0;i < array_size;++i)
value.string_array.push_back(array[i].asString()); value.string_array.push_back(array[i].asString());
...@@ -563,6 +563,11 @@ Requests::Param::Value Requests::Param::get_max(double timeout,bool lock) ...@@ -563,6 +563,11 @@ Requests::Param::Value Requests::Param::get_max(double timeout,bool lock)
return _get(timeout,lock,"max"); return _get(timeout,lock,"max");
} }
Requests::Param::Value Requests::Param::get_allowed_values(double timeout,bool lock)
{
return _get(timeout,lock,"allowed_values");
}
void Requests::Param::_fill_get_request() void Requests::Param::_fill_get_request()
{ {
curl_easy_setopt(m_handle,CURLOPT_WRITEFUNCTION,_write_callback); curl_easy_setopt(m_handle,CURLOPT_WRITEFUNCTION,_write_callback);
......
...@@ -30,7 +30,7 @@ namespace Eiger ...@@ -30,7 +30,7 @@ namespace Eiger
enum ApiGeneration { Eiger1, Eiger2 }; enum ApiGeneration { Eiger1, Eiger2 };
enum Status { Ready, Initialising, Exposure, Readout, Fault }; enum Status { Ready, Initialising, Exposure, Readout, Fault };
enum CompressionType {LZ4,BSLZ4}; enum CompressionType {NoCompression,LZ4,BSLZ4};
Camera(const std::string& detector_ip, Camera(const std::string& detector_ip,
Eiger::Camera::ApiGeneration api = Eiger::Camera::Eiger1); Eiger::Camera::ApiGeneration api = Eiger::Camera::Eiger1);
......
...@@ -1078,16 +1078,50 @@ void Camera::getCompressionType(Camera::CompressionType& type) const ...@@ -1078,16 +1078,50 @@ void Camera::getCompressionType(Camera::CompressionType& type) const
DEB_MEMBER_FUNCT(); DEB_MEMBER_FUNCT();
std::string compression_type; std::string compression_type;
EIGER_SYNC_GET_PARAM(Requests::COMPRESSION_TYPE,compression_type); EIGER_SYNC_GET_PARAM(Requests::COMPRESSION_TYPE,compression_type);
DEB_RETURN() << DEB_VAR1(compression_type); DEB_TRACE() << DEB_VAR1(compression_type);
type = compression_type == "lz4" ? LZ4 : BSLZ4; if (compression_type == "none")
type = NoCompression;
else if (compression_type == "lz4")
type = LZ4;
else if (compression_type == "bslz4")
type = BSLZ4;
else
THROW_HW_ERROR(InvalidValue) << "Invalid compression type: " << type;
DEB_RETURN() << DEB_VAR1(type);
} }
void Camera::setCompressionType(Camera::CompressionType type) void Camera::setCompressionType(Camera::CompressionType type)
{ {
DEB_MEMBER_FUNCT(); DEB_MEMBER_FUNCT();
EIGER_SYNC_SET_PARAM(Requests::COMPRESSION_TYPE, DEB_PARAM() << DEB_VAR1(type);
type == LZ4 ? "lz4" : "bslz4");
const char *s;
switch (type) {
case NoCompression: s = "none"; break;
case LZ4: s = "lz4"; break;
case BSLZ4: s = "bslz4"; break;
default:
THROW_HW_ERROR(InvalidValue) << "Invalid compression type: " << type;
}
DEB_TRACE() << DEB_VAR1(s);
std::shared_ptr<Requests::Param> type_req =
m_requests->get_param(Requests::COMPRESSION_TYPE);
Requests::Param::Value types_allowed = type_req->get_allowed_values();
const vector<string>& l = types_allowed.string_array;
if (DEB_CHECK_ANY(DebTypeTrace)) {
DEB_TRACE() << "allowed compression types:";
vector<string>::const_iterator it, end = l.end();
for (it = l.begin(); it != end; ++it)
DEB_TRACE() << " " << *it;
}
if (find(l.begin(), l.end(), s) == l.end())
THROW_HW_ERROR(NotSupported) << "Compression type " << type
<< " not allowed";
EIGER_SYNC_SET_PARAM(Requests::COMPRESSION_TYPE, s);
} }
void Camera::getSerieId(int& serie_id) void Camera::getSerieId(int& serie_id)
{ {
DEB_MEMBER_FUNCT(); DEB_MEMBER_FUNCT();
......
...@@ -42,7 +42,7 @@ private: ...@@ -42,7 +42,7 @@ private:
Stream& m_stream; Stream& m_stream;
}; };
void _expend(void *src,Data& dst) void _expand(void *src,Data& dst)
{ {
int nbItems = dst.size() / dst.depth(); int nbItems = dst.size() / dst.depth();
unsigned short* src_data = (unsigned short*)src; unsigned short* src_data = (unsigned short*)src;
...@@ -55,50 +55,54 @@ void _expend(void *src,Data& dst) ...@@ -55,50 +55,54 @@ void _expend(void *src,Data& dst)
dst.type = Data::UINT32; dst.type = Data::UINT32;
} }
Data _DecompressTask::process(Data& src) Data _DecompressTask::process(Data& out)
{ {
void *msg_data; void *msg_data;
size_t msg_size; size_t msg_size;
int depth; int depth;
if(!m_stream.get_msg(src.data(),msg_data,msg_size,depth)) void *lima_buffer = out.data();
if(!m_stream.get_msg(lima_buffer,msg_data,msg_size,depth))
throw ProcessException("_DecompressTask: can't find compressed message"); throw ProcessException("_DecompressTask: can't find compressed message");
void* dst;
int size; int size;
bool expand_16_to_32bit = ((src.depth() == 4) && (depth == 2)); bool expand_16_to_32bit = ((out.depth() == 4) && (depth == 2));
if(expand_16_to_32bit) size = out.size() / (expand_16_to_32bit ? 2 : 1);
{
if(posix_memalign(&dst,16,src.size() / 2))
throw ProcessException("Can't allocate temporary memory");
size = src.size() / 2;
}
else
dst = src.data(),size = src.size();
Camera::CompressionType type; Camera::CompressionType type;
m_stream.getCompressionType(type); m_stream.getCompressionType(type);
int return_code; bool decompress = (type != Camera::NoCompression);
void *aux_buffer = NULL;
if(expand_16_to_32bit && decompress)
if(posix_memalign(&aux_buffer,16,size))
throw ProcessException("Can't allocate temporary memory");
void *decompress_out = expand_16_to_32bit ? aux_buffer : lima_buffer;
int return_code = 0;
if (type == Camera::LZ4) { if (type == Camera::LZ4) {
return_code = LZ4_decompress_fast((const char*)msg_data,(char*)dst,size); return_code = LZ4_decompress_fast((const char*)msg_data,(char*)decompress_out,size);
} else { } else if (type == Camera::BSLZ4) {
size_t nb_elements = size / depth; size_t nb_elements = size / depth;
return_code = bshuf_decompress_lz4(msg_data, dst, nb_elements, depth, 0); return_code = bshuf_decompress_lz4(msg_data, decompress_out, nb_elements, depth, 0);
} }
if(return_code < 0) if(return_code < 0)
{ {
if(expand_16_to_32bit) free(dst); if(aux_buffer) free(aux_buffer);
char ErrorBuff[1024]; char ErrorBuff[1024];
snprintf(ErrorBuff,sizeof(ErrorBuff), snprintf(ErrorBuff,sizeof(ErrorBuff),
"_DecompressTask: decompression failed, (error code: %d) (data size %d)", "_DecompressTask: decompression failed, (error code: %d) (data size %d)",
return_code,src.size()); return_code,out.size());
throw ProcessException(ErrorBuff); throw ProcessException(ErrorBuff);
} }
if(expand_16_to_32bit)
{ if(expand_16_to_32bit) {
_expend(dst,src); void *expand_src = decompress ? decompress_out : msg_data;
free(dst); _expand(expand_src,out);
} } else if(!decompress)
return src; memcpy(lima_buffer, msg_data, size);
if(aux_buffer)
free(aux_buffer);
return out;
} }
Decompress::Decompress(Stream& stream) : Decompress::Decompress(Stream& stream) :
......
...@@ -35,6 +35,9 @@ ...@@ -35,6 +35,9 @@
#include "lima/Exceptions.h" #include "lima/Exceptions.h"
#include "EigerStream.h" #include "EigerStream.h"
#define _BSD_SOURCE
#include <endian.h>
using namespace lima; using namespace lima;
using namespace lima::Eiger; using namespace lima::Eiger;
using namespace eigerapi; using namespace eigerapi;
...@@ -157,6 +160,9 @@ Stream::Stream(Camera& cam) : ...@@ -157,6 +160,9 @@ Stream::Stream(Camera& cam) :
{ {
DEB_CONSTRUCTOR(); DEB_CONSTRUCTOR();
bool is_le = (htole16(0x1234) == 0x1234);
m_endianess = (is_le ? '<' : '>');
m_zmq_context = zmq_ctx_new(); m_zmq_context = zmq_ctx_new();
if(pipe(m_pipes)) if(pipe(m_pipes))
THROW_HW_ERROR(Error) << "Can't open pipe"; THROW_HW_ERROR(Error) << "Can't open pipe";
...@@ -313,8 +319,11 @@ void* Stream::_runFunc(void *streamPt) ...@@ -313,8 +319,11 @@ void* Stream::_runFunc(void *streamPt)
static inline bool _get_json_header(std::shared_ptr<Stream::Message> &msg, static inline bool _get_json_header(std::shared_ptr<Stream::Message> &msg,
Json::Value& header) Json::Value& header)
{ {
DEB_GLOBAL_FUNCT();
void* data = zmq_msg_data(msg->get_msg()); void* data = zmq_msg_data(msg->get_msg());
size_t data_size = zmq_msg_size(msg->get_msg()); size_t data_size = zmq_msg_size(msg->get_msg());
DEB_TRACE() << "json_header=" << std::string((char *) data, data_size);
const char* begin = (const char*)data; const char* begin = (const char*)data;
const char* end = begin + data_size; const char* end = begin + data_size;
Json::Reader reader; Json::Reader reader;
...@@ -445,25 +454,34 @@ void Stream::_run() ...@@ -445,25 +454,34 @@ void Stream::_run()
Json::Value data_header; Json::Value data_header;
if(!_get_json_header(pending_messages[1],data_header)) break; if(!_get_json_header(pending_messages[1],data_header)) break;
//Data size (width,height) //Data size (width,height)
Json::Value shape = data_header.get("shape",""); Json::Value shape = data_header.get("shape","");
if(!shape.isArray() || shape.size() != 2) break; if(!shape.isArray() || shape.size() != 2) break;
FrameDim anImageDim; FrameDim anImageDim;
anImageDim.setSize(Size(shape[0u].asInt(),shape[1u].asInt())); anImageDim.setSize(Size(shape[0u].asInt(),shape[1u].asInt()));
//data type //data type
ImageType image_type;
std::string dtype = data_header.get("type","none").asString(); std::string dtype = data_header.get("type","none").asString();
if(dtype == "int32") if(dtype == "int32")
anImageDim.setImageType(Bpp32S); image_type = Bpp32S;
else if(dtype == "uint32") else if(dtype == "uint32")
anImageDim.setImageType(Bpp32); image_type = Bpp32;
else if(dtype == "int16") else if(dtype == "int16")
anImageDim.setImageType(Bpp16S); image_type = Bpp16S;
else if(dtype == "uint16") else if(dtype == "uint16")
anImageDim.setImageType(Bpp16); image_type = Bpp16;
else else
break; break;
anImageDim.setImageType(image_type);
DEB_TRACE() << DEB_VAR1(anImageDim); DEB_TRACE() << DEB_VAR1(anImageDim);
if (frameid == 0) {
std::string encoding = data_header.get("encoding", "").asString();
Camera::CompressionType comp_type;
_checkEncoding(encoding, image_type, comp_type);
}
HwFrameInfoType frame_info; HwFrameInfoType frame_info;
frame_info.acq_frame_nb = frameid; frame_info.acq_frame_nb = frameid;
void* buffer_ptr = buffer_mgr.getFrameBufferPtr(frameid); void* buffer_ptr = buffer_mgr.getFrameBufferPtr(frameid);
...@@ -503,3 +521,45 @@ void Stream::_run() ...@@ -503,3 +521,45 @@ void Stream::_run()
} }
m_running = false; m_running = false;
} }
void Stream::_checkEncoding(const std::string& encoding,
ImageType image_type,
Camera::CompressionType& comp_type)
{
DEB_MEMBER_FUNCT();
DEB_PARAM() << DEB_VAR3(encoding, image_type, m_endianess);
char endianess = *encoding.rbegin();
if (endianess != m_endianess)
THROW_HW_ERROR(Error) << "Endianess mismatch: "
<< "got " << endianess << ", "
<< "expected " << m_endianess;
if (encoding == std::string(1, m_endianess)) {
comp_type = Camera::NoCompression;
} else {
const std::string lz4 = std::string("lz4") + m_endianess;
if (encoding == lz4) {
comp_type = Camera::LZ4;
} else {
const char *bs;
switch (image_type) {
case Bpp32S: case Bpp32: bs = "bs32-"; break;
case Bpp16S: case Bpp16: bs = "bs16-"; break;
case Bpp8S: case Bpp8: bs = "bs8-"; break;
}
if (encoding == std::string(bs) + lz4)
comp_type = Camera::BSLZ4;
else
THROW_HW_ERROR(Error) << "Unexpected encoding: " << encoding;
}
}
Camera::CompressionType expected_comp_type;
getCompressionType(expected_comp_type);
if (comp_type != expected_comp_type)
THROW_HW_ERROR(Error) << "Unexpected compression type: " << comp_type;
DEB_RETURN() << DEB_VAR1(comp_type);
}
...@@ -64,8 +64,13 @@ namespace lima ...@@ -64,8 +64,13 @@ namespace lima
static void* _runFunc(void*); static void* _runFunc(void*);
void _run(); void _run();
void _send_synchro(); void _send_synchro();
void _checkEncoding(const std::string& encoding,
ImageType image_type,
Camera::CompressionType& comp_type);
Camera& m_cam; Camera& m_cam;
char m_endianess;
bool m_active; bool m_active;
HeaderDetail m_header_detail; HeaderDetail m_header_detail;
bool m_dirty_flag; bool m_dirty_flag;
......
...@@ -83,7 +83,8 @@ class Eiger(PyTango.Device_4Impl): ...@@ -83,7 +83,8 @@ class Eiger(PyTango.Device_4Impl):
'OFF':False} 'OFF':False}
self.__PixelMask = {'ON':True, self.__PixelMask = {'ON':True,
'OFF':False} 'OFF':False}
self.__CompressionType = {'LZ4': EigerAcq.Camera.LZ4, self.__CompressionType = {'NONE': EigerAcq.Camera.NoCompression,
'LZ4': EigerAcq.Camera.LZ4,
'BSLZ4': EigerAcq.Camera.BSLZ4} 'BSLZ4': EigerAcq.Camera.BSLZ4}
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment