from __future__ import unicode_literals
from builtins import range, str
import socket # noqa
import struct
import sys
from datetime import datetime
from ..TimeseriesFactoryException import TimeseriesFactoryException
from ..TimeseriesUtility import round_usecs
import logging
from obspy.core import UTCDateTime
from time import sleep
"""
MAXINPUTSIZE: Edge uses a short for the data count, so the max input size
for a single data packet to edge is *supposed* to be 32767. However, this
was never tested thoroughly until we added support for 10Hz data, when
it was quickly discovered that setting this value caused Edge to hang.
Therefore, MAXINPUTSIZE was reduced to 10_000, which seems to work well.
HOURSECONDS: The number of seconds in an hour. Used as an arbitrary size
for sending seconds data.
DAYMINUTES: The numbers of minutes in a day. Used as the size for sending
minute data, since Edge stores by the day.
"""
MAXINPUTSIZE = 10_000
HOURSECONDS = 3600
DAYMINUTES = 1440
"""
PACKSTR, TAGSTR: String's used by pack.struct, to indicate the data format
for that packet.
PACKEHEAD: The code that leads a packet being sent to Edge.
"""
PACKSTR = "!1H1h12s4h4B3i"
TAGSTR = "!1H1h12s6i"
PACKETHEAD = 0xA1B2
"""
TAG, FORCEOUT: Flags that indicate to edge that a "data" packet has a specific
function. Goes in the nsamp position of the packet header.
"""
TAG = -1
FORCEOUT = -2
class RawInputClient:
"""RawInputClient for direct to edge data.
Parameters
----------
tag: str
A string used by edge to make certain a socket hasn't been
opened by a different user, and to log transactions.
host: str
The IP address of the target host RawInputServer
port: int
The port on the IP of the RawInputServer
station: str
station code.
channel: str
channel to be written
location: str
location code
network: str
network code
activity: int
The activity flags per the SEED manual
ioClock: int
The IO/CLOCK flags per the SEED manual
quality: int
The data Quality flags per the SEED manual
timingQuality: int [0-100]
The overall timing quality
Raises
------
TimeseriesFactoryException
NOTES
-----
Uses sockets to send data to an edge. See send method for packet encoding
"""
def __init__(
self,
tag="",
host="",
port=0,
station="",
channel="",
location="",
network="",
activity=0,
ioclock=0,
quality=0,
timingquality=0,
):
self.tag = tag
self.host = host
self.port = port
self.activity = activity
self.ioclock = ioclock
self.quality = quality
self.timingquality = timingquality
self.socket = None
self.buf = None
self.sequence = 0
self.seedname = self.create_seedname(station, channel, location, network)
if len(self.tag) > 10:
raise TimeseriesFactoryException("Tag limited to 10 characters")
def close(self):
"""close the open sockets"""
if self.socket is not None:
self.socket.close()
self.socket = None
def create_seedname(self, observatory, channel, location="R0", network="NT"):
"""create a seedname for communication with edge.
PARAMETERS
----------
observatory: str
observatory code.
channel: str
channel to be written
location: str
location code
network: str
network code
RETURNS
-------
str
the seedname
NOTES
-----
The seedname is in the form NNSSSSSCCCLL if a parameter is not
of the correct length, it should be padded with spaces to be of
the correct length. We only expect observatory to ever be of an
incorrect length.
"""
return str(network + observatory.ljust(5) + channel + location).encode()
def forceout(self):
"""force edge to recognize data
NOTES
-----
When sending data to edge it hangs on to the data, until either
enough data has accumulated, or enough time has passed. At that
point, it makes the new data available for reading.
Fourceout tells edge that we're done sending data for now, and
to go ahead and make it available
"""
buf = self._get_forceout(UTCDateTime(datetime.utcnow()), 0.0)
self._send(buf)
def send_trace(self, interval, trace):
"""send an obspy trace using send.
PARAMETERS
----------
interval: {'day', 'hour', 'minute', 'second', 'tenhertz'}
data interval.
trace: obspy.core.trace
NOTES
-----
Edge only takes a short as the max number of samples it takes at one
time. For ease of calculation, we break a trace into managable chunks
according to interval type.
"""
totalsamps = len(trace.data)
starttime = trace.stats.starttime
if interval == "tenhertz":
nsamp = MAXINPUTSIZE
timeoffset = 1 / 10.0
samplerate = 10.0
elif interval == "second":
nsamp = HOURSECONDS
timeoffset = 1
samplerate = 1.0
elif interval == "minute":
nsamp = DAYMINUTES
timeoffset = 60
samplerate = 1.0 / 60
elif interval == "hour":
nsamp = MAXINPUTSIZE
timeoffset = 3600
samplerate = 1.0 / 3600
elif interval == "day":
nsamp = MAXINPUTSIZE
timeoffset = 86400
samplerate = 1.0 / 86400
else:
raise TimeseriesFactoryException("Unsupported interval for RawInputClient")
for i in range(0, totalsamps, nsamp):
if totalsamps - i < nsamp:
endsample = totalsamps
else:
endsample = i + nsamp
nsamp = endsample - i
endtime = starttime + (nsamp - 1) * timeoffset
trace_send = trace.slice(starttime, endtime)
buf = self._get_data(trace_send.data, starttime, samplerate)
self._send(buf)
starttime += nsamp * timeoffset
def _send(self, buf):
"""Send a block of data to the Edge/CWB combination.
PARAMETERS
----------
samples: array like
An int array with the samples
time: UTCDateTime
time of the first sample
rate: int
The data rate in Hertz
forceout: boolean
indicates the packet to be sent will have a nsamp value of -2,
to tell edge to force the data to be written
Raises
------
TimeseriesFactoryException - if the socket will not open
"""
# Try and send the packet, if the socket doesn't exist open it.
try:
if self.socket is None:
self._open_socket()
self.socket.sendall(buf)
self.sequence += 1
except socket.error as v:
error = "Socket error %d" % v[0]
sys.stderr.write(error)
raise TimeseriesFactoryException(error)
def _get_forceout(self, time, rate):
"""
PARAMETERS
----------
time: UTCDateTime
time of the first sample
rate: int
The data rate in Hertz
RETURNS
-------
str
NOTES
-----
Data is encoded into a C style structure using struct.pack with the
following variables and type.
0xa1b2 (short)
nsamp (short)
seedname (12 char)
yr (short)
doy (short)
ratemantissa (short)
ratedivisor (short)
activity (byte)
ioClock (byte)
quality (byte)
timeingQuality (byte)
secs (int)
seq (int) Seems to be the miniseed sequence, but not certain.
basically we increment it for every new packet we send
The nsamp parameter is signed.
-2 is for a force out packet
"""
yr, doy, secs, usecs = self._get_time_values(time)
ratemantissa, ratedivisor = self._get_mantissa_divisor(rate)
buf = struct.pack(
PACKSTR,
PACKETHEAD,
FORCEOUT,
self.seedname,
yr,
doy,
ratemantissa,
ratedivisor,
self.activity,
self.ioclock,
self.quality,
self.timingquality,
secs,
usecs,
self.sequence,
)
return buf
def _get_data(self, samples, time, rate):
"""
PARAMETERS
----------
samples: array like
An int array with the samples
time: UTCDateTime
time of the first sample
rate: int
The data rate in Hertz
RETURNS
-------
str
NOTES
-----
Data is encoded into a C style structure using struct.pack with the
following variables and type.
0xa1b2 (short)
nsamp (short)
seedname (12 char)
yr (short)
doy (short)
ratemantissa (short)
ratedivisor (short)
activity (byte)
ioClock (byte)
quality (byte)
timeingQuality (byte)
secs (int)
seq (int) Seems to be the miniseed sequence, but not certain.
basically we increment it for every new set we send
data [int]
Notice that we expect the data to already be ints.
The nsamp parameter is signed. If it's positive we send a data packet.
"""
nsamp = len(samples)
if nsamp > 32767:
raise TimeseriesFactoryException(
"Edge input limited to 32767 integers per packet."
)
yr, doy, secs, usecs = self._get_time_values(time)
ratemantissa, ratedivisor = self._get_mantissa_divisor(rate)
packStr = "%s%d%s" % (PACKSTR, nsamp, "i")
bpackStr = str(packStr).encode()
buf = struct.pack(
bpackStr,
PACKETHEAD,
nsamp,
self.seedname,
yr,
doy,
ratemantissa,
ratedivisor,
self.activity,
self.ioclock,
self.quality,
self.timingquality,
secs,
usecs,
self.sequence,
*samples,
)
return buf
def _get_mantissa_divisor(self, rate):
"""
PARAMETERS
----------
rate: int
The data rate in Hertz
RETURNS
-------
tuple: (ratemantissa, ratedivosor)
ratemantissa: int
ratedivosor: int
NOTE that `ratemantissa` and `ratedivisor` map directly to the SEED
standard's "sample rate factor" and "sample rate multiplier", which
are stored in a miniseed block's fixed data header as 2-byte words;
earlier versions of this method did not handle sample rates lower
than 1/second properly (they did handle 1/minute as a special-case);
"""
if rate > 0.9999:
# sample rates greater or equal to 1/second
# (this will not handle >327 Hz, but for consistency
# with earlier versions, we leave it unchanged)
ratemantissa = int(rate * 100 + 0.001)
ratedivisor = -100
elif rate * 21600 > 1.0001:
# sample periods between 1 second and 6 hours
# (this should handle 1-minute data identically to
# earlier versions)
ratemantissa = -int(1 / rate + 0.001)
ratedivisor = -1
else:
# sample periods 6 hours and longer
# (this will not handle periods longer than ~22 years)
ratemantissa = -int(1 / rate / 21600)
ratedivisor = -21600
return (ratemantissa, ratedivisor)
def _get_tag(self):
"""Get tag struct
RETURNS
-------
str
NOTES
-----
The tag packet is used to by the edge server to log/determine a new
"user" has connected to the edge, not one who's connection dropped,
and is trying to continue sending data.
The packet uses -1 in the nsamp position to indicate it's a tag packet
The tag is right padded with spaces.
The Packet is right padded with zeros
The Packet must be 40 Bytes long.
"""
tg = str(self.tag + " ").encode()
tb = struct.pack(TAGSTR, PACKETHEAD, TAG, tg[:12], 0, 0, 0, 0, 0, 0)
return tb
def _get_time_values(self, time):
"""
PARAMETERS
----------
input_time: UTCDateTime
time of the first sample
RETURNS
-------
tuple: (yr, doy, secs, usecs)
yr: int
doy: int
secs: int
usecs: int
"""
# check for presence of residual microseconds
if (time.microsecond / 1000).is_integer() == False:
# create warning message when rounding is necessary
logging.warning(
"residual microsecond values encountered, rounding to nearest millisecond"
)
# round residual microsecond values
time = round_usecs(time)
yr = time.year
doy = time.datetime.timetuple().tm_yday
secs = time.hour * 3600 + time.minute * 60 + time.second
usecs = time.microsecond
return (yr, doy, secs, usecs)
def _open_socket(self):
"""Open a socket
NOTES
-----
Loops until a socket is opened, with a 1 second wait between attempts
Sends tag.
"""
done = False
newsocket = None
trys = 0
while not done:
try:
newsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
newsocket.connect((self.host, self.port))
done = True
except socket.error as v:
sys.stderr.write("Could not connect to socket, trying again")
sys.stderr.write("sockect error %d" % v[0])
sleep(1)
if trys > 2:
raise TimeseriesFactoryException("Could not open socket")
trys += 1
self.socket = newsocket
self.socket.sendall(self._get_tag())