Merge remote-tracking branch 'origin/add-emulators' into add-emulators

Author: benjimaclellan

src/oqd_cloud/client.py

@@ -13,10 +13,10 @@
 # limitations under the License.
 
 
-from typing import Literal, Optional, Sequence
+from typing import Optional, Sequence
+import urllib.request
 
 import requests
-import json
 from oqd_core.backend.task import Task
 from pydantic import BaseModel, ConfigDict
 
@@ -135,12 +135,7 @@ def connect(self, provider: Provider, username: str, password: str):
     #     self.connect(self, self.provider)
     #     pass
 
-    def submit_job(
-        self, 
-        task: Task, 
-        backend: str,
-        tags: str
-    ):
+    def submit_job(self, task: Task, backend: str, tags: str):
         """Submit a Task as an AnalogCircuit, DigitalCircuit, or AtomicCircuit to a backend."""
         response = requests.post(
             self.provider.job_submission_url(backend=backend),
@@ -167,6 +162,10 @@ def retrieve_job(self, job_id):
         job = Job.model_validate(response.json())
 
         if response.status_code == 200:
+            # download result file from temporary link
+            with urllib.request.urlopen(job.result) as f:
+                job.result = f.read().decode("utf-8")
+
             self._jobs[job_id] = job
             return self.jobs[job_id]
 

src/oqd_cloud/provider.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import requests 
+import requests
 from oqd_cloud.server.model import Backends
 
 
@@ -28,17 +28,15 @@ def __init__(self, host: str = "http://localhost", port: int = 8000):
 
         # get available backends
         self.backends = Backends(available=[])
-        response = requests.get(
-            self.url + "/available_backends"
-        )
+        response = requests.get(self.url + "/available_backends")
         backends = Backends.model_validate(response.json())
         if response.status_code == 200:
             self.backends = backends
 
     @property
     def available_backends(self):
         return self.backends.available
-    
+
     @property
     def registration_url(self):
         return self.url + "/auth/register"

src/oqd_cloud/server/jobqueue.py

@@ -39,11 +39,10 @@
 
 async def _report_success(job, connection, result, *args, **kwargs):
     async with asynccontextmanager(get_db)() as db:
-        
         save_obj(job, result)
         url = get_temp_link(job)
-        status_update = dict(status="finished", result=url)  
-        # status_update = dict(status="finished", result=result.model_dump_json())  
+        status_update = dict(status="finished", result=url)
+        # status_update = dict(status="finished", result=result.model_dump_json())
         query = await db.execute(select(JobInDB).filter(JobInDB.job_id == job.id))
         job_in_db = query.scalars().first()
         for k, v in status_update.items():

src/oqd_cloud/server/main.py

@@ -15,4 +15,4 @@
 if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run("app:app", host="0.0.0.0", port=8007, log_level='debug')
+    uvicorn.run("app:app", host="0.0.0.0", port=8007, log_level="debug")

src/oqd_cloud/server/model.py

@@ -45,7 +45,7 @@ class Job(BaseModel):
     result: Optional[str] = None
     user_id: str
     tags: Optional[str] = None
-    
+
 
 class Backends(BaseModel):
-    available: Sequence[str]
+    available: Sequence[str]

src/oqd_cloud/server/route/job.py

@@ -12,13 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Literal, Sequence, Optional, Annotated
+from typing import Literal, Annotated
 
 from fastapi import APIRouter, HTTPException, Body
 from fastapi import status as http_status
 
 ########################################################################################
-import oqd_analog_emulator #.qutip_backend import QutipBackend
+import oqd_analog_emulator  # .qutip_backend import QutipBackend
 import oqd_trical
 from oqd_core.backend.task import Task
 from rq.job import Callback
@@ -51,8 +51,8 @@
 @job_router.get("/available_backends")
 async def available_backends():
     return backends
-    
-    
+
+
 @job_router.post("/submit/{backend}", tags=["Job"])
 async def submit_job(
     backend: Literal[tuple(backends.available)],
@@ -61,7 +61,8 @@ async def submit_job(
     user: user_dependency,
     db: db_dependency,
 ):
-    print(f"Queueing {task} on server {backend} backend. {len(queue)} jobs in queue.")
+    print(f"Queueing task on server {backend} backend. {len(queue)} jobs in queue.")
+    # print(f"Queueing {task} on server {backend} backend. {len(queue)} jobs in queue.")
 
     job = queue.enqueue(
         _backends[backend].run,

src/oqd_cloud/server/storage.py

@@ -12,42 +12,48 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Literal
 from minio import Minio
 import os
 import io
 from datetime import timedelta
 
-from oqd_cloud.server.database import JobInDB, get_db
+from oqd_cloud.server.database import JobInDB
 
 ########################################################################################
 
 minio_client = Minio(
-    f"{os.getenv("MINIO_ENDPOINT")}:9000",
+    "localhost:9000",
     access_key=os.getenv("MINIO_ROOT_USER"),
     secret_key=os.getenv("MINIO_ROOT_PASSWORD"),
-    secure=False
+    secure=False,
 )
 
 DEFAULT_MINIO_BUCKET = os.getenv("MINIO_DEFAULT_BUCKETS")
 RESULT_FILENAME = "result.json"
 
+if not minio_client.bucket_exists(DEFAULT_MINIO_BUCKET):
+    minio_client.make_bucket(DEFAULT_MINIO_BUCKET)
+    print("Created bucket", DEFAULT_MINIO_BUCKET)
+else:
+    print("Bucket", DEFAULT_MINIO_BUCKET, "already exists")
+
+
 def save_obj(job: JobInDB, result):
     # if the file is already saved, fput can be used
     # minio_client.fput_object(
     #     BUCKET, destination_file, source_file,
     # )
-    
-    # here we dump to json, todo: future version should dump to HDF5 
-    json_bytes = result.model_dump_json().encode('utf-8')
+
+    # here we dump to json, todo: future version should dump to HDF5
+    json_bytes = result.model_dump_json().encode("utf-8")
     buffer = io.BytesIO(json_bytes)
 
     minio_client.put_object(
-        DEFAULT_MINIO_BUCKET, 
-        f"{job.id}/{RESULT_FILENAME}",  
-        data=buffer, 
-        length=len(json_bytes), 
-        content_type='application/json'
+        DEFAULT_MINIO_BUCKET,
+        f"{job.id}/{RESULT_FILENAME}",
+        data=buffer,
+        length=len(json_bytes),
+        content_type="application/json",
     )
 
     return

tests/test_client.py

@@ -23,11 +23,10 @@
 from oqd_core.interface.analog.operation import AnalogCircuit, AnalogGate
 from oqd_core.interface.analog.operator import PauliX, PauliZ
 
-from oqd_core.interface.atomic.circuit import AtomicCircuit
 
 from oqd_cloud.client import Client
 from oqd_cloud.provider import Provider
-from rich.pretty import pprint 
+from rich.pretty import pprint
 
 X = PauliX()
 Z = PauliZ()
@@ -67,20 +66,23 @@
 task.model_dump_json()
 
 
-#%%
+# %%
 client = Client()
 provider = Provider(port=8007)
 client.connect(provider=provider, username="ben", password="pwd")
 client.status_report
 
-#%%
+# %%
 backends = provider.available_backends
 print(backends)
 
 # %%
 print(client.jobs)
-job = client.submit_job(task=task, backend="oqd-analog-emulator-qutip", tags='a')
+job = client.submit_job(task=task, backend="oqd-analog-emulator-qutip", tags="a")
+pprint(job)
+
+# %%
+job = client.retrieve_job(job_id=job.job_id)
 pprint(job)
 
 # %%
-client.retrieve_job(job_id=job.job_id)