"""Pydantic models for the Ionworks API client.
These models use extra="allow" to accept any fields from the API response,
letting the API handle validation. Required fields are kept minimal.
"""
from __future__ import annotations
from collections.abc import Callable, Iterator
from datetime import UTC, datetime
from enum import StrEnum
from typing import TYPE_CHECKING, Any, Generic, TypeVar
from urllib.parse import urlencode
from pydantic import BaseModel, ConfigDict, Field, field_validator
from .validators import DataFrame, dict_to_df_validator
if TYPE_CHECKING:
from .errors import IonworksError
_T = TypeVar("_T")
[docs]
class PaginatedList(Generic[_T]): # noqa: UP046 - needs Python 3.11 compat
"""A list-like container that also carries pagination metadata.
Returned by ``list()`` methods when ``limit`` or ``offset`` is provided.
Behaves like a regular ``list`` for iteration, indexing, truthiness, and
``len()`` so callers can treat it interchangeably with ``list[T]``.
Parameters
----------
items : list[_T]
The page of results.
total : int
Total number of matching records across all pages.
"""
__slots__ = ("items", "total")
[docs]
def __init__(self, items: list[_T], total: int) -> None:
self.items = items
self.total = total
@property
def count(self) -> int:
"""Number of items in this page (same as ``len(self)``)."""
return len(self.items)
def __iter__(self) -> Iterator[_T]:
return iter(self.items)
def __len__(self) -> int:
return len(self.items)
def __getitem__(self, index: int | slice) -> Any:
return self.items[index]
def __bool__(self) -> bool:
return bool(self.items)
def __repr__(self) -> str:
return (
f"PaginatedList(items={self.items!r}, count={self.count}, "
f"total={self.total})"
)
def _build_endpoint(base: str, params: dict[str, Any]) -> str:
"""Append non-None query parameters to a base endpoint path.
List/tuple values are serialized as repeated params (``k=a&k=b``).
"""
pairs: list[tuple[str, str]] = []
for k, v in params.items():
if v is None:
continue
if isinstance(v, (list, tuple)):
pairs.extend((k, str(item)) for item in v)
else:
pairs.append((k, str(v)))
if not pairs:
return base
return f"{base}?{urlencode(pairs)}"
def _parse_list_response( # noqa: UP047
response_data: Any,
model_class: type[_T],
) -> PaginatedList[_T]:
"""Parse a list endpoint response into model instances.
Always returns a :class:`PaginatedList` which behaves like a regular
``list`` (iteration, indexing, ``len``, truthiness) so existing callers
are unaffected. Also exposes ``.count`` and ``.total`` for pagination.
Handles both the paginated dict format ``{"items": [...], "count": N,
"total": N}`` and legacy plain-array responses for backward
compatibility.
Parameters
----------
response_data : Any
Raw response from the API (list or paginated dict).
model_class : type[_T]
The Pydantic model class to instantiate per item.
Returns
-------
PaginatedList[_T]
A list-like result with ``.items``, ``.count``, and ``.total``.
"""
if isinstance(response_data, dict) and "items" in response_data:
items = [model_class(**item) for item in response_data["items"]]
return PaginatedList(
items=items,
total=response_data["total"],
)
# Legacy plain-array response (backward compat)
items = [model_class(**item) for item in response_data]
return PaginatedList(items=items, total=len(items))
def _build_filter_params(
*,
name: str | None = None,
name_exact: str | None = None,
created_by_email: str | None = None,
created_after: str | None = None,
created_before: str | None = None,
updated_after: str | None = None,
updated_before: str | None = None,
order_by: str | None = None,
order: str | None = None,
) -> dict[str, str]:
"""Translate Pythonic filter kwargs to backend query parameters.
Converts user-friendly parameter names to the operator syntax expected by
the backend API. For example, ``name="graphite"`` becomes
``name=ilike.%graphite%`` (a case-insensitive contains match).
Parameters
----------
name : str | None
Case-insensitive substring match on the name field.
name_exact : str | None
Exact match on the name field. Takes precedence over ``name`` if both
are provided.
created_by_email : str | None
Case-insensitive substring match on the creator's email.
created_after : str | None
ISO datetime string; return only records created after this time.
created_before : str | None
ISO datetime string; return only records created before this time.
updated_after : str | None
ISO datetime string; return only records updated after this time.
updated_before : str | None
ISO datetime string; return only records updated before this time.
order_by : str | None
Column to sort results by (e.g. ``"name"``, ``"created_at"``).
order : str | None
Sort direction: ``"asc"`` or ``"desc"``.
Returns
-------
dict[str, str]
Query parameters ready to pass to :func:`_build_endpoint`.
"""
params: dict[str, str] = {}
if name_exact is not None:
params["name"] = f"eq.{name_exact}"
elif name is not None:
params["name"] = f"ilike.%{name}%"
if created_by_email is not None:
params["created_by_email"] = f"ilike.%{created_by_email}%"
if created_after is not None:
params["created_at_gt"] = created_after
if created_before is not None:
params["created_at_lt"] = created_before
if updated_after is not None:
params["updated_at_gt"] = updated_after
if updated_before is not None:
params["updated_at_lt"] = updated_before
if order_by is not None:
params["order_by"] = order_by
if order is not None:
params["order"] = order
return params
def _extract_existing_id(e: IonworksError) -> str | None:
"""Extract ``existing_id`` from a CONFLICT error's detail payload.
The standardized error format nests the ID under
``{"detail": {"existing_id": "..."}}``. Returns ``None`` when the
field is missing or the payload has an unexpected shape.
"""
if e.data is None:
return None
detail = e.data.get("detail", {})
return detail.get("existing_id") if isinstance(detail, dict) else None
[docs]
def create_or_get( # noqa: UP047
*,
create: Callable[[], _T],
get_by_id: Callable[[str], _T],
find_by_name: Callable[[str], _T | None],
name: str | None,
resource_label: str,
) -> _T:
"""Create a resource, or return the existing one on a name conflict.
Shared conflict-resolution used by the equipment sub-clients. Calls
``create``; on an ``IonworksError`` that is a CONFLICT (``error_code ==
"CONFLICT"`` or HTTP 409), resolves the existing resource — first by the
``existing_id`` echoed in the error detail, then by ``find_by_name`` as a
fallback. Any non-conflict error propagates unchanged.
Parameters
----------
create : Callable[[], _T]
Zero-arg thunk that performs the create and returns the new resource.
get_by_id : Callable[[str], _T]
Fetch a resource by id (used with the conflict's ``existing_id``).
find_by_name : Callable[[str], _T | None]
Look up the existing resource by its (conflicting) name; returns
``None`` if not found.
name : str | None
The name that was being created, used for the fallback lookup and the
error message.
resource_label : str
Human-readable resource name for the "duplicate but not found" error
(e.g. ``"Cycler"``).
Returns
-------
_T
The newly created resource, or the pre-existing one on conflict.
Raises
------
ValueError
If the create reported a duplicate but the existing resource could not
be resolved by id or name.
"""
# Imported lazily to avoid a circular import at module load time.
from .errors import IonworksError
try:
return create()
except IonworksError as e:
if e.error_code == "CONFLICT" or e.status_code == 409:
existing_id = _extract_existing_id(e)
if existing_id:
return get_by_id(existing_id)
if name:
found = find_by_name(name)
if found is not None:
return found
raise ValueError(
f"{resource_label} '{name}' reported as duplicate but could "
"not be found"
) from e
raise
# --- Cell Specification Models --- #
[docs]
class CellSpecification(BaseModel):
"""Cell specification model - accepts any fields from the API.
The API returns nested component/material data and ratings objects.
This model is permissive to allow the API to define the schema.
"""
model_config = ConfigDict(extra="allow")
id: str
name: str
# --- Equipment Models --- #
[docs]
class Site(BaseModel):
"""Site model - accepts any fields from the API.
A site is an organization-scoped physical location that owns cyclers.
"""
model_config = ConfigDict(extra="allow")
id: str
name: str
[docs]
class Cycler(BaseModel):
"""Cycler model - accepts any fields from the API.
A cycler is battery test equipment that belongs to a site, is owned by
one project, and owns channels.
"""
model_config = ConfigDict(extra="allow")
id: str
name: str
site_id: str
project_id: str
#: Manufacturer (make) of the cycler, e.g. "Arbin". Nullable.
manufacturer: str | None = None
#: Model (type) of the cycler, e.g. "LBT-5V". Nullable.
model: str | None = None
#: Hardware/product version. Distinct from ``firmware_version``, which
#: changes without the hardware changing. Nullable.
version: str | None = None
#: Manufacturer serial number. Nullable.
serial_number: str | None = None
#: Firmware revision currently installed. Nullable.
firmware_version: str | None = None
#: DNS name of the control host on the lab network. Nullable.
hostname: str | None = None
#: IP address of the control host. Nullable.
ip_address: str | None = None
#: TCP port the control software listens on (1-65535). Nullable.
port: int | None = None
#: When the cycler was last calibrated. Null means never calibrated or not
#: tracked — read as unknown, not as overdue.
last_calibrated_at: datetime | None = None
#: Maximum time allowed between calibrations, in days. Null means the
#: cycler is not on a calibration schedule.
calibration_interval_days: int | None = None
#: When calibration next falls due: ``last_calibrated_at`` plus
#: ``calibration_interval_days``. Read-only — derived by the server, so it
#: cannot disagree with those two fields. Null when either is unset.
calibration_due_at: datetime | None = None
[docs]
class CyclerServiceEvent(BaseModel):
"""One instrument-level service on a cycler.
Calibration and preventive maintenance are performed on the *instrument*,
so a 40-channel cycler going in for its annual calibration is one event,
not 40 unrelated channel outages. Opening an event takes every channel out
of service together; completing it returns them together.
A cycler has at most one open event (``performed_at`` is None) at a time.
"""
model_config = ConfigDict(extra="allow")
id: str
cycler_id: str
project_id: str
#: One of "calibration", "preventive_maintenance", "firmware", "repair",
#: "other".
event_type: str
#: When the visit is booked. None on an event recorded after the fact.
scheduled_for: datetime | None = None
#: When the booked visit is expected to end, making the booking a span
#: rather than a start instant. None when no end was given. A plan, not a
#: guarantee — a visit may run past it.
scheduled_until: datetime | None = None
#: When the work was done. None means the event is still open.
performed_at: datetime | None = None
notes: str | None = None
created_by: str | None = None
performed_by: str | None = None
[docs]
class Channel(BaseModel):
"""Channel model - accepts any fields from the API.
A channel is an individual test channel belonging to a cycler; it
inherits its cycler's project.
"""
model_config = ConfigDict(extra="allow")
id: str
name: str
cycler_id: str
project_id: str
#: Free-text notes on the channel. Nullable.
notes: str | None = None
#: Whether the channel is deliberately out of service (broken /
#: maintenance). Shown as out-of-service in the Lab view.
out_of_commission: bool = False
#: Rated maximum current in amps (A). Nullable = unrated.
max_amps: float | None = None
#: Rated voltage window in volts (V). Nullable = unrated.
min_volts: float | None = None
max_volts: float | None = None
[docs]
class ChannelIncident(BaseModel):
"""One span during which a channel was out of service.
Opened when a channel is marked out of commission and closed when it
returns, so a channel's downtime is a history of spans rather than just the
current ``out_of_commission`` flag. A channel has at most one open incident
(``resolved_at`` is None) at a time.
An outage is either instrument-level or channel-local, and
``service_event_id`` is what tells them apart: it is set for every channel
taken down by a cycler calibration or PM visit, and None for a fault the
channel had on its own (a failed relay is not a cycler event).
"""
model_config = ConfigDict(extra="allow")
id: str
channel_id: str
project_id: str
#: The instrument-level service this outage belongs to, when it is one.
#: None for a channel-local fault.
service_event_id: str | None = None
#: One of "hardware_failure", "maintenance", "calibration",
#: "decommissioned", "other".
category: str = "other"
notes: str | None = None
#: When the booked service is due to happen. None on unplanned outages and
#: on backfilled rows, whose service was never scheduled.
service_scheduled_at: datetime | None = None
#: When the booked service is expected to end, making the outage a span
#: rather than a start instant. A plan, not a guarantee — an outage may run
#: past it, which is what :attr:`is_overdue_to_return` reports.
service_scheduled_until: datetime | None = None
#: When the channel went out of service.
started_at: datetime
#: When the channel returned. None means the outage is still open.
resolved_at: datetime | None = None
resolution_notes: str | None = None
started_by: str | None = None
resolved_by: str | None = None
#: True for backfilled rows whose ``started_at`` is derived from the
#: channel's last-modified time and is only an upper bound. Exclude these
#: from mean-time-to-repair arithmetic.
is_estimated: bool = False
@property
def is_overdue_to_return(self) -> bool:
"""Whether an open outage has run past the end its service was booked to.
Derived when you ask rather than stored, because overrunning a booked
window is a plan being wrong, not a record being invalid — the platform
deliberately accepts it. Only an *open* outage can be overdue: once the
channel is back it is back, however late.
Returns
-------
bool
True when the outage is still open, has a booked end, and that end
is in the past.
"""
if self.resolved_at is not None or self.service_scheduled_until is None:
return False
return self.service_scheduled_until < datetime.now(UTC)
# --- Lab view (occupancy) models --- #
[docs]
class ChannelState(StrEnum):
"""Derived occupancy state of a channel in the lab view."""
free = "free"
occupied = "occupied"
stale = "stale"
out_of_commission = "out_of_commission"
[docs]
class LabMeasurementSummary(BaseModel):
"""The measurement occupying a channel (slim view for the lab wall)."""
model_config = ConfigDict(extra="allow")
id: str
name: str
#: ISO time the test started, if known. Nullable.
start_time: str | None = None
#: ISO forecast finish time, if an estimate has been computed. Nullable.
#: Informational only -- doesn't affect the derived occupancy state or the
#: staleness window. The note explaining the estimate and the timestamp it
#: was computed at aren't part of this slim summary; fetch the full
#: measurement (``client.cell_measurement.get(measurement_id)``) for those.
estimated_end_time: str | None = None
#: ISO time the measurement was last updated (drives staleness).
updated_at: str
cell_instance_id: str
cell_instance_name: str | None = None
#: Name of the cell instance's cell specification, if known. Nullable.
cell_specification_name: str | None = None
protocol_name: str | None = None
#: Whether the current user is watching this measurement. A watch on the
#: live measurement reads as watching its channel (see ``LabClient.watch`` /
#: ``list_watched`` / ``watched_channels``).
watched: bool = False
[docs]
class LabChannel(BaseModel):
"""A channel with its derived occupancy state."""
model_config = ConfigDict(extra="allow")
id: str
name: str
state: ChannelState
#: The measurement driving the state (occupied/stale); None when free/OOC.
measurement: LabMeasurementSummary | None = None
notes: str | None = None
out_of_commission: bool = False
max_amps: float | None = None
min_volts: float | None = None
max_volts: float | None = None
[docs]
class LabCycler(BaseModel):
"""A cycler with its channels and per-cycler occupancy counts."""
model_config = ConfigDict(extra="allow")
id: str
name: str
manufacturer: str | None = None
model: str | None = None
channel_count: int
occupied: int
stale: int
free: int
out_of_commission: int = 0
channels: list[LabChannel] = []
[docs]
class LabSite(BaseModel):
"""A site grouping its cyclers for the lab wall."""
model_config = ConfigDict(extra="allow")
id: str
name: str
cyclers: list[LabCycler] = []
[docs]
class LabStatus(BaseModel):
"""The full lab-view tree for a project plus project-level counts."""
model_config = ConfigDict(extra="allow")
sites: list[LabSite] = []
occupied: int = 0
stale: int = 0
free: int = 0
out_of_commission: int = 0
[docs]
class FlatChannel(BaseModel):
"""A lab channel flattened with its parent cycler and site context.
Returned by ``LabClient.free_channels`` / ``stale_channels`` so an answer
reads as "CH3 on Maccor-1 at Boston Lab" without re-walking the tree.
"""
channel: LabChannel
cycler_id: str
cycler_name: str
site_id: str
site_name: str
[docs]
class Utilization(BaseModel):
"""Project lab utilization: the frontend headline percent plus raw counts.
``percent`` is the busy (occupied + stale) share of *all* channels
(out-of-commission included in the denominator), rounded to a whole
number to match the lab wall. ``0`` when there are no channels.
"""
percent: int
occupied: int
stale: int
free: int
out_of_commission: int
total: int
# --- Cell Instance Models --- #
[docs]
class CellInstance(BaseModel):
"""Cell instance model - accepts any fields from the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
cell_specification_id: str
# --- Cell Measurement Models --- #
[docs]
class MeasurementType(StrEnum):
"""Type of data stored in a cell measurement."""
time_series = "time_series"
file = "file"
properties = "properties"
[docs]
class CellMeasurement(BaseModel):
"""Cell measurement model - accepts any fields from the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
cell_instance_id: str
measurement_type: MeasurementType = MeasurementType.time_series
#: Optional channel the measurement was recorded on. Nullable. Only valid
#: on ``time_series`` measurements, and requires ``start_time`` to be set.
channel_id: str | None = None
#: ISO-formatted time the test started. Nullable, but required when
#: ``channel_id`` is set (it defines when the channel became occupied).
start_time: str | None = None
#: ISO-formatted time the test finished. Nullable — a null ``end_time``
#: means the test is still running (e.g. in the lab equipment view). Set it
#: once the measurement is complete. Must not precede ``start_time``.
end_time: str | None = None
#: ISO forecast finish time for a still-running test, if an estimate has
#: been computed. Nullable. Distinct from ``end_time`` (the actual
#: finish); informational only.
estimated_end_time: str | None = None
#: Free-text explanation of how ``estimated_end_time`` was derived.
#: Nullable.
estimated_end_time_note: str | None = None
#: ISO-formatted time ``estimated_end_time`` was last computed. Nullable.
estimated_end_time_calculated_at: str | None = None
#: Lifecycle of server-side step processing for uploaded ``time_series``
#: measurements: ``pending`` or ``running`` while the steps are still being
#: derived, ``ready`` once they are available, ``failed`` if the upload
#: could not be processed. ``file`` and ``properties`` measurements never
#: process steps and are always ``ready``.
processing_status: str | None = None
#: Why step processing failed, when ``processing_status`` is ``failed``.
processing_error: str | None = None
# --- Bundle Models --- #
[docs]
class CellMeasurementBundleResponse(CellMeasurement):
"""Flat response from creating a measurement bundle.
Measurement fields (id, name, measurement_type, etc.) are at the top level
alongside upload metadata.
"""
steps_created: int
[docs]
class UploadInfo(BaseModel):
"""Signed URL info for a single upload target."""
filename: str | None = None
signed_url: str
token: str
path: str
[docs]
class InitiateUploadResponse(BaseModel):
"""Response from the initiate-upload endpoint for signed URL uploads."""
measurement_id: str
uploads: list[UploadInfo]
[docs]
class InitiateRawDataUploadResponse(BaseModel):
"""Response from the raw-data initiate-upload endpoint."""
raw_data_id: str
uploads: list[UploadInfo]
# --- Detail Models --- #
[docs]
class CellMeasurementDetail(CellMeasurement):
"""Flat detail model for a measurement with steps and time series.
Measurement fields (id, name, measurement_type, etc.) are at the top level
alongside optional data payloads. Returns minimal data by default: foreign
keys for parent objects rather than nested objects. Use the spec/instance
clients to fetch parent objects if needed.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
specification_id: str | None = None
instance_id: str | None = None
steps: DataFrame | None = None
time_series: DataFrame | None = None
cycles: DataFrame | None = None
files: dict[str, bytes] | None = None
[docs]
@field_validator("steps", "time_series", "cycles", mode="before")
@classmethod
def convert_dict_to_df(cls, v: Any) -> Any:
"""Convert dictionary to DataFrame (polars or pandas based on config)."""
return dict_to_df_validator(v)
[docs]
class CellInstanceDetail(BaseModel):
"""Detail model for a cell instance with all measurements.
Returns a foreign key for the parent specification rather
than a nested object. Use ``client.cell_spec.get(detail
.specification_id)`` to fetch the full specification.
"""
instance: CellInstance
specification_id: str
measurements: list[CellMeasurementDetail]
# --- Equipment Detail Models --- #
[docs]
class CyclerDetail(BaseModel):
"""Detail model for a cycler with all its channels.
Returns a foreign key for the parent site rather than a nested object.
Use ``client.site.get(detail.site_id)`` to fetch the full site.
"""
cycler: Cycler
site_id: str
channels: list[Channel]
[docs]
class SiteDetail(BaseModel):
"""Detail model for a site with all its cyclers.
Each cycler is expanded to a :class:`CyclerDetail`, so its channels are
included too.
"""
site: Site
cyclers: list[CyclerDetail]
# --- Material Models --- #
[docs]
class ColumnSpec(BaseModel):
"""Column descriptor for a material property dataset."""
model_config = ConfigDict(extra="allow")
name: str
unit: str = ""
source_column_index: int
[docs]
class MaterialPropertyDataset(BaseModel):
"""Material property dataset record as returned by the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
columns: list[ColumnSpec] = []
data_version: int
nan_counts: dict[str, int] | None = None
created_at: str | None = None
source_pipeline_id: str | None = None
source_simple_pipeline_id: str | None = None
source_analysis_id: str | None = None
source_label: str | None = None
[docs]
class AnalysisType(StrEnum):
"""Well-known ``analysis_type`` values.
Members are plain strings (``StrEnum``), so they can be passed directly
wherever an ``analysis_type`` string is expected, e.g.
``client.analysis.create(..., analysis_type=AnalysisType.ECM_FROM_EIS)``.
This set is **advisory, not exhaustive**: the API does not restrict
``analysis_type`` to these values — any non-empty string is accepted, so a
new extractor can use a raw string without waiting for an SDK release.
"""
ECM_FROM_EIS = "ecm_from_eis"
LAM_LLI_FROM_RPT = "lam_lli_from_rpt"
DCIR_FROM_HPPC = "dcir_from_hppc"
#: The well-known analysis-type string values as a plain list, for iteration or
#: display. Derived from :class:`AnalysisType`; prefer the enum for authoring.
KNOWN_ANALYSIS_TYPES: list[str] = [t.value for t in AnalysisType]
[docs]
class AnalysisColumnSpec(BaseModel):
"""Column descriptor for an analysis parquet."""
model_config = ConfigDict(extra="allow")
name: str
unit: str = ""
dtype: str | None = None
[docs]
class Analysis(BaseModel):
"""Analysis record as returned by the API.
An analysis holds features extracted from a single ``cell_measurement``
(e.g. ECM parameters from EIS, LLI/LAM from RPT), stored as a parquet
table plus loose metadata.
"""
model_config = ConfigDict(extra="allow")
id: str
measurement_id: str
project_id: str | None = None
name: str
analysis_type: str
columns: list[AnalysisColumnSpec] = []
metadata: dict = {}
notes: str | None = None
source_pipeline_id: str | None = None
source_simple_pipeline_id: str | None = None
source_analysis_id: str | None = None
source_label: str | None = None
created_at: str | None = None
[docs]
class Material(BaseModel):
"""Material record as returned by the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
manufacturer: str | None = None
product_id: str | None = None
# --- Project Models --- #
[docs]
class Project(BaseModel):
"""Project model - accepts any fields from the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
organization_id: str
# --- Search Models --- #
[docs]
class SearchResult(BaseModel):
"""A single result from the global search endpoint.
Accepts any extra fields the API may add in future.
"""
model_config = ConfigDict(extra="allow")
id: str
entity_type: str
name: str
#: Project the entity belongs to (or a fallback project for org-scoped
#: entities). ``None`` when the API could not associate a project.
project_id: str | None = None
#: Parent entity ID needed for navigation (e.g. ``cell_specification_id``
#: for a ``cell_instance``). ``None`` when the ID alone is sufficient.
parent_id: str | None = None
[docs]
class SearchResponse(BaseModel):
"""Paginated response from the global search endpoint.
Iterates over :attr:`results` so callers can treat it like a list of
:class:`SearchResult` (``for hit in response``, ``response[0]``,
``len(response)``, ``if response:``) while still reading the pagination
metadata (:attr:`total`, :attr:`limit`, :attr:`offset`).
Notes
-----
Because :meth:`__iter__` yields :class:`SearchResult` objects (not
Pydantic's default ``(field_name, value)`` pairs), ``dict(response)``
does not work. Use :meth:`~pydantic.BaseModel.model_dump` to get a plain
dict of fields.
"""
model_config = ConfigDict(extra="allow")
results: list[SearchResult]
query: str
total: int
limit: int
offset: int
def __iter__(self) -> Iterator[SearchResult]: # type: ignore[override]
return iter(self.results)
def __len__(self) -> int:
return len(self.results)
def __getitem__(self, index: int) -> SearchResult:
return self.results[index]
def __bool__(self) -> bool:
return bool(self.results)
# --- Model (Custom Model) Models --- #
[docs]
class Model(BaseModel):
"""Custom model model - accepts any fields from the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
#: Model config (e.g. ``{"type": "SPMe"}``). The ``get`` response includes
#: it; the ``create`` response may omit it, in which case this is ``None``.
config: dict[str, Any] | None = None
#: Persistent simulation settings (mesh + solver) re-applied whenever the
#: model is simulated: a flat bag of ``var_pts`` / ``submesh_types`` /
#: ``spatial_methods`` / ``solver``. ``None`` uses the model defaults.
simulation_settings: dict[str, Any] | None = None
# --- Parameterized Model Models --- #
[docs]
class ParameterizedModel(BaseModel):
"""Parameterized model model - accepts any fields from the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
#: Parameter-specific persistent simulation settings (mesh + solver); takes
#: precedence over the base model's settings when the model is simulated.
#: ``None`` inherits the base model / defaults.
simulation_settings: dict[str, Any] | None = None
# --- Raw Data Models --- #
[docs]
class RawData(BaseModel):
"""A raw-data record.
Represents an original uploaded file stored as-is, scoped to an
organization and project. The API defines the full schema; extra fields
are accepted.
"""
model_config = ConfigDict(extra="allow")
id: str
project_id: str
name: str
filename: str
source: str | None = None
# --- Protocol Models --- #
[docs]
class Protocol(BaseModel):
"""A saved, project-scoped protocol.
Stored server-side as an ``experiment_template``: a named UCP protocol
plus the schema of any parameters it leaves open. Simulations and planned
measurements both reference a protocol by id rather than carrying the
protocol text, so the same protocol runs identically everywhere.
Heavy columns (``protocol_config``, ``parameters_schema``,
``source_protocol``, ...) are omitted from list responses unless requested
via ``include``; on a listed protocol they are ``None`` rather than empty.
Fetch the full record with :meth:`~ionworks.protocol.ProtocolClient.get`.
"""
model_config = ConfigDict(extra="allow")
id: str
name: str
organization_id: str
#: Project that owns the protocol. Every protocol reachable through the
#: API is project-scoped; ``None`` only appears on unmigrated legacy rows.
project_id: str | None = None
description: str | None = None
#: The protocol itself, in UCP form. ``None`` when not included in a list.
protocol_config: dict[str, Any] | None = None
#: Schema of the parameters the protocol leaves open (``input[...]``
#: references). ``None`` when not included in a list.
parameters_schema: dict[str, Any] | None = None
#: Original protocol text as the user entered it, when one was recorded.
source_protocol: str | None = None
created_by_email: str | None = None
[docs]
class ParsedProtocol(BaseModel):
"""Result of parsing a vendor protocol file into UCP.
Returned by :meth:`~ionworks.protocol.ProtocolClient.parse_file`. A parsed
protocol is not saved — pass ``ucp`` to
:meth:`~ionworks.protocol.ProtocolClient.create` to store it.
Drive cycles and subroutines referenced by the file may not be embedded in
it. Those the parser recovered are listed in ``available_drive_cycles`` /
``available_subroutines``; those still needed are in the ``required_``
lists and must be supplied before the protocol will simulate.
"""
model_config = ConfigDict(extra="allow", populate_by_name=True)
#: The parsed protocol as UCP YAML.
ucp: str = Field(alias="parsed_protocol_ucp")
#: Human-readable rendering of the same protocol.
human_readable: str = Field(default="", alias="parsed_protocol_human")
#: The original file's text, decoded with replacement for invalid bytes.
raw_protocol: str = ""
#: Detected cycler vendor, when the parser could identify one.
cycler_type: str | None = None
required_drive_cycles: list[str] = Field(default_factory=list)
available_drive_cycles: list[str] = Field(default_factory=list)
required_subroutines: list[str] = Field(default_factory=list)
available_subroutines: list[str] = Field(default_factory=list)
#: Set when the file parsed only partially; ``ucp`` may be incomplete.
error: str | None = None
# --- Study Models --- #
[docs]
class Study(BaseModel):
"""Study model - accepts any fields from the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str
# --- Planned Measurement Models --- #
[docs]
class PlannedMeasurementStatus(StrEnum):
"""Lifecycle state of a planned measurement.
A measurement is first ``requested``, then a scheduler assigns it a
channel and time window (``scheduled``). Once the real run starts it
becomes ``in_progress``, then ``completed``; it may be ``cancelled`` at
any point before completion.
"""
requested = "requested"
scheduled = "scheduled"
in_progress = "in_progress"
completed = "completed"
cancelled = "cancelled"
[docs]
class PlannedMeasurement(BaseModel):
"""Planned measurement model - accepts any fields from the API.
A planned measurement is a future, project-scoped test request. When
``status`` is ``scheduled`` it also reserves ``channel_id`` over the
``[planned_start_time, planned_end_time)`` window; a ``requested`` row
carries only an ``estimated_duration_seconds`` and no channel.
"""
model_config = ConfigDict(extra="allow")
id: str
name: str
project_id: str
status: PlannedMeasurementStatus
#: Named protocol (experiment_template) the measurement will run. Required on
#: create; resolve its display name via the linked protocol.
protocol_id: str | None = None
#: Cell specification the requester wants tested. Required on create.
cell_specification_id: str | None = None
#: Cell instance the future measurement will run on. Nullable.
cell_instance_id: str | None = None
#: Channel reserved once scheduled. Nullable until scheduled.
channel_id: str | None = None
#: Planned reservation window. Nullable until scheduled.
planned_start_time: datetime | None = None
planned_end_time: datetime | None = None
#: Expected run duration before scheduled times exist (requested rows).
estimated_duration_seconds: int | None = None
#: Email of the user who requested the plan (flattened from the embed).
requested_by_email: str | None = None
#: Email of the user who scheduled the plan; ``None`` until scheduled.
scheduled_by_email: str | None = None
[docs]
class AutoScheduleAssignment(BaseModel):
"""One proposed planned-measurement reservation.
A proposal may include an unscheduled assignment when the project has no
in-service channel able to take the test. Only assignments with
:attr:`is_scheduled` set should be passed to
:meth:`ionworks.planned_measurement.PlannedMeasurementClient.apply_auto_schedule_proposal`.
"""
planned_measurement_id: str
planned_measurement_name: str
planned_measurement_updated_at: datetime
channel_id: str | None = None
channel_name: str | None = None
#: Owning cycler's name. Channel names repeat across cyclers, so the pair
#: identifies the hardware being reserved.
cycler_name: str | None = None
planned_start_time: datetime | None = None
planned_end_time: datetime | None = None
unscheduled_reason: str | None = None
@property
def is_scheduled(self) -> bool:
"""Whether this assignment is complete and can be applied."""
return (
self.channel_id is not None
and self.planned_start_time is not None
and self.planned_end_time is not None
and self.unscheduled_reason is None
)
[docs]
class AutoScheduleProposal(BaseModel):
"""A transient earliest-gap proposal for selected requested tests."""
generated_at: datetime
assignments: list[AutoScheduleAssignment]
# --- Optimization Models --- #
[docs]
class Optimization(BaseModel):
"""Optimization model - accepts any fields from the API."""
model_config = ConfigDict(extra="allow")
id: str
name: str | None = None
job_id: str
project_id: str
# --- Organization Usage Models --- #
[docs]
class SimulationUsage(BaseModel):
"""Current simulation usage and its configured limit, in hours."""
model_config = ConfigDict(extra="allow")
#: Simulated battery time consumed this period, in hours.
usage: float = 0
#: Configured monthly limit in hours, or ``None`` when unconstrained.
limit: float | None = None
[docs]
class ComputeUsage(BaseModel):
"""Current compute usage and its configured limit, in hours.
``usage`` is the total across all job types; ``usage_by_type`` breaks it
down by job type (informational — the ``limit`` applies to the total).
"""
model_config = ConfigDict(extra="allow")
#: Total backend compute time consumed this period, in hours.
usage: float = 0
#: Compute time per job type (``simulation``, ``datafit``, ``optimization``,
#: ``validation``, ``pipeline``), in hours. Sums to ``usage``.
usage_by_type: dict[str, float] = {}
#: Configured monthly limit in hours, or ``None`` when unconstrained.
limit: float | None = None
[docs]
class OrganizationUsage(BaseModel):
"""An organization's usage and limits for the current billing period.
Returned by :meth:`~ionworks.organization.OrganizationClient.usage`. Usage
is aggregated across all members of the organization and resets on the first
of each month. Simulation usage is a single figure; compute usage carries a
per-job-type breakdown plus the total. All values are in hours. A ``None``
limit means that usage type is unconstrained.
"""
model_config = ConfigDict(extra="allow")
#: Start of the current billing period (inclusive).
period_start: datetime
#: End of the current billing period (exclusive) — the next reset.
period_end: datetime
simulation: SimulationUsage = SimulationUsage()
compute: ComputeUsage = ComputeUsage()
[docs]
class StepsAndCycles(BaseModel):
"""Steps and cycle metrics for a measurement.
Returned by the ``/steps_and_cycles`` endpoint which
fetches both in one call (cycles are derived from steps).
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
steps: DataFrame
cycles: DataFrame
[docs]
@field_validator("steps", "cycles", mode="before")
@classmethod
def convert_dict_to_df(cls, v: Any) -> Any:
"""Convert dictionary to DataFrame."""
return dict_to_df_validator(v)