refactor

onshape/README.md

@@ -0,0 +1,58 @@
+# Onshape API Utilities
+
+This module provides utilities for interacting with the Onshape API, including uploading and translating files and downloading translated files.
+
+## Prerequisites
+
+1. Set the following environment variables:
+   - `ONSHAPE_ACCESS_KEY`
+   - `ONSHAPE_SECRET_KEY`
+
+2. Install required Python packages:
+   ```bash
+   pip install requests loguru
+   ```
+
+## Modules
+
+### `onshape_base.py`
+
+Provides the `OnshapeBase` class for shared functionality like authentication and base URL construction.
+
+### `onshape_translation.py`
+
+Handles file upload and translation. Supports any file type and export format.
+
+#### Example Usage:
+```python
+from onshape_translation import OnshapeTranslation
+
+did = "your_document_id"
+wid = "your_workspace_id"
+file_path = "path_to_your_file"
+format_name = "desired_format"  # e.g., "STEP", "IGES", etc.
+
+translator = OnshapeTranslation(did, wid, file_path, format_name)
+translator.upload_and_translate()
+```
+
+### `onshape_download.py`
+
+Handles downloading translated files.
+
+#### Example Usage:
+```python
+from onshape_download import OnshapeDownload
+
+did = "your_document_id"
+wid = "your_workspace_id"
+eid = "your_element_id" # you can find it in `resultElementIds` when `requestState` of `TranslationStatusResponse` is `DONE`
+output_file = "output_file_name"
+
+downloader = OnshapeDownload(did, wid, eid, output_file)
+downloader.download()
+```
+
+## Logging
+
+This module uses `loguru` for logging. Logs will display detailed information about the operations performed.

onshape/onshape_base.py

@@ -0,0 +1,21 @@
+import os
+import base64
+
+
+class OnshapeBase:
+    def __init__(self):
+        self.access_key = os.environ.get("ONSHAPE_ACCESS_KEY")
+        self.secret_key = os.environ.get("ONSHAPE_SECRET_KEY")
+        self.base_url = "https://cad.onshape.com/api/v10"
+
+        if not self.access_key or not self.secret_key:
+            raise ValueError(
+                "ONSHAPE_ACCESS_KEY and ONSHAPE_SECRET_KEY must be set in environment variables."
+            )
+
+    def get_auth_header(self):
+        credentials_raw = f"{self.access_key}:{self.secret_key}"
+        credentials_base64 = base64.b64encode(credentials_raw.encode("utf-8")).decode(
+            "utf-8"
+        )
+        return {"Authorization": f"Basic {credentials_base64}"}

onshape/onshape_download.py

@@ -1,33 +1,39 @@
-import os
 import requests
-import base64
-
-access_key = os.environ.get("ONSHAPE_ACCESS_KEY")
-secret_key = os.environ.get("ONSHAPE_SECRET_KEY")
-did = "ef42d7639096f3e61a4d4f07"
-wid = "5fcd0f25ce3dee08bbb823bf"
-prt_file_path = "prt0002.prt"
-tid = "68382b8879834455c6382294"
-resulteid = "c46568af3e8bf618762a1db3"
-
-# Basic Auth header
-credentials_raw = f"{access_key}:{secret_key}"
-credentials_base64 = base64.b64encode(credentials_raw.encode("utf-8")).decode("utf-8")
-
-url = "https://cad.onshape.com/api/v10" + f"/blobelements/d/{did}/w/{wid}/e/{resulteid}"
-
-headers = {
-    "Accept": "application/json;charset=UTF-8; qs=0.09",
-    "Authorization": f"Basic {credentials_base64}",
-    # "X-XSRF-TOKEN": "QcscTFRgL6vE7h7UFHKmng==",
-    # DO NOT set Content-Type manually when sending files with requests
-}
-
-
-files = {"file": open(prt_file_path, "rb")}
-
-response = requests.get(url, headers=headers)
-
-with open("downloaded_file.step", "wb") as f:
-    f.write(response.content)
-    print("File downloaded successfully as 'downloaded_file.step'.")
+from loguru import logger
+from onshape_base import OnshapeBase
+
+
+class OnshapeDownload(OnshapeBase):
+    def __init__(self, did, wid, eid, output_file):
+        super().__init__()
+        self.did = did
+        self.wid = wid
+        self.eid = eid
+        self.output_file = output_file
+
+    def download(self):
+        url = f"{self.base_url}/blobelements/d/{self.did}/w/{self.wid}/e/{self.eid}"
+        headers = {
+            "Accept": "application/json;charset=UTF-8; qs=0.09",
+            **self.get_auth_header(),
+        }
+
+        logger.info(
+            f"Downloading file with element ID '{self.eid}' to '{self.output_file}'."
+        )
+        response = requests.get(url, headers=headers)
+
+        with open(self.output_file, "wb") as f:
+            f.write(response.content)
+            logger.success(f"File downloaded successfully as '{self.output_file}'.")
+
+
+# Example usage
+if __name__ == "__main__":
+    did = "ef42d7639096f3e61a4d4f07"
+    wid = "5fcd0f25ce3dee08bbb823bf"
+    eid = "fd0dbebc1ac05de1f1d1ed07"  # you can find it in `resultElementIds` when `requestState` of `TranslationStatusResponse` is `DONE`
+    output_file = "output_file.step"
+
+    downloader = OnshapeDownload(did, wid, eid, output_file)
+    downloader.download()

onshape/onshape_schema.py

@@ -0,0 +1,39 @@
+from pydantic import BaseModel, Field
+
+
+class TranslationResponse(BaseModel):
+    document_id: str = Field(alias="documentId")
+    failure_reason: str | None = Field(alias="failureReason", default=None)
+    href: str
+    id: str
+    name: str
+    request_element_id: str = Field(alias="requestElementId")
+    request_state: str = Field(alias="requestState")
+    result_document_id: str = Field(alias="resultDocumentId")
+    result_element_ids: list[str] | None = Field(alias="resultElementIds", default=None)
+    result_external_data_ids: list[str] | None = Field(
+        alias="resultExternalDataIds", default=None
+    )
+    result_workspace_id: str | None = Field(alias="resultWorkspaceId", default=None)
+    version_id: str | None = Field(alias="versionId", default=None)
+    view_ref: str | None = Field(alias="viewRef", default=None)
+    workspace_id: str = Field(alias="workspaceId")
+
+
+class TranslationStatusResponse(BaseModel):
+    id: str
+    request_state: str = Field(alias="requestState")
+    request_element_id: str = Field(alias="requestElementId")
+    result_external_data_ids: list[str] | None = Field(
+        alias="resultExternalDataIds", default=None
+    )
+    export_rule_file_name: str | None = Field(alias="exportRuleFileName", default=None)
+    version_id: str | None = Field(alias="versionId", default=None)
+    workspace_id: str = Field(alias="workspaceId")
+    document_id: str = Field(alias="documentId")
+    result_element_ids: list[str] | None = Field(alias="resultElementIds", default=None)
+    result_document_id: str = Field(alias="resultDocumentId")
+    failure_reason: str | None = Field(alias="failureReason", default=None)
+    result_workspace_id: str | None = Field(alias="resultWorkspaceId", default=None)
+    name: str
+    href: str

onshape/onshape_translation.py

@@ -1,35 +1,96 @@
-import os
+import time
+
 import requests
-import base64
+from loguru import logger
+from onshape_base import OnshapeBase
+from onshape_schema import TranslationResponse, TranslationStatusResponse
+
+
+class OnshapeTranslation(OnshapeBase):
+    def __init__(self, did: str, wid: str, file_path: str, format_name: str):
+        super().__init__()
+        self.did = did
+        self.wid = wid
+        self.file_path = file_path
+        self.format_name = format_name
+
+    def upload_and_translate(self):
+        url = f"{self.base_url}/translations/d/{self.did}/w/{self.wid}"
+        headers = {
+            "Accept": "application/json;charset=UTF-8; qs=0.09",
+            **self.get_auth_header(),
+        }
+        data = {"translate": "true", "formatName": self.format_name}
+        files = {"file": open(self.file_path, "rb")}
 
-access_key = os.environ.get("ONSHAPE_ACCESS_KEY")
-secret_key = os.environ.get("ONSHAPE_SECRET_KEY")
-did = "ef42d7639096f3e61a4d4f07"
-wid = "5fcd0f25ce3dee08bbb823bf"
-prt_file_path = "prt0002.prt"
+        logger.info(
+            f"Uploading file '{self.file_path}' for translation to '{self.format_name}' format."
+        )
+        response = requests.post(url, headers=headers, data=data, files=files)
 
-# Basic Auth header
-credentials_raw = f"{access_key}:{secret_key}"
-credentials_base64 = base64.b64encode(credentials_raw.encode("utf-8")).decode("utf-8")
+        try:
+            translate_response = response.json()
+            if response.status_code != 200:
+                logger.error(
+                    f"Failed to upload file. Status code: {response.status_code}"
+                )
+                logger.error(response.text)
+                raise Exception(
+                    f"Failed to upload file. Status code: {response.status_code}"
+                )
+            logger.success(f"Translate response: {translate_response}")
+            return TranslationResponse(**translate_response)
+        except Exception:
+            logger.error(f"Status: {response.status_code}")
+            logger.error(response.text)
+            raise Exception(
+                f"Failed to upload and translate file. Status code: {response.status_code}"
+            )
 
-url = "https://cad.onshape.com/api/v10" + f"/translations/d/{did}/w/{wid}"
+    def get_translation_status(self, translation_id: str):
+        url = f"{self.base_url}/translations/{translation_id}"
+        headers = {
+            "Accept": "application/json;charset=UTF-8; qs=0.09",
+            **self.get_auth_header(),
+        }
 
-headers = {
-    "Accept": "application/json;charset=UTF-8; qs=0.09",
-    "Authorization": f"Basic {credentials_base64}",
-    # "X-XSRF-TOKEN": "QcscTFRgL6vE7h7UFHKmng==",
-    # DO NOT set Content-Type manually when sending files with requests
-}
+        logger.info(f"Checking translation status for ID '{translation_id}'.")
+        response = requests.get(url, headers=headers)
 
-# All form fields as strings (empty strings '' where appropriate)
-data = {"translate": "true", "formatName": "STEP"}
+        try:
+            status = response.json()
+            if response.status_code != 200:
+                logger.error(
+                    f"Failed to get translation status. Status code: {response.status_code}"
+                )
+                logger.error(response.text)
+                raise Exception(
+                    f"Failed to get translation status. Status code: {response.status_code}"
+                )
+            logger.success(f"Translation status: {status}")
+            return TranslationStatusResponse(**status)
+        except Exception:
+            logger.error(f"Status: {response.status_code}")
+            logger.error(response.text)
+            raise Exception(
+                f"Failed to get translation status. Status code: {response.status_code}"
+            )
 
-files = {"file": open(prt_file_path, "rb")}
 
-response = requests.post(url, headers=headers, data=data, files=files)
+# Example usage
+if __name__ == "__main__":
+    did = "ef42d7639096f3e61a4d4f07"
+    wid = "5fcd0f25ce3dee08bbb823bf"
+    file_path = "onshape/example/prt0002.prt"
+    format_name = "STEP"
 
-try:
-    print(f"✅ Translate response: {response.json()}")
-except Exception:
-    print(f"Status: {response.status_code}")
-    print(response.text)
+    translator = OnshapeTranslation(did, wid, file_path, format_name)
+    response = translator.upload_and_translate()
+    response = translator.get_translation_status(response.id)
+    while response.request_state not in ["DONE", "FAILED"]:
+        logger.info(
+            f"Waiting for translation to complete. Current state: {response.request_state}"
+        )
+        response = translator.get_translation_status(response.id)
+        time.sleep(3)
+    logger.success(f"Translation completed with state: {response.request_state}")