Checkpoint; Disable publishing to test-pypi

.github/workflows/publish.yml

@@ -92,26 +92,26 @@ jobs:
         '${{ github.ref_name }}' dist/**
         --repo '${{ github.repository }}'
 
-  publish-to-testpypi:
-    name: Publish Python 🐍 distribution 📦 to TestPyPI
-    needs:
-    - build
-    runs-on: ubuntu-latest
+  # publish-to-testpypi:
+  #   name: Publish Python 🐍 distribution 📦 to TestPyPI
+  #   needs:
+  #   - build
+  #   runs-on: ubuntu-latest
 
-    environment:
-      name: testpypi
-      url: https://test.pypi.org/p/mzspeclib
+  #   environment:
+  #     name: testpypi
+  #     url: https://test.pypi.org/p/mzspeclib
 
-    permissions:
-      id-token: write  # IMPORTANT: mandatory for trusted publishing
+  #   permissions:
+  #     id-token: write  # IMPORTANT: mandatory for trusted publishing
 
-    steps:
-    - name: Download all the dists
-      uses: actions/download-artifact@v3
-      with:
-        name: python-package-distributions
-        path: dist/
-    - name: Publish distribution 📦 to TestPyPI
-      uses: pypa/gh-action-pypi-publish@release/v1
-      with:
-        repository-url: https://test.pypi.org/legacy/
+  #   steps:
+  #   - name: Download all the dists
+  #     uses: actions/download-artifact@v3
+  #     with:
+  #       name: python-package-distributions
+  #       path: dist/
+  #   - name: Publish distribution 📦 to TestPyPI
+  #     uses: pypa/gh-action-pypi-publish@release/v1
+  #     with:
+  #       repository-url: https://test.pypi.org/legacy/

mzspeclib/backends/base.py

@@ -3,9 +3,10 @@
 import csv
 import enum
 import logging
+import os
 import warnings
 
-from typing import Any, Callable, Dict, Iterable, Union, List, Type, Iterator
+from typing import Any, Callable, Dict, Iterable, Optional, Union, List, Type, Iterator
 from pathlib import Path
 
 
@@ -220,6 +221,17 @@ def __init__(self, filename: Union[str, Path, io.FileIO]):
 
         super().__init__(None)
 
+    def _infer_lib_name(self) -> Optional[str]:
+        if hasattr(self.filename, "name"):
+            name = self.filename.name.replace(".gz", "").rsplit(".", 1)[0].split(os.sep)[-1]
+        elif isinstance(self.filename, str):
+            name = self.filename.replace(".gz", "").rsplit(".", 1)[0].split(os.sep)[-1]
+        elif isinstance(self.filename, bytes):
+            name = self.filename.decode("utf8").replace(".gz", "").rsplit(".", 1)[0].split(os.sep)[-1]
+        else:
+            name = None
+        return name
+
     def read_header(self) -> bool:
         """
         Read just the header of the whole library

mzspeclib/backends/sptxt.py

@@ -409,21 +409,45 @@ def _parse_header_from_stream(self, stream: io.RawIOBase) -> Tuple[bool, int]:
         nbytes = 0
         attributes = AttributeManager()
         attributes.add_attribute(FORMAT_VERSION_TERM, DEFAULT_VERSION)
-        attributes.add_attribute(LIBRARY_NAME_TERM, self.filename)
-        self.attributes.clear()
-        self.attributes._from_iterable(attributes)
 
         line = stream.readline()
         nbytes += len(line)
         i = 0
+        header_buffer = []
         while not re.match(b"Name:", line) and i < HEADER_MAX_DEPTH:
+            header_buffer.append(line)
             line = stream.readline()
             nbytes += len(line)
             i += 1
+
+        self._parse_header_into_attributes(header_buffer, attributes)
+        if not attributes.has_attribute(LIBRARY_NAME_TERM):
+            name = self._infer_lib_name()
+            if name:
+                attributes.add_attribute(LIBRARY_NAME_TERM, name)
+
+        self.attributes.clear()
+        self.attributes._from_iterable(attributes)
         if re.match(b"Name:", line):
+            header_buffer.append(line)
+
             return True, nbytes - len(line)
         return False, 0
 
+    def _parse_header_into_attributes(self, lines: List[bytes], attributes: AttributeManager):
+        for i, line in enumerate(lines):
+            if line.startswith(b"###"):
+                if i == 0:
+                    name = line.rsplit(b'(', 1)[0].strip().decode('utf8').strip("# ")
+                    attributes.add_attribute(LIBRARY_NAME_TERM, name)
+                elif i == 1:
+                    version = line.split(b"(", 1)[1].strip().decode("utf8").split(",")[0].split(" ")[1]
+                    gid = attributes.get_next_group_identifier()
+                    attributes.add_attribute("MS:1003207|library creation software", "MS:1001477|SpectraST", gid)
+                    attributes.add_attribute("MS:1003200|software version", version, gid)
+            else:
+                continue
+
     def _make_peak_parsing_strategy(self) -> PeakParsingStrategy:
         return PeakParsingStrategy(parse_annotation)
 

mzspeclib/spectrum_library.py

@@ -2,8 +2,8 @@
 import os
 import pathlib
 
-from typing import Optional, Type, List, Union
-from mzspeclib.attributes import AttributeManagedProperty, AttributeManager
+from typing import Any, Optional, Type, List, Union
+from mzspeclib.attributes import Attribute, AttributeManagedProperty, AttributeManager
 from mzspeclib.backends.base import LIBRARY_DESCRIPTION_TERM, LIBRARY_NAME_TERM, LIBRARY_URI_TERM, LIBRARY_VERSION_TERM
 from mzspeclib.cluster import SpectrumCluster
 
@@ -327,25 +327,29 @@ def add_attribute(self, key, value, group_identifier=None):
         self._requires_backend()
         return self.backend.add_attribute(key, value, group_identifier=group_identifier)
 
-    def get_attribute(self, key, group_identifier=None):
+    def get_attribute(
+        self, key: str, group_identifier: Optional[str] = None, raw: bool = False
+    ) -> Union[Any, List[Any], Attribute, List[Attribute]]:
         """
         Get the value or values associated with a given
-        attribute key from the library level attribute store.
+        attribute key.
 
         Parameters
         ----------
         key : str
             The name of the attribute to retrieve
         group_identifier : str, optional
             The specific group identifier to return from.
+        raw : bool
+            Whether to return the :class:`Attribute` object or unwrap the value
 
         Returns
         -------
         attribute_value: object or list[object]
             Returns single or multiple values for the requested attribute.
         """
         self._requires_backend()
-        return self.backend.get_attribute(key, group_identifier=group_identifier)
+        return self.backend.get_attribute(key, group_identifier=group_identifier, raw=raw)
 
     def remove_attribute(self, key, group_identifier=None):
         """

mzspeclib/validate/rules/convert_to_xlsx.py

@@ -42,7 +42,7 @@ def rule_to_table_rows(rule: dict):
             units = ";".join([f"{rel.accession}|{rel.comment}" for rel in term.get("has_units", [])])
             if units:
                 row["has_units"] = units
-            row["notes"] = attr.get("notes")
+            row["attr_notes"] = attr.get("notes")
             if "value" in attr:
                 value_rule = attr["value"]
                 if isinstance(value_rule, str):