Merge pull request #90 from Exabyte-io/feature/SOF-7768

Feature/SOF-7768

Author: VsevolodX

src/py/mat3ra/__init__.py

@@ -0,0 +1 @@
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)

src/py/mat3ra/standata/applications.py

@@ -0,0 +1,37 @@
+from collections import defaultdict
+from typing import Dict, List
+
+from .base import Standata, StandataData
+from .data.applications import applications_data
+
+
+class ApplicationStandata(Standata):
+    data_dict: Dict = applications_data
+    data: StandataData = StandataData(data_dict)
+
+    @classmethod
+    def list_all(cls) -> Dict[str, List[dict]]:
+        """
+        Lists all applications with their versions and build information and prints in a human-readable format.
+        Returns a dict grouped by application name.
+        """
+        grouped = defaultdict(list)
+        for app in cls.get_as_list():
+            version_info = {
+                "version": app.get("version"),
+                "build": app.get("build"),
+            }
+            if app.get("isLicensed"):
+                version_info["isLicensed"] = True
+            grouped[app.get("name")].append(version_info)
+
+        lines = []
+        for app_name in sorted(grouped.keys()):
+            for info in grouped[app_name]:
+                licensed = " (licensed)" if info.get("isLicensed") else ""
+                lines.append(f"{app_name}:\n      version: {info['version']}, build: {info['build']}{licensed}")
+
+        print("\n".join(lines))
+        return dict(grouped)
+
+

src/py/mat3ra/standata/base.py

@@ -1,4 +1,5 @@
 import re
+from enum import Enum
 from typing import Dict, List
 
 import pandas as pd
@@ -211,8 +212,11 @@ def get_by_name_first_match(cls, name: str) -> dict:
             name: Name of the entity.
         """
         matching_filenames = cls.data.standataConfig.get_filenames_by_regex(name)
-        return cls.data.filesMapByName.get_objects_by_filenames(matching_filenames)[0]
-
+        objects = cls.data.filesMapByName.get_objects_by_filenames(matching_filenames)
+        if not objects:
+            raise ValueError(f"No matches found for name '{name}'")
+        return objects[0]
+    
     @classmethod
     def get_by_categories(cls, *tags: str) -> List[dict]:
         """
@@ -246,3 +250,37 @@ def get_by_name_and_categories(cls, name: str, *tags: str) -> dict:
             raise ValueError(f"No matches found for name '{name}' and categories {tags}")
 
         return cls.data.filesMapByName.get_objects_by_filenames(matching_filenames)[0]
+
+    @classmethod
+    def _create_filtered_data(cls, filenames: List[str]) -> StandataData:
+        filtered_files_map = {k: v for k, v in cls.data.filesMapByName.dictionary.items() if k in filenames}
+        filtered_entities = [e for e in cls.data.standataConfig.entities if e.filename in filenames]
+        return StandataData({
+            "filesMapByName": filtered_files_map,
+            "standataConfig": {
+                "categories": cls.data.standataConfig.categories,
+                "entities": [{"filename": e.filename, "categories": e.categories} for e in filtered_entities]
+            }
+        })
+
+    @classmethod
+    def _normalize_enum_name(cls, name: str) -> str:
+        return name.upper().replace("-", "_")
+
+    @classmethod
+    def _create_enum_from_values(cls, values: List[str], enum_name: str) -> type[Enum]:
+        enum_dict = {cls._normalize_enum_name(value): value for value in values}
+        return Enum(enum_name, enum_dict)
+
+    @classmethod
+    def filter_by_name(cls, name: str) -> "Standata":
+        matching_filenames = cls.data.standataConfig.get_filenames_by_regex(name)
+        filtered_data = cls._create_filtered_data(matching_filenames)
+        return type(cls.__name__, (cls,), {"data": filtered_data})
+
+    @classmethod
+    def filter_by_tags(cls, *tags: str) -> "Standata":
+        categories = cls.data.standataConfig.convert_tags_to_categories_list(*tags)
+        matching_filenames = cls.data.standataConfig.get_filenames_by_categories(*categories)
+        filtered_data = cls._create_filtered_data(matching_filenames)
+        return type(cls.__name__, (cls,), {"data": filtered_data})

src/py/mat3ra/standata/model_tree.py

@@ -1,3 +1,4 @@
+from enum import Enum
 from typing import Any, Dict, List, Optional
 
 from mat3ra.esse.models.method.categorized_method import SlugifiedEntry
@@ -32,3 +33,47 @@ def get_default_model_type_for_application(self, application: Dict[str, Any]) ->
         tree = self.get_tree_by_application_name_and_version(name, application.get("version", ""))
         keys = list(tree.keys())
         return keys[0] if keys else None
+
+    @classmethod
+    def get_subtypes_by_model_type(cls, model_type: str) -> type[Enum]:
+        model_tree = MODEL_TREE.get(model_type, {})
+        subtypes = list(model_tree.keys())
+        return cls._create_enum_from_values(subtypes, f"{model_type.upper()}Subtypes")
+
+    @classmethod
+    def get_functionals_by_subtype(cls, model_type: str, subtype_enum: Enum) -> type[Enum]:
+        model_tree = MODEL_TREE.get(model_type, {})
+        subtype_value = subtype_enum.value if isinstance(subtype_enum, Enum) else subtype_enum
+        subtype_tree = model_tree.get(subtype_value, {})
+        functionals = subtype_tree.get("functionals", [])
+        enum_name = f"{model_type.upper()}{cls._normalize_enum_name(subtype_value)}Functionals"
+        return cls._create_enum_from_values(functionals, enum_name)
+
+    @classmethod
+    def get_default_subtype(cls, model_tree: Dict[str, Any]) -> Optional[str]:
+        subtypes = [key for key in model_tree.keys() if key not in ["refiners", "modifiers", "methods"]]
+        return subtypes[0] if subtypes else None
+
+    @classmethod
+    def get_model_by_parameters(cls, type: str, subtype: Optional[str], functional: Optional[str]) -> Dict[str, Any]:
+        model_tree = MODEL_TREE.get(type, {})
+        if not model_tree:
+            return {}
+
+        result = {"type": type}
+
+        resolved_subtype = subtype or cls.get_default_subtype(model_tree)
+        subtype_tree = model_tree.get(resolved_subtype, {}) if resolved_subtype else {}
+        if not subtype_tree:
+            return result
+
+        result["subtype"] = resolved_subtype
+
+        functionals_from_tree = subtype_tree.get("functionals", [])
+        if functionals_from_tree:
+            if functional and functional in functionals_from_tree:
+                result["functional"] = functional
+            else:
+                result["functional"] = functionals_from_tree[0]
+
+        return result

src/py/mat3ra/standata/subworkflows.py

@@ -0,0 +1,14 @@
+from typing import Dict
+
+from .base import Standata, StandataData
+from .data.subworkflows import subworkflows_data
+
+
+class SubworkflowStandata(Standata):
+    data_dict: Dict = subworkflows_data
+    data: StandataData = StandataData(data_dict)
+
+    @classmethod
+    def filter_by_application(cls, application: str) -> "SubworkflowStandata":
+        return cls.filter_by_tags(application)
+

src/py/mat3ra/standata/workflows.py

@@ -4,6 +4,15 @@
 from .data.workflows import workflows_data
 
 
-class Workflows(Standata):
+class WorkflowStandata(Standata):
     data_dict: Dict = workflows_data
     data: StandataData = StandataData(data_dict)
+
+    @classmethod
+    def filter_by_application(cls, application: str) -> "WorkflowStandata":
+        return cls.filter_by_tags(application)
+
+    @classmethod
+    def filter_by_application_config(cls, application_config: Dict) -> "WorkflowStandata":
+        application_name = application_config.get("name", "")
+        return cls.filter_by_application(application_name)

tests/py/unit/test_applications_data.py

@@ -0,0 +1,51 @@
+from mat3ra.standata.data.applications import applications_data
+from mat3ra.standata.applications import ApplicationStandata
+
+
+def test_get_by_name():
+    application = ApplicationStandata.get_by_name_first_match("espresso")
+    assert type(application) == dict
+    assert application["name"] == "espresso"
+    assert application["version"] == "6.3"
+
+
+def test_get_by_categories():
+    applications = ApplicationStandata.get_by_categories("quantum-mechanical")
+    assert isinstance(applications, list)
+    assert applications[0]["name"] == "espresso"
+
+
+def test_get_application_data():
+    application = applications_data["filesMapByName"]["espresso/espresso_gnu_6.3.json"]
+    assert type(application) == dict
+    assert application["name"] == "espresso"
+    assert application["version"] == "6.3"
+
+
+def test_get_by_name_and_categories():
+    application = ApplicationStandata.get_by_name_and_categories("vasp", "quantum-mechanical")
+    assert type(application) == dict
+    assert application["name"] == "vasp"
+    assert application["version"] == "5.4.4"
+
+
+def test_list_all():
+    applications = ApplicationStandata.list_all()
+    assert isinstance(applications, dict)
+    assert len(applications) >= 1
+    assert "espresso" in applications
+    assert isinstance(applications["espresso"], list)
+    assert len(applications["espresso"]) >= 1
+    assert isinstance(applications["espresso"][0], dict)
+    assert "version" in applications["espresso"][0]
+    assert "build" in applications["espresso"][0]
+    assert applications["espresso"][0]["version"] == "6.3"
+    assert applications["espresso"][0]["build"] == "GNU"
+
+def test_get_as_list():
+    applications_list = ApplicationStandata.get_as_list()
+    assert isinstance(applications_list, list)
+    assert len(applications_list) >= 1
+    assert isinstance(applications_list[0], dict)
+    assert applications_list[0]["name"] == "espresso"
+

tests/py/unit/test_model_tree.py

@@ -1,3 +1,4 @@
+from enum import Enum
 from types import SimpleNamespace
 
 import pytest
@@ -10,6 +11,22 @@
 VERSION = SimpleNamespace(V6_0="6.0", V1_0="1.0")
 PSEUDOPOTENTIAL_TYPE = SimpleNamespace(PAW="paw", NC="nc", NC_FR="nc-fr", US="us")
 MODEL_TYPE = SimpleNamespace(DFT="dft")
+SUBTYPE = SimpleNamespace(GGA="gga", LDA="lda", HYBRID="hybrid", OTHER="other", INVALID="invalid_subtype")
+FUNCTIONAL = SimpleNamespace(
+    PBE="pbe",
+    PBESOL="pbesol",
+    PW91="pw91",
+    PZ="pz",
+    PW="pw",
+    VWN="vwn",
+    OTHER="other",
+    INVALID="invalid_functional",
+)
+GGA_FUNCTIONALS = {"PBE": FUNCTIONAL.PBE, "PBESOL": FUNCTIONAL.PBESOL, "PW91": FUNCTIONAL.PW91,
+                   "OTHER": FUNCTIONAL.OTHER}
+LDA_FUNCTIONALS = {"PZ": FUNCTIONAL.PZ, "PW": FUNCTIONAL.PW, "VWN": FUNCTIONAL.VWN, "OTHER": FUNCTIONAL.OTHER}
+EXPECTED_MODEL_BY_PARAMETERS_GGA = {"type": MODEL.DFT, "subtype": SUBTYPE.GGA, "functional": FUNCTIONAL.PBE}
+EXPECTED_MODEL_BY_PARAMETERS_LDA = {"type": MODEL.DFT, "subtype": SUBTYPE.LDA, "functional": FUNCTIONAL.PZ}
 
 
 @pytest.mark.parametrize(
@@ -77,3 +94,65 @@ def test_get_default_model_type_for_application(application, expected):
         assert result == expected
     else:
         assert result is None
+
+
+@pytest.mark.parametrize(
+    "model_type,expected_subtypes",
+    [
+        ("dft", {"GGA": "gga", "LDA": "lda", "HYBRID": "hybrid", "OTHER": "other"}),
+        ("invalid_model", {}),
+    ],
+)
+def test_get_subtypes_by_model_type(model_type, expected_subtypes):
+    subtypes = ModelTreeStandata.get_subtypes_by_model_type(model_type)
+    assert issubclass(subtypes, Enum)
+    assert len(list(subtypes)) == len(expected_subtypes)
+    for enum_name, expected_value in expected_subtypes.items():
+        assert hasattr(subtypes, enum_name)
+        assert getattr(subtypes, enum_name).value == expected_value
+
+
+@pytest.mark.parametrize(
+    "model_type,subtype_input,use_string,expected_functionals,excluded_functionals",
+    [
+        (MODEL.DFT, SUBTYPE.LDA, False, LDA_FUNCTIONALS, [FUNCTIONAL.PBE]),
+        (MODEL.DFT, SUBTYPE.GGA, False, GGA_FUNCTIONALS, [FUNCTIONAL.PZ]),
+        (MODEL.DFT, SUBTYPE.LDA, True, LDA_FUNCTIONALS, [FUNCTIONAL.PBE]),
+    ],
+)
+def test_get_functionals_by_subtype(model_type, subtype_input, use_string, expected_functionals, excluded_functionals):
+    if use_string:
+        subtype_arg = subtype_input
+    else:
+        subtypes = ModelTreeStandata.get_subtypes_by_model_type(model_type)
+        subtype_arg = getattr(subtypes, subtype_input.upper())
+
+    functionals = ModelTreeStandata.get_functionals_by_subtype(model_type, subtype_arg)
+    assert issubclass(functionals, Enum)
+
+    for enum_name, expected_value in expected_functionals.items():
+        assert hasattr(functionals, enum_name)
+        assert getattr(functionals, enum_name).value == expected_value
+
+    functional_values = [f.value for f in functionals]
+    for excluded in excluded_functionals:
+        assert excluded not in functional_values
+
+
+@pytest.mark.parametrize(
+    "type,subtype,functional,expected",
+    [
+        (MODEL.DFT, SUBTYPE.GGA, FUNCTIONAL.PBE, EXPECTED_MODEL_BY_PARAMETERS_GGA),
+        (MODEL.DFT, SUBTYPE.LDA, FUNCTIONAL.PZ, EXPECTED_MODEL_BY_PARAMETERS_LDA),
+        (MODEL.DFT, SUBTYPE.GGA, None, EXPECTED_MODEL_BY_PARAMETERS_GGA),
+        (MODEL.DFT, None, None, EXPECTED_MODEL_BY_PARAMETERS_GGA),
+        (MODEL.DFT, None, FUNCTIONAL.PBE, EXPECTED_MODEL_BY_PARAMETERS_GGA),
+        (MODEL.DFT, SUBTYPE.LDA, None, EXPECTED_MODEL_BY_PARAMETERS_LDA),
+        (MODEL.INVALID, None, None, {}),
+        (MODEL.DFT, SUBTYPE.INVALID, None, {"type": MODEL.DFT}),
+        (MODEL.DFT, SUBTYPE.GGA, FUNCTIONAL.INVALID, EXPECTED_MODEL_BY_PARAMETERS_GGA),
+    ],
+)
+def test_get_model_by_parameters(type, subtype, functional, expected):
+    result = ModelTreeStandata.get_model_by_parameters(type, subtype, functional)
+    assert result == expected

tests/py/unit/test_subworkflows_data.py

@@ -0,0 +1,56 @@
+from types import SimpleNamespace
+
+from mat3ra.standata.data.subworkflows import subworkflows_data
+from mat3ra.standata.subworkflows import SubworkflowStandata
+
+APP = SimpleNamespace(ESPRESSO="espresso")
+SUBWORKFLOW = SimpleNamespace(
+    SEARCH_NAME="pw_scf",
+    FILENAME="espresso/pw_scf.json",
+    EXACT_NAME="Preliminary SCF Calculation",
+)
+
+
+def test_get_by_name():
+    subworkflow = SubworkflowStandata.get_by_name_first_match(SUBWORKFLOW.SEARCH_NAME)
+    assert type(subworkflow) == dict
+    assert "name" in subworkflow
+    assert SUBWORKFLOW.EXACT_NAME in subworkflow["name"]
+
+
+def test_get_by_categories():
+    subworkflows = SubworkflowStandata.get_by_categories(APP.ESPRESSO)
+    assert isinstance(subworkflows, list)
+    assert len(subworkflows) >= 1
+    assert isinstance(subworkflows[0], dict)
+
+
+def test_get_subworkflow_data():
+    subworkflow = subworkflows_data["filesMapByName"][SUBWORKFLOW.FILENAME]
+    assert type(subworkflow) == dict
+    assert "name" in subworkflow
+    assert subworkflow["name"] == SUBWORKFLOW.EXACT_NAME
+
+
+def test_get_by_name_and_categories():
+    subworkflow = SubworkflowStandata.get_by_name_and_categories(SUBWORKFLOW.SEARCH_NAME, APP.ESPRESSO)
+    assert type(subworkflow) == dict
+    assert "name" in subworkflow
+    assert APP.ESPRESSO in str(subworkflow.get("application", {})).lower() or APP.ESPRESSO in str(subworkflow)
+
+
+def test_get_as_list():
+    subworkflows_list = SubworkflowStandata.get_as_list()
+    assert isinstance(subworkflows_list, list)
+    assert len(subworkflows_list) >= 1
+    assert isinstance(subworkflows_list[0], dict)
+    assert "name" in subworkflows_list[0]
+
+
+def test_filter_by_application_and_get_by_name():
+    subworkflow = SubworkflowStandata.filter_by_application(APP.ESPRESSO).get_by_name_first_match(
+        SUBWORKFLOW.SEARCH_NAME)
+    assert type(subworkflow) == dict
+    assert "name" in subworkflow
+    assert subworkflow["name"] == SUBWORKFLOW.EXACT_NAME
+    assert APP.ESPRESSO in str(subworkflow.get("application", {})).lower()