Sanjgupt/onlineproblem

azure-quantum/azure/quantum/optimization/__init__.py

@@ -8,4 +8,4 @@
 from .problem import *
 from .solvers import *
 from .streaming_problem import *
-
+from .online_problem import *

azure-quantum/azure/quantum/optimization/online_problem.py

@@ -0,0 +1,33 @@
+import logging
+from typing import TYPE_CHECKING, Union, Dict 
+from azure.quantum.optimization import Problem
+
+logger = logging.getLogger(__name__)
+
+__all__ = ['OnlineProblem']
+
+if TYPE_CHECKING:
+    from azure.quantum.workspace import Workspace
+
+class OnlineProblem(Problem):
+    def __init__(self, name:str, blob_uri:str):
+        super().__init__(self)
+        self.name = name
+        self.uploaded_blob_uri = blob_uri
+
+    def evaluate(self, configuration: Union[Dict[int, int], Dict[str, int]]) -> float:
+        """
+        An OnlineProblem cannot be evaluated on client side. Calling this function will raise a user exception
+        """
+        raise Exception('An Online Problem cannot be evaluated. Please download the problem to do this operation')
+
+    def set_fixed_variables(self, fixed_variables: Union[Dict[int, int], Dict[str, int]]) -> Problem:
+        """
+         An OnlineProblem cannot be evaluated on client side. Calling this function will raise a user exception
+        """
+        raise Exception('An Online Problem cannot set fixed terms. Please download the problem to do this operation')
+
+    def download(self)-> Problem:
+        logger.warning('The problem will be downloaded to the client')
+        return super().download()
+

azure-quantum/azure/quantum/optimization/problem.py

@@ -14,8 +14,8 @@
 from typing import List, Union, Dict, Optional, TYPE_CHECKING
 from enum import Enum
 from azure.quantum.optimization import Term
-from azure.quantum.storage import upload_blob, ContainerClient
-
+from azure.quantum.storage import upload_blob, ContainerClient, BlobClient, download_blob
+#import azure.quantum.storage
 
 logger = logging.getLogger(__name__)
 
@@ -230,3 +230,12 @@ def is_large(self) -> bool:
             set_vars.update(term.ids)
 
         return (len(set_vars) >= Problem.NUM_VARIABLES_LARGE and len(self.terms) >= Problem.NUM_TERMS_LARGE)
+
+    def download(self):
+        """Dowloads an uploaded problem as an instance of 'Problem'
+        """
+        if not self.uploaded_blob_uri: 
+            raise Exception("Problem must be uploaded before it can be downloaded")
+        contents = download_blob(self.uploaded_blob_uri)
+        return Problem.deserialize(contents, self.name)
+

azure-quantum/azure/quantum/optimization/solvers.py

@@ -101,7 +101,10 @@ def submit(self, problem: Union[str, Problem], compress: bool = True) -> Job:
 
         if isinstance(problem, str):
             name = "Optimization problem"
-            problem_uri = problem  
+            problem_uri = problem 
+        elif not problem.terms:
+            name = problem.name
+            problem_uri = problem.uploaded_blob_uri
         else:
             name = problem.name
             problem_uri = problem.upload(self.workspace, compress=compress, container_name=container_name, blob_name="inputData")

azure-quantum/tests/unit/quantum_test_base.py → azure-quantum/tests/unit/common.py

@@ -18,7 +18,7 @@
 from azure_devtools.scenario_tests.base import ReplayableTest
 from azure_devtools.scenario_tests.recording_processors import RecordingProcessor, is_text_payload, AccessTokenReplacer
 from azure_devtools.scenario_tests.utilities import _get_content_type
-
+import json
 class QuantumTestBase(ReplayableTest):
     """QuantumTestBase
 
@@ -196,3 +196,16 @@ def process_response(self, response):
 
         return response
 
+#helper functions
+
+# helpers
+def expected_terms():
+    expected = json.dumps({
+        "cost_function": {
+        "version": "1.1",
+        "type":  "ising",
+        "terms": [ { 'c':3, 'ids':[1,0] }, {'c':5, 'ids':[2,0]} ],
+        "initial_configuration": {"0":-1 , "1": 1, "2": -1},
+        }
+    })
+    return expected

azure-quantum/tests/unit/test_job.py

@@ -15,7 +15,7 @@
 from azure.quantum.optimization import Problem
 from azure.quantum.optimization.solvers import SimulatedAnnealing
 from azure.quantum import Job
-from quantum_test_base import QuantumTestBase
+from common import QuantumTestBase
 
 
 class TestJob(QuantumTestBase):

azure-quantum/tests/unit/test_online_problem.py

@@ -0,0 +1,39 @@
+import unittest
+from unittest.mock import Mock
+from azure.quantum.optimization import Problem, Term, OnlineProblem
+import azure.quantum.optimization.problem 
+from common import expected_terms
+import json
+
+
+class TestOnlineProblemClass(unittest.TestCase):
+
+    def setUp(self):
+        self.o_problem = OnlineProblem(name = "test", blob_uri = "mock_blob_uri")
+
+
+    def test_download(self):
+        azure.quantum.optimization.problem.download_blob = Mock(return_value = expected_terms())
+        acutal_result = self.o_problem.download()
+        #to-do add test that user warning was registered in log
+        assert acutal_result.name == 'test'
+        azure.quantum.optimization.problem.download_blob.assert_called_once()
+        assert isinstance(acutal_result, Problem)
+
+    def test_evaluate(self):
+        config_dict = {
+            1:1,
+            0:1
+        }
+        with self.assertRaises(Exception):
+            self.o_problem.evaluate(config_dict)
+
+    def test_set_fixed_variables(self):
+        config_dict = {
+            1:1,
+            0:1
+        }
+        with self.assertRaises(Exception):
+            self.o_problem.set_fixed_variables(config_dict)
+
+

azure-quantum/tests/unit/test_optimization.py

@@ -11,11 +11,11 @@
 ## IMPORTS ##
 
 import json
-
+import unittest
 from azure.quantum import Workspace
 from azure.quantum.optimization import Problem, ProblemType, Term
 from azure.quantum.optimization.solvers import ParallelTempering, PopulationAnnealing, RangeSchedule, SimulatedAnnealing, HardwarePlatform, QuantumMonteCarlo, SubstochasticMonteCarlo
-from quantum_test_base import QuantumTestBase 
+from common import QuantumTestBase 
 
 class TestProblem(QuantumTestBase):
     def test_add_terms(self):

azure-quantum/tests/unit/test_problem.py

@@ -0,0 +1,26 @@
+import unittest
+from unittest.mock import Mock
+from azure.quantum.optimization import Problem, Term
+import azure.quantum.optimization.problem 
+from common import expected_terms
+import json
+
+
+class TestProblemClass(unittest.TestCase):
+
+    def setUp(self):
+        self.problem = Problem(name = "test")
+        self.problem.terms = [
+            Term(c = 3, indices=[1,0]),
+            Term(c = 5, indices=[2,0])
+        ]
+        self.problem.uploaded_blob_uri = "mock_blob_uri"
+
+    def test_download(self):
+        azure.quantum.optimization.problem.download_blob = Mock(return_value = expected_terms())
+        acutal_result = self.problem.download()
+        assert acutal_result.name == 'test'
+        azure.quantum.optimization.problem.download_blob.assert_called_once()
+
+
+

azure-quantum/tests/unit/test_solvers.py

@@ -0,0 +1,21 @@
+import unittest
+from unittest.mock import Mock, patch
+from azure.quantum import Workspace, storage
+from azure.quantum.optimization import Solver, OnlineProblem
+import azure.quantum.storage
+
+class TestSolvers(unittest.TestCase):
+    def setUp(self):
+        self.mock_ws = Mock(spec = Workspace)
+        self.mock_ws.storage = "mock_storage"
+        #self.mock_ws.submit_job = Mock(return_value = )
+        self.testsolver = Solver(self.mock_ws, 'Microsoft', 'SimulatedAnnealing', 'json', 'json')
+
+    def test_submit_online_problem(self):
+        #Arrange
+        o_problem = OnlineProblem(name  = "test", blob_uri = "mock_blob_uri")
+        #Act
+        azure.quantum.storage.get_container_uri = Mock(return_value = "mock_container_uri")
+        _ = self.testsolver.submit(o_problem)
+        #Assert
+        self.testsolver.workspace.submit_job.assert_called_once()

azure-quantum/tests/unit/test_workspace.py

@@ -13,7 +13,7 @@
 import pytest
 
 from azure.quantum import Workspace
-from quantum_test_base import QuantumTestBase 
+from common import QuantumTestBase 
 
 class TestWorkspace(QuantumTestBase):
     def test_workspace_login(self):