add ifc export

APSToolkitPython/src/aps_toolkit/Derivative.py

@@ -30,7 +30,7 @@
 from .Token import Token
 import json
 import pandas as pd
-
+import time
 
 class Derivative:
     def __init__(self, urn: str, token: Token, region: str = "US"):
@@ -76,6 +76,93 @@ def translate_job(self, root_file_name: str, type: str = "svf", generate_master_
         response = requests.post(url, headers=headers, data=payload)
         return response.text
 
+    def download_stream_ifc(self,ifc_setting_name="IFC4 Reference View"):
+        response = self.check_job_status()
+        derivatives = response["derivatives"]
+        # find any outputType is "ifc"
+        flag = any(derivative["outputType"] == "ifc" for derivative in derivatives)
+        if not flag:
+            # start call translate job
+            response = self.translate_to_ifc(ifc_setting_name)
+            if response.status_code != 200:
+                raise Exception("Translate job failed.Return: " + response.text)
+            # check job status
+            response = self.check_job_status()
+            process = response["progress"]
+            while process != "complete" and response['status']=='inprogress':
+                time.sleep(5)
+                response = self.check_job_status()
+                process = response["progress"]
+            # get all derivatives
+            derivatives = response["derivatives"]
+        # find in list derivatives any item has outputType is "ifc"
+        for derivative in derivatives:
+            if derivative["outputType"] == "ifc":
+                # get the manifest of the derivative
+                children = derivative["children"][0] if "children" in derivative else []
+                if not children:
+                    raise Exception("The derivative has no Urn IFC.")
+                guid = children["guid"]
+                mime = children["mime"]
+                path_info = self._decompose_urn(children["urn"])
+                urn_json = children["urn"]
+                manifest_item = ManifestItem(guid, mime, path_info, urn_json)
+                local_path = path_info.local_path
+                myUri = "file://" + path_info.base_path + path_info.root_file_name
+                remote_path = join("derivativeservice/v2/derivatives/", unquote(myUri)[len("file://"):])
+                resource = Resource(manifest_item.path_info.root_file_name, remote_path, local_path)
+                # download the IFC file
+                stream = self.download_stream_resource(resource)
+                return {"file_name": manifest_item.path_info.root_file_name, "stream": stream}
+        return None
+    def download_ifc(self, file_path=None,ifc_setting_name="IFC4 Reference View"):
+        result_dict = self.download_stream_ifc(ifc_setting_name)
+        if result_dict is None:
+            raise Exception("Can not download IFC file.")
+        if file_path is None:
+            file_path = result_dict["file_name"]
+        # in case file exits, remove it
+        if os.path.exists(file_path):
+            try:
+                os.remove(file_path)
+            except Exception as e:
+                raise Exception(f"Can not remove file {file_path}. Reason: {e}")
+        bytes_io = result_dict["stream"]
+        with open(file_path, "wb") as f:
+            f.write(bytes_io.read())
+        return file_path
+
+    def translate_to_ifc(self, ifc_setting_name="IFC4 Reference View")-> requests.Response:
+        """
+        Run a Job to translate the model to IFC format:
+        https://aps.autodesk.com/blog/export-ifc-rvt-using-model-derivative-api
+        :return:
+        """
+        url = 'https://developer.api.autodesk.com/modelderivative/v2/designdata/job'
+        headers = {
+            'Authorization': f'Bearer {self.token.access_token}',
+            'Content-Type': 'application/json',
+            'x-ads-force': 'true'
+        }
+        data = {
+            "input": {
+                "urn": self.urn
+            },
+            "output": {
+                "formats": [
+                    {
+                        "type": "ifc",
+                        "advanced": {
+                            "exportSettingName": f"{ifc_setting_name}"
+                        }
+                    }
+                ]
+            }
+        }
+        # Make the POST request
+        response = requests.post(url, headers=headers, json=data)
+        return response
+
     def check_job_status(self):
         url = f"{self.host}/modelderivative/v2/designdata/{self.urn}/manifest"
         access_token = self.token.access_token

APSToolkitPython/src/test/test_derivative.py

@@ -1,7 +1,7 @@
 from unittest import TestCase
-
 from .context import Derivative
 from .context import Auth
+import os
 
 
 class TestDerivative(TestCase):
@@ -15,8 +15,25 @@ def test_translate_job(self):
         response = derivative.translate_job("Project Completion.ifc")
         self.assertNotEquals(response, "")
 
+    def test_translate_to_ifc(self):
+        ## Wrong Urn
+        #self.urn = "dXJuOmFkc2sud2lwcHJvZDpmcy5maWxlOnZmLktXNWVBZ25oUjNpRVJaVnh1bEVrb3c_dmVyc2lvbj0x"
+        ## Right Urn
+        self.urn = "dXJuOmFkc2sud2lwcHJvZDpmcy5maWxlOnZmLmRnRkswLXZqVFlLRS1tUDA3Z3o3WUE_dmVyc2lvbj0z"
+        derivative = Derivative(self.urn, self.token)
+        response = derivative.translate_to_ifc()
+        status_code = response.status_code
+        self.assertEqual(status_code, 200)
+    def test_download_ifc(self):
+        self.urn = "dXJuOmFkc2sud2lwcHJvZDpmcy5maWxlOnZmLnc3cjBxY1BRUzNXVDczeGV6dC13SEE_dmVyc2lvbj0z"
+        derivative = Derivative(self.urn, self.token)
+        filepath = derivative.download_ifc()
+        # check size
+        self.assertNotEqual(os.path.getsize(filepath), 0)
+
+
     def test_check_job_status(self):
-        self.urn = "dXJuOmFkc2sub2JqZWN0czpvcy5vYmplY3Q6Y2h1b25nX2J1Y2tldC9NeUhvdXNlLm53Yw"
+        self.urn = "dXJuOmFkc2sud2lwcHJvZDpmcy5maWxlOnZmLmRnRkswLXZqVFlLRS1tUDA3Z3o3WUE_dmVyc2lvbj0z"
         derivative = Derivative(self.urn, self.token)
         response = derivative.check_job_status()
         self.assertNotEquals(response, "")