ups

cobrakbase/Workspace/staging.py

@@ -0,0 +1,145 @@
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+
+"""
+From: /kb/dev_container/narrative/src/biokbase/narrative/staging/helper.py
+KBase staging.helper class Tianhao Gu <tgu@anl.gov>
+"""
+
+STAGING_URL = 'https://kbase.us/services/staging_service'
+
+CHUNK_SIZE = 16 * 1024
+
+
+class Staging:
+
+    def __fetch_url(self, end_point, values=None, headers=None, method="GET", save_path=None):
+        """Fetching URL
+        By default, it sends a GET request with {"Authorization": $KB_AUTH_TOKEN} header
+        """
+        if save_path and os.path.exists(save_path):
+            raise ValueError(
+                f"A file exists at {save_path} but this method does not overwrite files"
+            )
+
+        data = None
+        if values:
+            data = urllib.parse.urlencode(values)
+
+        if not headers:
+            headers = {"Authorization": self._token}
+
+        req = urllib.request.Request(end_point, data, headers)
+        req.get_method = lambda: method
+        try:
+            response = urllib.request.urlopen(req)
+        except urllib.error.URLError as e:
+            error_msg = "The server could not fulfill the request.\n"
+
+            server_msg = e.read()
+            if server_msg:
+                error_msg += f"Server message: {server_msg}\n"
+
+            if hasattr(e, "reason"):
+                error_msg += f"Reason: {e.reason}\n"
+
+            if hasattr(e, "code"):
+                error_msg += f"Error code: {e.code}\n"
+
+            raise ValueError(error_msg) from e
+
+        if not save_path:
+            return response.read()
+
+        with open(save_path, "wb") as f:
+            while True:
+                chunk = response.read(CHUNK_SIZE)
+                if not chunk:
+                    break
+                f.write(chunk)
+
+    def __init__(self, token):
+        """Initializes a new Helper instance."""
+        self._token = token
+        self._staging_url = STAGING_URL
+
+    def list(self, directory=""):
+        """Calling LIST endpoint and return a list of file path"""
+        end_point = self._staging_url + "/list/" + directory
+        response = self.__fetch_url(end_point)
+
+        resp_json = json.loads(response)
+
+        file_list = []
+        for file in resp_json:
+            if not file.get("isFolder"):
+                file_list.append(file.get("path"))
+
+        return sorted(file_list)
+
+    def metadata(self, path=""):
+        """Calling METADATA endpoint and return metadata in JSON format"""
+        if not path:
+            raise ValueError("Must provide path argument")
+
+        end_point = self._staging_url + "/metadata/" + path
+        response = self.__fetch_url(end_point)
+
+        return json.loads(response)
+
+    def jgi_metadata(self, path=""):
+        """Calling JGI-METADATA endpoint and return metadata in JSON format"""
+        if not path:
+            raise ValueError("Must provide path argument")
+
+        end_point = self._staging_url + "/jgi-metadata/" + path
+        response = self.__fetch_url(end_point)
+
+        return json.loads(response)
+
+    def search(self, path=""):
+        """Calling SEARCH endpoint and return server response in JSON format"""
+        if not path:
+            raise ValueError("Must provide path argument")
+
+        end_point = self._staging_url + "/search/" + path
+        response = self.__fetch_url(end_point)
+
+        return json.loads(response)
+
+    def delete(self, path=""):
+        """Calling DELETE endpoint and return server response in JSON format"""
+        if not path:
+            raise ValueError("Must provide path argument")
+
+        end_point = self._staging_url + "/delete/" + path
+        response = self.__fetch_url(end_point, method="DELETE")
+
+        return {"server_response": response}
+
+    def download(self, path, save_location=None):
+        """Calling DOWNLOAD endpoint and saving the resulting file"""
+        if not save_location:
+            save_location = "./" + os.path.basename(path)
+
+        end_point = self._staging_url + "/download/" + path
+        self.__fetch_url(end_point, save_path=save_location)
+
+        return save_location
+
+    def mv(self, path="", new_path=""):
+        """Calling MV endpoint and return server response in JSON format"""
+        if not path:
+            raise ValueError("Must provide path argument")
+
+        if not new_path:
+            raise ValueError("Must provide new_path argument")
+
+        end_point = self._staging_url + "/mv/" + path
+        body_values = {"newPath": new_path}
+        response = self.__fetch_url(end_point, values=body_values, method="PATCH")
+
+        return {"server_response": response}

cobrakbase/core/kbasefba/fbamodel.py

@@ -95,6 +95,9 @@ class FBAModel(KBaseObject, Model):
           float ATPSynthaseStoichiometry;
           float ATPMaintenance;
           list<ModelQuantOpt> quantopts;
+
+
+
     """
 
     OBJECT_TYPE = "KBaseFBA.FBAModel"
@@ -232,7 +235,53 @@ def _set_gapgens(self, value):
     gapgens = property(_get_gapgens, _set_gapgens)
 
     def future__init__(
-        self,
+            self,
+            id_or_model: Union[str, "Model", None] = None,
+            name: Optional[str] = None,
+            source: str = None,
+            source_id: str = None,
+            type: str = None,
+            genome = None,
+            metagenome = None,
+            template = None,
+            ATPSynthaseStoichiometry: float = None,
+            ATPMaintenance: float = None,
+            attributes: dict = None,
+            drain_list: dict = None,
+            info=None,
+            args=None,
+    ) -> None:
+        """
+
+        @param id_or_model:
+        @param name:
+        @param source:
+        @param source_id:
+        @param type:
+        @param genome:
+        @param metagenome:
+        @param template:
+        @param ATPSynthaseStoichiometry:
+        @param ATPMaintenance:
+        @param attributes:
+        @param drain_list:
+        @param info:
+        @param args:
+        @return:
+        """
+        if info is None:
+            info = KBaseObjectInfo(object_type=FBAModel.OBJECT_TYPE)
+        KBaseObject.__init__(self, {}, info, args, None)
+        Model.__init__(self, id_or_model, name)
+        self.source = None
+        self.source_id = None
+        self.type = None
+        self.template = None
+        self.gap_fillings = None
+        self.gap_gens = None
+
+    @staticmethod
+    def actual_init(
         id_or_model: Union[str, "Model", None] = None,
         name: Optional[str] = None,
         info=None,
@@ -343,34 +392,7 @@ def _to_json(self):
         data = {}
         ignore = {"modelcompounds", "modelreactions", "biomasses", "modelcompartments"}
 
-        data["modelcompartments"] = [converter.convert_compartment(x) for x in self.compartments]
-        """
-        for cmp_id, name in self.compartments:
-            model_compartment = {
-                "id": cmp_id,
-                "label": name,
-                "compartment_ref": "~/template/compartments/id/" + cmp_id[:-1],
-            }
-            cmp_meta_data_key = f"kbase_compartment_data_{cmp_id}"
-            if cmp_meta_data_key in self.notes:
-                m = self.notes[cmp_meta_data_key]
-                if type(m) == str:
-                    extra_data = json.loads(m.replace("&apos;", '"'))
-                elif type(m) == dict:
-                    extra_data = m
-                else:
-                    raise ValueError(
-                        f"note field for {cmp_meta_data_key} must be either str or dict, found: {type(m)}"
-                    )
-                if "compartmentIndex" in extra_data:
-                    extra_data["compartmentIndex"] = int(extra_data["compartmentIndex"])
-                if "pH" in extra_data:
-                    extra_data["pH"] = float(extra_data["pH"])
-                if "potential" in extra_data:
-                    extra_data["potential"] = float(extra_data["potential"])
-                model_compartment.update(extra_data)
-            data["modelcompartments"].append(model_compartment)
-        """
+        data["modelcompartments"] = converter.build_model_compartments()
 
         for key in self.data_keys:
             if key not in ignore: