Session

Session(
    resource_ids,
    agent,
    ttl_ms=4000,
    tentative_ttl_ms=1000,
    token=None,
    trigger_count_proposer=None,
    skip_lock_check=False,
)

A lease over a set of resources on a QuEL system.

Opening a session locks its resources on the server and yields a token used for subsequent operations such as deploying instruments and triggering. Create sessions with QuelwareClient.create_session() and use them as an async context manager, so they are opened on entry and closed on exit:

async with qc.create_session(["unit0:port0"]) as session:
    await session.deploy_instruments("unit0:port0", definitions)
    await session.trigger(instrument_ids)

Build a session over a set of resources.

Normally created by QuelwareClient.create_session() rather than directly.

Parameters:

Name	Type	Description	Default
resource_ids	Collection[ResourceId]	Resources to lock for the session.	required
agent	AgentContainer	Container providing the session and per-unit agents.	required
ttl_ms	int	Time-to-live, in milliseconds, of the committed lease.	4000
tentative_ttl_ms	int	Time-to-live, in milliseconds, of the tentative lease held while opening.	1000
token	SessionToken | None	Pre-existing session token, if resuming a session.	None
trigger_count_proposer	TriggerCountProposer | None	Strategy for choosing the clock count of a synchronized multi-unit trigger. Defaults to a fixed-offset proposer aligned to a 32-count grid.	None
skip_lock_check	bool	When True, skip verifying that the requested resources are locked after opening.	False

Source code in quelware-client/src/quelware_client/core/_session.py

def __init__(  # noqa: PLR0913
    self,
    resource_ids: Collection[ResourceId],
    agent: AgentContainer,
    ttl_ms: int = 4000,
    tentative_ttl_ms: int = 1000,
    token: SessionToken | None = None,
    trigger_count_proposer: TriggerCountProposer | None = None,
    skip_lock_check: bool = False,
):
    """Build a session over a set of resources.

    Normally created by `QuelwareClient.create_session()` rather than
    directly.

    Args:
        resource_ids: Resources to lock for the session.
        agent: Container providing the session and per-unit agents.
        ttl_ms: Time-to-live, in milliseconds, of the committed lease.
        tentative_ttl_ms: Time-to-live, in milliseconds, of the tentative
            lease held while opening.
        token: Pre-existing session token, if resuming a session.
        trigger_count_proposer: Strategy for choosing the clock count of a
            synchronized multi-unit trigger. Defaults to a fixed-offset
            proposer aligned to a 32-count grid.
        skip_lock_check: When True, skip verifying that the requested
            resources are locked after opening.
    """
    self._rsrc_ids = set(resource_ids)
    self._ttl_ms = ttl_ms
    self._tentative_ttl_ms = tentative_ttl_ms
    self._agent = agent
    self._token = token
    if trigger_count_proposer is None:
        trigger_count_proposer = _default_count_proposer
    self._trigger_count_proposer = trigger_count_proposer

    self._unit_to_ids: dict[UnitLabel, list[ResourceId]] = {}
    for rid in self._rsrc_ids:
        ul = extract_unit_label(rid)
        self._unit_to_ids.setdefault(ul, []).append(rid)

    self._check_lock = not skip_lock_check

agent_container property

agent_container

The underlying container of agents used by this session.

available_resource_ids property

available_resource_ids

The set of resource ids this session was created for.

token property

token

The session token obtained when the session was opened.

Raises:

Type	Description
ValueError	If the session has not been opened yet.

unit_labels property

unit_labels

The labels of the units spanned by this session's resources.

close async

close()

Close the session and release its resources.

Source code in quelware-client/src/quelware_client/core/_session.py

async def close(self):
    """Close the session and release its resources."""
    await self._agent.session.close_session(self.token)
    logger.info(f"Session closed. session_token={self.token}")

configure_unit async

configure_unit(unit_label, controls)

Apply unit-wide controls to a unit and return the resulting values.

The session must hold locks on every port of unit_label and carry the required capability; the unit must have no deployed instruments.

Parameters:

Name	Type	Description	Default
unit_label	UnitLabel	The unit to configure.	required
controls	Mapping[str, str]	Vendor-namespaced control key/values to apply.	required

Returns:

Type	Description
dict[str, str]	The unit's control values after applying the change.

Source code in quelware-client/src/quelware_client/core/_session.py

async def configure_unit(
    self, unit_label: UnitLabel, controls: Mapping[str, str]
) -> dict[str, str]:
    """Apply unit-wide controls to a unit and return the resulting values.

    The session must hold locks on every port of ``unit_label`` and carry
    the required capability; the unit must have no deployed instruments.

    Args:
        unit_label: The unit to configure.
        controls: Vendor-namespaced control key/values to apply.

    Returns:
        The unit's control values after applying the change.
    """
    return await self._agent.worker(unit_label).configure_unit(controls, self.token)

deploy_instruments async

deploy_instruments(port_id, definitions, append=False)

Deploy instrument definitions onto a port.

Each definition's alias is automatically prefixed with the port's unit label, so the aliases passed in must not contain a ':'.

Parameters:

Name	Type	Description	Default
port_id	str | ResourceId	Port to deploy onto. Its unit label selects the unit.	required
definitions	Collection[InstrumentDefinition]	Instrument definitions to deploy.	required
append	bool	When True, add to the port's existing instruments instead of replacing them.	False

Returns:

Type	Description
list[InstrumentInfo]	Information about the deployed instruments.

Raises:

Type	Description
ValueError	If any definition's alias contains a ':'.

Source code in quelware-client/src/quelware_client/core/_session.py

async def deploy_instruments(
    self,
    port_id: str | ResourceId,
    definitions: Collection[InstrumentDefinition],
    append: bool = False,
) -> list[InstrumentInfo]:
    """Deploy instrument definitions onto a port.

    Each definition's alias is automatically prefixed with the port's unit
    label, so the aliases passed in must not contain a ``':'``.

    Args:
        port_id: Port to deploy onto. Its unit label selects the unit.
        definitions: Instrument definitions to deploy.
        append: When True, add to the port's existing instruments instead
            of replacing them.

    Returns:
        Information about the deployed instruments.

    Raises:
        ValueError: If any definition's alias contains a ``':'``.
    """
    port_id = ResourceId(port_id)
    unit_label = extract_unit_label(port_id)
    prefixed_definitions = []
    for d in definitions:
        if ":" in d.alias:
            raise ValueError(f"alias must not contain ':' (got '{d.alias}')")
        prefixed = InstrumentDefinition(
            alias=f"{unit_label}:{d.alias}",
            mode=d.mode,
            role=d.role,
            profile=d.profile,
        )
        prefixed_definitions.append(prefixed)
    insts = await self._agent.resource(unit_label).deploy_instruments(
        port_id, prefixed_definitions, append, self.token
    )
    return insts

discard_instruments async

discard_instruments(port_id)

Remove all instruments deployed on a port.

Parameters:

Name	Type	Description	Default
port_id	str | ResourceId	Port whose instruments to discard. Its unit label selects the unit.	required

Source code in quelware-client/src/quelware_client/core/_session.py

async def discard_instruments(self, port_id: str | ResourceId) -> None:
    """Remove all instruments deployed on a port.

    Args:
        port_id: Port whose instruments to discard. Its unit label selects
            the unit.
    """
    port_id = ResourceId(port_id)
    unit_label = extract_unit_label(port_id)
    await self._agent.resource(unit_label).discard_instruments(port_id, self.token)

extend async

extend(new_ttl_ms)

Extend the session's lease.

Parameters:

Name	Type	Description	Default
new_ttl_ms	int	New time-to-live, in milliseconds, from now.	required

Returns:

Type	Description
bool	True if the server accepted the extension.

Source code in quelware-client/src/quelware_client/core/_session.py

async def extend(self, new_ttl_ms: int) -> bool:
    """Extend the session's lease.

    Args:
        new_ttl_ms: New time-to-live, in milliseconds, from now.

    Returns:
        True if the server accepted the extension.
    """
    success = await self._agent.session.extend_session(self.token, new_ttl_ms)
    logger.info(
        f"Session extended. session_token={self.token} new_ttl_ms={new_ttl_ms}"
    )
    return success

open async

open()

Open the session, locking its resources and obtaining a token.

Unless lock checking is disabled, this also verifies that every requested resource is actually locked.

Raises:

Type	Description
ValueError	If some requested resources could not be locked.

Source code in quelware-client/src/quelware_client/core/_session.py

async def open(self):
    """Open the session, locking its resources and obtaining a token.

    Unless lock checking is disabled, this also verifies that every
    requested resource is actually locked.

    Raises:
        ValueError: If some requested resources could not be locked.
    """
    token, _ = await self._agent.session.open_session(
        self._rsrc_ids,
        tentative_ttl_ms=self._tentative_ttl_ms,
        committed_ttl_ms=self._ttl_ms,
    )
    self._token = token
    if self._check_lock:
        await self._ensure_target_resources_locked()
    logger.info(f"Session opened successfully. session_token={token}")

trigger async

trigger(instrument_ids, wait_ms=None)

Apply pending configuration and trigger the given instruments.

The instruments' configuration is applied first, then a trigger is scheduled. If the manager-side trigger service is unavailable, the client falls back to a client-side trigger: a self-timed trigger for a single unit, or a clock-synchronized trigger across multiple units.

Parameters:

Name	Type	Description	Default
instrument_ids	Collection[ResourceId]	Instruments to trigger.	required
wait_ms	int | None	Minimum delay, in milliseconds, before the trigger fires. Gives all units time to be armed; a lower bound is enforced on the client-side fallback path.	None

Returns:

Type	Description
int	The clock count at which the trigger was scheduled.

Source code in quelware-client/src/quelware_client/core/_session.py

async def trigger(
    self,
    instrument_ids: Collection[ResourceId],
    wait_ms: int | None = None,
) -> int:
    """Apply pending configuration and trigger the given instruments.

    The instruments' configuration is applied first, then a trigger is
    scheduled. If the manager-side trigger service is unavailable, the
    client falls back to a client-side trigger: a self-timed trigger for a
    single unit, or a clock-synchronized trigger across multiple units.

    Args:
        instrument_ids: Instruments to trigger.
        wait_ms: Minimum delay, in milliseconds, before the trigger fires.
            Gives all units time to be armed; a lower bound is enforced on
            the client-side fallback path.

    Returns:
        The clock count at which the trigger was scheduled.
    """
    unit_to_ids = create_unit_to_ids_map(instrument_ids)

    logger.info(f"starting application (token= {self.token} )")
    apply_coros = [
        self._agent.instrument(unit_label).apply(self.token, ids)
        for unit_label, ids in unit_to_ids.items()
    ]
    await asyncio.gather(*apply_coros)
    logger.info(f"finished application (token= {self.token} )")

    try:
        scheduled = await self._agent.trigger.trigger(
            self.token,
            list(instrument_ids),
            requested_min_wait_ms=wait_ms,
        )
        logger.info(f"trigger scheduled via manager at clock_count={scheduled}")
        return scheduled
    except ServiceUnavailableError:
        fallback_wait_ms = max(wait_ms or 0, _FALLBACK_MIN_WAIT_MS)
        logger.warning(
            f"Manager-side TriggerService unavailable; falling back to "
            f"client-side trigger (wait_ms={fallback_wait_ms})."
        )

    return await self._client_side_trigger_fallback(unit_to_ids, fallback_wait_ms)