Commit ef5e6fdf8e5638abc0a93d86c105bf26c83a424a
1 parent
2b11025d
Exists in
rhitier-dev
and in
1 other branch
New test-read-nc
Showing
1 changed file
with
27 additions
and
2 deletions
Show diff stats
tests/BackendTests.py
| 1 | +import gzip | ||
| 2 | +import os.path | ||
| 1 | import sys | 3 | import sys |
| 2 | import unittest | 4 | import unittest |
| 3 | from datetime import datetime as ddatetime, datetime | 5 | from datetime import datetime as ddatetime, datetime |
| 4 | from pprint import pprint | 6 | from pprint import pprint |
| 5 | 7 | ||
| 6 | -from web.run import retrieve_amda_netcdf, get_data_for_target, _sta_sto | 8 | +from netCDF4 import Dataset |
| 9 | + | ||
| 10 | +from web.run import retrieve_amda_netcdf, get_data_for_target, _sta_sto, _read_var | ||
| 7 | 11 | ||
| 8 | FILE_DATE_FMT = "%Y-%m-%d %H:%M:%S" | 12 | FILE_DATE_FMT = "%Y-%m-%d %H:%M:%S" |
| 9 | 13 | ||
| @@ -55,7 +59,7 @@ class AmdaTestCase(BaseTestCase): | @@ -55,7 +59,7 @@ class AmdaTestCase(BaseTestCase): | ||
| 55 | stopped_at = ddatetime.strptime("2021-12-18 00:00:00", FILE_DATE_FMT) | 59 | stopped_at = ddatetime.strptime("2021-12-18 00:00:00", FILE_DATE_FMT) |
| 56 | model = {'parameters': {'dens': 'Dens', 'temp': 'Temp', 'vtot': 'Vel'}, 'slug': 'ace_swepam_real_1h'} | 60 | model = {'parameters': {'dens': 'Dens', 'temp': 'Temp', 'vtot': 'Vel'}, 'slug': 'ace_swepam_real_1h'} |
| 57 | s0, s1 = _sta_sto(model, started_at, stopped_at) | 61 | s0, s1 = _sta_sto(model, started_at, stopped_at) |
| 58 | - all_data = get_data_for_target(target_config, 'om', s0, s1 ) | 62 | + all_data = get_data_for_target(target_config, 'om', s0, s1) |
| 59 | self.assertEqual(1, len(all_data)) | 63 | self.assertEqual(1, len(all_data)) |
| 60 | 64 | ||
| 61 | def test_sta_sto(self): | 65 | def test_sta_sto(self): |
| @@ -68,3 +72,24 @@ class AmdaTestCase(BaseTestCase): | @@ -68,3 +72,24 @@ class AmdaTestCase(BaseTestCase): | ||
| 68 | self.assertIsInstance(s0, datetime) | 72 | self.assertIsInstance(s0, datetime) |
| 69 | self.assertIsInstance(s1, datetime) | 73 | self.assertIsInstance(s1, datetime) |
| 70 | 74 | ||
| 75 | + def test_read_nc(self): | ||
| 76 | + SCRIPT_PATH = os.path.dirname(__file__) | ||
| 77 | + PROJECT_DIR = os.path.abspath(os.path.join(SCRIPT_PATH, os.pardir)) | ||
| 78 | + print("PROJECT_DIR", PROJECT_DIR) | ||
| 79 | + print("SCRIPT_PATH", SCRIPT_PATH) | ||
| 80 | + # return | ||
| 81 | + local_netc_file = os.path.join(PROJECT_DIR,'tests-resources', 'amda-irap-omp-eu-ddservice-base-data-omni-hour-omni202107010-nc') | ||
| 82 | + cdf_handle = Dataset(local_netc_file, "r", format="NETCDF4") | ||
| 83 | + default_nc_keys = { | ||
| 84 | + 'hee': 'HEE', | ||
| 85 | + 'vtot': 'V', | ||
| 86 | + 'magn': 'B', | ||
| 87 | + 'temp': 'T', | ||
| 88 | + 'dens': 'N', | ||
| 89 | + 'pdyn': 'P_dyn', | ||
| 90 | + 'atse': 'Delta_angle', | ||
| 91 | + } | ||
| 92 | + nc_keys = default_nc_keys.copy() | ||
| 93 | + data_v = _read_var(cdf_handle, nc_keys, 'vtot') | ||
| 94 | + pprint(data_v) | ||
| 95 | + pprint(data_v[0]) |