# Sources

## timenet\_connectors.sources

Shared source access and media conversion helpers for dataset connectors.

### huggingface\_hub

Helpers for downloading files from pinned Hugging Face dataset revisions.

:func:`download_hub_files` downloads with `huggingface_hub.snapshot_download`, limited to the
requested files. The Hub client runs the downloads in its own thread pool. Its progress bars stay
hidden: the bar that counts file bytes reaches the ambient :mod:`~timenet_connectors.download.progress`
sink instead, so the build CLI shows the download like any other connector download.
`huggingface_hub` is declared in the connector's `requirements.txt` and imported lazily, which
keeps `--no-isolation` usable. The Hub client reads `HF_TOKEN` from the environment, so gated
datasets work with no extra wiring.

#### download\_hub\_files `async`

```
download_hub_files(
    repo: str,
    filenames: Sequence[str],
    *,
    revision: str,
    cache_dir: Path,
) -> list[Path]
```

Download files from a dataset revision with one `snapshot_download` call.

Parameters:

| Name | Type | Description | Default |
| --- | --- | --- | --- |
| `repo` | `str` | The Hub dataset id, such as `"BASH-Lab/SensorCaps"`. | *required* |
| `filenames` | `Sequence[str]` | Repository-relative paths of the files to download. | *required* |
| `revision` | `str` | The revision to read, normally a pinned commit. | *required* |
| `cache_dir` | `Path` | Directory where the connector caches Hub files. | *required* |

Returns:

| Type | Description |
| --- | --- |
| `list[Path]` | The local paths, in the order of `filenames`. |

Raises:

| Type | Description |
| --- | --- |
| `ImportError` | If `huggingface_hub`, declared in the connector's requirements, is missing. |
| `TimeNetDownloadError` | If the download fails, or a file is absent from the revision. |

### images

Represent source images as RGB or RGBA tensor Signals for Zarr.

#### image\_signal

```
image_signal(source: Path | bytes, *, name: str) -> Signal
```

Create a lazy one-frame RGB or RGBA Signal with a shape-specific spec of the image modality.

Returns:

| Type | Description |
| --- | --- |
| `Signal` | A Signal whose single value is the original-size image tensor. |

### json\_stream

Stream large JSON documents entry by entry, keeping each value's byte range.

Some sources publish one JSON document of hundreds of megabytes instead of JSON Lines. Decoding it
whole holds the entire corpus in memory. Instead, decode one value at a time, record its UTF-8 byte
range, and let a lazy loader reread just that range with :func:`read_entry`.

#### JsonCursor

Read JSON syntax incrementally while tracking consumed UTF-8 bytes.

The cursor reads raw bytes and decodes UTF-8 itself, so its offsets always count the file's own
bytes, CRLF included. Dataset-specific key and structure validation belongs to the caller.

##### buffer `instance-attribute`

```
buffer = ''
```

##### decoder `instance-attribute`

```
decoder = json.JSONDecoder(
    object_pairs_hook=_distinct_fields
)
```

##### handle `instance-attribute`

```
handle = handle
```

##### offset `instance-attribute`

```
offset = offset
```

##### position `instance-attribute`

```
position = 0
```

##### utf8 `instance-attribute`

```
utf8 = codecs.getincrementaldecoder('utf-8')()
```

##### consume

```
consume(length: int) -> None
```

Consume `length` buffered characters, advancing the byte offset past them.

##### expect

```
expect(token: str) -> None
```

Consume one expected structural character.

Raises:

| Type | Description |
| --- | --- |
| `TimeFValidationError` | If the next character is different. |

##### expect\_end

```
expect_end() -> None
```

Check that nothing but whitespace follows the document.

Raises:

| Type | Description |
| --- | --- |
| `TimeFValidationError` | If content follows the document. |

##### object\_keys

```
object_keys() -> Iterator[str]
```

Yield object keys, leaving each value for the caller to consume.

The caller must consume the entire value before requesting the next key. This allows
nested objects to be traversed without decoding them all at once.

Yields:

| Type | Description |
| --- | --- |
| `str` | Distinct string keys, with the cursor positioned after each colon. |

Raises:

| Type | Description |
| --- | --- |
| `TimeFValidationError` | If object syntax or keys are invalid. |

##### open `classmethod`

```
open(path: Path) -> Generator[Self]
```

Open a file for reading from its first byte.

Yields:

| Type | Description |
| --- | --- |
| `Generator[Self]` | A cursor at the start of the file. |

##### peek

```
peek() -> str
```

Skip whitespace and return the next character, or `""` at the end of the document.

Returns:

| Type | Description |
| --- | --- |
| `str` | The next non-whitespace character. |

##### read

```
read(size: int) -> bool
```

Append up to `size` more bytes of the file to the buffer.

Returns:

| Type | Description |
| --- | --- |
| `bool` | False at the end of the file. |

Raises:

| Type | Description |
| --- | --- |
| `TimeFValidationError` | If the bytes are not valid UTF-8. |

##### value

```
value() -> tuple[Any, int, int]
```

Decode the next complete value.

Returns:

| Type | Description |
| --- | --- |
| `tuple[Any, int, int]` | The value, and its UTF-8 byte offset and length. |

Raises:

| Type | Description |
| --- | --- |
| `TimeFValidationError` | If the value is not valid UTF-8 or JSON, with the error's byte offset in the message, or repeats a field. |

#### iter\_object

```
iter_object(
    path: Path,
) -> Iterator[tuple[str, Any, int, int]]
```

Traverse a document that is one object, without retaining decoded values.

Yields:

| Type | Description |
| --- | --- |
| `tuple[str, Any, int, int]` | Each key, its decoded value, and the value's UTF-8 byte offset and length. |

Raises:

| Type | Description |
| --- | --- |
| `TimeFValidationError` | If the document is not one object with distinct string keys, with the file and key in the message. |

#### read\_entry

```
read_entry(path: Path, offset: int, length: int) -> Any
```

Reread one value at the byte range a traversal recorded.

Decodes with the traversal's own cursor, so a reread accepts exactly what the traversal did:
strict UTF-8 without a byte order mark, and no duplicate fields. Errors name their byte in the
file.

Nothing records the file's identity, so a change that leaves a valid value in the range goes
unnoticed. Pin the file's content, by checksum say, where a stale reread would matter.

Returns:

| Type | Description |
| --- | --- |
| `Any` | The decoded value. |

Raises:

| Type | Description |
| --- | --- |
| `TimeFFormatError` | If the range cannot be read, ends early, or does not decode as one value with distinct fields, as happens when the file changed since the traversal. |

### parquet

Read parquet source files row by row.

#### iter\_parquet\_rows

```
iter_parquet_rows(
    path: Path, columns: Sequence[str] | None = None
) -> Iterator[dict[str, Any]]
```

Yield one plain dict per row of a parquet file, decoding a batch at a time.

Parameters:

| Name | Type | Description | Default |
| --- | --- | --- | --- |
| `path` | `Path` | The parquet file. | *required* |
| `columns` | `Sequence[str] | None` | The columns to read, which leaves out the others' decoding cost. All columns by default. | `None` |

Yields:

| Type | Description |
| --- | --- |
| `dict[str, Any]` | Each row, keyed by column name, in file order. |

Raises:

| Type | Description |
| --- | --- |
| `TimeFFormatError` | If `columns` names a column the file doesn't have. |