diff --git a/sohstationviewer/controller/util.py b/sohstationviewer/controller/util.py
index 04fd4408d5b700c70fcbd8a87bcbd5ca6f935f14..33693f35f9f97e34b736867444042279cf44ec5b 100644
--- a/sohstationviewer/controller/util.py
+++ b/sohstationviewer/controller/util.py
@@ -149,29 +149,24 @@ def get_time_6_4y(time_str: str) -> Tuple[float, int]:
return utc_time.timestamp, time.year
-def get_time_4(time_str: str, tracking_year: int, y_added: bool
- ) -> Tuple[float, int, bool]:
+def get_time_4(time_str: str, nearest_prior_epoch: float) -> float:
"""
- Get time from 4 parts string. (day of year:hour:minute:second)
+ Get time from 4 parts string (day of year:hour:minute:second). If the day
+ of year in the time string is smaller than the
Ex: 253:19:41:42
:param time_str: time string
- :param tracking_year: year that has been detected
- :param y_added: flag to tell if year has been plussed 1 or not
- :return:
- + utc_time.timestamp: epoch time
- + time.year: year
- + y_added: flag to tell if year has been plussed 1 or not
- """
- if not y_added:
- # first day => move to next year
- doy = int(time_str.split(':')[0])
- if doy == 1:
- tracking_year += 1
- y_added = True
- time_str = f'{str(tracking_year)}:{time_str}'
+ :param nearest_prior_epoch: the nearest epoch time detected
+ :return: the epoch time parsed from time_str
+ """
+ nearest_utcdatetime = UTCDateTime(nearest_prior_epoch)
+ year = nearest_utcdatetime.year
+ nearest_doy = nearest_utcdatetime.julday
+ time_str_doy = int(time_str[:3])
+ if nearest_doy > time_str_doy:
+ year += 1
+ time_str = f'{str(year)}:{time_str}'
time = datetime.strptime(time_str, "%Y:%j:%H:%M:%S")
- utc_time = UTCDateTime(time)
- return utc_time.timestamp, time.year, y_added
+ return UTCDateTime(time).timestamp
def get_val(text: str) -> float:
diff --git a/sohstationviewer/database/extract_data.py b/sohstationviewer/database/extract_data.py
index eef25836e57b2ca3a557aa49edbf2a35b8adeb89..183bfcf19a9deb89f5a2f4a422a783808cdb3ec5 100755
--- a/sohstationviewer/database/extract_data.py
+++ b/sohstationviewer/database/extract_data.py
@@ -25,7 +25,7 @@ def get_chan_plot_info(org_chan_id: str, data_type: str,
chan = org_chan_id
chan = convert_actual_channel_to_db_channel_w_question_mark(chan)
- if org_chan_id.startswith('DS'):
+ if len(org_chan_id) == 3 and org_chan_id.startswith('DS'):
chan = 'SEISMIC'
if dbConf['seisRE'].match(chan):
chan = 'SEISMIC'
diff --git a/sohstationviewer/model/reftek_data/log_info.py b/sohstationviewer/model/reftek_data/log_info.py
index 8c2f46b46baa199d564df483364fe01d6197a235..c2f1c40ddd919181969e97c3fd6fb4450313bd34 100644
--- a/sohstationviewer/model/reftek_data/log_info.py
+++ b/sohstationviewer/model/reftek_data/log_info.py
@@ -123,10 +123,12 @@ class LogInfo():
False
return epoch
- def simple_read(self, line: str) -> Union[Tuple[List[str], float], bool]:
+ def simple_read(self, line: str, header_epoch: float
+ ) -> Union[Tuple[List[str], float], bool]:
"""
Read parts and epoch from an SOH line
:param line: a line of evt message
+ :param header_epoch: the epoch time gotten from the SOH header
:return parts: parts of line with space delim
:return epoch: time when info is recorded
max_epoch is updated with the epoch time.
@@ -134,23 +136,23 @@ class LogInfo():
# Ex: 186:21:41:35 <content>
parts = line.split()
try:
- epoch, self.track_year, self.y_added = get_time_4(
- parts[0], self.track_year, self.y_added)
+ epoch = get_time_4(parts[0], header_epoch)
except AttributeError:
self.parent.processing_log.append(line, LogType.ERROR)
return False
return parts, epoch
- def read_int_clock_phase_err(self, line: str
+ def read_int_clock_phase_err(self, line: str, header_epoch: float
) -> Union[bool, Tuple[float, float]]:
"""
Read internal clock phase error
:param line: str - a line of evt message
+ :param header_epoch: the epoch time gotten from the SOH header
:return epoch: float - time when info is recorded
:return error: float - time of ICP error in microseconds
"""
# Ex: 253:19:41:42 INTERNAL CLOCK PHASE ERROR OF 4823 USECONDS
- ret = self.simple_read(line)
+ ret = self.simple_read(line, header_epoch)
if not ret:
return False
parts, epoch = ret
@@ -159,11 +161,12 @@ class LogInfo():
error *= 1000000.0
return epoch, error
- def read_bat_tem_bkup(self, line: str
+ def read_bat_tem_bkup(self, line: str, header_epoch: float
) -> Union[bool, Tuple[float, float, float, float]]:
"""
Read battery voltage, temperature, backup voltage
:param line: str - a line of evt message
+ :param header_epoch: the epoch time gotten from the SOH header
:return epoch: float - time when info is recorded
:return volts: float - battery voltage.
:return temp: float - temperature of battery in Celsius
@@ -175,7 +178,7 @@ class LogInfo():
# RT130:
# Ex: 186:14:33:58 BATTERY VOLTAGE = 13.6V, TEMPERATURE = 26C,
# BACKUP = 3.3V
- parts, epoch = self.simple_read(line)
+ parts, epoch = self.simple_read(line, header_epoch)
try:
volts = get_val(parts[4])
temp = get_val(parts[7])
@@ -188,11 +191,12 @@ class LogInfo():
bkup_v = 0.0
return epoch, volts, temp, bkup_v
- def read_disk_usage(self, line: str
+ def read_disk_usage(self, line: str, header_epoch: float
) -> Union[bool, Tuple[float, float, float]]:
"""
Read disk usage
:param line: str - a line of evt message
+ :param header_epoch: the epoch time gotten from the SOH header
:return epoch: float - time when info is recorded
:return disk: int - disk number
:return val: int - memory used in disk
@@ -201,7 +205,7 @@ class LogInfo():
# RT130:
# Ex: 186:14:33:58 DISK 1: USED: 89744 AVAIL:...
# Ex: 186:14:33:58 DISK 2* USED: 89744 AVAIL:...
- parts, epoch = self.simple_read(line)
+ parts, epoch = self.simple_read(line, header_epoch)
try:
disk = get_val(parts[2])
val = get_val(parts[4])
@@ -210,16 +214,17 @@ class LogInfo():
return False
return epoch, disk, val
- def read_dps_clock_diff(self, line: str
+ def read_dsp_clock_diff(self, line: str, header_epoch: float
) -> Union[bool, Tuple[float, float]]:
"""
Read DPS clock difference
:param line: str - a line of evt message
+ :param header_epoch: the epoch time gotten from the SOH header
:return epoch: float - time when info is recorded
:return total: float - total difference time in milliseconds
"""
# Ex: 245:07:41:45 DSP CLOCK DIFFERENCE: 0 SECS AND -989 MSECS
- parts, epoch = self.simple_read(line)
+ parts, epoch = self.simple_read(line, header_epoch)
try:
secs = get_val(parts[4])
msecs = get_val(parts[-2])
@@ -302,6 +307,169 @@ class LogInfo():
gps_ver = " ".join(ver_parts[2:])
return gps_ver
+ def read_soh_line(self, line: str, line_idx: int, header_epoch: float):
+ """
+ Read an SOH line and store the parsed data point.
+ :param line: the line being read
+ :param line_idx: the index of the line being read
+ :param header_epoch: the time of the SOH header in epoch time
+ """
+ if "INTERNAL CLOCK PHASE ERROR" in line:
+ ret = self.read_int_clock_phase_err(line, header_epoch)
+ if ret:
+ epoch, err = ret
+ self.add_chan_info('Clk Phase Err', epoch, err, line_idx)
+
+ elif "POSSIBLE DISCREPANCY" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Discrepancies', epoch, 1, line_idx)
+
+ elif "BATTERY VOLTAGE" in line:
+ ret = self.read_bat_tem_bkup(line, header_epoch)
+ if ret:
+ epoch, volts, temp, bkup_v = ret
+ self.add_chan_info('Battery Volt', epoch, volts, line_idx)
+ self.add_chan_info('DAS Temp', epoch, temp, line_idx)
+ # bkup_v: x<3, 3<=x<3.3, >3.3
+ self.add_chan_info('Backup Volt', epoch, bkup_v, line_idx)
+
+ elif all(x in line for x in ['DISK', 'USED']):
+ ret = self.read_disk_usage(line, header_epoch)
+ if ret:
+ epoch, disk, val = ret
+ self.add_chan_info(
+ f'Disk Usage{int(disk)}', epoch, val, line_idx
+ )
+
+ elif "AUTO DUMP CALLED" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Dump Called/Comp', epoch, 1, line_idx)
+ elif "AUTO DUMP COMPLETE" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Dump Called/Comp', epoch, 0, line_idx)
+
+ elif "DSP CLOCK SET" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Jerks/DSP Sets', epoch, 1, line_idx)
+ elif 'JERK' in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Jerks/DSP Sets', epoch, 0, line_idx)
+
+ elif "DSP CLOCK DIFF" in line:
+ ret = self.read_dsp_clock_diff(line, header_epoch)
+ if ret:
+ epoch, total = ret
+ self.add_chan_info('DSP Clock Diff', epoch, total, line_idx)
+
+ elif "ACQUISITION STARTED" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('ACQ On/Off', epoch, 1, line_idx)
+ elif "ACQUISITION STOPPED" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('ACQ On/Off', epoch, 0, line_idx)
+
+ elif "NETWORK LAYER IS UP" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Net Up/Down', epoch, 0, line_idx)
+ elif "NETWORK LAYER IS DOWN" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Net Up/Down', epoch, 1, line_idx)
+
+ elif "MASS RE-CENTER" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Mass Re-center', epoch, 0, line_idx)
+
+ elif any(x in line for x in ["SYSTEM RESET", "FORCE RESET"]):
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Reset/Power-up', epoch, 0, line_idx)
+ elif "SYSTEM POWERUP" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Reset/Power-up', epoch, 1, line_idx)
+
+ # ================= GPS ==================================
+ elif "EXTERNAL CLOCK IS UNLOCKED" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('GPS Lk/Unlk', epoch, -1, line_idx)
+ elif "EXTERNAL CLOCK IS LOCKED" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('GPS Lk/Unlk', epoch, 1, line_idx)
+ elif "EXTERNAL CLOCK CYCLE" in line:
+ # GPS Clock Power
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('GPS Lk/Unlk', epoch, 0, line_idx)
+
+ elif any(x in line for x in ["EXTERNAL CLOCK POWER IS TURNED ON",
+ "EXTERNAL CLOCK WAKEUP",
+ "GPS: POWER IS TURNED ON",
+ "GPS: POWER IS CONTINUOUS"]
+ ):
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('GPS On/Off/Err', epoch, 1, line_idx)
+ elif any(x in line for x in ["EXTERNAL CLOCK POWER IS TURNED OFF",
+ "EXTERNAL CLOCK SLEEP",
+ "GPS: POWER IS TURNED OFF",
+ "NO EXTERNAL CLOCK"]
+ ):
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('GPS On/Off/Err', epoch, 0, line_idx)
+ elif "EXTERNAL CLOCK ERROR" in line:
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('GPS On/Off/Err', epoch, -1, line_idx)
+
+ # ================= VERSIONS ==============================
+ elif any(x in line for x in ["REF TEK", "CPU SOFTWARE"]):
+ cpu_ver = self.read_cpu_ver(line)
+ self.cpu_vers.add(cpu_ver)
+
+ elif any(x in line for x in ["GPS FIRMWARE VERSION:", "GPS: V"]):
+ gps_ver = self.read_gps_ver(line)
+ self.gps_vers.add(gps_ver)
+
+ # ================= ERROR/WARNING =========================
+ elif any(x in line for x in ["ERROR:", "NO VALID DISK", "FAILED"]):
+ # These lines with "ERROR:" are generated by programs like
+ # ref2segy and do not have any time associated with them so
+ # just use whatever time was last in Time.
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Error/Warning', epoch, 1, line_idx)
+
+ elif "WARNING" in line:
+ # Warings come in lines with the time from the RT130 and lines
+ # created by programs without times.
+ if rtn_pattern(line[:12]) == "000:00:00:00":
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Error/Warning', epoch, 0, line_idx)
+ else:
+ # Just use whatever time was last in Time.
+ # The 2 means that the dots will be a little different.
+ self.add_chan_info('Error/Warning', header_epoch, -1, line_idx)
+
+ elif (all(x in line for x in ["BAD", "MAKER"]) or
+ any(x in line for x in ["ERTFS", "IDE BUSY"])):
+ epoch = self.simple_read(line, header_epoch)[1]
+ if epoch:
+ self.add_chan_info('Error/Warning', epoch, 0, line_idx)
+
def add_chan_info(self, chan_id: str, t: float, d: float, idx: int
) -> None:
"""
@@ -351,13 +519,16 @@ class LogInfo():
SOH channels' tracesInfo using add_chan_info()
"""
lines = [ln.strip() for ln in self.log_text.splitlines()]
- soh_epoch = 0
+ is_in_soh_packet = False
+ current_soh_packet_epoch = 0
for idx, line in enumerate(lines):
if line == '':
+ is_in_soh_packet = False
continue
line = line.upper()
+
if 'FST' in line:
ret = self.read_evt(line)
if ret is not False:
@@ -372,167 +543,16 @@ class LogInfo():
self.parent.processing_log.append(
(line, LogType.ERROR))
- elif line.startswith("STATE OF HEALTH"):
- epoch = self.read_sh_header(line)
- if epoch:
- self.add_chan_info('SOH/Data Def', epoch, 1, idx)
elif "DEFINITION" in line:
epoch = self.read_defs(line)
if epoch:
self.add_chan_info('SOH/Data Def', epoch, 0, idx)
- if "INTERNAL CLOCK PHASE ERROR" in line:
- ret = self.read_int_clock_phase_err(line)
- if ret:
- epoch, err = ret
- self.add_chan_info('Clk Phase Err', epoch, err, idx)
-
- elif "POSSIBLE DISCREPANCY" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Discrepancies', epoch, 1, idx)
-
- elif "BATTERY VOLTAGE" in line:
- ret = self.read_bat_tem_bkup(line)
- if ret:
- epoch, volts, temp, bkup_v = ret
- self.add_chan_info('Battery Volt', epoch, volts, idx)
- self.add_chan_info('DAS Temp', epoch, temp, idx)
- # bkup_v: x<3, 3<=x<3.3, >3.3
- self.add_chan_info('Backup Volt', epoch, bkup_v, idx)
-
- elif all(x in line for x in ['DISK', 'USED']):
- ret = self.read_disk_usage(line)
- if ret:
- epoch, disk, val = ret
- self.add_chan_info(
- f'Disk Usage{int(disk)}', epoch, val, idx
- )
-
- elif "AUTO DUMP CALLED" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Dump Called/Comp', epoch, 1, idx)
- elif "AUTO DUMP COMPLETE" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Dump Called/Comp', epoch, 0, idx)
-
- elif "DSP CLOCK SET" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Jerks/DSP Sets', epoch, 1, idx)
- elif 'JERK' in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Jerks/DSP Sets', epoch, 0, idx)
-
- elif "DPS clock diff" in line:
- ret = self.read_dps_clock_diff()
- if ret:
- epoch, total = ret
- self.add_chan_info('DPS Clock Diff', epoch, total, idx)
-
- elif "ACQUISITION STARTED" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('ACQ On/Off', epoch, 1, idx)
- elif "ACQUISITION STOPPED" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('ACQ On/Off', epoch, 0, idx)
-
- elif "NETWORK LAYER IS UP" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Net Up/Down', epoch, 0, idx)
- elif "NETWORK LAYER IS DOWN" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Net Up/Down', epoch, 1, idx)
-
- elif "MASS RE-CENTER" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Mass Re-center', epoch, 0, idx)
-
- elif any(x in line for x in ["SYSTEM RESET", "FORCE RESET"]):
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Reset/Power-up', epoch, 0, idx)
- elif "SYSTEM POWERUP" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Reset/Power-up', epoch, 1, idx)
-
- # ================= GPS ==================================
- elif "EXTERNAL CLOCK IS UNLOCKED" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('GPS Lk/Unlk', epoch, -1, idx)
- elif "EXTERNAL CLOCK IS LOCKED" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('GPS Lk/Unlk', epoch, 1, idx)
- elif "EXTERNAL CLOCK CYCLE" in line:
- # GPS Clock Power
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('GPS Lk/Unlk', epoch, 0, idx)
-
- elif any(x in line for x in ["EXTERNAL CLOCK POWER IS TURNED ON",
- "EXTERNAL CLOCK WAKEUP",
- "GPS: POWER IS TURNED ON",
- "GPS: POWER IS CONTINUOUS"]
- ):
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('GPS On/Off/Err', epoch, 1, idx)
- elif any(x in line for x in ["EXTERNAL CLOCK POWER IS TURNED OFF",
- "EXTERNAL CLOCK SLEEP",
- "GPS: POWER IS TURNED OFF",
- "NO EXTERNAL CLOCK"]
- ):
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('GPS On/Off/Err', epoch, 0, idx)
- elif "EXTERNAL CLOCK ERROR" in line:
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('GPS On/Off/Err', epoch, -1, idx)
-
- # ================= VERSIONS ==============================
- elif any(x in line for x in ["REF TEK", "CPU SOFTWARE"]):
- cpu_ver = self.read_cpu_ver(line)
- self.cpu_vers.add(cpu_ver)
-
- elif any(x in line for x in ["GPS FIRMWARE VERSION:", "GPS: V"]):
- gps_ver = self.read_gps_ver(line)
- self.gps_vers.add(gps_ver)
-
- # ================= ERROR/WARNING =========================
- elif any(x in line for x in ["ERROR:", "NO VALID DISK", "FAILED"]):
- # These lines with "ERROR:" are generated by programs like
- # ref2segy and do not have any time associated with them so
- # just use whatever time was last in Time.
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Error/Warning', epoch, 1, idx)
-
- elif "WARNING" in line:
- # Warings come in lines with the time from the RT130 and lines
- # created by programs without times.
- if rtn_pattern(line[:12]) == "000:00:00:00":
- epoch = self.simple_read(line)[1]
- if epoch:
- self.add_chan_info('Error/Warning', epoch, 0, idx)
- else:
- # Just use whatever time was last in Time.
- # The 2 means that the dots will be a little different.
- self.add_chan_info('Error/Warning', soh_epoch, -1, idx)
-
- elif (all(x in line for x in ["BAD", "MAKER"]) or
- any(x in line for x in ["ERTFS", "IDE BUSY"])):
- epoch = self.simple_read(line)[1]
+ elif line.startswith("STATE OF HEALTH"):
+ is_in_soh_packet = True
+ epoch = self.read_sh_header(line)
if epoch:
- self.add_chan_info('Error/Warning', epoch, 0, idx)
+ current_soh_packet_epoch = epoch
+ self.add_chan_info('SOH/Data Def', epoch, 1, idx)
+ elif is_in_soh_packet:
+ self.read_soh_line(line, idx, current_soh_packet_epoch)
diff --git a/tests/controller/test_util.py b/tests/controller/test_util.py
index 9be6e7c23305c8d28af0a3921daec3c6ca1d303b..ab29a57b4e40d8a13a172237245eb139fbaef29a 100644
--- a/tests/controller/test_util.py
+++ b/tests/controller/test_util.py
@@ -37,7 +37,7 @@ class TestGetTime(BaseTestCase):
"""Set up text fixtures."""
self.time6_2y = '01:251:09:41:35:656'
self.time6_4y = '2001:251:09:41:35:656'
- self.time4_day_1 = '1:09:41:35'
+ self.time4_day_1 = '001:09:41:35'
self.time4 = '251:09:41:35'
@patch('sohstationviewer.controller.util.get_time_6_4y')
@@ -82,45 +82,49 @@ class TestGetTime(BaseTestCase):
self.assertAlmostEqual(epoch_time, 999942095.656)
self.assertEqual(year, 2001)
- def test_get_time4_year_added(self):
- """Test get_time_4 - a year has been added."""
- year = 2001
- year_added = True
- with self.subTest('test_first_day_of_year'):
- epoch_time, ret_year, ret_year_added = (
- get_time_4(self.time4_day_1, year, year_added)
- )
- self.assertEqual(epoch_time, 978342095)
- self.assertEqual(ret_year, year)
- self.assertTrue(ret_year_added)
-
- with self.subTest('test_other_days'):
- epoch_time, ret_year, ret_year_added = (
- get_time_4(self.time4, year, year_added)
- )
- self.assertEqual(epoch_time, 999942095)
- self.assertEqual(ret_year, year)
- self.assertTrue(ret_year_added)
-
- def test_get_time4_year_not_added(self):
- """Test get_time_4 - a year has not been added."""
- year = 2001
- year_added = False
+ def test_get_time4_nearest_epoch_doy_greater_than_time_str_doy(self):
+ """
+ Test get_time_4 - the day of year extracted from the nearest epoch is
+ greater than the day of year extracted from the time string.
+ """
+ # This time is in day 359 of 2017.
+ nearest_epoch = 1514226373
+ with self.subTest('test_time_str_first_day_of_year'):
+ epoch_time = get_time_4(self.time4_day_1, nearest_epoch)
+ self.assertEqual(epoch_time, 1514799695)
+ with self.subTest('test_time_str_other_days'):
+ epoch_time = get_time_4(self.time4, nearest_epoch)
+ self.assertEqual(epoch_time, 1536399695)
+
+ def test_get_time4_nearest_epoch_doy_less_than_time_string_doy(self):
+ """
+ Test get_time_4 - the day of year extracted from the nearest epoch is
+ less than the day of year extracted from the time string.
+ """
+ with self.subTest('test_nearest_epoch_first_day_of_year'):
+ nearest_epoch = 1483272391
+ epoch_time = get_time_4(self.time4, nearest_epoch)
+ self.assertEqual(epoch_time, 1504863695)
+ with self.subTest('test_nearest_epoch_other_days'):
+ # This time is in day 124 of 2017.
+ nearest_epoch = 1493812392
+ epoch_time = get_time_4(self.time4, nearest_epoch)
+ self.assertEqual(epoch_time, 1504863695)
+
+ def test_get_time4_nearest_epoch_doy_equal_to_time_string_doy(self):
+ """
+ Test get_time_4 - the day of year extracted from the nearest epoch is
+ equal to the day of year extracted from the time string.
+ """
with self.subTest('test_first_day_of_year'):
- epoch_time, ret_year, ret_year_added = (
- get_time_4(self.time4_day_1, year, year_added)
- )
- self.assertEqual(epoch_time, 1009878095)
- self.assertEqual(ret_year, year + 1)
- self.assertTrue(ret_year_added)
-
- with self.subTest('test_other_days'):
- epoch_time, ret_year, ret_year_added = (
- get_time_4(self.time4, year, year_added)
- )
- self.assertEqual(epoch_time, 999942095)
- self.assertEqual(ret_year, year)
- self.assertFalse(ret_year_added)
+ nearest_epoch = 1483272391
+ epoch_time = get_time_4(self.time4_day_1, nearest_epoch)
+ self.assertEqual(epoch_time, 1483263695)
+ with self.subTest('test_nearest_epoch_other_days'):
+ # This time is in day 251 of 2017.
+ nearest_epoch = 1504788145
+ epoch_time = get_time_4(self.time4, nearest_epoch)
+ self.assertEqual(epoch_time, 1504863695)
class TestValidateFile(BaseTestCase):