1import json
2import requests
3import time
4from typing import Optional, TypeVar, Callable
6T = TypeVar("T")
9def _retry_with_backoff(
10 func: Callable[..., T], num_retries: int = 4, *args, **kwargs
11) -> T:
12 """Execute a function with exponential backoff retry logic.
14 Args:
15 func: Function to execute
16 num_retries: Maximum number of retries
17 args: Positional arguments for the function
18 kwargs: Keyword arguments for the function
20 Returns:
21 The function's return value
23 Raises:
24 The last exception encountered after all retries are exhausted
25 """
26 last_exception = None
27 for attempt in range(num_retries):
28 try:
29 return func(*args, **kwargs)
30 except Exception as e:
31 last_exception = e
32 if attempt < num_retries - 1:
33 sleep_time = 2**attempt # 1, 2, 4, 8 seconds
34 time.sleep(sleep_time)
35 else:
36 raise last_exception
37 raise RuntimeError("Unexpected: retry loop completed without return or raise")
40def create_signed_upload_url(
41 url: str, size: int, user_id: str, memobin_api_key: str, num_retries: int = 4
42) -> str:
43 """Create a signed upload URL for memobin.
45 Args:
46 url: The target URL for the file
47 size: Size of the file in bytes
48 user_id: User ID for memobin
49 memobin_api_key: API key for memobin authentication
51 Returns:
52 The signed upload URL
54 Raises:
55 ValueError: If the URL prefix is invalid
56 requests.RequestException: If the API request fails
57 """
59 def _create_url() -> str:
60 prefix = "https://tempory.net/f/memobin/"
61 if not url.startswith(prefix):
62 raise ValueError("Invalid url. Does not have proper prefix")
64 file_path = url[len(prefix) :]
65 tempory_api_url = "https://hub.tempory.net/api/uploadFile"
67 response = requests.post(
68 tempory_api_url,
69 headers={
70 "Content-Type": "application/json",
71 "Authorization": f"Bearer {memobin_api_key}",
72 },
73 json={
74 "appName": "memobin",
75 "filePath": file_path,
76 "size": size,
77 "userId": user_id,
78 },
79 )
81 if not response.ok:
82 raise requests.RequestException("Failed to get signed url")
84 result = response.json()
85 upload_url = result["uploadUrl"]
86 download_url = result["downloadUrl"]
88 if download_url != url:
89 raise ValueError("Mismatch between download url and url")
91 return upload_url
93 return _retry_with_backoff(_create_url, num_retries)
96def construct_memobin_url(
97 alg_name: str,
98 dataset_name: str,
99 alg_version: str,
100 dataset_version: str,
101 system_version: str,
102 file_type: str = "metadata.json",
103) -> str:
104 """Construct the memobin URL for a specific benchmark result or dataset.
106 Args:
107 alg_name: Name of the algorithm
108 dataset_name: Name of the dataset
109 alg_version: Version of the algorithm
110 dataset_version: Version of the dataset
111 system_version: Version of the system
112 file_type: Type of file (metadata.json or data.bin)
114 Returns:
115 The constructed memobin URL
116 """
117 path = f"{alg_name}/{dataset_name}/{alg_version}/{dataset_version}/{system_version}/{file_type}"
118 return f"https://tempory.net/f/memobin/{path}"
121def construct_dataset_url(
122 dataset_name: str, dataset_version: str, format: str = "dat"
123) -> str:
124 """Construct the memobin URL for a dataset.
126 Args:
127 dataset_name: Name of the dataset
128 dataset_version: Version of the dataset
129 format: File format ("dat", "npy", or "json")
131 Returns:
132 The constructed memobin URL for the dataset
133 """
134 path = f"datasets/{dataset_name}/{dataset_version}/{dataset_name}-{dataset_version}.{format}"
135 return f"https://tempory.net/f/memobin/{path}"
138def upload_to_memobin(
139 data: dict | bytes,
140 url: str,
141 memobin_api_key: str,
142 content_type: str = "application/json",
143 num_retries: int = 4,
144) -> None:
145 """Upload data to memobin.
147 Args:
148 data: The data to upload (dict for JSON or bytes for binary)
149 url: The target URL for the file
150 memobin_api_key: API key for memobin authentication
151 content_type: Content type of the data
153 Raises:
154 requests.RequestException: If the upload fails
155 """
156 if isinstance(data, dict):
157 data_bytes = json.dumps(data).encode("utf-8")
158 else:
159 data_bytes = data
160 size = len(data_bytes)
162 def _do_upload() -> None:
163 upload_url = create_signed_upload_url(
164 url, size, "ephys_compression_tests", memobin_api_key, num_retries
165 )
167 response = requests.put(
168 upload_url, data=data_bytes, headers={"Content-Type": content_type}
169 )
171 if not response.ok:
172 raise requests.RequestException("Failed to upload data to memobin")
174 _retry_with_backoff(_do_upload, num_retries)
177def exists_in_memobin(url: str, num_retries: int = 4) -> bool:
178 """Check if a file exists in memobin using a HEAD request.
180 Args:
181 url: The URL to check
183 Returns:
184 True if the file exists, False otherwise
185 """
187 def _check_exists() -> bool:
188 try:
189 response = requests.head(url)
190 return (
191 200 <= response.status_code < 300
192 ) # Any 2xx status code indicates success
193 except requests.RequestException:
194 return False
196 return _retry_with_backoff(_check_exists, num_retries)
199def download_from_memobin(
200 url: str, as_json: bool = True, num_retries: int = 4
201) -> Optional[dict | bytes]:
202 """Download data from memobin.
204 Args:
205 url: The URL to download from
206 as_json: Whether to parse the response as JSON
208 Returns:
209 The downloaded data as a dictionary or bytes, or None if not found
211 Raises:
212 requests.RequestException: If the download fails for a reason other than 404
213 """
215 def _do_download() -> Optional[dict | bytes]:
216 response = None
217 try:
218 response = requests.get(url)
219 if response.status_code == 404:
220 return None
221 response.raise_for_status()
222 return response.json() if as_json else response.content
223 except requests.RequestException as e:
224 if response and response.status_code == 404:
225 return None
226 raise e
228 return _retry_with_backoff(_do_download, num_retries)