/*
* Copyright (C) 2011-2015 Karlsruhe Institute of Technology
*
* This file is part of Ufo.
*
* This library is free software: you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation, either
* version 3 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library. If not, see .
*/
#define _GNU_SOURCE
#include
#ifdef __APPLE__
#include
#else
#include
#endif
#include "ufo-roof.h"
#include "ufo-roof-read-socket.h"
#include "ufo-roof-read-file.h"
#include "ufo-roof-read-task.h"
struct _UfoRoofReadTaskPrivate {
gchar *config; // ROOF configuration file name
UfoRoofConfig *cfg; // Parsed ROOF parameters
UfoRoofReadInterface *reader;
guint id; // Reader ID (defince sequential port number)
gboolean stop; // Flag requiring termination
gboolean simulate; // Indicates if we are running in network or simulation modes
gchar *path; // UFO file path for simulation mode
guint first_file_number; // Number of a first simulated file (0 or 1)
};
static void ufo_task_interface_init (UfoTaskIface *iface);
G_DEFINE_TYPE_WITH_CODE (UfoRoofReadTask, ufo_roof_read_task, UFO_TYPE_TASK_NODE,
G_IMPLEMENT_INTERFACE (UFO_TYPE_TASK,
ufo_task_interface_init))
#define UFO_ROOF_READ_TASK_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE((obj), UFO_TYPE_ROOF_READ_TASK, UfoRoofReadTaskPrivate))
enum {
PROP_0,
PROP_ID,
PROP_STOP,
PROP_CONFIG,
PROP_SIMULATE,
PROP_PATH,
PROP_FIRST,
N_PROPERTIES
};
static GParamSpec *properties[N_PROPERTIES] = { NULL, };
UfoNode *
ufo_roof_read_task_new (void)
{
return UFO_NODE (g_object_new (UFO_TYPE_ROOF_READ_TASK, NULL));
}
static void
ufo_roof_read_task_setup (UfoTask *task,
UfoResources *resources,
GError **error)
{
GError *gerr = NULL;
UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (task);
if (!priv->config)
roof_setup_error(error, "ROOF configuration is not specified");
priv->cfg = ufo_roof_config_new(priv->config, priv->simulate?UFO_ROOF_CONFIG_SIMULATION:UFO_ROOF_CONFIG_DEFAULT, &gerr);
if (!priv->cfg) roof_propagate_error(error, gerr, "roof_config_new: ");
// Consistency checks
if (priv->id >= priv->cfg->n_streams)
roof_setup_error(error, "Specified Stream ID is %u, but only %u data streams is configured", priv->id, priv->cfg->n_streams);
// Start actual reader
if (priv->simulate) {
if (!priv->path)
roof_setup_error(error, "Path to simulated data should be specified");
priv->reader = ufo_roof_read_file_new(priv->cfg, priv->path, priv->id + priv->first_file_number, &gerr);
} else
priv->reader = ufo_roof_read_socket_new(priv->cfg, priv->id, &gerr);
if (!priv->reader)
roof_propagate_error(error, gerr, "roof_read_new: ");
}
static void
ufo_roof_read_task_finalize (GObject *object)
{
UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (object);
if (priv->reader) {
priv->reader->close(priv->reader);
}
if (priv->cfg) {
ufo_roof_config_free(priv->cfg);
priv->cfg = NULL;
}
if (priv->config) {
g_free(priv->config);
priv->config = NULL;
}
if (priv->path) {
g_free(priv->path);
priv->path = NULL;
}
G_OBJECT_CLASS (ufo_roof_read_task_parent_class)->finalize (object);
}
static void
ufo_roof_read_task_get_requisition (UfoTask *task,
UfoBuffer **inputs,
UfoRequisition *requisition,
GError **error)
{
UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (task);
UfoRoofConfig *cfg = priv->cfg;
guint bytes = cfg->max_packets * cfg->max_packet_size + sizeof(UfoRoofPacketBlockHeader);
// FIXME: Can this be made more elegant?
requisition->n_dims = 1;
requisition->dims[0] = bytes / sizeof(float) + ((bytes%sizeof(float))?1:0);
}
static guint
ufo_roof_read_task_get_num_inputs (UfoTask *task)
{
return 0;
}
static guint
ufo_roof_read_task_get_num_dimensions (UfoTask *task,
guint input)
{
return 0;
}
static UfoTaskMode
ufo_roof_read_task_get_mode (UfoTask *task)
{
return UFO_TASK_MODE_CPU | UFO_TASK_MODE_GENERATOR;
}
static gboolean
ufo_roof_read_task_generate (UfoTask *task,
UfoBuffer *output,
UfoRequisition *requisition)
{
GError *gerr = NULL;
UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (task);
UfoRoofConfig *cfg = priv->cfg;
void *output_buffer = ufo_buffer_get_host_array(output, NULL);
UfoRoofPacketBlockHeader *header = UFO_ROOF_PACKET_BLOCK_HEADER(output_buffer, cfg);
if (priv->stop)
return FALSE;
guint packets = priv->reader->read(priv->reader, output_buffer, &gerr);
if (gerr) {
g_warning("Error reciving data: %s", gerr->message);
g_error_free(gerr);
return FALSE;
}
#ifdef UFO_ROOF_DEBUG
// Store first received packet on each channel...
static int debug = 1;
if (debug) {
char fname[256];
sprintf(fname, "channel%i_packet0.raw", priv->id);
FILE *f = fopen(fname, "w");
if (f) {
fwrite(output_buffer, 1, cfg->max_packets * cfg->max_packet_size, f);
fclose(f);
}
debug = 0;
}
#endif /* UFO_ROOF_DEBUG */
// FIXME: End of data (shall we restart in the network case?)
if (!packets)
return FALSE;
// Shall I use UFO metadata (ufo_buffer_set_metadata) insead?
header->channel_id = priv->id;
header->n_packets = packets;
return TRUE;
}
static void
ufo_roof_read_task_set_property (GObject *object,
guint property_id,
const GValue *value,
GParamSpec *pspec)
{
UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (object);
switch (property_id) {
case PROP_CONFIG:
if (priv->config) g_free(priv->config);
priv->config = g_value_dup_string(value);
break;
case PROP_ID:
priv->id = g_value_get_uint (value);
break;
case PROP_STOP:
priv->stop = g_value_get_boolean (value);
break;
case PROP_SIMULATE:
priv->simulate = g_value_get_boolean (value);
break;
case PROP_PATH:
if (priv->path) g_free(priv->path);
priv->path = g_value_dup_string(value);
break;
case PROP_FIRST:
priv->first_file_number = g_value_get_uint (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
break;
}
}
static void
ufo_roof_read_task_get_property (GObject *object,
guint property_id,
GValue *value,
GParamSpec *pspec)
{
UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (object);
switch (property_id) {
case PROP_CONFIG:
g_value_set_string(value, priv->config);
break;
case PROP_ID:
g_value_set_uint (value, priv->id);
break;
case PROP_STOP:
g_value_set_boolean (value, priv->stop);
break;
case PROP_SIMULATE:
g_value_set_boolean (value, priv->simulate);
break;
case PROP_PATH:
g_value_set_string(value, priv->path?priv->path:"");
break;
case PROP_FIRST:
g_value_set_uint (value, priv->first_file_number);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
break;
}
}
static void
ufo_task_interface_init (UfoTaskIface *iface)
{
iface->setup = ufo_roof_read_task_setup;
iface->get_num_inputs = ufo_roof_read_task_get_num_inputs;
iface->get_num_dimensions = ufo_roof_read_task_get_num_dimensions;
iface->get_mode = ufo_roof_read_task_get_mode;
iface->get_requisition = ufo_roof_read_task_get_requisition;
iface->generate = ufo_roof_read_task_generate;
}
static void
ufo_roof_read_task_class_init (UfoRoofReadTaskClass *klass)
{
GObjectClass *oclass = G_OBJECT_CLASS (klass);
oclass->set_property = ufo_roof_read_task_set_property;
oclass->get_property = ufo_roof_read_task_get_property;
oclass->finalize = ufo_roof_read_task_finalize;
properties[PROP_CONFIG] =
g_param_spec_string ("config",
"ROOF configuration",
"Path to ROOF configuration file",
"",
G_PARAM_READWRITE);
properties[PROP_ID] =
g_param_spec_uint ("id",
"Reader ID",
"ID for multi-port servers",
0, G_MAXUINT, 1,
G_PARAM_READWRITE);
properties[PROP_STOP] =
g_param_spec_boolean ("stop",
"Stop flag",
"Stop socket servers and terminates filter execution",
FALSE,
G_PARAM_READWRITE);
properties[PROP_SIMULATE] =
g_param_spec_boolean ("simulate",
"Simulation mode",
"Read data from the specified files instead of network",
FALSE,
G_PARAM_READWRITE);
properties[PROP_PATH] =
g_param_spec_string ("path",
"Input files for simulation mode",
"Optional path to input files for simulation mode (parameter from configuration file is used if not specified)",
"",
G_PARAM_READWRITE);
properties[PROP_FIRST] =
g_param_spec_uint ("first_file_number",
"Offset to the first read file",
"Offset to the first read file",
0, G_MAXUINT, 0,
G_PARAM_READWRITE);
for (guint i = PROP_0 + 1; i < N_PROPERTIES; i++)
g_object_class_install_property (oclass, i, properties[i]);
g_type_class_add_private (oclass, sizeof(UfoRoofReadTaskPrivate));
}
static void
ufo_roof_read_task_init(UfoRoofReadTask *self)
{
self->priv = UFO_ROOF_READ_TASK_GET_PRIVATE(self);
}