Skip to content

File Systems API

Pluggable storage backends for simulation I/O.

FileSystem Protocol

idfkit.simulation.fs.FileSystem

Bases: Protocol

Protocol for file system operations used by the simulation module.

All methods accept str | Path for path arguments.

Source code in src/idfkit/simulation/fs.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@runtime_checkable
class FileSystem(Protocol):
    """Protocol for file system operations used by the simulation module.

    All methods accept ``str | Path`` for path arguments.
    """

    def read_bytes(self, path: str | Path) -> bytes:
        """Read a file as raw bytes.

        Args:
            path: Path to the file.

        Returns:
            The file contents as bytes.
        """
        ...

    def write_bytes(self, path: str | Path, data: bytes) -> None:
        """Write raw bytes to a file.

        Args:
            path: Path to the file.
            data: Bytes to write.
        """
        ...

    def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
        """Read a file as text.

        Args:
            path: Path to the file.
            encoding: Text encoding (default ``"utf-8"``).

        Returns:
            The file contents as a string.
        """
        ...

    def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
        """Write text to a file.

        Args:
            path: Path to the file.
            text: Text to write.
            encoding: Text encoding (default ``"utf-8"``).
        """
        ...

    def exists(self, path: str | Path) -> bool:
        """Check whether a file exists.

        Args:
            path: Path to check.

        Returns:
            True if the file exists.
        """
        ...

    def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
        """Create directories recursively.

        Args:
            path: Directory path to create.
            exist_ok: If True, do not raise if the directory already exists.
        """
        ...

    def copy(self, src: str | Path, dst: str | Path) -> None:
        """Copy a file from *src* to *dst*.

        Args:
            src: Source file path.
            dst: Destination file path.
        """
        ...

    def glob(self, path: str | Path, pattern: str) -> list[str]:
        """List files matching a glob pattern under *path*.

        Args:
            path: Base directory.
            pattern: Glob pattern (e.g. ``"*.sql"``).

        Returns:
            List of matching file paths as strings.
        """
        ...

    def remove(self, path: str | Path) -> None:
        """Remove a file.

        Args:
            path: Path to the file to remove.
        """
        ...

read_bytes(path)

Read a file as raw bytes.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required

Returns:

Type Description
bytes

The file contents as bytes.
Source code in src/idfkit/simulation/fs.py 
71
72
73
74
75
76
77
78
79
80
def read_bytes(self, path: str | Path) -> bytes:
    """Read a file as raw bytes.

    Args:
        path: Path to the file.

    Returns:
        The file contents as bytes.
    """
    ...

write_bytes(path, data)

Write raw bytes to a file.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required
data bytes

Bytes to write.

 required
Source code in src/idfkit/simulation/fs.py 
82
83
84
85
86
87
88
89
def write_bytes(self, path: str | Path, data: bytes) -> None:
    """Write raw bytes to a file.

    Args:
        path: Path to the file.
        data: Bytes to write.
    """
    ...

read_text(path, encoding='utf-8')

Read a file as text.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required
encoding str

Text encoding (default "utf-8").

 'utf-8'

Returns:

Type Description
str

The file contents as a string.
Source code in src/idfkit/simulation/fs.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
    """Read a file as text.

    Args:
        path: Path to the file.
        encoding: Text encoding (default ``"utf-8"``).

    Returns:
        The file contents as a string.
    """
    ...

write_text(path, text, encoding='utf-8')

Write text to a file.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required
text str

Text to write.

 required
encoding str

Text encoding (default "utf-8").

 'utf-8'
Source code in src/idfkit/simulation/fs.py 
103
104
105
106
107
108
109
110
111
def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
    """Write text to a file.

    Args:
        path: Path to the file.
        text: Text to write.
        encoding: Text encoding (default ``"utf-8"``).
    """
    ...

exists(path)

Check whether a file exists.

Parameters:

Name Type Description Default
path str | Path

Path to check.

 required

Returns:

Type Description
bool

True if the file exists.
Source code in src/idfkit/simulation/fs.py 
113
114
115
116
117
118
119
120
121
122
def exists(self, path: str | Path) -> bool:
    """Check whether a file exists.

    Args:
        path: Path to check.

    Returns:
        True if the file exists.
    """
    ...

makedirs(path, *, exist_ok=False)

Create directories recursively.

Parameters:

Name Type Description Default
path str | Path

Directory path to create.

 required
exist_ok bool

If True, do not raise if the directory already exists.

 False
Source code in src/idfkit/simulation/fs.py 
124
125
126
127
128
129
130
131
def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
    """Create directories recursively.

    Args:
        path: Directory path to create.
        exist_ok: If True, do not raise if the directory already exists.
    """
    ...

copy(src, dst)

Copy a file from src to dst.

Parameters:

Name Type Description Default
src str | Path

Source file path.

 required
dst str | Path

Destination file path.

 required
Source code in src/idfkit/simulation/fs.py 
133
134
135
136
137
138
139
140
def copy(self, src: str | Path, dst: str | Path) -> None:
    """Copy a file from *src* to *dst*.

    Args:
        src: Source file path.
        dst: Destination file path.
    """
    ...

glob(path, pattern)

List files matching a glob pattern under path.

Parameters:

Name Type Description Default
path str | Path

Base directory.

 required
pattern str

Glob pattern (e.g. "*.sql").

 required

Returns:

Type Description
list[str]

List of matching file paths as strings.
Source code in src/idfkit/simulation/fs.py 
142
143
144
145
146
147
148
149
150
151
152
def glob(self, path: str | Path, pattern: str) -> list[str]:
    """List files matching a glob pattern under *path*.

    Args:
        path: Base directory.
        pattern: Glob pattern (e.g. ``"*.sql"``).

    Returns:
        List of matching file paths as strings.
    """
    ...

remove(path)

Remove a file.

Parameters:

Name Type Description Default
path str | Path

Path to the file to remove.

 required
Source code in src/idfkit/simulation/fs.py 
154
155
156
157
158
159
160
def remove(self, path: str | Path) -> None:
    """Remove a file.

    Args:
        path: Path to the file to remove.
    """
    ...

AsyncFileSystem Protocol

idfkit.simulation.fs.AsyncFileSystem

Bases: Protocol

Protocol for async file system operations used by the async simulation module.

This is the async counterpart to FileSystem. Use this with async_simulate and the async batch functions to avoid blocking the event loop during file I/O — especially important for network-backed storage like S3.

All methods accept str | Path for path arguments.

Source code in src/idfkit/simulation/fs.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
@runtime_checkable
class AsyncFileSystem(Protocol):
    """Protocol for async file system operations used by the async simulation module.

    This is the async counterpart to [FileSystem][idfkit.simulation.fs.FileSystem].  Use this with
    [async_simulate][idfkit.simulation.async_runner.async_simulate] and the async batch
    functions to avoid blocking the event loop during file I/O — especially
    important for network-backed storage like S3.

    All methods accept ``str | Path`` for path arguments.
    """

    async def read_bytes(self, path: str | Path) -> bytes:
        """Read a file as raw bytes.

        Args:
            path: Path to the file.

        Returns:
            The file contents as bytes.
        """
        ...

    async def write_bytes(self, path: str | Path, data: bytes) -> None:
        """Write raw bytes to a file.

        Args:
            path: Path to the file.
            data: Bytes to write.
        """
        ...

    async def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
        """Read a file as text.

        Args:
            path: Path to the file.
            encoding: Text encoding (default ``"utf-8"``).

        Returns:
            The file contents as a string.
        """
        ...

    async def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
        """Write text to a file.

        Args:
            path: Path to the file.
            text: Text to write.
            encoding: Text encoding (default ``"utf-8"``).
        """
        ...

    async def exists(self, path: str | Path) -> bool:
        """Check whether a file exists.

        Args:
            path: Path to check.

        Returns:
            True if the file exists.
        """
        ...

    async def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
        """Create directories recursively.

        Args:
            path: Directory path to create.
            exist_ok: If True, do not raise if the directory already exists.
        """
        ...

    async def copy(self, src: str | Path, dst: str | Path) -> None:
        """Copy a file from *src* to *dst*.

        Args:
            src: Source file path.
            dst: Destination file path.
        """
        ...

    async def glob(self, path: str | Path, pattern: str) -> list[str]:
        """List files matching a glob pattern under *path*.

        Args:
            path: Base directory.
            pattern: Glob pattern (e.g. ``"*.sql"``).

        Returns:
            List of matching file paths as strings.
        """
        ...

    async def remove(self, path: str | Path) -> None:
        """Remove a file.

        Args:
            path: Path to the file to remove.
        """
        ...

read_bytes(path) async

Read a file as raw bytes.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required

Returns:

Type Description
bytes

The file contents as bytes.
Source code in src/idfkit/simulation/fs.py 
215
216
217
218
219
220
221
222
223
224
async def read_bytes(self, path: str | Path) -> bytes:
    """Read a file as raw bytes.

    Args:
        path: Path to the file.

    Returns:
        The file contents as bytes.
    """
    ...

write_bytes(path, data) async

Write raw bytes to a file.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required
data bytes

Bytes to write.

 required
Source code in src/idfkit/simulation/fs.py 
226
227
228
229
230
231
232
233
async def write_bytes(self, path: str | Path, data: bytes) -> None:
    """Write raw bytes to a file.

    Args:
        path: Path to the file.
        data: Bytes to write.
    """
    ...

read_text(path, encoding='utf-8') async

Read a file as text.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required
encoding str

Text encoding (default "utf-8").

 'utf-8'

Returns:

Type Description
str

The file contents as a string.
Source code in src/idfkit/simulation/fs.py 
235
236
237
238
239
240
241
242
243
244
245
async def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
    """Read a file as text.

    Args:
        path: Path to the file.
        encoding: Text encoding (default ``"utf-8"``).

    Returns:
        The file contents as a string.
    """
    ...

write_text(path, text, encoding='utf-8') async

Write text to a file.

Parameters:

Name Type Description Default
path str | Path

Path to the file.

 required
text str

Text to write.

 required
encoding str

Text encoding (default "utf-8").

 'utf-8'
Source code in src/idfkit/simulation/fs.py 
247
248
249
250
251
252
253
254
255
async def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
    """Write text to a file.

    Args:
        path: Path to the file.
        text: Text to write.
        encoding: Text encoding (default ``"utf-8"``).
    """
    ...

exists(path) async

Check whether a file exists.

Parameters:

Name Type Description Default
path str | Path

Path to check.

 required

Returns:

Type Description
bool

True if the file exists.
Source code in src/idfkit/simulation/fs.py 
257
258
259
260
261
262
263
264
265
266
async def exists(self, path: str | Path) -> bool:
    """Check whether a file exists.

    Args:
        path: Path to check.

    Returns:
        True if the file exists.
    """
    ...

makedirs(path, *, exist_ok=False) async

Create directories recursively.

Parameters:

Name Type Description Default
path str | Path

Directory path to create.

 required
exist_ok bool

If True, do not raise if the directory already exists.

 False
Source code in src/idfkit/simulation/fs.py 
268
269
270
271
272
273
274
275
async def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
    """Create directories recursively.

    Args:
        path: Directory path to create.
        exist_ok: If True, do not raise if the directory already exists.
    """
    ...

copy(src, dst) async

Copy a file from src to dst.

Parameters:

Name Type Description Default
src str | Path

Source file path.

 required
dst str | Path

Destination file path.

 required
Source code in src/idfkit/simulation/fs.py 
277
278
279
280
281
282
283
284
async def copy(self, src: str | Path, dst: str | Path) -> None:
    """Copy a file from *src* to *dst*.

    Args:
        src: Source file path.
        dst: Destination file path.
    """
    ...

glob(path, pattern) async

List files matching a glob pattern under path.

Parameters:

Name Type Description Default
path str | Path

Base directory.

 required
pattern str

Glob pattern (e.g. "*.sql").

 required

Returns:

Type Description
list[str]

List of matching file paths as strings.
Source code in src/idfkit/simulation/fs.py 
286
287
288
289
290
291
292
293
294
295
296
async def glob(self, path: str | Path, pattern: str) -> list[str]:
    """List files matching a glob pattern under *path*.

    Args:
        path: Base directory.
        pattern: Glob pattern (e.g. ``"*.sql"``).

    Returns:
        List of matching file paths as strings.
    """
    ...

remove(path) async

Remove a file.

Parameters:

Name Type Description Default
path str | Path

Path to the file to remove.

 required
Source code in src/idfkit/simulation/fs.py 
298
299
300
301
302
303
304
async def remove(self, path: str | Path) -> None:
    """Remove a file.

    Args:
        path: Path to the file to remove.
    """
    ...

LocalFileSystem

idfkit.simulation.fs.LocalFileSystem

File system implementation backed by pathlib and shutil.

Source code in src/idfkit/simulation/fs.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class LocalFileSystem:
    """File system implementation backed by [pathlib][] and [shutil][]."""

    def read_bytes(self, path: str | Path) -> bytes:
        """Read a file as raw bytes."""
        return Path(path).read_bytes()

    def write_bytes(self, path: str | Path, data: bytes) -> None:
        """Write raw bytes to a file."""
        Path(path).write_bytes(data)

    def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
        """Read a file as text."""
        return Path(path).read_text(encoding=encoding)

    def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
        """Write text to a file."""
        Path(path).write_text(text, encoding=encoding)

    def exists(self, path: str | Path) -> bool:
        """Check whether a file exists."""
        return Path(path).exists()

    def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
        """Create directories recursively."""
        Path(path).mkdir(parents=True, exist_ok=exist_ok)

    def copy(self, src: str | Path, dst: str | Path) -> None:
        """Copy a file from *src* to *dst*."""
        shutil.copy2(str(src), str(dst))

    def glob(self, path: str | Path, pattern: str) -> list[str]:
        """List files matching a glob pattern under *path*."""
        return [str(p) for p in Path(path).glob(pattern)]

    def remove(self, path: str | Path) -> None:
        """Remove a file."""
        Path(path).unlink()

read_bytes(path)

Read a file as raw bytes.

Source code in src/idfkit/simulation/fs.py 
166
167
168
def read_bytes(self, path: str | Path) -> bytes:
    """Read a file as raw bytes."""
    return Path(path).read_bytes()

write_bytes(path, data)

Write raw bytes to a file.

Source code in src/idfkit/simulation/fs.py 
170
171
172
def write_bytes(self, path: str | Path, data: bytes) -> None:
    """Write raw bytes to a file."""
    Path(path).write_bytes(data)

read_text(path, encoding='utf-8')

Read a file as text.

Source code in src/idfkit/simulation/fs.py 
174
175
176
def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
    """Read a file as text."""
    return Path(path).read_text(encoding=encoding)

write_text(path, text, encoding='utf-8')

Write text to a file.

Source code in src/idfkit/simulation/fs.py 
178
179
180
def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
    """Write text to a file."""
    Path(path).write_text(text, encoding=encoding)

exists(path)

Check whether a file exists.

Source code in src/idfkit/simulation/fs.py 
182
183
184
def exists(self, path: str | Path) -> bool:
    """Check whether a file exists."""
    return Path(path).exists()

makedirs(path, *, exist_ok=False)

Create directories recursively.

Source code in src/idfkit/simulation/fs.py 
186
187
188
def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
    """Create directories recursively."""
    Path(path).mkdir(parents=True, exist_ok=exist_ok)

copy(src, dst)

Copy a file from src to dst.

Source code in src/idfkit/simulation/fs.py 
190
191
192
def copy(self, src: str | Path, dst: str | Path) -> None:
    """Copy a file from *src* to *dst*."""
    shutil.copy2(str(src), str(dst))

glob(path, pattern)

List files matching a glob pattern under path.

Source code in src/idfkit/simulation/fs.py 
194
195
196
def glob(self, path: str | Path, pattern: str) -> list[str]:
    """List files matching a glob pattern under *path*."""
    return [str(p) for p in Path(path).glob(pattern)]

remove(path)

Remove a file.

Source code in src/idfkit/simulation/fs.py 
198
199
200
def remove(self, path: str | Path) -> None:
    """Remove a file."""
    Path(path).unlink()

AsyncLocalFileSystem

idfkit.simulation.fs.AsyncLocalFileSystem

Non-blocking local file system using asyncio.to_thread.

Wraps LocalFileSystem so that each blocking I/O call runs in the default executor, keeping the event loop free.

Source code in src/idfkit/simulation/fs.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
class AsyncLocalFileSystem:
    """Non-blocking local file system using [asyncio.to_thread][].

    Wraps [LocalFileSystem][idfkit.simulation.fs.LocalFileSystem] so that each blocking I/O call runs in
    the default executor, keeping the event loop free.
    """

    def __init__(self) -> None:
        self._sync = LocalFileSystem()

    async def read_bytes(self, path: str | Path) -> bytes:
        """Read a file as raw bytes without blocking the event loop."""
        return await asyncio.to_thread(self._sync.read_bytes, path)

    async def write_bytes(self, path: str | Path, data: bytes) -> None:
        """Write raw bytes to a file without blocking the event loop."""
        await asyncio.to_thread(self._sync.write_bytes, path, data)

    async def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
        """Read a file as text without blocking the event loop."""
        return await asyncio.to_thread(self._sync.read_text, path, encoding)

    async def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
        """Write text to a file without blocking the event loop."""
        await asyncio.to_thread(self._sync.write_text, path, text, encoding)

    async def exists(self, path: str | Path) -> bool:
        """Check whether a file exists without blocking the event loop."""
        return await asyncio.to_thread(self._sync.exists, path)

    async def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
        """Create directories recursively without blocking the event loop."""
        await asyncio.to_thread(self._sync.makedirs, path, exist_ok=exist_ok)

    async def copy(self, src: str | Path, dst: str | Path) -> None:
        """Copy a file without blocking the event loop."""
        await asyncio.to_thread(self._sync.copy, src, dst)

    async def glob(self, path: str | Path, pattern: str) -> list[str]:
        """List files matching a glob pattern without blocking the event loop."""
        return await asyncio.to_thread(self._sync.glob, path, pattern)

    async def remove(self, path: str | Path) -> None:
        """Remove a file without blocking the event loop."""
        await asyncio.to_thread(self._sync.remove, path)

read_bytes(path) async

Read a file as raw bytes without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
317
318
319
async def read_bytes(self, path: str | Path) -> bytes:
    """Read a file as raw bytes without blocking the event loop."""
    return await asyncio.to_thread(self._sync.read_bytes, path)

write_bytes(path, data) async

Write raw bytes to a file without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
321
322
323
async def write_bytes(self, path: str | Path, data: bytes) -> None:
    """Write raw bytes to a file without blocking the event loop."""
    await asyncio.to_thread(self._sync.write_bytes, path, data)

read_text(path, encoding='utf-8') async

Read a file as text without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
325
326
327
async def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
    """Read a file as text without blocking the event loop."""
    return await asyncio.to_thread(self._sync.read_text, path, encoding)

write_text(path, text, encoding='utf-8') async

Write text to a file without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
329
330
331
async def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
    """Write text to a file without blocking the event loop."""
    await asyncio.to_thread(self._sync.write_text, path, text, encoding)

exists(path) async

Check whether a file exists without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
333
334
335
async def exists(self, path: str | Path) -> bool:
    """Check whether a file exists without blocking the event loop."""
    return await asyncio.to_thread(self._sync.exists, path)

makedirs(path, *, exist_ok=False) async

Create directories recursively without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
337
338
339
async def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
    """Create directories recursively without blocking the event loop."""
    await asyncio.to_thread(self._sync.makedirs, path, exist_ok=exist_ok)

copy(src, dst) async

Copy a file without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
341
342
343
async def copy(self, src: str | Path, dst: str | Path) -> None:
    """Copy a file without blocking the event loop."""
    await asyncio.to_thread(self._sync.copy, src, dst)

glob(path, pattern) async

List files matching a glob pattern without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
345
346
347
async def glob(self, path: str | Path, pattern: str) -> list[str]:
    """List files matching a glob pattern without blocking the event loop."""
    return await asyncio.to_thread(self._sync.glob, path, pattern)

remove(path) async

Remove a file without blocking the event loop.

Source code in src/idfkit/simulation/fs.py 
349
350
351
async def remove(self, path: str | Path) -> None:
    """Remove a file without blocking the event loop."""
    await asyncio.to_thread(self._sync.remove, path)

S3FileSystem

idfkit.simulation.fs.S3FileSystem

File system implementation backed by Amazon S3.

Requires the boto3 package (install via pip install idfkit[s3]).

This backend enables cloud-native simulation workflows where results are stored directly in S3 for later retrieval. EnergyPlus runs locally in a temporary directory, then results are uploaded to S3 after completion.

Parameters:

Name Type Description Default
bucket str

S3 bucket name.

 required
prefix str

Optional key prefix prepended to all paths. Use this to namespace simulations (e.g., "project-x/batch-42/").

 ''
**boto_kwargs Any

Additional keyword arguments passed to boto3.client("s3", ...). Common options include:

  • region_name: AWS region (e.g., "us-east-1")
  • endpoint_url: Custom endpoint for S3-compatible services (MinIO, LocalStack, etc.)
  • aws_access_key_id, aws_secret_access_key: Explicit credentials (normally use IAM roles or environment variables)
 {}

Examples:

# Basic usage
fs = S3FileSystem(bucket="my-bucket", prefix="simulations/")

# With MinIO (S3-compatible)
fs = S3FileSystem(
    bucket="local-bucket",
    endpoint_url="http://localhost:9000",
    aws_access_key_id="minioadmin",
    aws_secret_access_key="minioadmin",
)

# Use with simulate()
result = simulate(model, weather, output_dir="run-001", fs=fs)
Source code in src/idfkit/simulation/fs.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
class S3FileSystem:
    """File system implementation backed by Amazon S3.

    Requires the ``boto3`` package (install via ``pip install idfkit[s3]``).

    This backend enables cloud-native simulation workflows where results are
    stored directly in S3 for later retrieval. EnergyPlus runs locally in a
    temporary directory, then results are uploaded to S3 after completion.

    Args:
        bucket: S3 bucket name.
        prefix: Optional key prefix prepended to all paths. Use this to
            namespace simulations (e.g., ``"project-x/batch-42/"``).
        **boto_kwargs: Additional keyword arguments passed to
            ``boto3.client("s3", ...)``. Common options include:

            - ``region_name``: AWS region (e.g., ``"us-east-1"``)
            - ``endpoint_url``: Custom endpoint for S3-compatible services
              (MinIO, LocalStack, etc.)
            - ``aws_access_key_id``, ``aws_secret_access_key``: Explicit
              credentials (normally use IAM roles or environment variables)

    Examples:
        ```python
        # Basic usage
        fs = S3FileSystem(bucket="my-bucket", prefix="simulations/")

        # With MinIO (S3-compatible)
        fs = S3FileSystem(
            bucket="local-bucket",
            endpoint_url="http://localhost:9000",
            aws_access_key_id="minioadmin",
            aws_secret_access_key="minioadmin",
        )

        # Use with simulate()
        result = simulate(model, weather, output_dir="run-001", fs=fs)
        ```
    """

    def __init__(self, bucket: str, prefix: str = "", **boto_kwargs: Any) -> None:
        try:
            import boto3  # type: ignore[import-not-found]
        except ImportError:
            msg = "boto3 is required for S3FileSystem. Install it with: pip install idfkit[s3]"
            raise ImportError(msg) from None
        _boto3: Any = boto3
        self._bucket = bucket
        self._prefix = prefix.strip("/")
        self._client: S3Client = _boto3.client("s3", **boto_kwargs)

    def _key(self, path: str | Path) -> str:
        """Build an S3 key by prepending the configured prefix.

        Args:
            path: Logical file path.

        Returns:
            The full S3 object key.
        """
        raw = str(path).lstrip("/")
        if self._prefix:
            return f"{self._prefix}/{raw}"
        return raw

    def read_bytes(self, path: str | Path) -> bytes:
        """Read a file as raw bytes from S3."""
        resp = self._client.get_object(Bucket=self._bucket, Key=self._key(path))
        return resp["Body"].read()  # type: ignore[no-any-return]

    def write_bytes(self, path: str | Path, data: bytes) -> None:
        """Write raw bytes to S3."""
        self._client.put_object(Bucket=self._bucket, Key=self._key(path), Body=data)

    def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
        """Read a file as text from S3."""
        return self.read_bytes(path).decode(encoding)

    def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
        """Write text to S3."""
        self.write_bytes(path, text.encode(encoding))

    def exists(self, path: str | Path) -> bool:
        """Check whether an object exists in S3."""
        try:
            self._client.head_object(Bucket=self._bucket, Key=self._key(path))
        except Exception:
            return False
        return True

    def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
        """No-op — S3 has no directory concept."""

    def copy(self, src: str | Path, dst: str | Path) -> None:
        """Copy an object within the same bucket."""
        self._client.copy_object(
            Bucket=self._bucket,
            CopySource={"Bucket": self._bucket, "Key": self._key(src)},
            Key=self._key(dst),
        )

    def glob(self, path: str | Path, pattern: str) -> list[str]:
        """List objects matching a glob pattern under *path*.

        Returns logical paths (without the configured S3 prefix) so that
        they can be passed back to other ``S3FileSystem`` methods which
        prepend the prefix automatically via ``_key()``.
        """
        prefix = self._key(path).rstrip("/") + "/"
        paginator = self._client.get_paginator("list_objects_v2")
        matches: list[str] = []
        # Compute how much of the key is the S3 prefix, so we can strip it.
        logical_base = str(path).strip("/")
        for page in paginator.paginate(Bucket=self._bucket, Prefix=prefix):
            for obj in page.get("Contents", []):
                key = obj.get("Key", "")
                if not key:
                    continue
                # Match only the filename portion against the pattern
                name = key[len(prefix) :]
                if fnmatch.fnmatch(name, pattern):
                    # Return logical path (base/name) that _key() can prefix
                    matches.append(f"{logical_base}/{name}" if logical_base else name)
        return matches

    def remove(self, path: str | Path) -> None:
        """Delete an object from S3."""
        self._client.delete_object(Bucket=self._bucket, Key=self._key(path))

read_bytes(path)

Read a file as raw bytes from S3.

Source code in src/idfkit/simulation/fs.py 
419
420
421
422
def read_bytes(self, path: str | Path) -> bytes:
    """Read a file as raw bytes from S3."""
    resp = self._client.get_object(Bucket=self._bucket, Key=self._key(path))
    return resp["Body"].read()  # type: ignore[no-any-return]

write_bytes(path, data)

Write raw bytes to S3.

Source code in src/idfkit/simulation/fs.py 
424
425
426
def write_bytes(self, path: str | Path, data: bytes) -> None:
    """Write raw bytes to S3."""
    self._client.put_object(Bucket=self._bucket, Key=self._key(path), Body=data)

read_text(path, encoding='utf-8')

Read a file as text from S3.

Source code in src/idfkit/simulation/fs.py 
428
429
430
def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
    """Read a file as text from S3."""
    return self.read_bytes(path).decode(encoding)

write_text(path, text, encoding='utf-8')

Write text to S3.

Source code in src/idfkit/simulation/fs.py 
432
433
434
def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
    """Write text to S3."""
    self.write_bytes(path, text.encode(encoding))

exists(path)

Check whether an object exists in S3.

Source code in src/idfkit/simulation/fs.py 
436
437
438
439
440
441
442
def exists(self, path: str | Path) -> bool:
    """Check whether an object exists in S3."""
    try:
        self._client.head_object(Bucket=self._bucket, Key=self._key(path))
    except Exception:
        return False
    return True

makedirs(path, *, exist_ok=False)

No-op — S3 has no directory concept.

Source code in src/idfkit/simulation/fs.py 
444
445
def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
    """No-op — S3 has no directory concept."""

copy(src, dst)

Copy an object within the same bucket.

Source code in src/idfkit/simulation/fs.py 
447
448
449
450
451
452
453
def copy(self, src: str | Path, dst: str | Path) -> None:
    """Copy an object within the same bucket."""
    self._client.copy_object(
        Bucket=self._bucket,
        CopySource={"Bucket": self._bucket, "Key": self._key(src)},
        Key=self._key(dst),
    )

glob(path, pattern)

List objects matching a glob pattern under path.

Returns logical paths (without the configured S3 prefix) so that they can be passed back to other S3FileSystem methods which prepend the prefix automatically via _key().

Source code in src/idfkit/simulation/fs.py 
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
def glob(self, path: str | Path, pattern: str) -> list[str]:
    """List objects matching a glob pattern under *path*.

    Returns logical paths (without the configured S3 prefix) so that
    they can be passed back to other ``S3FileSystem`` methods which
    prepend the prefix automatically via ``_key()``.
    """
    prefix = self._key(path).rstrip("/") + "/"
    paginator = self._client.get_paginator("list_objects_v2")
    matches: list[str] = []
    # Compute how much of the key is the S3 prefix, so we can strip it.
    logical_base = str(path).strip("/")
    for page in paginator.paginate(Bucket=self._bucket, Prefix=prefix):
        for obj in page.get("Contents", []):
            key = obj.get("Key", "")
            if not key:
                continue
            # Match only the filename portion against the pattern
            name = key[len(prefix) :]
            if fnmatch.fnmatch(name, pattern):
                # Return logical path (base/name) that _key() can prefix
                matches.append(f"{logical_base}/{name}" if logical_base else name)
    return matches

remove(path)

Delete an object from S3.

Source code in src/idfkit/simulation/fs.py 
479
480
481
def remove(self, path: str | Path) -> None:
    """Delete an object from S3."""
    self._client.delete_object(Bucket=self._bucket, Key=self._key(path))

AsyncS3FileSystem

idfkit.simulation.fs.AsyncS3FileSystem

Async file system implementation backed by Amazon S3 via aiobotocore.

Requires the aiobotocore package (install via pip install idfkit[async-s3]).

This is the non-blocking counterpart to S3FileSystem. Use it with async_simulate and the async batch functions to avoid blocking the event loop during S3 I/O.

The client must be initialised via the async context manager protocol:

```python
async with AsyncS3FileSystem(bucket="my-bucket") as fs:
    result = await async_simulate(model, weather, output_dir="run-001", fs=fs)
```

Parameters:

Name Type Description Default
bucket str

S3 bucket name.

 required
prefix str

Optional key prefix prepended to all paths. Use this to namespace simulations (e.g., "project-x/batch-42/").

 ''
**boto_kwargs Any

Additional keyword arguments passed to session.create_client("s3", ...). Common options include:

  • region_name: AWS region (e.g., "us-east-1")
  • endpoint_url: Custom endpoint for S3-compatible services (MinIO, LocalStack, etc.)
  • aws_access_key_id, aws_secret_access_key: Explicit credentials (normally use IAM roles or environment variables)
 {}

Examples:

from idfkit.simulation import AsyncS3FileSystem, async_simulate

async with AsyncS3FileSystem(bucket="my-bucket", prefix="sims/") as fs:
    result = await async_simulate(model, weather, output_dir="run-001", fs=fs)
    errors = await result.async_errors()
Source code in src/idfkit/simulation/fs.py 
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
class AsyncS3FileSystem:
    """Async file system implementation backed by Amazon S3 via ``aiobotocore``.

    Requires the ``aiobotocore`` package
    (install via ``pip install idfkit[async-s3]``).

    This is the non-blocking counterpart to [S3FileSystem][idfkit.simulation.fs.S3FileSystem].  Use it
    with [async_simulate][idfkit.simulation.async_runner.async_simulate] and the
    async batch functions to avoid blocking the event loop during S3 I/O.

    The client must be initialised via the async context manager protocol:

        ```python
        async with AsyncS3FileSystem(bucket="my-bucket") as fs:
            result = await async_simulate(model, weather, output_dir="run-001", fs=fs)
        ```

    Args:
        bucket: S3 bucket name.
        prefix: Optional key prefix prepended to all paths. Use this to
            namespace simulations (e.g., ``"project-x/batch-42/"``).
        **boto_kwargs: Additional keyword arguments passed to
            ``session.create_client("s3", ...)``. Common options include:

            - ``region_name``: AWS region (e.g., ``"us-east-1"``)
            - ``endpoint_url``: Custom endpoint for S3-compatible services
              (MinIO, LocalStack, etc.)
            - ``aws_access_key_id``, ``aws_secret_access_key``: Explicit
              credentials (normally use IAM roles or environment variables)

    Examples:
        ```python
        from idfkit.simulation import AsyncS3FileSystem, async_simulate

        async with AsyncS3FileSystem(bucket="my-bucket", prefix="sims/") as fs:
            result = await async_simulate(model, weather, output_dir="run-001", fs=fs)
            errors = await result.async_errors()
        ```
    """

    def __init__(self, bucket: str, prefix: str = "", **boto_kwargs: Any) -> None:
        try:
            from aiobotocore.session import get_session  # type: ignore[import-not-found]
        except ImportError:
            msg = "aiobotocore is required for AsyncS3FileSystem. Install it with: pip install idfkit[async-s3]"
            raise ImportError(msg) from None
        self._bucket = bucket
        self._prefix = prefix.strip("/")
        self._session: Any = get_session()
        self._boto_kwargs = boto_kwargs
        self._client: AsyncS3Client | None = None
        self._client_ctx: Any = None

    async def __aenter__(self) -> AsyncS3FileSystem:
        """Create the aiobotocore S3 client."""
        self._client_ctx = self._session.create_client("s3", **self._boto_kwargs)
        self._client = await self._client_ctx.__aenter__()
        return self

    async def __aexit__(self, *args: Any) -> None:
        """Close the aiobotocore S3 client."""
        if self._client_ctx is not None:
            await self._client_ctx.__aexit__(*args)
            self._client = None
            self._client_ctx = None

    def _ensure_client(self) -> AsyncS3Client:
        if self._client is None:
            msg = (
                "AsyncS3FileSystem client is not initialised. "
                "Use 'async with AsyncS3FileSystem(...) as fs:' to create the client."
            )
            raise RuntimeError(msg)
        return self._client

    def _key(self, path: str | Path) -> str:
        """Build an S3 key by prepending the configured prefix.

        Args:
            path: Logical file path.

        Returns:
            The full S3 object key.
        """
        raw = str(path).lstrip("/")
        if self._prefix:
            return f"{self._prefix}/{raw}"
        return raw

    async def read_bytes(self, path: str | Path) -> bytes:
        """Read a file as raw bytes from S3."""
        client = self._ensure_client()
        resp = await client.get_object(Bucket=self._bucket, Key=self._key(path))
        async with resp["Body"] as stream:
            return await stream.read()  # type: ignore[no-any-return]

    async def write_bytes(self, path: str | Path, data: bytes) -> None:
        """Write raw bytes to S3."""
        client = self._ensure_client()
        await client.put_object(Bucket=self._bucket, Key=self._key(path), Body=data)

    async def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
        """Read a file as text from S3."""
        return (await self.read_bytes(path)).decode(encoding)

    async def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
        """Write text to S3."""
        await self.write_bytes(path, text.encode(encoding))

    async def exists(self, path: str | Path) -> bool:
        """Check whether an object exists in S3."""
        client = self._ensure_client()
        try:
            await client.head_object(Bucket=self._bucket, Key=self._key(path))
        except Exception:
            return False
        return True

    async def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
        """No-op — S3 has no directory concept."""

    async def copy(self, src: str | Path, dst: str | Path) -> None:
        """Copy an object within the same bucket."""
        client = self._ensure_client()
        await client.copy_object(
            Bucket=self._bucket,
            CopySource={"Bucket": self._bucket, "Key": self._key(src)},
            Key=self._key(dst),
        )

    async def glob(self, path: str | Path, pattern: str) -> list[str]:
        """List objects matching a glob pattern under *path*.

        Returns logical paths (without the configured S3 prefix) so that
        they can be passed back to other ``AsyncS3FileSystem`` methods which
        prepend the prefix automatically via ``_key()``.
        """
        client = self._ensure_client()
        prefix = self._key(path).rstrip("/") + "/"
        paginator = client.get_paginator("list_objects_v2")
        logical_base = str(path).strip("/")
        return await self._collect_glob_matches(paginator, prefix, pattern, logical_base)

    async def _collect_glob_matches(
        self,
        paginator: Any,
        prefix: str,
        pattern: str,
        logical_base: str,
    ) -> list[str]:
        """Iterate the paginator and collect matching keys.

        Separated to keep the ``glob`` method's return type fully known
        to the type checker (aiobotocore's paginator yields untyped pages).
        """
        matches: list[str] = []
        async for raw_page in paginator.paginate(Bucket=self._bucket, Prefix=prefix):
            page: dict[str, Any] = raw_page
            contents: list[dict[str, Any]] = page.get("Contents", [])
            for obj in contents:
                key: str = obj.get("Key", "")
                if not key:
                    continue
                name: str = key[len(prefix) :]
                if fnmatch.fnmatch(name, pattern):
                    matches.append(f"{logical_base}/{name}" if logical_base else name)
        return matches

    async def remove(self, path: str | Path) -> None:
        """Delete an object from S3."""
        client = self._ensure_client()
        await client.delete_object(Bucket=self._bucket, Key=self._key(path))

read_bytes(path) async

Read a file as raw bytes from S3.

Source code in src/idfkit/simulation/fs.py 
573
574
575
576
577
578
async def read_bytes(self, path: str | Path) -> bytes:
    """Read a file as raw bytes from S3."""
    client = self._ensure_client()
    resp = await client.get_object(Bucket=self._bucket, Key=self._key(path))
    async with resp["Body"] as stream:
        return await stream.read()  # type: ignore[no-any-return]

write_bytes(path, data) async

Write raw bytes to S3.

Source code in src/idfkit/simulation/fs.py 
580
581
582
583
async def write_bytes(self, path: str | Path, data: bytes) -> None:
    """Write raw bytes to S3."""
    client = self._ensure_client()
    await client.put_object(Bucket=self._bucket, Key=self._key(path), Body=data)

read_text(path, encoding='utf-8') async

Read a file as text from S3.

Source code in src/idfkit/simulation/fs.py 
585
586
587
async def read_text(self, path: str | Path, encoding: str = "utf-8") -> str:
    """Read a file as text from S3."""
    return (await self.read_bytes(path)).decode(encoding)

write_text(path, text, encoding='utf-8') async

Write text to S3.

Source code in src/idfkit/simulation/fs.py 
589
590
591
async def write_text(self, path: str | Path, text: str, encoding: str = "utf-8") -> None:
    """Write text to S3."""
    await self.write_bytes(path, text.encode(encoding))

exists(path) async

Check whether an object exists in S3.

Source code in src/idfkit/simulation/fs.py 
593
594
595
596
597
598
599
600
async def exists(self, path: str | Path) -> bool:
    """Check whether an object exists in S3."""
    client = self._ensure_client()
    try:
        await client.head_object(Bucket=self._bucket, Key=self._key(path))
    except Exception:
        return False
    return True

makedirs(path, *, exist_ok=False) async

No-op — S3 has no directory concept.

Source code in src/idfkit/simulation/fs.py 
602
603
async def makedirs(self, path: str | Path, *, exist_ok: bool = False) -> None:
    """No-op — S3 has no directory concept."""

copy(src, dst) async

Copy an object within the same bucket.

Source code in src/idfkit/simulation/fs.py 
605
606
607
608
609
610
611
612
async def copy(self, src: str | Path, dst: str | Path) -> None:
    """Copy an object within the same bucket."""
    client = self._ensure_client()
    await client.copy_object(
        Bucket=self._bucket,
        CopySource={"Bucket": self._bucket, "Key": self._key(src)},
        Key=self._key(dst),
    )

glob(path, pattern) async

List objects matching a glob pattern under path.

Returns logical paths (without the configured S3 prefix) so that they can be passed back to other AsyncS3FileSystem methods which prepend the prefix automatically via _key().

Source code in src/idfkit/simulation/fs.py 
614
615
616
617
618
619
620
621
622
623
624
625
async def glob(self, path: str | Path, pattern: str) -> list[str]:
    """List objects matching a glob pattern under *path*.

    Returns logical paths (without the configured S3 prefix) so that
    they can be passed back to other ``AsyncS3FileSystem`` methods which
    prepend the prefix automatically via ``_key()``.
    """
    client = self._ensure_client()
    prefix = self._key(path).rstrip("/") + "/"
    paginator = client.get_paginator("list_objects_v2")
    logical_base = str(path).strip("/")
    return await self._collect_glob_matches(paginator, prefix, pattern, logical_base)

remove(path) async

Delete an object from S3.

Source code in src/idfkit/simulation/fs.py 
652
653
654
655
async def remove(self, path: str | Path) -> None:
    """Delete an object from S3."""
    client = self._ensure_client()
    await client.delete_object(Bucket=self._bucket, Key=self._key(path))