Metadata-Version: 2.4
Name: etiket-sdk
Version: 0.3.0b1
Summary: Python SDK for the eTiKeT platform
Author: QHarbor team
License-Expression: LicenseRef-Proprietary
Project-URL: Homepage, https://qharbor.nl
Project-URL: Documentation, https://docs.qharbor.nl
Keywords: etiket,sync,sdk,client
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: POSIX :: Linux
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Operating System :: Microsoft :: Windows
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx<1.0,>=0.24
Requires-Dist: pydantic<3.0,>=2.0
Requires-Dist: etiket-service-manager>=0.3.0b1
Dynamic: license-file

# eTiKeT SDK

Python SDK for the eTiKeT platform. The Sync Agent runs as a background service on your computer, automatically synchronizing your experimental data to qHarbor. This SDK allows you to easily interact with the Sync Agent and manage your sync sources.

## Installation

```bash
pip install etiket-sdk
```

---

## Sync Agent

The `SyncAgent` class provides methods to check the status of the background sync service, start and stop it, and view recent errors. For more information about debugging, see the [Debugging](#debugging) section.

### Checking Status

```python
from etiket_sdk.sync import SyncAgent

# Get the current status
status = SyncAgent.status()
print(status)
```

Output:
```
Sync Agent Status:
  Agent Service: running
  API Service: running
  Agent Status: running
  Iterations: 142
  Last Update: 2025-12-16 10:30:00
```

The status shows:
- **Agent Service**: Whether the sync agent service is installed and running
- **API Service**: Whether the sync API service is running (the service that allows the SDK to interact with the Sync Agent)
- **Agent Status**: Current state (`running`, `stopped`, `no_connection`, `unavailable`). When the sync agent is not running, the API service is not available.
- **Iterations**: Number of sync cycles completed
- **Last Update**: When the status was last updated

### Starting and Stopping

```python
# Stop the sync agent
SyncAgent.stop()

# Start the sync agent
SyncAgent.start()
```

### Viewing Errors

If sync isn't working, check for agent-level errors. Note: always check the timestamps to see if errors are recent.

```python
# Get recent errors
errors = SyncAgent.errors()
for error in errors:
    print(error)

# Or fetch only the unread inbox
unread = SyncAgent.errors(viewed=False)
```

Errors carry a `viewed` flag so you can acknowledge what you've already triaged:

```python
# Mark a single error as viewed (or unviewed)
unread[0].mark_viewed()

# Bulk-acknowledge — returns the count of rows actually flipped
flipped = SyncAgent.mark_all_errors_viewed()
```

These are system-level errors that block sync from running entirely (e.g., not logged in, no network connection).

---

## Sync Sources

A **Sync Source** defines where data comes from and how it should be synchronized. Each source uses a **backend** that knows how to read data from a specific format (e.g., QCoDeS, Quantify, or a folder of files).

### Listing Sources

```python
from etiket_sdk.sync import SyncSources

# List all configured sync sources
sources = SyncSources.list()
print(sources)
```
Output:
```
['QH datasets' (backend='etiket_sync_agent_native'), 'my_sync_source' (backend='etiket_sync_agent_folderbase')]
```

This lists all configured sync sources. Use `SyncSources.get(name)` to get full details for a specific source.

### Getting Source Details

```python
# Get a specific source by name
source = SyncSources.get("my_sync_source")
print(source)
```

Output:
```
SyncSource: my_sync_source
  id: 2
  backend: etiket_sync_agent_folderbase
  status: scanning
  items: 28/28 synced, 0 failed
  last_update: 2025-12-15 14:54:00
```

The output shows sync progress at a glance.

### Viewing Source Errors

When the sync source is unable to run because of an error, the status will be `error`. You can fetch and view errors for a source:

```python
source = SyncSources.get("my_source")
source.print_errors()

# Filter by viewed flag — pass viewed=False to print only unread errors
source.print_errors(viewed=False)
```

Errors carry a `viewed` flag so you can acknowledge what you've already triaged:

```python
unread = source.errors(viewed=False)
unread[0].mark_viewed()                # mark one
unread[0].mark_viewed(viewed=False)    # un-view if you change your mind

# Bulk-acknowledge — returns the count of rows actually flipped
flipped = source.mark_all_errors_viewed()
```

### Listing Available Backends

Before creating a source, see what backends are available:

```python
# List backend identifiers
print(SyncSources.backends())
# ['etiket_sync_agent_folderbase', 'etiket_sync_agent_quantify', 'etiket_sync_agent_qcodes', ...]

# Get details for a specific backend
backend = SyncSources.backend("etiket_sync_agent_folderbase")
print(backend)
```

### Generating a Create Example

Each backend can generate a copy-paste ready example:

```python
SyncSources.backend("etiket_sync_agent_folderbase").example()
```

Output:
```python
# Copy-paste this example to create a sync source:
SyncSources.create(
    name="my_folderbase_source",
    backend_identifier="etiket_sync_agent_folderbase",
    config_data={
        "root_directory": "<path to folder>",
        "is_server_folder": False,
    },
)
```

### Creating a Sync Source

```python
from etiket_sdk.sync import SyncSources

source = SyncSources.create(
    name="my_folderbase_source",
    backend_identifier="etiket_sync_agent_folderbase",
    config_data={
        "root_directory": "~/Downloads/MyData",
        "is_server_folder": False,
    },
    default_scope="6c319227-f6f8-4bc6-b7f9-ed278a8fb040",
)
```

**Scope Options:**

The `default_scope` and `autoassign_scope_mapping` options depend on the backend:

- **Required default scope** (e.g., FolderBase): The `default_scope` must be provided. All items sync to this scope.

- **Optional/No default scope** (e.g., core_tools, QCoDeS): Items may have scope identifiers that need to be mapped to scopes. Use `autoassign_scope_mapping=True` to automatically match identifiers to scope names:

```python
source = SyncSources.create(
    name="my_coretools_source",
    backend_identifier="etiket_sync_agent_coretools",
    config_data={"db_path": "/path/to/db"},
    autoassign_scope_mapping=True,  # auto-map identifiers to scopes
)
```

See [Scope Mappings](#scope-mappings) for more details on managing scope mappings.

### Updating a Sync Source

```python
source = SyncSources.get("my_source")

# Rename a source
source.update(rename_to="new_name")

# Update configuration
source.update(config_data={"is_server_folder": True})
```

> **Note:** Some configuration changes may require a sync agent restart to take effect.

### Deleting a Sync Source

```python
source = SyncSources.get("my_source")
source.delete()
```

### Source Ownership

Some backends (e.g. those that need a per-user API token to upload) attach an **owner** to each sync source — that's the user whose stored credentials get used for uploads. You can check, claim, or release ownership from the SDK:

```python
source = SyncSources.get("my_source")
print(source.owner)            # the current owner's user_sub, or None

# Claim ownership for the currently logged-in user. 
source.take_ownership()

# Release current ownership.
source.release_ownership()
```

`release_ownership()` is idempotent — calling it on an already-unowned source is a no-op. Backends that don't have an owner concept (e.g. `etiket_sync_agent_native`) will reject `take_ownership()` with `SyncSourceBackendHasNoOwner`:

```python
from etiket_sdk.sync import SyncSourceBackendHasNoOwner

try:
    source.take_ownership()
except SyncSourceBackendHasNoOwner:
    print("This backend doesn't support per-user ownership")
```

### Starting and Stopping Sync Sources

You can pause and resume individual sync sources without affecting others:

```python
source = SyncSources.get("my_source")

# Stop a running sync source
source.stop()

# Start a stopped sync source
source.start()
```

**Status transitions:**
- **Start**: Changes status from `stopped` or `error` to `syncing`. Raises `SyncSourceAlreadyRunning` if the source is already `syncing` or `scanning`.
- **Stop**: Changes status from `syncing`, `scanning`, or `error` to `stopped`. Raises `SyncSourceAlreadyStopped` if the source is already `stopped`.

```python
from etiket_sdk.sync import (
    SyncSourceAlreadyRunning,
    SyncSourceAlreadyStopped,
)

source = SyncSources.get("my_source")

try:
    source.start()
except SyncSourceAlreadyRunning:
    print("Source is already running!")

try:
    source.stop()
except SyncSourceAlreadyStopped:
    print("Source is already stopped!")
```

### Resetting Sync Items

If many items have failed and you want to retry them all at once (instead of resetting each individually), you can reset all failed sync items for a source:

```python
source = SyncSources.get("my_source")

# Reset only failed items (retry them)
count = source.reset_sync_items()
print(f"Reset {count} failed items")

# Reset ALL items (including successful ones, forcing re-sync)
count = source.reset_sync_items(include_successful=True)
print(f"Reset {count} items")
```

This is useful when:
- Multiple items failed due to a temporary issue (e.g. permissions, ...)
- You need to re-sync everything after changing the default scope
- You want to force a full re-sync of all data

### Reporting Errors


If something isn't working, you can send an error report to qHarbor support:

```python
source = SyncSources.get("my_source")
source.report_error("Sync keeps failing with permission denied")
```

### Scope Mappings

For sync sources that don't have a **required default scope** (like core_tools), sync items may have scope identifiers (project names, etc.) that need to be mapped to specific scopes in qHarbor. This allows items from different folders to be synced to different scopes.

> **Note:** If your sync source has a required default scope, all items are synced to that scope and mappings are not needed.

**Enabling Auto-Assign:**

When creating or updating a sync source, you can enable automatic scope mapping. When enabled, the sync agent will automatically create mappings for identifiers that match exactly one scope name (case-insensitive):

```python
# Enable auto-assign when creating a source
source = SyncSources.create(
    name="my_source",
    backend_identifier="etiket_sync_agent_folderbase",
    config_data={"root_directory": "/data", "is_server_folder": False},
    autoassign_scope_mapping=True,  # Enable automatic scope mapping
)

# Or enable it on an existing source
source = SyncSources.get("my_source")
source.update(autoassign_scope_mapping=True)
```

**Viewing Current Mappings:**

```python
source = SyncSources.get("my_source")

# View existing mappings
for mapping in source.scope_mappings:
    print(f"{mapping.scope_identifier} -> {mapping.scope_uuid}")

# Check how many items need mapping
print(f"Items needing mapping: {source.items_unmapped_scope_count}")
```

**Finding Unmapped Identifiers:**

```python
source = SyncSources.get("my_source")
unmapped = source.unmapped_identifiers()
print(f"Need mapping: {unmapped}")
# ['scope_A', 'scope_B', ...]
```

**Manual Auto-Assign:**

If `autoassign_scope_mapping` is not enabled on the source, you can manually trigger auto-assignment. This creates mappings for identifiers that match exactly one scope name (case-insensitive):

```python
source = SyncSources.get("my_source")
count = source.autoassign_mappings()
print(f"Created {count} mappings automatically")
```

> **Note:** When `autoassign_scope_mapping=True` is set on the sync source, this happens automatically during each sync cycle, so manual calls are not needed.

**Manual Mapping Management:**

For identifiers that don't match a scope name or match multiple scopes, you can create mappings manually:

```python
import uuid

source = SyncSources.get("my_source")

# Create a new mapping
source.create_mapping(
    scope_identifier="scope_A",
    scope_uuid=uuid.UUID("12345678-1234-1234-1234-123456789abc"),
)

# Update an existing mapping
source.update_mapping(
    scope_identifier="scope_A",
    scope_uuid=uuid.UUID("87654321-4321-4321-4321-cba987654321"),
)

# Delete a mapping
source.delete_mapping(scope_identifier="scope_A")
```

**Handling Mapping Errors:**

```python
from etiket_sdk.sync import (
    ScopeMappingNotFound,
    ScopeMappingAlreadyExists,
    ScopeMappingInvalidScope,
)

source = SyncSources.get("my_source")

try:
    source.create_mapping("scope_A", some_uuid)
except ScopeMappingAlreadyExists:
    print("Mapping already exists - use update_mapping instead")
except ScopeMappingInvalidScope:
    print("The target scope UUID doesn't exist")
```


---

## Sync Items

A **Sync Item** represents a single dataset that needs to be synchronized. Sync items are automatically created when the sync agent scans your sources.

### Listing Items

```python
from etiket_sdk.sync import SyncSources

source = SyncSources.get("my_source")

# List all items for a source
items = source.items()
for item in items:
    print(item)

# Filter by sync status
failed_items = source.items(is_synchronized=False)

# Search by data identifier
items = source.items(query="measurement_2024")
```

### Getting Item Details

```python
source = SyncSources.get("my_source")

# Get a specific item by ID
item = source.item(123)
print(item)

# View the sync record with detailed logs
print(item.sync_record)
```

### Prioritizing an Item

If you need an item to sync immediately:

```python
source = SyncSources.get("my_source")
item = source.item(123)
item.prioritize()
```

### Resetting a Failed Item

If an item failed and you want to retry:

```python
source = SyncSources.get("my_source")
item = source.item(123)
item.reset()
```

This clears the attempt counter and marks it for retry.

### Reporting Errors

```python
source = SyncSources.get("my_source")
item = source.item(123)
item.report_error("This file keeps failing to upload")
```

---

## Backends

In the [Sync Sources](#sync-sources) section, we used backends to create sync sources. The `Backends` class lets you manage installed backend packages — install new ones, update existing ones, or uninstall them.

**Backends** are implementations that know how to synchronize data from specific data acquisition software. We provide pre-built backends for popular packages:

| Backend | Package | Description |
|---------|---------|-------------|
| `etiket_sync_agent_qcodes` | QCoDeS | Sync QCoDeS datasets |
| `etiket_sync_agent_quantify` | Quantify | Sync Quantify datasets |
| `etiket_sync_agent_labber` | Labber | Sync Labber log files |
| `etiket_sync_agent_coretools` | Core Tools | Sync Core Tools datasets |
| `etiket_sync_agent_folderbase` | FolderBase | Sync files from a folder |

### Installing Backends

```python
from etiket_sdk.sync import Backends

# List installed backends
print(Backends.list())

# Get a specific backend and print its details
backend = Backends.get("etiket_sync_agent_qcodes")
print(backend)

# Install from PyPI
Backends.install_from_pypi("etiket-sync-agent-qcodes")

# Install a specific version
Backends.install_from_pypi("qcodes", version="0.45.0")

# Install from local development path
Backends.install_from_local("/path/to/my-backend")

# Install in editable mode (for development)
Backends.install_from_local("/path/to/my-backend", editable=True)

# Update a backend
Backends.update_from_pypi("etiket-sync-agent-qcodes")

# Uninstall a backend
Backends.uninstall("etiket_sync_agent_qcodes")
```

You can also create your own backends. See the [Backend Development Guide](https://docs.qharbor.nl/sync/backends) for more information.

---

## Converters

**Converters** automatically transform files from one format to another during sync. For example, you can convert CSV files to HDF5 so they can be plotted automatically in qHarbor.

Converters are supported when using the [FolderBase backend](https://docs.qharbor.nl/sync/folderbase).

### Installing Converters

```python
from etiket_sdk.sync import Converters

# List installed converters
print(Converters.list())

# Get a specific converter and print its details
converter = Converters.get("my_converter")
print(converter)

# Install from PyPI
Converters.install_from_pypi("my-converter-package")

# Install from local development path
Converters.install_from_local("/path/to/my-converter")

# Update a converter
Converters.update_from_pypi("my-converter-package")

# Uninstall a converter
Converters.uninstall("my_converter")
```

For information on creating your own converters, see the [Converter Development Guide](https://docs.qharbor.nl/sync/converters).

---

## Debugging

When sync isn't working, errors are caught at three levels. Debug from top to bottom:

### 1. Sync Agent Errors

These are **system-level errors** that block sync entirely (e.g., not logged in, no network).

```python
from etiket_sdk.sync import SyncAgent

# Check if the agent is running
status = SyncAgent.status()
print(status)

# If not running or status shows issues, check errors
errors = SyncAgent.errors()
for error in errors:
    print(error)
```

### 2. Sync Source Errors

These are **source-level errors** that prevent items from being discovered or processed.

```python
from etiket_sdk.sync import SyncSources

# Check the source status
source = SyncSources.get("my_source")
print(source)  # Shows status and item counts

# View recent errors
source.print_errors()
```

Look for:
- Configuration errors
- Backend connection issues
- Permission problems

### 3. Sync Item Errors

These are **item-level errors** that occur while syncing individual items.

```python
from etiket_sdk.sync import SyncSources

source = SyncSources.get("my_source")

# List failed items
failed = source.items(is_synchronized=False)
for item in failed:
    print(item)

# Get detailed error info for a specific item
item = source.item(123)
print(item.sync_record)  # Shows detailed logs and errors
```

### Debugging Workflow

1. **Start at the top**: Check `SyncAgent.status()` — is it running? Any login or network issues?
2. **Check agent errors**: `SyncAgent.errors()` — any agent-level errors?
3. **Check source status**: `SyncSources.get("name")` — is the source in an error state?
4. **Check source errors**: `source.print_errors()` — any configuration issues?
5. **Check failed items**: `source.items(is_synchronized=False)` — which items failed?
6. **Get item details**: `source.item(id).sync_record` — what went wrong?

### Reporting Issues

If you can't resolve an issue, send an error report to qHarbor support. This sends an automatic email to qHarbor support with error details attached:

```python
# Report agent issues
SyncAgent.report_error("Sync agent keeps crashing on startup")

# Report source issues
source = SyncSources.get("my_source")
source.report_error("Sync keeps failing with permission denied")

# Report item issues
item = source.item(123)
item.report_error("This file always fails to upload")
```

---

## Admin / Logs

Both local services (sync API and sync agent) write to a rotating log file in the shared data directory. The SDK exposes them via the `Logs` class so you don't need to know where they live on disk.

```python
from etiket_sdk.admin import Logs, LogService

# Last 500 records of the API service log
for line in Logs.tail(LogService.api, n=500):
    print(line)

# Full sync-agent log (current file plus the rotation backup, up to ~20 MB)
all_lines = Logs.tail(LogService.sync_agent, n=-1)
```

`n` counts logical records, not physical lines: multi-line entries (e.g. stack traces) come back as a single string with embedded `\n`, so a tail request will never slice through the middle of a traceback.

Raises `LogFileNotFound` if the requested service hasn't run yet (its log file doesn't exist on disk).

---

## Configuration

By default, the SDK connects to the local sync service at `localhost:10410`. You can change this if needed (though this is generally not necessary):

```python
from etiket_sdk._client import EtiketClient

# Configure once at application startup
EtiketClient.configure(base_url="http://my-server:10410")
```
