Skip to content
Snippets Groups Projects
Commit 18696738 authored by Maeva Pourpoint's avatar Maeva Pourpoint
Browse files

Update variable naming in utils module

parent b9f1d292
No related branches found
No related tags found
1 merge request!17Latest updates to the utils and lemi_data modules before MVP release
Hide whitespace changes
Inline Side-by-side
lemi2seed/utils.py
+ 33
− 33
View file @ 18696738
......@@ -5,6 +5,7 @@ validation.
Maeva Pourpoint - IRIS/PASSCAL
"""
from __future__ import annotations
import re
......@@ -17,16 +18,16 @@ from lemi2seed.logging import setup_logger
# Set up logging
logger = setup_logger(__name__)
ELECTRODE_DEFAULT = 'Borin STELTH 4 - Silver-Silver Chloride'
ELEC_DEFAULT = 'Borin STELTH 4 - Silver-Silver Chloride'
FLUXGATE_DEFAULT = 'LEMI-039'
LOC = ['00', '01', '02', '03']
MSG_E_CHA = ("If you did not use that channel number to record data, please "
"update your list of recorded components at the station/runs "
"levels!")
NBR_E_CHANNEL_MAX = 4
NUM_E_CHA_MAX = 4
def check_email_formatting(email: str) -> bool:
def check_email_format(email: str) -> bool:
"""
Basic email check. Check for proper formatting.
Will not catch typos or fake email addresses.
......@@ -45,31 +46,31 @@ def check_email_formatting(email: str) -> bool:
return bool(valid)
def check_instrument_specs(specs: str, equipment: str, channel_info: str = "") -> bool:
def check_inst_specs(specs: str, equipment: str, cha_info: str = "") -> bool:
"""
Check instrument specifications provided by the user.
Both manufacturer and model info are required inputs. Type is optional.
"""
if specs in [ELECTRODE_DEFAULT, FLUXGATE_DEFAULT]:
if specs in [ELEC_DEFAULT, FLUXGATE_DEFAULT]:
return True
elif re.match(r"^Manufacturer: \w* - Model: \w* - Type: \w*$", specs):
manufacturer, model, _ = [x.split(':')[-1].strip() for x in specs.split('-')]
if manufacturer == '':
logger.error("Please provide {0} manufacturer{1}. Required "
"metadata field.".format(equipment, channel_info))
"metadata field.".format(equipment, cha_info))
return False
if model == '':
logger.error("Please provide {0} model{1}. Required metadata "
"field.".format(equipment, channel_info))
"field.".format(equipment, cha_info))
return False
return True
else:
logger.error("Please provide {0} specs for{1}. Required metadata "
"field.".format(equipment, channel_info))
"field.".format(equipment, cha_info))
return False
def compare_serial_number(equipment: str, serial_number: str, data_input: Optional[str] = None) -> bool:
def compare_sn(equipment: str, sn: str, data_input: Optional[str] = None) -> bool:
"""
Compare user provided serial number for the data logger or fluxgate against
the serial number recorded in the .INF file.
......@@ -77,29 +78,28 @@ def compare_serial_number(equipment: str, serial_number: str, data_input: Option
long.
"""
if data_input is not None:
valid = (str(serial_number) == data_input)
valid = (str(sn) == data_input)
if not valid:
logger.error("The serial number of the {} differs from the one "
"automatically recorded by the logger in the field "
"({} != {}). Please correct!"
.format(equipment, serial_number, data_input))
"({} != {}). Please correct!".format(equipment, sn, data_input))
else:
valid = re.match(r"^\d{3}$", str(serial_number)) # type: ignore
valid = re.match(r"^\d{3}$", str(sn)) # type: ignore
if valid is None:
logger.error("The serial number of the {} should be 3 digits "
"long.".format(equipment))
return valid
def check_serial_number(equipment: str, serial_number: str, data_input: Optional[str] = None) -> bool:
def check_sn(equipment: str, sn: str, data_input: Optional[str] = None) -> bool:
"""
Check user provided serial number for their data logger, fluxgate or
electrodes.
"""
if serial_number is not None:
if sn is not None:
valid = True
if 'data logger' in equipment or 'fluxgate' in equipment:
valid = compare_serial_number(equipment, serial_number, data_input)
valid = compare_sn(equipment, sn, data_input)
return valid
else:
msg = "Please provide a serial number for the {}!".format(equipment)
......@@ -109,19 +109,19 @@ def check_serial_number(equipment: str, serial_number: str, data_input: Optional
return False
def convert_coordinate(coordinate: str, hemisphere: str) -> Optional[float]:
def convert_coord(coord: str, hemisphere: str) -> Optional[float]:
"""Convert coordinates outputted by LEMI to decimal degrees."""
if hemisphere not in ['N', 'S', 'E', 'W']:
logger.error("Unexpected hemisphere - {} - listed in data file!"
.format(hemisphere))
return None
try:
coordinate = float(coordinate) / 100 # type: ignore
coord = float(coord) / 100 # type: ignore
except ValueError:
logger.error("Failed to convert geographic coordinate - {} - to "
"decimal degrees!".format(coordinate))
"decimal degrees!".format(coord))
return None
return -coordinate if hemisphere in ["S", "W"] else coordinate # type: ignore
return -coord if hemisphere in ["S", "W"] else coord # type: ignore
def convert_time(time_stamp: str) -> Optional[UTCDateTime]:
......@@ -140,7 +140,7 @@ def convert_time(time_stamp: str) -> Optional[UTCDateTime]:
return None
def evaluate_location_code(e_infos: Dict) -> bool:
def eval_loc_code(e_infos: Dict) -> bool:
"""
Get location code boolean - Needed to define whether location code
should be '' (i.e. undefined)
......@@ -151,10 +151,10 @@ def evaluate_location_code(e_infos: Dict) -> bool:
return False
def get_e_ids(nbr_runs: int) -> List[Tuple[str, int]]:
def get_e_ids(num_runs: int) -> List[Tuple[str, int]]:
"""Get permutation of run ids and channel numbers."""
run_list = get_run_list(nbr_runs)
return [(i, j) for i in run_list for j in range(1, NBR_E_CHANNEL_MAX + 1)]
run_list = get_run_list(num_runs)
return [(i, j) for i in run_list for j in range(1, NUM_E_CHA_MAX + 1)]
def get_e_loc(e_info: Dict) -> Dict:
......@@ -163,23 +163,23 @@ def get_e_loc(e_info: Dict) -> Dict:
Only useful when more than two electrode pairs are deployed at a station.
"""
e_loc = {}
components = {x for x in e_info.values() if x}
for component in components:
channel_nbr = [key for key, val in e_info.items() if val == component]
e_loc.update({key: LOC[ind] for ind, key in enumerate(channel_nbr)})
comps = {x for x in e_info.values() if x}
for comp in comps:
cha_num = [key for key, val in e_info.items() if val == comp]
e_loc.update({key: LOC[ind] for ind, key in enumerate(cha_num)})
return e_loc
def get_run_list(nbr_runs: int) -> List[str]:
def get_run_list(num_runs: int) -> List[str]:
"""Get list of run ids."""
return [chr(x) for x in range(ord('a'), ord('a') + nbr_runs)]
return [chr(x) for x in range(ord('a'), ord('a') + num_runs)]
def get_run_nbr(run_id: str) -> int:
def get_run_num(run_id: str) -> int:
"""Get the integer representation of the string run_id."""
integer_representation_unicode_character_a = 97
run_nbr = ord(run_id) - integer_representation_unicode_character_a + 1
return run_nbr
run_num = ord(run_id) - integer_representation_unicode_character_a + 1
return run_num
def is_empty(val: Any) -> bool:
......
tests/test_utils.py
+ 105
− 105
View file @ 18696738
......@@ -7,10 +7,10 @@ import unittest
from obspy import UTCDateTime
from lemi2seed.logging import setup_logger
from lemi2seed.utils import (convert_coordinate, convert_time, str2list,
check_email_formatting, check_instrument_specs,
check_serial_number, is_empty, evaluate_location_code,
get_e_loc, get_e_ids, get_run_list, get_run_nbr)
from lemi2seed.utils import (convert_coord, convert_time, str2list,
check_email_format, check_inst_specs, check_sn,
is_empty, eval_loc_code, get_e_loc, get_e_ids,
get_run_list, get_run_num)
SCR_DIR = "lemi2seed.utils"
......@@ -18,48 +18,48 @@ SCR_DIR = "lemi2seed.utils"
logger = setup_logger(SCR_DIR)
class TestConvertCoordinate(unittest.TestCase):
"""Test suite for convert_coordinate."""
class TestConvertCoord(unittest.TestCase):
"""Test suite for convert_coord."""
def test_convert_coordinate_positive_lat(self):
"""Test basic functionality of convert_coordinate."""
def test_convert_coord_positive_lat(self):
"""Test basic functionality of convert_coord."""
coordinate = 3404.83926
hemisphere = 'N'
self.assertEqual(convert_coordinate(coordinate, hemisphere),
self.assertEqual(convert_coord(coordinate, hemisphere),
34.0483926)
def test_convert_coordinate_negative_lat(self):
"""Test basic functionality of convert_coordinate."""
def test_convert_coord_negative_lat(self):
"""Test basic functionality of convert_coord."""
coordinate = 2456.43250
hemisphere = 'S'
self.assertEqual(convert_coordinate(coordinate, hemisphere),
self.assertEqual(convert_coord(coordinate, hemisphere),
-24.564325)
def test_convert_coordinate_positive_lon(self):
"""Test basic functionality of convert_coordinate."""
def test_convert_coord_positive_lon(self):
"""Test basic functionality of convert_coord."""
coordinate = 4303.74563
hemisphere = 'E'
self.assertEqual(convert_coordinate(coordinate, hemisphere),
self.assertEqual(convert_coord(coordinate, hemisphere),
43.0374563)
def test_convert_coordinate_negative_lon(self):
"""Test basic functionality of convert_coordinate."""
def test_convert_coord_negative_lon(self):
"""Test basic functionality of convert_coord."""
coordinate = 1071.84450
hemisphere = 'W'
self.assertEqual(convert_coordinate(coordinate, hemisphere),
self.assertEqual(convert_coord(coordinate, hemisphere),
-10.718445)
def test_convert_coordinate_erroneous_hemisphere(self):
"""Test basic functionality of convert_coordinate."""
def test_convert_coord_erroneous_hemisphere(self):
"""Test basic functionality of convert_coord."""
coordinate = 1071.84450
hemisphere = 'B'
self.assertEqual(convert_coordinate(coordinate, hemisphere), None)
self.assertEqual(convert_coord(coordinate, hemisphere), None)
def test_convert_coordinate_erroneous_type(self):
"""Test basic functionality of convert_coordinate."""
def test_convert_coord_erroneous_type(self):
"""Test basic functionality of convert_coord."""
coordinate = '1071.84450'
hemisphere = 'B'
self.assertEqual(convert_coordinate(coordinate, hemisphere), None)
self.assertEqual(convert_coord(coordinate, hemisphere), None)
class TestConvertTime(unittest.TestCase):
......@@ -96,128 +96,128 @@ class TestStr2list(unittest.TestCase):
self.assertEqual(str2list(str_input), ['a', 'b', 'c', 'd', 'e'])
class TestCheckEmailFormatting(unittest.TestCase):
"""Test suite for check_email_formatting."""
class TestCheckEmailformat(unittest.TestCase):
"""Test suite for check_email_format."""
def test_check_email_formatting_valid_email(self):
"""Test basic functionality of check_email_formatting."""
def test_check_email_format_valid_email(self):
"""Test basic functionality of check_email_format."""
email = 'm@passcal.nmt.edu'
self.assertTrue(check_email_formatting(email))
self.assertTrue(check_email_format(email))
def test_check_email_formatting_erroneous_type(self):
"""Test basic functionality of check_email_formatting."""
def test_check_email_format_erroneous_type(self):
"""Test basic functionality of check_email_format."""
email = 12
with self.assertLogs(logger, level='ERROR') as cmd:
check_email_formatting(email)
check_email_format(email)
msg = "The provided email '{}' should be a string.".format(email)
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_check_email_formatting_erroneous_email(self):
"""Test basic functionality of check_email_formatting."""
def test_check_email_format_erroneous_email(self):
"""Test basic functionality of check_email_format."""
email = 'm_passcal.nmt.edu'
with self.assertLogs(logger, level='ERROR') as cmd:
check_email_formatting(email)
check_email_format(email)
msg = ("Invalid email. The provided email '{}' does not meet minimum "
"formatting requirements: account@domain.".format(email))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
class TestCheckInstrumentSpecs(unittest.TestCase):
"""Test suite for check_instrument_specs."""
class TestCheckInstSpecs(unittest.TestCase):
"""Test suite for check_inst_specs."""
def test_check_instrument_specs_default_value(self):
"""Test basic functionality of check_instrument_specs."""
def test_check_inst_specs_default_value(self):
"""Test basic functionality of check_inst_specs."""
specs = 'LEMI-039'
self.assertTrue(check_instrument_specs(specs, 'fluxgate', ''))
self.assertTrue(check_inst_specs(specs, 'fluxgate', ''))
def test_check_instrument_specs_valid_user_input(self):
"""Test basic functionality of check_instrument_specs."""
def test_check_inst_specs_valid_user_input(self):
"""Test basic functionality of check_inst_specs."""
specs = 'Manufacturer: a - Model: b - Type: c'
self.assertTrue(check_instrument_specs(specs, 'fluxgate', ''))
self.assertTrue(check_inst_specs(specs, 'fluxgate', ''))
def test_check_instrument_specs_no_type(self):
"""Test basic functionality of check_instrument_specs."""
def test_check_inst_specs_no_type(self):
"""Test basic functionality of check_inst_specs."""
specs = 'Manufacturer: a - Model: b - Type: '
self.assertTrue(check_instrument_specs(specs, 'fluxgate', ''))
self.assertTrue(check_inst_specs(specs, 'fluxgate', ''))
def test_check_instrument_specs_no_manufacturer(self):
"""Test basic functionality of check_instrument_specs."""
def test_check_inst_specs_no_manufacturer(self):
"""Test basic functionality of check_inst_specs."""
specs = 'Manufacturer: - Model: b - Type: c'
with self.assertLogs(logger, level='ERROR') as cmd:
check_instrument_specs(specs, 'fluxgate', '')
check_inst_specs(specs, 'fluxgate', '')
msg = "Please provide fluxgate manufacturer. Required metadata field."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_check_instrument_specs_no_model(self):
"""Test basic functionality of check_instrument_specs."""
def test_check_inst_specs_no_model(self):
"""Test basic functionality of check_inst_specs."""
specs = 'Manufacturer: a - Model: - Type: c'
with self.assertLogs(logger, level='ERROR') as cmd:
check_instrument_specs(specs, 'fluxgate', '')
check_inst_specs(specs, 'fluxgate', '')
msg = "Please provide fluxgate model. Required metadata field."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_check_instrument_specs_erroneous_specs(self):
"""Test basic functionality of check_instrument_specs."""
def test_check_inst_specs_erroneous_specs(self):
"""Test basic functionality of check_inst_specs."""
specs = 'Other'
with self.assertLogs(logger, level='ERROR') as cmd:
check_instrument_specs(specs, 'fluxgate', '')
check_inst_specs(specs, 'fluxgate', '')
msg = "Please provide fluxgate specs for. Required metadata field."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
class TestCheckSerialNumber(unittest.TestCase):
"""Test suite for check_serial_number."""
class TestCheckSN(unittest.TestCase):
"""Test suite for check_sn."""
def test_check_serial_number_no_input(self):
"""Test basic functionality of check_serial_number."""
serial_number = None
def test_check_sn_no_input(self):
"""Test basic functionality of check_sn."""
sn = None
with self.assertLogs(logger, level='ERROR') as cmd:
valid = check_serial_number('data logger', serial_number)
valid = check_sn('data logger', sn)
msg = "Please provide a serial number for the data logger!"
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertFalse(valid)
def test_check_serial_number_invalid_input_data_logger(self):
"""Test basic functionality of check_serial_number."""
serial_number = 1234
def test_check_sn_invalid_input_data_logger(self):
"""Test basic functionality of check_sn."""
sn = 1234
with self.assertLogs(logger, level='ERROR') as cmd:
valid = check_serial_number('data logger', serial_number)
valid = check_sn('data logger', sn)
msg = "The serial number of the data logger should be 3 digits long."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertFalse(valid)
def test_check_serial_number_mismatch_between_user_and_data_inputs(self):
"""Test basic functionality of check_serial_number."""
serial_number = 110
def test_check_sn_mismatch_between_user_and_data_inputs(self):
"""Test basic functionality of check_sn."""
sn = 110
data_input = '132'
with self.assertLogs(logger, level='ERROR') as cmd:
valid = check_serial_number('data logger', serial_number, data_input)
valid = check_sn('data logger', sn, data_input)
msg = ("The serial number of the data logger differs from the one "
"automatically recorded by the logger in the field (110 != 132). "
"Please correct!")
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertFalse(valid)
def test_check_serial_number_match_between_user_and_data_inputs(self):
"""Test basic functionality of check_serial_number."""
serial_number = 110
def test_check_sn_match_between_user_and_data_inputs(self):
"""Test basic functionality of check_sn."""
sn = 110
data_input = '110'
self.assertTrue(check_serial_number("data logger", serial_number, data_input))
self.assertTrue(check_sn("data logger", sn, data_input))
def test_check_serial_number_valid_input_data_logger(self):
"""Test basic functionality of check_serial_number."""
serial_number = 110
self.assertTrue(check_serial_number("data logger", serial_number))
def test_check_sn_valid_input_data_logger(self):
"""Test basic functionality of check_sn."""
sn = 110
self.assertTrue(check_sn("data logger", sn))
def test_check_serial_number_valid_input_fluxgate(self):
"""Test basic functionality of check_serial_number."""
serial_number = 110
self.assertTrue(check_serial_number("fluxgate", serial_number))
def test_check_sn_valid_input_fluxgate(self):
"""Test basic functionality of check_sn."""
sn = 110
self.assertTrue(check_sn("fluxgate", sn))
def test_check_serial_number_valid_input_electrode(self):
"""Test basic functionality of check_serial_number."""
serial_number = 1103455
self.assertTrue(check_serial_number("electrode", serial_number))
def test_check_sn_valid_input_electrode(self):
"""Test basic functionality of check_sn."""
sn = 1103455
self.assertTrue(check_sn("electrode", sn))
class TestIsEmpty(unittest.TestCase):
......@@ -244,24 +244,24 @@ class TestIsEmpty(unittest.TestCase):
self.assertFalse(is_empty(val))
class TestGetBoolLoc(unittest.TestCase):
"""Test suite for evaluate_location_code."""
class TestEvalLocCode(unittest.TestCase):
"""Test suite for eval_loc_code."""
def test_evaluate_location_code_two_electrode_pairs_different_direction(self):
"""Test basic functionality of evaluate_location_code."""
def test_eval_loc_code_two_electrode_pairs_different_direction(self):
"""Test basic functionality of eval_loc_code."""
e_infos = {'a': {'E1': 'Ex', 'E2': 'Ey'}, 'b': {'E1': 'Ex', 'E2': 'Ey'}}
self.assertFalse(evaluate_location_code(e_infos))
self.assertFalse(eval_loc_code(e_infos))
def test_evaluate_location_code_two_electrode_pairs_same_direction(self):
"""Test basic functionality of evaluate_location_code."""
def test_eval_loc_code_two_electrode_pairs_same_direction(self):
"""Test basic functionality of eval_loc_code."""
e_infos = {'a': {'E1': 'Ex', 'E2': 'Ex'}, 'b': {'E1': 'Ex', 'E2': 'Ex'}}
self.assertTrue(evaluate_location_code(e_infos))
self.assertTrue(eval_loc_code(e_infos))
def test_evaluate_location_code_more_than_two_electrode_pairs(self):
"""Test basic functionality of evaluate_location_code."""
def test_eval_loc_code_more_than_two_electrode_pairs(self):
"""Test basic functionality of eval_loc_code."""
e_infos = {'a': {'E1': 'Ex', 'E2': 'Ey', 'E3': 'Ex', 'E4': 'Ey'},
'b': {'E1': 'Ex', 'E2': 'Ey'}}
self.assertTrue(evaluate_location_code(e_infos))
self.assertTrue(eval_loc_code(e_infos))
class TestGetELoc(unittest.TestCase):
......@@ -282,20 +282,20 @@ class TestGetELoc(unittest.TestCase):
class TestGetRunListGetEIds(unittest.TestCase):
"""Test suite for get_run_list and get_e_ids."""
def test_get_run_nbr(self):
"""Test basic functionality of get_run_nbr."""
self.assertEqual(get_run_nbr('a'), 1)
self.assertEqual(get_run_nbr('b'), 2)
self.assertEqual(get_run_nbr('c'), 3)
def test_get_run_num(self):
"""Test basic functionality of get_run_num."""
self.assertEqual(get_run_num('a'), 1)
self.assertEqual(get_run_num('b'), 2)
self.assertEqual(get_run_num('c'), 3)
def test_get_run_list(self):
"""Test basic functionality of get_run_list."""
nbr_runs = 3
self.assertListEqual(get_run_list(nbr_runs), ['a', 'b', 'c'])
num_runs = 3
self.assertListEqual(get_run_list(num_runs), ['a', 'b', 'c'])
def test_get_e_ids(self):
"""Test basic functionality of get_e_ids."""
nbr_runs = 2
num_runs = 2
e_ids = [('a', 1), ('a', 2), ('a', 3), ('a', 4), ('b', 1), ('b', 2),
('b', 3), ('b', 4)]
self.assertListEqual(get_e_ids(nbr_runs), e_ids)
self.assertListEqual(get_e_ids(num_runs), e_ids)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment