/* * 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 . */ #include #ifdef __APPLE__ #include #else #include #endif #include "ufo-roof.h" #include "ufo-roof-buffer.h" #include "ufo-roof-build-task.h" typedef enum { BUILD_AUTO = 0, BUILD_RAW, BUILD_SINO, BUILD_UFO } BuildType; struct _UfoRoofBuildTaskPrivate { gchar *config; // ROOF configuration file name UfoRoofConfig *cfg; // Parsed ROOF parameters UfoRoofBuffer *buf; // Ring buffer for incomming UDP packet BuildType build; // What dataset do we build: ROOF sinogram or raw network data guint number; // Number of datasets to read gboolean stop; // Stop flag gboolean simulate; // Indicates if we are running in network or simulation modes guint announced; // For debugging struct timespec last_fragment_timestamp; }; static void ufo_task_interface_init (UfoTaskIface *iface); G_DEFINE_TYPE_WITH_CODE (UfoRoofBuildTask, ufo_roof_build_task, UFO_TYPE_TASK_NODE, G_IMPLEMENT_INTERFACE (UFO_TYPE_TASK, ufo_task_interface_init)) #define UFO_ROOF_BUILD_TASK_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE((obj), UFO_TYPE_ROOF_BUILD_TASK, UfoRoofBuildTaskPrivate)) static GEnumValue build_values[] = { { BUILD_AUTO, "BUILD_AUTO", "auto" }, { BUILD_RAW, "BUILD_RAW", "raw" }, { BUILD_SINO, "BUILD_SINO", "sino" }, { BUILD_UFO, "BUILD_UFO", "ufo" }, { 0, NULL, NULL} }; enum { PROP_0, PROP_STOP, PROP_SIMULATE, PROP_NUMBER, PROP_BUILD, PROP_CONFIG, N_PROPERTIES }; static GParamSpec *properties[N_PROPERTIES] = { NULL, }; UfoNode * ufo_roof_build_task_new (void) { return UFO_NODE (g_object_new (UFO_TYPE_ROOF_BUILD_TASK, NULL)); } static void ufo_roof_build_task_setup (UfoTask *task, UfoResources *resources, GError **error) { GError *gerr = NULL; UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_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-build-setup: "); if (priv->build == BUILD_AUTO) { if (priv->cfg->roof_mode) priv->build = BUILD_SINO; else priv->build = BUILD_RAW; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_BUILD]); } priv->buf = ufo_roof_buffer_new(priv->cfg, (priv->build == BUILD_RAW)?1:2, priv->number, &gerr); if (!priv->buf) roof_propagate_error(error, gerr, "roof-build-setup: "); clock_gettime(CLOCK_REALTIME, &priv->last_fragment_timestamp); } static void ufo_roof_build_task_finalize (GObject *object) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (object); if (priv->buf) { ufo_roof_buffer_free(priv->buf); priv->buf = NULL; } if (priv->cfg) { ufo_roof_config_free(priv->cfg); priv->cfg = NULL; } if (priv->config) { g_free(priv->config); priv->config = NULL; } G_OBJECT_CLASS (ufo_roof_build_task_parent_class)->finalize (object); } static void ufo_roof_build_task_get_requisition (UfoTask *task, UfoBuffer **inputs, UfoRequisition *requisition, GError **error) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); // FIXME: Can we handle data types more elegant? if (priv->build == BUILD_RAW) { guint bytes = priv->cfg->dataset_size; requisition->n_dims = 1; requisition->dims[0] = bytes / sizeof(float) + ((bytes%sizeof(float))?1:0); } else if (priv->build == BUILD_SINO) { guint bytes = priv->cfg->fan_bins * priv->cfg->bit_depth / 8; requisition->n_dims = 2; requisition->dims[0] = bytes / sizeof(float) + ((bytes%sizeof(float))?1:0); requisition->dims[1] = priv->cfg->fan_projections; } else if (priv->build == BUILD_UFO) { requisition->n_dims = 2; requisition->dims[0] = priv->cfg->fan_bins; requisition->dims[1] = priv->cfg->fan_projections; } } static guint ufo_roof_build_task_get_num_inputs (UfoTask *task) { return 1; } static guint ufo_roof_build_task_get_num_dimensions (UfoTask *task, guint input) { return 1; } static UfoTaskMode ufo_roof_build_task_get_mode (UfoTask *task) { return UFO_TASK_MODE_CPU | UFO_TASK_MODE_REDUCTOR; } static gboolean ufo_roof_build_task_process (UfoTask *task, UfoBuffer **inputs, UfoBuffer *output, UfoRequisition *requisition) { GError *gerr = NULL; UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); UfoRoofConfig *cfg = priv->cfg; UfoRoofBuffer *buf = priv->buf; gboolean ready = FALSE; // UfoRequisition in_req; // ufo_buffer_get_requisition (inputs[0], &in_req); const uint8_t *data = (uint8_t*)ufo_buffer_get_host_array(inputs[0], NULL); UfoRoofPacketBlockHeader *header = UFO_ROOF_PACKET_BLOCK_HEADER(data, cfg); if (priv->stop) return FALSE; const uint8_t *fragment = data; for (guint i = 0; i < header->n_packets; i++) { guint packet_id = 0; // Otherwise considered consecutive and handled by the buffer if (cfg->header_size >= sizeof(UfoRoofPacketHeader)) { UfoRoofPacketHeader *pheader = UFO_ROOF_PACKET_HEADER(fragment); packet_id = pheader->packet_id + 1; } // FIXME: Can we kill here the dataset finished during the previous step of iteration ready |= ufo_roof_buffer_set_fragment(buf, header->channel_id, packet_id, fragment + cfg->header_size, &gerr); if (gerr) roof_print_error(gerr); fragment += cfg->max_packet_size; } // FIXME: if 2nd dataset is ready (2nd and 3rd?), skip the first one? if (ready) { clock_gettime(CLOCK_REALTIME, &priv->last_fragment_timestamp); } else { struct timespec current_time; clock_gettime(CLOCK_REALTIME, ¤t_time); // No new accepted events within timeout if (((current_time.tv_sec - priv->last_fragment_timestamp.tv_sec) * 1000000 + (current_time.tv_nsec - priv->last_fragment_timestamp.tv_nsec) / 1000) > cfg->network_timeout) { priv->stop = TRUE; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]); } } /* printf("proc (%s - %u of %u) - channel: %i, packets: %i, first dataset: %i\n", ready?"yes":" no", buf->n_fragments[(buf->current_id)%buf->ring_size], buf->fragments_per_dataset, header->channel_id, header->n_packets, (cfg->header_size >= sizeof(UfoRoofPacketHeader))?UFO_ROOF_PACKET_HEADER(data)->packet_id / (cfg->dataset_size / cfg->payload_size / cfg->n_streams):0); */ return !ready; } static gboolean ufo_roof_build_task_generate (UfoTask *task, UfoBuffer *output, UfoRequisition *requisition) { gboolean ready = FALSE; gulong seqid; GError *gerr = NULL; GValue ival = G_VALUE_INIT; GValue lval = G_VALUE_INIT; UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); UfoRoofConfig *cfg = priv->cfg; UfoRoofBuffer *buf = priv->buf; void *output_buffer = ufo_buffer_get_host_array(output, NULL); if (priv->stop) return FALSE; ready = ufo_roof_buffer_get_dataset(buf, output_buffer, &seqid, &gerr); if (gerr) roof_print_error(gerr); if (priv->build == BUILD_UFO) { switch (cfg->bit_depth) { case 8: ufo_buffer_convert(output, UFO_BUFFER_DEPTH_8U); break; case 16: ufo_buffer_convert(output, UFO_BUFFER_DEPTH_16U); break; case 32: ufo_buffer_convert(output, UFO_BUFFER_DEPTH_32U); break; default: printf("Usupported bit-depth %u\n", cfg->bit_depth); } } // Metadata: plane and sequential number within the plane g_value_init (&ival, G_TYPE_UINT); g_value_init (&lval, G_TYPE_ULONG); if (priv->build != BUILD_UFO) { g_value_set_uint (&ival, cfg->bit_depth); ufo_buffer_set_metadata (output, "bpp", &ival); } g_value_set_uint (&ival, 1 + seqid % cfg->n_planes); ufo_buffer_set_metadata (output, "plane", &ival); g_value_set_ulong (&lval, seqid / cfg->n_planes); ufo_buffer_set_metadata (output, "plane_id", &lval); g_value_set_ulong (&lval, seqid); ufo_buffer_set_metadata (output, "seqid", &lval); g_value_unset(&lval); g_value_unset(&ival); // FIXME: Or shall we start from counting from the ID of the first registerd dataset if ((priv->number)&&(buf->current_id >= priv->number)) { // printf("%u datasets processed, stopping\n", buf->current_id); priv->stop = TRUE; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]); } if (((priv->number > 0)&&(priv->number <= 100))||((buf->current_id - priv->announced) > 1000)) { printf("Generating dataset %i (%s), next: %u out of %u)\n", buf->current_id, ready?"yes":" no", buf->n_fragments[buf->current_id%buf->ring_size], buf->fragments_per_dataset); priv->announced = buf->current_id; } return ready; } static void ufo_roof_build_task_set_property (GObject *object, guint property_id, const GValue *value, GParamSpec *pspec) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_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_STOP: priv->stop = g_value_get_boolean (value); break; case PROP_SIMULATE: priv->simulate = g_value_get_boolean (value); break; case PROP_NUMBER: priv->number = g_value_get_uint (value); break; case PROP_BUILD: priv->build = g_value_get_enum (value); if ((priv->build == BUILD_AUTO)&&(priv->cfg)) { if (priv->cfg->roof_mode) priv->build = BUILD_SINO; else priv->build = BUILD_RAW; } break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec); break; } } static void ufo_roof_build_task_get_property (GObject *object, guint property_id, GValue *value, GParamSpec *pspec) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (object); switch (property_id) { case PROP_CONFIG: g_value_set_string(value, priv->config); 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_NUMBER: g_value_set_uint (value, priv->number); break; case PROP_BUILD: g_value_set_enum (value, priv->build); 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_build_task_setup; iface->get_num_inputs = ufo_roof_build_task_get_num_inputs; iface->get_num_dimensions = ufo_roof_build_task_get_num_dimensions; iface->get_mode = ufo_roof_build_task_get_mode; iface->get_requisition = ufo_roof_build_task_get_requisition; iface->process = ufo_roof_build_task_process; iface->generate = ufo_roof_build_task_generate; } static void ufo_roof_build_task_class_init (UfoRoofBuildTaskClass *klass) { GObjectClass *oclass = G_OBJECT_CLASS (klass); oclass->set_property = ufo_roof_build_task_set_property; oclass->get_property = ufo_roof_build_task_get_property; oclass->finalize = ufo_roof_build_task_finalize; properties[PROP_CONFIG] = g_param_spec_string ("config", "ROOF configuration", "Path to ROOF configuration file", "", 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_NUMBER] = g_param_spec_uint("number", "Number of datasets to receive", "Number of datasets to receive", 0, G_MAXUINT, 0, G_PARAM_READWRITE); properties[PROP_BUILD] = g_param_spec_enum ("build", "Build type (\"raw\", \"sino\", \"ufo\")", "Build type (\"raw\" - raw data, \"sino\" - arrange in sinogram, \"ufo\" - arrange in sinogram and convert UFO floating-point format)", g_enum_register_static ("build", build_values), 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(UfoRoofBuildTaskPrivate)); } static void ufo_roof_build_task_init(UfoRoofBuildTask *self) { self->priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE(self); }