Merge branch 'main' into reductionist_count_as_bytes

Author: valeriupredoi

.github/workflows/run-test-push.yml

@@ -33,4 +33,4 @@ jobs:
       - run: conda list
       - run: flake8 --exclude tests,doc --max-line-length 120 --ignore F405,F401
       - run: pytest -n 2 -m "not slow" --junitxml=report-1.xml
-      - uses: codecov/codecov-action@v5
+      - uses: codecov/codecov-action@v6

activestorage/active.py

@@ -143,7 +143,8 @@ def hfix(x):
     missing_value = ds.attrs.get('missing_value')
     # see https://github.com/NCAS-CMS/PyActiveStorage/pull/303
     if isinstance(missing_value, np.ndarray):
-        missing_value = missing_value[0]
+        if missing_value.size == 1:
+            missing_value = missing_value[0]
     valid_min = hfix(ds.attrs.get('valid_min'))
     valid_max = hfix(ds.attrs.get('valid_max'))
     valid_range = hfix(ds.attrs.get('valid_range'))
@@ -519,21 +520,22 @@ def _from_storage(self, ds, indexer, chunks, out_shape, out_dtype,
         # Create a shared session object.
         if self.interface_type == "s3" and self._version == 2:
             if self.storage_options is not None:
-                key, secret = None, None
                 if self.storage_options.get("anon", None) is True:
                     print("Reductionist session for Anon S3 bucket.")
                     session = reductionist.get_session(
                         None, None, S3_ACTIVE_STORAGE_CACERT)
-                if "key" in self.storage_options:
-                    key = self.storage_options["key"]
-                if "secret" in self.storage_options:
-                    secret = self.storage_options["secret"]
-                if key and secret:
-                    session = reductionist.get_session(
-                        key, secret, S3_ACTIVE_STORAGE_CACERT)
                 else:
-                    session = reductionist.get_session(
-                        S3_ACCESS_KEY, S3_SECRET_KEY, S3_ACTIVE_STORAGE_CACERT)
+                    key, secret = None, None
+                    if "key" in self.storage_options:
+                        key = self.storage_options["key"]
+                    if "secret" in self.storage_options:
+                        secret = self.storage_options["secret"]
+                    if key and secret:
+                        session = reductionist.get_session(
+                            key, secret, S3_ACTIVE_STORAGE_CACERT)
+                    else:
+                        session = reductionist.get_session(
+                            S3_ACCESS_KEY, S3_SECRET_KEY, S3_ACTIVE_STORAGE_CACERT)
             else:
                 session = reductionist.get_session(S3_ACCESS_KEY,
                                                    S3_SECRET_KEY,

activestorage/reductionist.py

@@ -24,13 +24,10 @@ def get_session(username: str, password: str,
     :returns: a client session object.
     """
     session = requests.Session()
-    # TODO Stack-HPC
-    # we need to allow Anon buckets. though this
-    # will break connection to data server
-    # if username is None and password is None:
-    #     return session
-    session.auth = (username, password)
     session.verify = cacert or False
+    if username is None and password is None:
+        return session
+    session.auth = (username, password)
     return session
 
 
@@ -272,8 +269,13 @@ def __init__(self, status_code, error):
 
 def decode_and_raise_error(response):
     """Decode an error response and raise ReductionistError."""
-    try:
-        error = json.dumps(response.json())
+    if response.status_code == http.client.INTERNAL_SERVER_ERROR:
+        try:
+            error = json.dumps(response.json())
+        except requests.exceptions.JSONDecodeError as exc:
+            error = http.client.responses.get(response.status_code, "-")
+            raise ReductionistError(response.status_code, error) from exc
         raise ReductionistError(response.status_code, error)
-    except requests.exceptions.JSONDecodeError as exc:
-        raise ReductionistError(response.status_code, "-") from exc
+
+    error = http.client.responses.get(response.status_code, "-")
+    raise ReductionistError(response.status_code, error)

activestorage/storage.py

@@ -130,10 +130,19 @@ def mask_missing(data, missing):
     fill_value, missing_value, valid_min, valid_max = missing
 
     if fill_value is not None:
-        data = np.ma.masked_equal(data, fill_value)
+        if isinstance(fill_value, np.ndarray) or isinstance(fill_value, list):
+            data = np.ma.masked_where(data == fill_value, data)
+        else:
+            data = np.ma.masked_equal(data, fill_value)
 
     if missing_value is not None:
-        data = np.ma.masked_equal(data, missing_value)
+        if isinstance(missing_value, np.ndarray) or isinstance(missing_value, list):
+            try:
+                data = np.ma.masked_where(data == missing_value, data)
+            except ValueError:  # not broadcastable
+                raise ValueError("Data and missing_value arrays are not brodcastable!")
+        else:
+            data = np.ma.masked_equal(data, missing_value)
 
     if valid_max is not None:
         data = np.ma.masked_greater(data, valid_max)