48-smc_read_input

.vscode/settings.json

@@ -79,16 +79,7 @@
         "xema"
     ],
     "python.pythonPath": "~/anaconda3/envs/pymica/bin/python",
-    "python.testing.unittestArgs": [
-        "-v",
-        "-s",
-        "./test",
-        "-p",
-        "*_test.py"
-    ],
-    "python.testing.pytestEnabled": false,
-    "python.testing.nosetestsEnabled": false,
-    "python.testing.unittestEnabled": true,
+
     "python.linting.flake8Enabled": true,
     "python.linting.pycodestyleEnabled": false,
     "python.linting.pylintPath": "~/anaconda3/envs/pymica_test/bin/python",

pymica/io/importers.py

@@ -0,0 +1,66 @@
+"""Subroutine for reading data from SMC API 
+"""
+
+import json
+import datetime
+import urllib.request
+
+
+from datetime import timedelta, datetime
+
+def import_api_smc(variable: str, url_meta: str, 
+                   date: datetime, url: str, station_list: list) -> list:
+    """_summary_
+
+    Args:
+        variables (str): String containing the name of the variables.
+        url_meta (str): String which contains the url for download the metadata
+        date (datetime): Is the time for which we obtained the data 
+        url (str): String which contains the url for download the metadata
+        station_list (list): List with the codes of stations to be extracted. 
+                             If its None returns all the available stations.
+
+    Raises:
+        ValueError: Check if the variables are available.
+
+    Returns:
+        dict: dictionary containing the data ready to be interpolated.
+    """
+
+    str_variables = {'2t':'32', '2r':'33', 'ws':'30', 'wdir':'31', 'ws':'46',
+                     'wdir':'47', 'ws':'48', 'wdir':'49'}
+
+    # Mirem que la variable estigui dins les possible
+    if not set(map(str, [variable])) <= set(str_variables.keys()):
+            raise ValueError('Les variables disponibles són ' +
+                            str(list(str_variables.keys())))
+
+    # Mirem que station_list realment sigui una llista i ens assegurem
+    if not isinstance(station_list,list):
+        station_list = [None]
+
+    json_var=urllib.request.urlopen(url_meta).read()    
+    metadata=json.loads(json_var.decode("utf-8"))
+
+
+    # The LOOP begins
+    data = []
+    # We add 5 minutes arbitrarely
+    json_data = urllib.request.urlopen(
+        url + "/" + str_variables[variable] + "?din=" +
+        str(date.strftime("%Y-%m-%dT%H:%MZ")) + "&dfi=" +
+        str(date.strftime("%Y-%m-%dT%H:%MZ"))).read()
+    var_data = json.loads(json_data)
+
+    for reg in var_data:
+        if station_list == [None] or (reg['codi'] in station_list):
+            for lec in reg['variables'][0]['lectures']:
+                for k in range(len(metadata)):
+                    if metadata[k]['codi']==reg['codi']:
+                        data.append({'id': reg['codi'], 'altitude': metadata[k]['altitud'],
+                                            'lat': metadata[k]['coordenades']['latitud'], 
+                                            'lon': metadata[k]['coordenades']['longitud'],
+                                            'value': lec['valor']})       
+
+    return data
+

pymica_tests/importers_pymica_test.py

@@ -0,0 +1,44 @@
+'''Tests for importers.py
+'''
+import unittest
+
+
+import dateutil.parser
+
+from pymica.io.importers import \
+    import_api_smc  # pylint: disable=E0611
+
+
+class TestImporters(unittest.TestCase):
+
+    url = 'http://smcawrest01:8080/ApiRestInterna/xema/v1/mesurades/' \
+          'dades/variables'
+    url_meta = "http://smcawrest01:8080/ApiRestInterna/xema/v1/mesurades/metadades/estacions"
+
+    station_list = ["C6","C7","C8","CE","CW","D1","D7","D8","D9","DI","DK","DQ","H1","MR","U3","UH","UJ","UM","V1","V8",
+                "VA","VB","VC","VD","VM","VY","VQ","W4","W8","WB","WC","WD","WJ","WL","WP","WV","WZ","XA","XB","XD",
+                "X1","XE","XI","XM","XP","XR","XX","Y6","YD","YJ","YF","YH","YE","YL","YO"]
+
+    #station_list = ["ZD"]
+    station_list = None
+
+    variable = '2t'
+
+    date = "20210401T0000Z"
+    date = dateutil.parser.parse(date)
+
+
+    def test_import_api_smc(self):
+
+        dict_out = import_api_smc(self.variable, self.url_meta,
+                           self.date, self.url, self.station_list)
+
+        if self.station_list == None:
+            self.assertAlmostEqual(len(dict_out), 186)
+            self.assertEqual(dict_out[0]['id'],'C6')
+            self.assertEqual(dict_out[0]['altitude'], 264.0) 
+        else:
+            self.assertAlmostEqual(len(dict_out), 1)
+            self.assertEqual(dict_out[0]['id'],'ZD')
+            self.assertEqual(dict_out[0]['altitude'], 2478.0) 
+