Add feature for querying dedicated deployment usage (per workspace)

roboflow/adapters/deploymentapi.py

@@ -41,6 +41,22 @@ def list_deployment(api_key):
     return response.status_code, response.json()
 
 
+def get_workspace_usage(api_key, from_timestamp, to_timestamp):
+    url = f"{DEDICATED_DEPLOYMENT_URL}/usage_workspace?api_key={api_key}&from_timestamp={from_timestamp.isoformat()}&to_timestamp={to_timestamp.isoformat()}"
+    response = requests.get(url)
+    if response.status_code != 200:
+        return response.status_code, response.text
+    return response.status_code, response.json()
+
+
+def get_deployment_usage(api_key, deployment_name, from_timestamp, to_timestamp):
+    url = f"{DEDICATED_DEPLOYMENT_URL}/usage_deployment?api_key={api_key}&deployment_name={deployment_name}&from_timestamp={from_timestamp.isoformat()}&to_timestamp={to_timestamp.isoformat()}"
+    response = requests.get(url)
+    if response.status_code != 200:
+        return response.status_code, response.text
+    return response.status_code, response.json()
+
+
 def delete_deployment(api_key, deployment_name):
     url = f"{DEDICATED_DEPLOYMENT_URL}/delete"
     response = requests.post(url, json={"api_key": api_key, "deployment_name": deployment_name})

roboflow/deployment.py

@@ -6,6 +6,43 @@
 from roboflow.config import load_roboflow_api_key
 
 
+def is_valid_ISO8601_timestamp(ts):
+    try:
+        datetime.fromisoformat(ts)
+        return True
+    except:
+        return False
+
+
+def check_from_to_timestamp(from_timestamp, to_timestamp, default_timedelta):
+    if from_timestamp and not is_valid_ISO8601_timestamp(from_timestamp):
+        print("Please provide a valid from_timestamp in ISO8601 format")
+        exit(1)
+
+    if to_timestamp and not is_valid_ISO8601_timestamp(to_timestamp):
+        print("Please provide a valid to_timestamp in ISO8601 format")
+        exit(1)
+
+    time_now = datetime.now().replace(tzinfo=None)
+    if from_timestamp is None and to_timestamp is None:
+        from_timestamp = time_now - default_timedelta
+        to_timestamp = time_now
+    elif from_timestamp is not None and to_timestamp is None:
+        from_timestamp = datetime.fromisoformat(from_timestamp).replace(tzinfo=None)
+        to_timestamp = from_timestamp + default_timedelta
+    elif from_timestamp is None and to_timestamp is not None:
+        to_timestamp = datetime.fromisoformat(to_timestamp).replace(tzinfo=None)
+        from_timestamp = to_timestamp - default_timedelta
+    else:
+        from_timestamp = datetime.fromisoformat(from_timestamp).replace(tzinfo=None)
+        to_timestamp = datetime.fromisoformat(to_timestamp).replace(tzinfo=None)
+        if from_timestamp >= to_timestamp:
+            print("from_timestamp should be earlier than to_timestamp")
+            exit(1)
+
+    return from_timestamp, to_timestamp
+
+
 def add_deployment_parser(subparsers):
     deployment_parser = subparsers.add_parser(
         "deployment",
@@ -18,6 +55,12 @@ def add_deployment_parser(subparsers):
         "get", help="show detailed info for a dedicated deployment"
     )
     deployment_list_parser = deployment_subparsers.add_parser("list", help="list dedicated deployments in a workspace")
+    deployment_usage_workspace_parser = deployment_subparsers.add_parser(
+        "usage_workspace", help="get all dedicated deployments usage in a workspace"
+    )
+    deployment_usage_deployment_parser = deployment_subparsers.add_parser(
+        "usage_deployment", help="get usage of a specific dedicated deployments"
+    )
     deployment_delete_parser = deployment_subparsers.add_parser("delete", help="delete a dedicated deployment")
     deployment_log_parser = deployment_subparsers.add_parser("log", help="show log info for a dedicated deployment")
 
@@ -66,6 +109,25 @@ def add_deployment_parser(subparsers):
     deployment_list_parser.set_defaults(func=list_deployment)
     deployment_list_parser.add_argument("-a", "--api_key", help="api key")
 
+    deployment_usage_workspace_parser.set_defaults(func=get_workspace_usage)
+    deployment_usage_workspace_parser.add_argument("-a", "--api_key", help="api key")
+    deployment_usage_workspace_parser.add_argument(
+        "-f", "--from_timestamp", help="begin time stamp in ISO8601 format (YYYY-MM-DD HH:MM:SS)", default=None
+    )
+    deployment_usage_workspace_parser.add_argument(
+        "-t", "--to_timestamp", help="end time stamp in ISO8601 format (YYYY-MM-DD HH:MM:SS)", default=None
+    )
+
+    deployment_usage_deployment_parser.set_defaults(func=get_deployment_usage)
+    deployment_usage_deployment_parser.add_argument("-a", "--api_key", help="api key")
+    deployment_usage_deployment_parser.add_argument("deployment_name", help="deployment name")
+    deployment_usage_deployment_parser.add_argument(
+        "-f", "--from_timestamp", help="begin time stamp in ISO8601 format (YYYY-MM-DD HH:MM:SS)", default=None
+    )
+    deployment_usage_deployment_parser.add_argument(
+        "-t", "--to_timestamp", help="end time stamp in ISO8601 format (YYYY-MM-DD HH:MM:SS)", default=None
+    )
+
     deployment_delete_parser.set_defaults(func=delete_deployment)
     deployment_delete_parser.add_argument("-a", "--api_key", help="api key")
     deployment_delete_parser.add_argument("deployment_name", help="deployment name")
@@ -151,6 +213,34 @@ def list_deployment(args):
     print(json.dumps(msg, indent=2))
 
 
+def get_workspace_usage(args):
+    api_key = args.api_key or load_roboflow_api_key(None)
+    if api_key is None:
+        print("Please provide an api key")
+        exit(1)
+
+    from_timestamp, to_timestamp = check_from_to_timestamp(args.from_timestamp, args.to_timestamp, timedelta(days=1))
+    status_code, msg = deploymentapi.get_workspace_usage(api_key, from_timestamp, to_timestamp)
+    if status_code != 200:
+        print(f"{status_code}: {msg}")
+        exit(status_code)
+    print(json.dumps(msg, indent=2))
+
+
+def get_deployment_usage(args):
+    api_key = args.api_key or load_roboflow_api_key(None)
+    if api_key is None:
+        print("Please provide an api key")
+        exit(1)
+
+    from_timestamp, to_timestamp = check_from_to_timestamp(args.from_timestamp, args.to_timestamp, timedelta(days=1))
+    status_code, msg = deploymentapi.get_deployment_usage(api_key, args.deployment_name, from_timestamp, to_timestamp)
+    if status_code != 200:
+        print(f"{status_code}: {msg}")
+        exit(status_code)
+    print(json.dumps(msg, indent=2))
+
+
 def delete_deployment(args):
     api_key = args.api_key or load_roboflow_api_key(None)
     if api_key is None: