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 print(f" Attempt {attempt + 1} failed with error: {str(e)}")
32 last_exception = e
33 if attempt < num_retries - 1:
34 sleep_time = 2**attempt # 1, 2, 4, 8 seconds
35 time.sleep(sleep_time)
36 else:
37 raise last_exception
38 raise RuntimeError("Unexpected: retry loop completed without return or raise")
41def create_signed_upload_url(
42 url: str, size: int, user_id: str, memobin_api_key: str, num_retries: int = 4
43) -> str:
44 """Create a signed upload URL for memobin.
46 Args:
47 url: The target URL for the file
48 size: Size of the file in bytes
49 user_id: User ID for memobin
50 memobin_api_key: API key for memobin authentication
52 Returns:
53 The signed upload URL
55 Raises:
56 ValueError: If the URL prefix is invalid
57 requests.RequestException: If the API request fails
58 """
60 def _create_url() -> str:
61 prefix = "https://tempory.net/f/memobin/"
62 if not url.startswith(prefix):
63 raise ValueError("Invalid url. Does not have proper prefix")
65 file_path = url[len(prefix) :]
66 tempory_api_url = "https://hub.tempory.net/api/uploadFile"
68 response = requests.post(
69 tempory_api_url,
70 headers={
71 "Content-Type": "application/json",
72 "Authorization": f"Bearer {memobin_api_key}",
73 },
74 json={
75 "appName": "memobin",
76 "filePath": file_path,
77 "size": size,
78 "userId": user_id,
79 },
80 )
82 if not response.ok:
83 raise requests.RequestException("Failed to get signed url")
85 result = response.json()
86 upload_url = result["uploadUrl"]
87 download_url = result["downloadUrl"]
89 if download_url != url:
90 raise ValueError(f"Mismatch between download url and url: {download_url} != {url}")
92 return upload_url
94 return _retry_with_backoff(_create_url, num_retries)
97def construct_memobin_url(
98 alg_name: str,
99 dataset_name: str,
100 alg_version: str,
101 dataset_version: str,
102 system_version: str,
103 file_type: str = "metadata.json",
104) -> str:
105 """Construct the memobin URL for a specific benchmark result or dataset.
107 Args:
108 alg_name: Name of the algorithm
109 dataset_name: Name of the dataset
110 alg_version: Version of the algorithm
111 dataset_version: Version of the dataset
112 system_version: Version of the system
113 file_type: Type of file (metadata.json or data.bin)
115 Returns:
116 The constructed memobin URL
117 """
118 path = f"{alg_name}/{dataset_name}/{alg_version}/{dataset_version}/{system_version}/{file_type}"
119 return f"https://tempory.net/f/memobin/ephys_compression_tests/{path}"
122def construct_dataset_url(
123 dataset_name: str, dataset_version: str, format: str = "dat"
124) -> str:
125 """Construct the memobin URL for a dataset.
127 Args:
128 dataset_name: Name of the dataset
129 dataset_version: Version of the dataset
130 format: File format ("dat", "npy", or "json")
132 Returns:
133 The constructed memobin URL for the dataset
134 """
135 path = f"datasets/{dataset_name}/{dataset_version}/{dataset_name}-{dataset_version}.{format}"
136 return f"https://tempory.net/f/memobin/ephys_compression_tests/{path}"
139def construct_reconstructed_url(
140 algorithm_name: str,
141 dataset_name: str,
142 algorithm_version: str,
143 dataset_version: str,
144 system_version: str,
145 format: str = "dat",
146) -> str:
147 """Construct the memobin URL for a reconstructed array.
149 Args:
150 algorithm_name: Name of the algorithm
151 dataset_name: Name of the dataset
152 algorithm_version: Version of the algorithm
153 dataset_version: Version of the dataset
154 system_version: Version of the system
155 format: File format ("dat", "npy", or "json")
157 Returns:
158 The constructed memobin URL for the reconstructed array
159 """
160 version_str = f"v{algorithm_version}-{dataset_version}-{system_version}"
161 path = f"reconstructed/{algorithm_name}/{dataset_name}/{version_str}/reconstructed.{format}"
162 return f"https://tempory.net/f/memobin/ephys_compression_tests/{path}"
165def upload_to_memobin(
166 data: dict | bytes,
167 url: str,
168 memobin_api_key: str,
169 content_type: str = "application/json",
170 num_retries: int = 4,
171) -> None:
172 """Upload data to memobin.
174 Args:
175 data: The data to upload (dict for JSON or bytes for binary)
176 url: The target URL for the file
177 memobin_api_key: API key for memobin authentication
178 content_type: Content type of the data
180 Raises:
181 requests.RequestException: If the upload fails
182 """
183 if isinstance(data, dict):
184 data_bytes = json.dumps(data).encode("utf-8")
185 else:
186 data_bytes = data
187 size = len(data_bytes)
189 def _do_upload() -> None:
190 upload_url = create_signed_upload_url(
191 url, size, "ephys_compression_tests", memobin_api_key, num_retries
192 )
194 response = requests.put(
195 upload_url, data=data_bytes, headers={"Content-Type": content_type}
196 )
198 if not response.ok:
199 raise requests.RequestException("Failed to upload data to memobin")
201 _retry_with_backoff(_do_upload, num_retries)
204def exists_in_memobin(url: str, num_retries: int = 4) -> bool:
205 """Check if a file exists in memobin using a HEAD request.
207 Args:
208 url: The URL to check
210 Returns:
211 True if the file exists, False otherwise
212 """
214 def _check_exists() -> bool:
215 try:
216 response = requests.head(url)
217 return (
218 200 <= response.status_code < 300
219 ) # Any 2xx status code indicates success
220 except requests.RequestException:
221 return False
223 return _retry_with_backoff(_check_exists, num_retries)
226def download_from_memobin(
227 url: str, as_json: bool = True, num_retries: int = 4
228) -> Optional[dict | bytes]:
229 """Download data from memobin.
231 Args:
232 url: The URL to download from
233 as_json: Whether to parse the response as JSON
235 Returns:
236 The downloaded data as a dictionary or bytes, or None if not found
238 Raises:
239 requests.RequestException: If the download fails for a reason other than 404
240 """
242 def _do_download() -> Optional[dict | bytes]:
243 response = None
244 try:
245 response = requests.get(url)
246 if response.status_code == 404:
247 return None
248 response.raise_for_status()
249 return response.json() if as_json else response.content
250 except requests.RequestException as e:
251 if response and response.status_code == 404:
252 return None
253 raise e
255 return _retry_with_backoff(_do_download, num_retries)