Models¶
pyconvexity.models
¶
Model management module for PyConvexity.
Contains high-level operations for networks, components, and attributes.
Scenario
dataclass
¶
Represents a scenario (single network per database).
SolveResults
dataclass
¶
Represents solve results for a scenario.
YearlyResults
dataclass
¶
Represents yearly solve results.
Carrier
dataclass
¶
Represents an energy carrier in the network (single network per database).
get_component_type(conn: sqlite3.Connection, component_id: int) -> str
¶
Get component type for a component ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
ID of the component |
required |
Returns:
| Type | Description |
|---|---|
str
|
Component type string (e.g., "BUS", "GENERATOR") |
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
get_component(conn: sqlite3.Connection, component_id: int) -> Component
¶
Get component by ID (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
ID of the component |
required |
Returns:
| Type | Description |
|---|---|
Component
|
Component object with all fields populated |
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
list_components_by_type(conn: sqlite3.Connection, component_type: Optional[str] = None) -> List[Component]
¶
List components by type (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_type
|
Optional[str]
|
Optional component type filter (e.g., "BUS", "GENERATOR") |
None
|
Returns:
| Type | Description |
|---|---|
List[Component]
|
List of Component objects |
insert_component(conn: sqlite3.Connection, request: CreateComponentRequest) -> int
¶
Insert a new component (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
request
|
CreateComponentRequest
|
Component creation request with all necessary fields |
required |
Returns:
| Type | Description |
|---|---|
int
|
ID of the newly created component |
Raises:
| Type | Description |
|---|---|
DatabaseError
|
If insertion fails |
ValidationError
|
If required fields are missing |
create_component(conn: sqlite3.Connection, component_type: str, name: str, description: Optional[str] = None, longitude: Optional[float] = None, latitude: Optional[float] = None, carrier_id: Optional[int] = None, bus_id: Optional[int] = None, bus0_id: Optional[int] = None, bus1_id: Optional[int] = None) -> int
¶
Create a component and return its ID - convenience function (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_type
|
str
|
Type of component (e.g., "BUS", "GENERATOR") |
required |
name
|
str
|
Component name |
required |
description
|
Optional[str]
|
Optional description |
None
|
longitude
|
Optional[float]
|
Optional longitude coordinate |
None
|
latitude
|
Optional[float]
|
Optional latitude coordinate |
None
|
carrier_id
|
Optional[int]
|
Optional carrier ID |
None
|
bus_id
|
Optional[int]
|
Optional bus ID (for single-bus components) |
None
|
bus0_id
|
Optional[int]
|
Optional first bus ID (for two-bus components) |
None
|
bus1_id
|
Optional[int]
|
Optional second bus ID (for two-bus components) |
None
|
Returns:
| Type | Description |
|---|---|
int
|
ID of the newly created component |
update_component(conn: sqlite3.Connection, component_id: int, name: Optional[str] = None, description: Optional[str] = None, longitude: Optional[float] = None, latitude: Optional[float] = None, carrier_id: Optional[int] = None, bus_id: Optional[int] = None, bus0_id: Optional[int] = None, bus1_id: Optional[int] = None) -> None
¶
Update a component.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
ID of component to update |
required |
name
|
Optional[str]
|
New name (optional) |
None
|
description
|
Optional[str]
|
New description (optional) |
None
|
longitude
|
Optional[float]
|
New longitude (optional) |
None
|
latitude
|
Optional[float]
|
New latitude (optional) |
None
|
carrier_id
|
Optional[int]
|
New carrier ID (optional) |
None
|
bus_id
|
Optional[int]
|
New bus ID (optional) |
None
|
bus0_id
|
Optional[int]
|
New first bus ID (optional) |
None
|
bus1_id
|
Optional[int]
|
New second bus ID (optional) |
None
|
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
delete_component(conn: sqlite3.Connection, component_id: int) -> None
¶
Delete a component and all its attributes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
ID of component to delete |
required |
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
list_component_attributes(conn: sqlite3.Connection, component_id: int) -> List[str]
¶
List all attribute names for a component.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
Component ID |
required |
Returns:
| Type | Description |
|---|---|
List[str]
|
List of attribute names |
get_default_carrier_id(conn: sqlite3.Connection, component_type: str) -> int
¶
Get default carrier ID for a component type (single network per database).
get_bus_name_to_id_map(conn: sqlite3.Connection) -> Dict[str, int]
¶
Get mapping from bus names to component IDs (single network per database).
get_component_by_name(conn: sqlite3.Connection, name: str) -> Component
¶
Get a component by name (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
name
|
str
|
Component name |
required |
Returns:
| Type | Description |
|---|---|
Component
|
Component object |
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
get_component_id(conn: sqlite3.Connection, name: str) -> int
¶
Get component ID by name (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
name
|
str
|
Component name |
required |
Returns:
| Type | Description |
|---|---|
int
|
Component ID |
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
component_exists(conn: sqlite3.Connection, name: str) -> bool
¶
Check if a component exists (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
name
|
str
|
Component name |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if component exists, False otherwise |
get_component_carrier_map(conn: sqlite3.Connection, component_type: Optional[str] = None) -> Dict[str, str]
¶
Get mapping from component names to carrier names (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_type
|
Optional[str]
|
Optional component type filter |
None
|
Returns:
| Type | Description |
|---|---|
Dict[str, str]
|
Dictionary mapping component names to carrier names |
set_static_attribute(conn: sqlite3.Connection, component_id: int, attribute_name: str, value: StaticValue, scenario_id: Optional[int] = None) -> None
¶
Set a static attribute value for a component in a specific scenario.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
Component ID |
required |
attribute_name
|
str
|
Name of the attribute |
required |
value
|
StaticValue
|
Static value to set |
required |
scenario_id
|
Optional[int]
|
Scenario ID (uses master scenario if None) |
None
|
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
ValidationError
|
If attribute doesn't allow static values or validation fails |
set_timeseries_attribute(conn: sqlite3.Connection, component_id: int, attribute_name: str, timeseries: Union[Timeseries, List[float]], scenario_id: Optional[int] = None) -> None
¶
Set a timeseries attribute value for a component in a specific scenario.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
Component ID |
required |
attribute_name
|
str
|
Name of the attribute |
required |
timeseries
|
Union[Timeseries, List[float]]
|
Timeseries object or list of float values |
required |
scenario_id
|
Optional[int]
|
Scenario ID (uses master scenario if None) |
None
|
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
ValidationError
|
If attribute doesn't allow timeseries values |
TimeseriesError
|
If timeseries serialization fails |
get_attribute(conn: sqlite3.Connection, component_id: int, attribute_name: str, scenario_id: Optional[int] = None) -> AttributeValue
¶
Get an attribute value with scenario fallback logic.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
Component ID |
required |
attribute_name
|
str
|
Name of the attribute |
required |
scenario_id
|
Optional[int]
|
Scenario ID (uses master scenario if None) |
None
|
Returns:
| Type | Description |
|---|---|
AttributeValue
|
AttributeValue containing either static or timeseries data |
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
AttributeNotFound
|
If attribute doesn't exist |
delete_attribute(conn: sqlite3.Connection, component_id: int, attribute_name: str, scenario_id: Optional[int] = None) -> None
¶
Delete an attribute from a specific scenario.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
Component ID |
required |
attribute_name
|
str
|
Name of the attribute |
required |
scenario_id
|
Optional[int]
|
Scenario ID (uses master scenario if None) |
None
|
Raises:
| Type | Description |
|---|---|
AttributeNotFound
|
If attribute doesn't exist |
get_timeseries(conn: sqlite3.Connection, component_id: int, attribute_name: str, scenario_id: Optional[int] = None, start_index: Optional[int] = None, end_index: Optional[int] = None, max_points: Optional[int] = None) -> Timeseries
¶
Get timeseries data with unified interface matching Rust implementation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
Component ID |
required |
attribute_name
|
str
|
Name of the attribute |
required |
scenario_id
|
Optional[int]
|
Scenario ID (uses master scenario if None) |
None
|
start_index
|
Optional[int]
|
Start index for range queries |
None
|
end_index
|
Optional[int]
|
End index for range queries |
None
|
max_points
|
Optional[int]
|
Maximum number of points (for sampling) |
None
|
Returns:
| Type | Description |
|---|---|
Timeseries
|
Timeseries object with efficient array-based data |
Raises:
| Type | Description |
|---|---|
ComponentNotFound
|
If component doesn't exist |
AttributeNotFound
|
If attribute doesn't exist |
get_timeseries_metadata(conn: sqlite3.Connection, component_id: int, attribute_name: str, scenario_id: Optional[int] = None) -> TimeseriesMetadata
¶
Get timeseries metadata without loading the full data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
component_id
|
int
|
Component ID |
required |
attribute_name
|
str
|
Name of the attribute |
required |
scenario_id
|
Optional[int]
|
Scenario ID (uses master scenario if None) |
None
|
Returns:
| Type | Description |
|---|---|
TimeseriesMetadata
|
TimeseriesMetadata with length and type information |
create_network(conn: sqlite3.Connection, request: CreateNetworkRequest) -> None
¶
Create network metadata (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
request
|
CreateNetworkRequest
|
Network creation request |
required |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If required fields are missing |
DatabaseError
|
If creation fails |
get_network_info(conn: sqlite3.Connection) -> Dict[str, Any]
¶
Get network information (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dictionary with network information |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If network metadata doesn't exist |
get_network_time_periods(conn: sqlite3.Connection) -> List[TimePeriod]
¶
Get network time periods using optimized storage (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
List[TimePeriod]
|
List of TimePeriod objects ordered by period_index |
list_networks(conn: sqlite3.Connection) -> List[Dict[str, Any]]
¶
Get network information (returns single network in list for backward compatibility).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List with single network dictionary (for backward compatibility) |
create_carrier(conn: sqlite3.Connection, name: str, co2_emissions: float = 0.0, color: Optional[str] = None, nice_name: Optional[str] = None) -> int
¶
Create a carrier record and return carrier ID (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
name
|
str
|
Carrier name |
required |
co2_emissions
|
float
|
CO2 emissions factor |
0.0
|
color
|
Optional[str]
|
Display color |
None
|
nice_name
|
Optional[str]
|
Human-readable name |
None
|
Returns:
| Type | Description |
|---|---|
int
|
ID of the newly created carrier |
get_network_config(conn: sqlite3.Connection, scenario_id: Optional[int] = None) -> Dict[str, Any]
¶
Get network configuration with scenario-aware fallback (single network per database).
Priority order: 1. Scenario-specific config (network_config WHERE scenario_id = X) 2. Network default config (network_config WHERE scenario_id IS NULL) 3. System default value
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
scenario_id
|
Optional[int]
|
Optional scenario ID |
None
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dictionary with network configuration |
set_network_config(conn: sqlite3.Connection, param_name: str, param_value: Any, param_type: str, scenario_id: Optional[int] = None, description: Optional[str] = None) -> None
¶
Set network configuration parameter (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
param_name
|
str
|
Parameter name |
required |
param_value
|
Any
|
Parameter value |
required |
param_type
|
str
|
Parameter type ('boolean', 'real', 'integer', 'string', 'json') |
required |
scenario_id
|
Optional[int]
|
Optional scenario ID (NULL for base network) |
None
|
description
|
Optional[str]
|
Optional parameter description |
None
|
Raises:
| Type | Description |
|---|---|
ValidationError
|
If parameter type is invalid or serialization fails |
get_component_counts(conn: sqlite3.Connection) -> Dict[str, int]
¶
Get component counts by type (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
Dict[str, int]
|
Dictionary mapping component types to counts |
get_first_network(conn: sqlite3.Connection) -> Optional[Dict[str, Any]]
¶
Get network (for backward compatibility with single-network-per-database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
Optional[Dict[str, Any]]
|
Network dictionary or None if no network exists |
get_network_by_name(conn: sqlite3.Connection, name: str) -> Optional[Dict[str, Any]]
¶
Get network by name (for backward compatibility - checks if name matches).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
name
|
str
|
Network name to match |
required |
Returns:
| Type | Description |
|---|---|
Optional[Dict[str, Any]]
|
Network dictionary if name matches, None otherwise |
list_scenarios(conn: sqlite3.Connection) -> List[Scenario]
¶
List all scenarios (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
List[Scenario]
|
List of Scenario objects ordered by creation date |
get_scenario_by_name(conn: sqlite3.Connection, name: str) -> Scenario
¶
Get a scenario by name (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
name
|
str
|
Scenario name |
required |
Returns:
| Type | Description |
|---|---|
Scenario
|
Scenario object |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If scenario doesn't exist |
get_scenario_by_id(conn: sqlite3.Connection, scenario_id: int) -> Scenario
¶
Get a scenario by ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
scenario_id
|
int
|
Scenario ID |
required |
Returns:
| Type | Description |
|---|---|
Scenario
|
Scenario object |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If scenario doesn't exist |
get_solve_results(conn: sqlite3.Connection, scenario_id: Optional[int] = None) -> Optional[SolveResults]
¶
Get overall solve results for a scenario (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
scenario_id
|
Optional[int]
|
Scenario ID (NULL for base network) |
None
|
Returns:
| Type | Description |
|---|---|
Optional[SolveResults]
|
SolveResults object or None if no results found |
get_yearly_results(conn: sqlite3.Connection, scenario_id: Optional[int] = None) -> Dict[int, YearlyResults]
¶
Get year-by-year solve results for a scenario (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
scenario_id
|
Optional[int]
|
Scenario ID (NULL for base network) |
None
|
Returns:
| Type | Description |
|---|---|
Dict[int, YearlyResults]
|
Dictionary mapping years to YearlyResults objects |
list_carriers(conn: sqlite3.Connection) -> List[Carrier]
¶
List all carriers for the network (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
List[Carrier]
|
List of Carrier objects ordered by name |
get_carrier_by_name(conn: sqlite3.Connection, name: str) -> Carrier
¶
Get a carrier by name (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
name
|
str
|
Carrier name |
required |
Returns:
| Type | Description |
|---|---|
Carrier
|
Carrier object |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If carrier doesn't exist |
get_carrier_by_id(conn: sqlite3.Connection, carrier_id: int) -> Carrier
¶
Get a carrier by ID (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
carrier_id
|
int
|
Carrier ID |
required |
Returns:
| Type | Description |
|---|---|
Carrier
|
Carrier object |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If carrier doesn't exist |
get_carrier_colors(conn: sqlite3.Connection) -> Dict[str, str]
¶
Get carrier colors for visualization (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Connection
|
Database connection |
required |
Returns:
| Type | Description |
|---|---|
Dict[str, str]
|
Dictionary mapping carrier names to color strings |
get_scenario(conn, scenario_id)
¶
Backward compatible alias for get_scenario_by_id.
create_scenario(conn, name, description=None, probability=None)
¶
Create a new scenario (single network per database).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Database connection |
required | |
name
|
Scenario name |
required | |
description
|
Optional description |
None
|
|
probability
|
Optional probability for stochastic optimization |
None
|
Returns:
| Type | Description |
|---|---|
|
Scenario ID |
delete_scenario(conn, scenario_id)
¶
Delete a scenario.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conn
|
Database connection |
required | |
scenario_id
|
Scenario ID to delete |
required |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If scenario doesn't exist |