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 upload_to_memobin(
140 data: dict | bytes,
141 url: str,
142 memobin_api_key: str,
143 content_type: str = "application/json",
144 num_retries: int = 4,
145) -> None:
146 """Upload data to memobin.
148 Args:
149 data: The data to upload (dict for JSON or bytes for binary)
150 url: The target URL for the file
151 memobin_api_key: API key for memobin authentication
152 content_type: Content type of the data
154 Raises:
155 requests.RequestException: If the upload fails
156 """
157 if isinstance(data, dict):
158 data_bytes = json.dumps(data).encode("utf-8")
159 else:
160 data_bytes = data
161 size = len(data_bytes)
163 def _do_upload() -> None:
164 upload_url = create_signed_upload_url(
165 url, size, "ephys_compression_tests", memobin_api_key, num_retries
166 )
168 response = requests.put(
169 upload_url, data=data_bytes, headers={"Content-Type": content_type}
170 )
172 if not response.ok:
173 raise requests.RequestException("Failed to upload data to memobin")
175 _retry_with_backoff(_do_upload, num_retries)
178def exists_in_memobin(url: str, num_retries: int = 4) -> bool:
179 """Check if a file exists in memobin using a HEAD request.
181 Args:
182 url: The URL to check
184 Returns:
185 True if the file exists, False otherwise
186 """
188 def _check_exists() -> bool:
189 try:
190 response = requests.head(url)
191 return (
192 200 <= response.status_code < 300
193 ) # Any 2xx status code indicates success
194 except requests.RequestException:
195 return False
197 return _retry_with_backoff(_check_exists, num_retries)
200def download_from_memobin(
201 url: str, as_json: bool = True, num_retries: int = 4
202) -> Optional[dict | bytes]:
203 """Download data from memobin.
205 Args:
206 url: The URL to download from
207 as_json: Whether to parse the response as JSON
209 Returns:
210 The downloaded data as a dictionary or bytes, or None if not found
212 Raises:
213 requests.RequestException: If the download fails for a reason other than 404
214 """
216 def _do_download() -> Optional[dict | bytes]:
217 response = None
218 try:
219 response = requests.get(url)
220 if response.status_code == 404:
221 return None
222 response.raise_for_status()
223 return response.json() if as_json else response.content
224 except requests.RequestException as e:
225 if response and response.status_code == 404:
226 return None
227 raise e
229 return _retry_with_backoff(_do_download, num_retries)