Skip to content

wattsworth / ethstream

Go to a project
Commit a57c0c22 by zacharyc
Options

Massive cleanup and refactoring of data_stream function. 


git-svn-id: https://bucket.mit.edu/svn/nilm/acquisition/ethstream@7431 ddd99763-3ecb-0310-9145-efcb8ce7c51f
parent 2e3f2f4b
Inline Side-by-side
Showing with 122 additions and 193 deletions
ethstream.c
...@@ -146,7 +146,7 @@ main (int argc, char *argv[]) ...@@ -146,7 +146,7 @@ main (int argc, char *argv[])
} }
//We do not want to overflow channel_list, so we need the check here //We do not want to overflow channel_list, so we need the check here
//The rest of the sanity checking can come later after we know //The rest of the sanity checking can come later after we know
//whether this is a //whether this is a
//LabJack or a NerdJack //LabJack or a NerdJack
#if UE9_CHANNELS > NERDJACK_CHANNELS #if UE9_CHANNELS > NERDJACK_CHANNELS
if (channel_count >= UE9_CHANNELS) if (channel_count >= UE9_CHANNELS)
...@@ -227,7 +227,7 @@ main (int argc, char *argv[]) ...@@ -227,7 +227,7 @@ main (int argc, char *argv[])
case 'V': case 'V':
printf ("etherstream " VERSION "\n"); printf ("etherstream " VERSION "\n");
printf ("Written by Jim Paris <jim@jtan.com>\n"); printf ("Written by Jim Paris <jim@jtan.com>\n");
printf ("and Zachary Clifford <zacharyc@mit.edu>\n"); printf ("and Zachary Clifford <zacharyc@mit.edu>\n");
printf ("This program comes with no warranty and is " printf ("This program comes with no warranty and is "
"provided under the GPLv2.\n"); "provided under the GPLv2.\n");
return 0; return 0;
...@@ -445,31 +445,34 @@ tryagain: ...@@ -445,31 +445,34 @@ tryagain:
goto out; goto out;
} }
//We've tried communicating, so this is not the first call anymore
first_call = 0;
if (nerd_send_command (address, &command, sizeof (command)) < 0) if (nerd_send_command (address, &command, sizeof (command)) < 0)
{ {
info ("Failed to send GET command\n"); info ("Failed to send GET command\n");
goto out; goto out;
} }
} else {
//If we had a transmission in progress, send a command to resume from there }
else
{
//If we had a transmission in progress, send a command to resume from there
char cmdbuf[10]; char cmdbuf[10];
sprintf (cmdbuf, "SETC%05hd", currentcount); sprintf (cmdbuf, "SETC%05hd", currentcount);
retval = nerd_send_command (address, cmdbuf, strlen (cmdbuf)); retval = nerd_send_command (address, cmdbuf, strlen (cmdbuf));
if(retval == -4) { if (retval == -4)
info("NerdJack was reset\n"); {
//Assume we have not started yet, reset on this side. info ("NerdJack was reset\n");
//If this routine is retried, start over //Assume we have not started yet, reset on this side.
printf("# NerdJack was reset here\n"); //If this routine is retried, start over
currentcount = 0; printf ("# NerdJack was reset here\n");
started = 0; currentcount = 0;
wasreset = 1; started = 0;
goto tryagain; wasreset = 1;
} else if(retval < 0) { goto tryagain;
info ("Failed to send SETC command\n"); }
goto out; else if (retval < 0)
{
info ("Failed to send SETC command\n");
goto out;
} }
} }
...@@ -485,14 +488,14 @@ tryagain: ...@@ -485,14 +488,14 @@ tryagain:
} }
retval = nerd_data_stream retval = nerd_data_stream
(fd_data, channel_count, channel_list, precision, convert, lines, (fd_data, channel_count, channel_list, precision, convert, lines,
showmem, &currentcount, period, wasreset); showmem, &currentcount, period, wasreset);
wasreset = 0; wasreset = 0;
if(retval == -3) if (retval == -3)
{ {
retval = 0; retval = 0;
} }
if(retval < 0) if (retval < 0)
{ {
info ("Failed to open data stream\n"); info ("Failed to open data stream\n");
goto out1; goto out1;
...@@ -504,6 +507,8 @@ tryagain: ...@@ -504,6 +507,8 @@ tryagain:
out1: out1:
nerd_close_conn (fd_data); nerd_close_conn (fd_data);
out: out:
//We've tried communicating, so this is not the first call anymore
first_call = 0;
return retval; return retval;
} }
......
nerdjack.c
...@@ -27,13 +27,6 @@ ...@@ -27,13 +27,6 @@
#define NERDJACK_TIMEOUT 5 /* Timeout for connect/send/recv, in seconds */ #define NERDJACK_TIMEOUT 5 /* Timeout for connect/send/recv, in seconds */
//Struct holding information about how channels should be reordered for output
typedef struct
{
int numCopies;
int *destlist;
} deststruct;
#define NERD_HEADER_SIZE 8 #define NERD_HEADER_SIZE 8
typedef struct __attribute__ ((__packed__)) typedef struct __attribute__ ((__packed__))
...@@ -211,85 +204,45 @@ nerd_send_command (const char *address, void *command, int length) ...@@ -211,85 +204,45 @@ nerd_send_command (const char *address, void *command, int length)
return 0; return 0;
} }
/*
* Initialize the channel structure to distill how data should be displayed
*/
static void
nerd_init_channels (deststruct * destination, int numChannels,
int numChannelsSampled, int *channel_list)
{
int channelprocessing = 0;
int currentalign = 0; //Index into sampled channels
int i;
int tempdestlist[NERDJACK_CHANNELS];
//Clear out destination stuff
for (i = 0; i < numChannelsSampled; i++)
{
destination[i].numCopies = 0;
}
for (channelprocessing = 0; channelprocessing < numChannelsSampled;
channelprocessing++)
{
//Find out how many copies of each channel so we malloc the right things
currentalign = 0;
for (i = 0; i < numChannels; i++)
{
if (channelprocessing == channel_list[i])
{
tempdestlist[currentalign] = i;
currentalign++;
}
}
//If this channel is wanted, set it up.
if (currentalign > 0)
{
destination[channelprocessing].numCopies = currentalign;
destination[channelprocessing].destlist =
malloc (destination[channelprocessing].numCopies * sizeof (int));
memcpy (destination[channelprocessing].destlist, tempdestlist,
destination[channelprocessing].numCopies * sizeof (int));
}
}
return;
}
int int
nerd_data_stream (int data_fd, int numChannels, int *channel_list, nerd_data_stream (int data_fd, int numChannels, int *channel_list,
int precision, int convert, int lines, int showmem, int precision, int convert, int lines, int showmem,
unsigned short *currentcount, unsigned int period, unsigned short *currentcount, unsigned int period,
int wasreset) int wasreset)
{ {
//Variables that should persist across retries //Variables that should persist across retries
static dataPacket buf; static dataPacket buf;
static int linesleft = 0; static int linesleft = 0;
static int linesdumped = 0; static int linesdumped = 0;
int index = 0; //Variables essential to packet processing
int charsprocessed = 0;
int alignment = 0;
signed short datapoint = 0; signed short datapoint = 0;
unsigned short dataline[NERDJACK_CHANNELS];
long double voltline[NERDJACK_CHANNELS];
int i; int i;
int numChannelsSampled = channel_list[0] + 1;
//The number sampled will be the highest channel requested plus 1
//(i.e. channel 0 requested means 1 sampled)
for (i = 0; i < numChannels; i++)
{
if (channel_list[i] + 1 > numChannelsSampled)
numChannelsSampled = channel_list[i] + 1;
}
long double voltline[numChannels];
unsigned short dataline[numChannels];
//unsigned long memused = 0;
unsigned short packetsready = 0; unsigned short packetsready = 0;
unsigned short adcused = 0; unsigned short adcused = 0;
unsigned short tempshort = 0; unsigned short tempshort = 0;
int charsread = 0; int charsread = 0;
int numgroups = 0; int numgroupsProcessed = 0;
long double volts; long double volts;
//The timeout should be the expected time plus two seconds //The timeout should be the expected time plus two seconds
//This permits slower speeds to work properly //This permits slower speeds to work properly
unsigned int expectedtimeout = unsigned int expectedtimeout =
...@@ -302,26 +255,12 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list, ...@@ -302,26 +255,12 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list,
linesleft = lines; linesleft = lines;
} }
if(wasreset) { //If there was a reset, we still need to dump a line because of faulty PDCA start
linesdumped = 0; if (wasreset)
}
int numChannelsSampled = channel_list[0] + 1;
//The number sampled will be the highest channel requested plus 1
//(i.e. channel 0 requested means 1 sampled)
for (i = 0; i < numChannels; i++)
{ {
if (channel_list[i] + 1 > numChannelsSampled) linesdumped = 0;
numChannelsSampled = channel_list[i] + 1;
} }
deststruct destination[numChannelsSampled];
nerd_init_channels (destination, numChannels, numChannelsSampled,
channel_list);
//If this is the first time called, warn the user if we're too fast //If this is the first time called, warn the user if we're too fast
if (linesdumped == 0) if (linesdumped == 0)
{ {
...@@ -334,7 +273,7 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list, ...@@ -334,7 +273,7 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list,
//Now destination structure array is set as well as numDuplicates. //Now destination structure array is set as well as numDuplicates.
int numGroups = NERDJACK_NUM_SAMPLES / numChannelsSampled; int totalGroups = NERDJACK_NUM_SAMPLES / numChannelsSampled;
//Loop forever to grab data //Loop forever to grab data
...@@ -372,18 +311,9 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list, ...@@ -372,18 +311,9 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list,
//Increment number of packets received //Increment number of packets received
*currentcount = *currentcount + 1; *currentcount = *currentcount + 1;
//Process the rest of the header and update the index value to be pointing after it
charsprocessed = NERD_HEADER_SIZE;
//memused = ntohl (buf.lwipmemoryused);
adcused = ntohs (buf.adcused); adcused = ntohs (buf.adcused);
packetsready = ntohs (buf.packetsready); packetsready = ntohs (buf.packetsready);
alignment = 0; numgroupsProcessed = 0;
numgroups = 0;
//if(memused) {
// printf("#Packet lost\n");
// continue;
//}
if (showmem) if (showmem)
{ {
...@@ -391,18 +321,35 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list, ...@@ -391,18 +321,35 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list,
continue; continue;
} }
index = 0;
//While there is still more data in the packet, process it //While there is still more data in the packet, process it
//use the destination structure to load the line before printing while (numgroupsProcessed < totalGroups)
while (charsread > charsprocessed)
{ {
datapoint = ntohs (buf.data[index]); //Poison the data structure
if (destination[alignment].numCopies != 0) switch (convert)
{
case CONVERT_VOLTS:
memset (voltline, 0, numChannels * sizeof (long double));
break;
default:
case CONVERT_HEX:
case CONVERT_DEC:
memset (dataline, 0, numChannels * sizeof (unsigned char));
}
//Read in each group
for (i = 0; i < numChannels; i++)
{ {
//Get the datapoint associated with the desired channel
datapoint =
ntohs (buf.
data[channel_list[i] +
numgroupsProcessed * numChannelsSampled]);
//Place it into the line
switch (convert) switch (convert)
{ {
case CONVERT_VOLTS: case CONVERT_VOLTS:
if (alignment <= 5) if (channel_list[i] <= 5)
{ {
volts = volts =
(long double) (datapoint / 32767.0) * (long double) (datapoint / 32767.0) *
...@@ -414,74 +361,48 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list, ...@@ -414,74 +361,48 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list,
(long double) (datapoint / 32767.0) * (long double) (datapoint / 32767.0) *
((precision & 0x02) ? 5.0 : 10.0); ((precision & 0x02) ? 5.0 : 10.0);
} }
for (i = 0; i < destination[alignment].numCopies; i++) voltline[i] = volts;
{
voltline[destination[alignment].destlist[i]] = volts;
}
break; break;
default: default:
case CONVERT_HEX: case CONVERT_HEX:
case CONVERT_DEC: case CONVERT_DEC:
for (i = 0; i < destination[alignment].numCopies; i++) dataline[i] = (unsigned short) (datapoint - INT16_MIN);
{
dataline[destination[alignment].destlist[i]] =
(unsigned short) (datapoint - INT16_MIN);
}
break; break;
} }
} }
//We want to dump the first line because it's usually spurious
//Each point is two bytes, so increment index and total bytes read if (linesdumped != 0)
charsprocessed++;
charsprocessed++;
index++;
alignment++;
//Since channel data is packed, we need to know when to insert a newline
if (alignment == numChannelsSampled)
{ {
//We want to dump the first line because it's usually spurious //Now print the group
if (linesdumped != 0) switch (convert)
{ {
switch (convert) case CONVERT_VOLTS:
for (i = 0; i < numChannels; i++)
{ {
case CONVERT_VOLTS: if (printf ("%Lf ", voltline[i]) < 0)
for (i = 0; i < numChannels; i++) goto bad;
{
if (printf ("%Lf ", voltline[i]) < 0)
goto bad;
}
break;
case CONVERT_HEX:
for (i = 0; i < numChannels; i++)
{
if (printf ("%04hX", dataline[i]) < 0)
goto bad;
}
break;
default:
case CONVERT_DEC:
for (i = 0; i < numChannels; i++)
{
if (printf ("%hu ", dataline[i]) < 0)
goto bad;
}
break;
} }
if (printf ("\n") < 0) break;
goto bad; case CONVERT_HEX:
} for (i = 0; i < numChannels; i++)
else
{
linesdumped = linesdumped + 1;
if (lines != 0)
{ {
linesleft++; if (printf ("%04hX", dataline[i]) < 0)
goto bad;
} }
break;
default:
case CONVERT_DEC:
for (i = 0; i < numChannels; i++)
{
if (printf ("%hu ", dataline[i]) < 0)
goto bad;
}
break;
} }
alignment = 0; if (printf ("\n") < 0)
numgroups++; goto bad;
//If we're counting lines, decrement them
if (lines != 0) if (lines != 0)
{ {
linesleft--; linesleft--;
...@@ -490,15 +411,17 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list, ...@@ -490,15 +411,17 @@ nerd_data_stream (int data_fd, int numChannels, int *channel_list,
return 0; return 0;
} }
} }
//If numgroups so far is equal to the numGroups in a packet, this packet is done
if (numgroups == numGroups) }
{ else
break; {
} linesdumped = linesdumped + 1;
} }
//We've processed this group, so advance the counter
numgroupsProcessed++;
} }
index = 0;
charsprocessed = 0;
} }
return 0; return 0;
...@@ -576,15 +499,17 @@ nerd_generate_command (getPacket * command, int *channel_list, ...@@ -576,15 +499,17 @@ nerd_generate_command (getPacket * command, int *channel_list,
for (i = 0; i < channel_count; i++) for (i = 0; i < channel_count; i++)
{ {
if (channel_list[i] > highestchannel) { if (channel_list[i] > highestchannel)
highestchannel = channel_list[i]; {
} highestchannel = channel_list[i];
}
//channelbit = channelbit | (0x1 << channel_list[i]); //channelbit = channelbit | (0x1 << channel_list[i]);
} }
for( i = 0; i <= highestchannel; i++) { for (i = 0; i <= highestchannel; i++)
channelbit = channelbit | (0x01 << i); {
} channelbit = channelbit | (0x01 << i);
}
command->word[0] = 'G'; command->word[0] = 'G';
command->word[1] = 'E'; command->word[1] = 'E';
...@@ -606,4 +531,3 @@ nerd_close_conn (int data_fd) ...@@ -606,4 +531,3 @@ nerd_close_conn (int data_fd)
close (data_fd); close (data_fd);
return 0; return 0;
} }
nerdjack.h
...@@ -50,7 +50,7 @@ int nerd_send_command (const char *address, void *command, int length); ...@@ -50,7 +50,7 @@ int nerd_send_command (const char *address, void *command, int length);
int nerd_data_stream (int data_fd, int numChannels, int *channel_list, int nerd_data_stream (int data_fd, int numChannels, int *channel_list,
int precision, int convert, int lines, int showmem, int precision, int convert, int lines, int showmem,
unsigned short *currentcount, unsigned int period, unsigned short *currentcount, unsigned int period,
int wasreset); int wasreset);
/* Detect the IP Address of the NerdJack and return in ipAddress */ /* Detect the IP Address of the NerdJack and return in ipAddress */
int nerdjack_detect (char *ipAddress); int nerdjack_detect (char *ipAddress);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment