/* * 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); }