Incomplete modifications

.gitignore

@@ -4,6 +4,7 @@ test_pybamm.py
 
 # Ignore the virtual environment folder
 venv/
+env/
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

BatteryModelMapper/__init__.py

@@ -3,3 +3,5 @@
 from .json_validator import JSONValidator
 from .json_writer import JSONWriter
 from .parameter_mapper import ParameterMapper
+from .jsonld_exporter import export_jsonld
+from .preprocess_input import PreprocessInput

BatteryModelMapper/json_loader.py

@@ -1,8 +1,21 @@
+from pathlib import Path
+from urllib.parse import urlparse
+import json
 import requests
 
+
 class JSONLoader:
     @staticmethod
-    def load(json_url):
-        response = requests.get(json_url)
-        response.raise_for_status()
-        return response.json()
+    def load(source):
+        source = Path(source)
+
+        if urlparse(str(source)).scheme in ("http", "https"):
+            # Read from URL
+            response = requests.get(str(source))
+            response.raise_for_status()
+            return response.json()
+        elif source.is_file():
+            # Read from file
+            return json.loads(source.read_text(encoding="utf-8"))
+        else:
+            raise ValueError(f"File does not exist: {source}")

BatteryModelMapper/jsonld_exporter.py

@@ -0,0 +1,219 @@
+import json
+from typing import Any, Dict, List, Optional, Set
+from rdflib import BNode, URIRef
+from rdflib.namespace import RDF, RDFS, OWL, SKOS
+
+
+def _is_number_like(v: Any) -> bool:
+    if isinstance(v, (int, float)) and not isinstance(v, bool):
+        return True
+    if isinstance(v, str):
+        s = v.strip()
+        if not s:
+            return False
+        try:
+            float(s)
+            return True
+        except ValueError:
+            return False
+    return False
+
+
+def _get_modellib_hash(u: URIRef) -> str:
+    s = str(u)
+    return s.rsplit("#", 1)[-1].rsplit("/", 1)[-1]
+
+
+def _curie(g, term: URIRef) -> str:
+    try:
+        return g.namespace_manager.normalizeUri(term)
+    except Exception:
+        return str(term)
+
+
+def _first_literal_str(g, subj: URIRef, pred: URIRef) -> Optional[str]:
+    for o in g.objects(subj, pred):
+        return str(o)
+    return None
+
+
+def _get_skos_prefLabel(g, term: URIRef) -> str:
+    return (
+        _first_literal_str(g, term, SKOS.prefLabel)
+        or _first_literal_str(g, term, RDFS.label)
+        or _curie(g, term)
+    )
+
+
+def _find_any_predicate_by_localname(g, candidates: Set[str]) -> Optional[URIRef]:
+    for p in set(g.predicates()):
+        if _get_modellib_hash(p) in candidates:
+            return p
+    return None
+
+
+def _get_value_from_path(data: Any, keys: List[Any]) -> Any:
+    cur = data
+    try:
+        for k in keys:
+            if isinstance(k, str):
+                k = k.strip()
+            if isinstance(cur, dict):
+                cur = cur[k]
+            elif isinstance(cur, list):
+                cur = cur[int(k)]
+            else:
+                return None
+        return cur
+    except (KeyError, IndexError, ValueError, TypeError):
+        return None
+
+
+def _iter_restrictions(g, cls: URIRef):
+    for sc in g.objects(cls, RDFS.subClassOf):
+        if isinstance(sc, BNode) and (sc, RDF.type, OWL.Restriction) in g:
+            yield sc
+    for ec in g.objects(cls, OWL.equivalentClass):
+        if isinstance(ec, BNode) and (ec, RDF.type, OWL.Restriction) in g:
+            yield ec
+
+
+def _find_missing_values(ontology_parser, input_data, input_type):
+    mapped_paths = set()
+    key = ontology_parser.key_map.get(input_type)
+    for s in ontology_parser.graph.subjects():
+        for p, o in ontology_parser.graph.predicate_objects(s):
+            if p == key:
+                mapped_paths.add(tuple(ontology_parser.parse_key(str(o))))
+
+    def collect_json_paths(data, prefix=()):
+        paths = set()
+        if isinstance(data, dict):
+            for k, v in data.items():
+                paths |= collect_json_paths(v, prefix + (k,))
+        elif isinstance(data, list):
+            for i, v in enumerate(data):
+                paths |= collect_json_paths(v, prefix + (i,))
+        else:
+            paths.add(prefix)
+        return paths
+
+    input_paths = collect_json_paths(input_data)
+    missing = sorted(p for p in input_paths if p not in mapped_paths)
+    return input_paths, mapped_paths, missing
+
+
+def _find_any_predicate_by_localname(g, candidates: Set[str]) -> Optional[URIRef]:
+    for p in set(g.predicates()):
+        if _get_modellib_hash(p) in candidates:
+            return p
+    return None
+
+
+def _get_unit_for_subject(g, subject: URIRef) -> Optional[str]:
+    # Get the unit for a given subject if defined
+    unit_predicates = {"hasMeasurementUnit", "hasUnit", "unit"}
+    unit_pred = _find_any_predicate_by_localname(g, unit_predicates)
+    if unit_pred:
+        for unit in g.objects(subject, unit_pred):
+            return _curie(g, unit)
+    else:
+        # Assume SI units based on common property names
+        # breakpoint()
+
+        # Get skos preflabel if exists
+        label = _get_skos_prefLabel(g, subject)
+        breakpoint()
+
+
+def export_jsonld(
+    ontology_parser,
+    input_type: str,
+    input_data: Dict[str, Any],
+    output_path: str,
+    cell_id: str = "BattMo",
+    cell_type: str = "PouchCell",
+):
+    g = ontology_parser.graph
+    input_key = ontology_parser.key_map.get(input_type)
+    if not input_key:
+        raise ValueError(f"Invalid input type: {input_type}")
+
+    out = {
+        "@context": "https://w3id.org/emmo/domain/battery/context",
+        "@graph": {
+            "@id": cell_id,
+            "@type": cell_type,
+            "hasProperty": [],
+        },
+    }
+    has_property = out["@graph"]["hasProperty"]
+
+    for subject in set(g.subjects(input_key, None)):
+        path = None
+        for p, o in g.predicate_objects(subject):
+            if p == input_key:
+                path = ontology_parser.parse_key(str(o))
+                break
+        if not path:
+            continue
+        value = _get_value_from_path(input_data, path)
+        if value is None:
+            continue
+
+        prop_obj = {
+            "@type": _curie(g, subject),
+            "rdfs:label": _get_skos_prefLabel(g, subject),
+        }
+
+        if _is_number_like(value):
+            prop_obj["hasNumericalPart"] = {
+                "@type": "Real",
+                "hasNumericalValue": float(value),
+            }
+        elif isinstance(value, str):
+            prop_obj["hasStringPart"] = {
+                "@type": "String",
+                "hasStringValue": value,
+            }
+        elif isinstance(value, (list, dict)):
+            if isinstance(value, dict) and "functionname" in value:
+                func_name = str(value["functionname"])
+                prop_obj["hasStringPart"] = {
+                    "@type": "String",
+                    "hasStringValue": func_name,
+                }
+            else:
+                prop_obj["hasStringPart"] = {
+                    "@type": "String",
+                    "hasStringValue": str(value),
+                }
+        else:
+            prop_obj["hasStringPart"] = {
+                "@type": "String",
+                "hasStringValue": str(value),
+            }
+        has_property.append(prop_obj)
+
+        # Add units
+        unit = _get_unit_for_subject(g, subject)
+        if unit:
+            prop_obj["emmo:hasMeasurementUnit"] = unit
+
+    with open(output_path, "w", encoding="utf-8") as f:
+        json.dump(out, f, indent=2)
+
+    # Find values not mapped
+    input_paths, mapped_paths, missing = _find_missing_values(
+        ontology_parser, input_data, input_type
+    )
+
+    print("Number of JSON leaf values:", len(input_paths))
+    print("Number of mapped values:", len(mapped_paths))
+    print("Missing values:", len(missing))
+    print("These are missing values from the ontology mapping:")
+    for p in missing:
+        print(" ", ".".join(str(x) for x in p))
+    print("Write these missing values to 'missing_values.json'")
+    with open("missing_values.json", "w", encoding="utf-8") as f:
+        json.dump([".".join(str(x) for x in p) for p in missing], f, indent=2)

BatteryModelMapper/ontology_parser.py

@@ -1,17 +1,41 @@
+from pathlib import Path
+from urllib.parse import urlparse
+
 import ast
+import json
 import requests
-from rdflib import Graph, URIRef
+
+from rdflib import Graph, URIRef, OWL
+from rdflib.namespace import RDF
+
 
 class OntologyParser:
-    def __init__(self, ontology_url):
+    def __init__(self, ontology_ref):
         self.graph = Graph()
-        response = requests.get(ontology_url)
-        response.raise_for_status()
-        self.graph.parse(data=response.text, format='ttl')
+        ontology_ref = Path(ontology_ref)
+
+        if urlparse(str(ontology_ref)).scheme in ("http", "https"):
+            response = requests.get(ontology_ref)
+            response.raise_for_status()
+            response_text = response.text
+        elif ontology_ref.is_file():
+            with open(ontology_ref, "r", encoding="utf-8") as f:
+                response_text = f.read().replace("\r\n", "\n")
+        else:
+            raise ValueError(f"File does not exist: {ontology_ref}")
+
+        self.graph.parse(data=response_text, format="ttl")
+
         self.key_map = {
-            'bpx': URIRef("https://w3id.org/emmo/domain/battery-model-lithium-ion#bmli_0a5b99ee_995b_4899_a79b_925a4086da37"),
-            'cidemod': URIRef("https://w3id.org/emmo/domain/battery-model-lithium-ion#bmli_1b718841_5d72_4071_bb71_fc4a754f5e30"),
-            'battmo': URIRef("https://w3id.org/emmo/domain/battery-model-lithium-ion#bmli_2c718841_6d73_5082_bb81_gc5b754f6e40")  # Placeholder URI
+            "bpx": URIRef(
+                "https://w3id.org/emmo/domain/battery-model-lithium-ion#bmli_0a5b99ee_995b_4899_a79b_925a4086da37"
+            ),
+            "cidemod": URIRef(
+                "https://w3id.org/emmo/domain/battery-model-lithium-ion#bmli_1b718841_5d72_4071_bb71_fc4a754f5e30"
+            ),
+            "battmo.m": URIRef(
+                "https://w3id.org/emmo/domain/battery-model-lithium-ion#bmli_e5e86474_8623_48ea_a1cf_502bdb10aa14"
+            ),
         }
 
     def parse_key(self, key):
@@ -25,7 +49,9 @@ def get_mappings(self, input_type, output_type):
         input_key = self.key_map.get(input_type)
         output_key = self.key_map.get(output_type)
         if not input_key or not output_key:
-            raise ValueError(f"Invalid input or output type: {input_type}, {output_type}")
+            raise ValueError(
+                f"Invalid input or output type: {input_type}, {output_type}"
+            )
 
         mappings = {}
         for subject in self.graph.subjects():

BatteryModelMapper/parameter_mapper.py

@@ -1,6 +1,7 @@
 import json
 import re
 
+
 class ParameterMapper:
     def __init__(self, mappings, template, input_url, output_type, input_type):
         self.mappings = mappings
@@ -17,8 +18,6 @@ def map_parameters(self, input_data):
             if value is not None:
                 if isinstance(value, str):
                     value = self.replace_variables(value)
-                if self.input_type == 'cidemod' and 'kinetic_constant' in input_key:
-                    value = self.scale_kinetic_constant(value)
                 self.set_value_from_path(output_data, output_key, value)
                 self.remove_default_from_used(output_key)
         self.set_bpx_header(output_data)
@@ -27,17 +26,10 @@ def map_parameters(self, input_data):
 
     def replace_variables(self, value):
         if isinstance(value, str):
-            value = re.sub(r'\bx_s\b', 'x', value)
-            value = re.sub(r'\bc_e\b', 'x', value)
+            value = re.sub(r"\bx_s\b", "x", value)
+            value = re.sub(r"\bc_e\b", "x", value)
         return value
 
-    def scale_kinetic_constant(self, value):
-        try:
-            return value * 1e6
-        except TypeError:
-            print(f"Error scaling kinetic_constant value: {value}")
-            return value
-
     def get_all_paths(self, data, path=""):
         paths = set()
         if isinstance(data, dict):
@@ -66,7 +58,7 @@ def get_value_from_path(self, data, keys):
                     return None
             return data
         except (KeyError, IndexError, ValueError, TypeError) as e:
-            print(f"Error accessing key {key} in path {keys}: {e}")
+            print(f"Warning: accessing key {key} in path {keys}: {e}")
             return None
 
     def set_value_from_path(self, data, keys, value):
@@ -86,7 +78,9 @@ def set_value_from_path(self, data, keys, value):
             final_key = keys[-1]
             if isinstance(final_key, str):
                 final_key = final_key.strip()
-            if isinstance(final_key, int) or (isinstance(final_key, str) and final_key.isdigit()):
+            if isinstance(final_key, int) or (
+                isinstance(final_key, str) and final_key.isdigit()
+            ):
                 final_key = int(final_key)
             data[final_key] = value
             print(f"Set value for path {keys}: {value}")
@@ -107,9 +101,13 @@ def set_bpx_header(self, data):
             "BPX": 0.1,
             "Title": "An autoconverted parameter set using BatteryModelMapper",
             "Description": f"This data set was automatically generated from {self.input_url}. Please check carefully.",
-            "Model": "DFN"
+            "Model": "DFN",
         }
         data.pop("Validation", None)
 
     def remove_high_level_defaults(self):
-        self.defaults_used = {path for path in self.defaults_used if not any(k in path for k in ["Parameterisation", "Header"])}
+        self.defaults_used = {
+            path
+            for path in self.defaults_used
+            if not any(k in path for k in ["Parameterisation", "Header"])
+        }