"""
Cell instance client for managing individual cell records.
This module provides the :class:`CellInstanceClient` for creating, reading,
updating, and deleting cell instances, which represent individual physical
battery cells linked to a cell specification.
"""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
from typing import Any
import warnings
from .errors import IonworksError
from .models import (
CellInstance,
CellInstanceDetail,
CellMeasurement,
CellMeasurementDetail,
)
[docs]
class CellInstanceClient:
"""Client for managing cell instances."""
[docs]
def __init__(self, client: Any) -> None:
"""Initialize the CellInstanceClient.
Parameters
----------
client : Any
The HTTP client instance used for API calls.
"""
self.client = client
[docs]
def get(self, cell_instance_id: str) -> CellInstance:
"""Get a specific cell instance by ID."""
endpoint = f"/cell_instances/{cell_instance_id}"
response_data = self.client.get(endpoint)
return CellInstance(**response_data)
[docs]
def list(self, cell_spec_id: str) -> list[CellInstance]:
"""List all cell instances for a specification.
Parameters
----------
cell_spec_id : str
The ID of the cell specification.
Returns
-------
list[CellInstance]
All instances belonging to the specification.
"""
endpoint = f"/cell_specifications/{cell_spec_id}/cell_instances"
response_data = self.client.get(endpoint)
return [CellInstance(**item) for item in response_data]
[docs]
def update(
self,
cell_instance_id: str,
data: dict[str, Any],
) -> CellInstance:
"""Update an existing cell instance.
Parameters
----------
cell_instance_id : str
The ID of the cell instance to update.
data : dict[str, Any]
Dictionary containing the fields to update.
Returns
-------
CellInstance
The updated cell instance.
"""
endpoint = f"/cell_instances/{cell_instance_id}"
response_data = self.client.put(endpoint, data)
return CellInstance(**response_data)
[docs]
def delete(self, cell_instance_id: str) -> None:
"""Delete a cell instance by ID."""
endpoint = f"/cell_instances/{cell_instance_id}"
self.client.delete(endpoint)
[docs]
def detail(
self,
cell_instance_id: str,
include_steps: bool = True,
include_cycles: bool = True,
include_time_series: bool = True,
) -> CellInstanceDetail:
"""Get full details for a cell instance.
Composes flat API endpoints in parallel to build
the complete detail view. For each measurement,
fetches steps/cycles and time series according to
the ``include_*`` flags.
Parameters
----------
cell_instance_id : str
The cell instance ID.
include_steps : bool, optional
Whether to fetch step data for each measurement.
Defaults to True.
include_cycles : bool, optional
Whether to fetch cycle metrics for each
measurement. Defaults to True.
include_time_series : bool, optional
Whether to fetch time series for each
measurement. Defaults to True.
Returns
-------
CellInstanceDetail
Instance metadata, specification foreign key,
and a list of measurement details.
"""
# Step 1: fetch instance + measurement list in
# parallel
with ThreadPoolExecutor(max_workers=2) as pool:
inst_future = pool.submit(
self.client.get,
f"/cell_instances/{cell_instance_id}",
)
meas_future = pool.submit(
self.client.get,
f"/cell_instances/{cell_instance_id}/cell_measurements",
)
instance = CellInstance(**inst_future.result())
measurements_raw: list[dict] = meas_future.result()
measurements_meta = [CellMeasurement(**m) for m in measurements_raw]
# Step 2: for each measurement, fetch detail data
# in parallel using the measurement client
measurement_client = self.client.cell_measurement
measurement_details: list[CellMeasurementDetail] = []
if measurements_meta:
with ThreadPoolExecutor(max_workers=min(len(measurements_meta), 6)) as pool:
detail_futures = [
pool.submit(
measurement_client.detail,
m.id,
include_steps=include_steps,
include_cycles=include_cycles,
include_time_series=include_time_series,
)
for m in measurements_meta
]
measurement_details = [f.result() for f in detail_futures]
return CellInstanceDetail(
instance=instance,
specification_id=(instance.cell_specification_id),
measurements=measurement_details,
)
[docs]
def create(
self,
cell_spec_id: str,
data: dict[str, Any],
) -> CellInstance:
"""Create a new cell instance under a specification.
Parameters
----------
cell_spec_id : str
The ID of the parent cell specification.
data : dict[str, Any]
Dictionary containing the cell instance data.
Returns
-------
CellInstance
The newly created cell instance.
"""
endpoint = f"/cell_specifications/{cell_spec_id}/cell_instances"
response_data = self.client.post(endpoint, data)
return CellInstance(**response_data)
[docs]
def create_or_get(
self,
cell_spec_id: str,
data: dict[str, Any],
) -> CellInstance:
"""Create a new cell instance or get existing.
Parameters
----------
cell_spec_id : str
The ID of the parent cell specification.
data : dict[str, Any]
Dictionary containing the cell instance data.
Returns
-------
CellInstance
The cell instance (newly created or existing).
"""
try:
return self.create(cell_spec_id, data)
except IonworksError as e:
if e.error_code == "CONFLICT" or e.status_code == 409:
# Try to get existing instance by ID from error detail
if e.data is not None:
detail = e.data.get("detail", {})
existing_id = (
detail.get("existing_id") if isinstance(detail, dict) else None
)
if existing_id:
return self.get(existing_id)
# Deprecated: legacy error format fallback
legacy_id = e.data.get("existing_cell_instance_id")
if legacy_id:
warnings.warn(
"Received legacy error key "
"'existing_cell_instance_id'. "
"Update the backend to use the "
"standardized error format.",
DeprecationWarning,
stacklevel=2,
)
return self.get(legacy_id)
# Fall back to listing and matching by name
instance_name = data.get("name")
if instance_name:
for instance in self.list(cell_spec_id):
if instance.name == instance_name:
return instance
raise ValueError(
f"Cell instance '{instance_name}' reported as "
"duplicate but could not be found"
) from e
raise