IV Curve Diagnostics¶
IVCurveDiagnosticTypes(perfdb)
¶
Class used for handling IV Curve Diagnostic Types data. Can be accessed via perfdb.ivcurves.diagnostics.types.
Parameters:
Source code in echo_postgres/perfdb_root.py
def __init__(self, perfdb: e_pg.PerfDB) -> None:
"""Base class that all subclasses should inherit from.
Parameters
----------
perfdb : PerfDB
Top level object carrying all functionality and the connection handler.
"""
self._perfdb: e_pg.PerfDB = perfdb
get(names=None, output_type='dict')
¶
Retrieves IV Curve Diagnostic Types from the database.
The most useful keys/columns returned are:
- id
- name (index in case of DataFrame)
- description
Parameters:
-
(names¶list[str] | None, default:None) –List of IV Curve Diagnostic names to filter the retrieval.
If None, retrieves all IV Curve Diagnostics.
-
(output_type¶Literal['dict', 'DataFrame'], default:'dict') –Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"
Returns:
-
dict[str, dict[str, Any]]–In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
-
DataFrame–In case output_type is "pl.DataFrame", returns a Polars DataFrame
Source code in echo_postgres/ivcurve_diagnostic_types.py
@validate_call
def get(
self,
names: list[str] | None = None,
output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
"""Retrieves IV Curve Diagnostic Types from the database.
The most useful keys/columns returned are:
- id
- name (index in case of DataFrame)
- description
Parameters
----------
names : list[str] | None, optional
List of IV Curve Diagnostic names to filter the retrieval.
If None, retrieves all IV Curve Diagnostics.
output_type : Literal["dict", "DataFrame"], optional
Output type of the data. Can be one of ["dict", "DataFrame"]
By default "dict"
Returns
-------
dict[str, dict[str, Any]]
In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
pl.DataFrame
In case output_type is "pl.DataFrame", returns a Polars DataFrame
"""
where_query = (
sql.SQL("")
if not names
else sql.SQL(
"WHERE name = ANY({names})",
).format(
names=sql.Literal(names),
)
)
query = sql.SQL(
"SELECT * FROM performance.iv_curve_diagnostic_types {where_query} ORDER BY name",
).format(
where_query=where_query,
)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_polars(query)
if output_type == "pl.DataFrame":
return df
df = df.to_pandas(use_pyarrow_extension_array=True).set_index("name")
return df.to_dict(orient="index") if output_type == "dict" else df
IVCurveDiagnosticInstances(perfdb)
¶
Class used for handling IV Curve Diagnostics data. Can be accessed via perfdb.ivcurves.diagnostics.instances.
Parameters:
Source code in echo_postgres/perfdb_root.py
def __init__(self, perfdb: e_pg.PerfDB) -> None:
"""Base class that all subclasses should inherit from.
Parameters
----------
perfdb : PerfDB
Top level object carrying all functionality and the connection handler.
"""
self._perfdb: e_pg.PerfDB = perfdb
delete(ids=None, names=None)
¶
Deletes IV Curve Diagnostics from the database.
Parameters:
-
(ids¶list[int] | None, default:None) –List of IV Curve Diagnostic IDs to delete.
If None, no filtering by IDs is applied.
-
(names¶list[str] | None, default:None) –List of IV Curve Diagnostic names to delete.
If None, no filtering by names is applied.
Source code in echo_postgres/ivcurve_diagnostic_instances.py
@validate_call
def delete(
self,
ids: list[int] | None = None,
names: list[str] | None = None,
) -> None:
"""Deletes IV Curve Diagnostics from the database.
Parameters
----------
ids : list[int] | None, optional
List of IV Curve Diagnostic IDs to delete.
If None, no filtering by IDs is applied.
names : list[str] | None, optional
List of IV Curve Diagnostic names to delete.
If None, no filtering by names is applied.
"""
where_conditions = []
if ids:
where_conditions.append(
sql.SQL("id = ANY({ids})").format(
ids=sql.Literal(ids),
),
)
if names:
where_conditions.append(
sql.SQL("name = ANY({names})").format(
names=sql.Literal(names),
),
)
if len(where_conditions) == 0:
raise ValueError("At least one filter must be provided to delete IV Curve Diagnostics data.")
where_query = sql.SQL("WHERE ") + sql.SQL(" AND ").join(where_conditions) if where_conditions else sql.SQL("")
query = sql.SQL(
"DELETE FROM performance.iv_curve_diagnostics {where_query}",
).format(where_query=where_query)
with self._perfdb.conn.reconnect() as conn:
# deleting
conn.execute(query)
logger.debug(f"Deleted {conn.rowcount} IV Curve Diagnostics records from the database")
get(names=None, types=None, filter_type='and', output_type='dict')
¶
Retrieves IV Curve Diagnostics from the database.
The most useful keys/columns returned are:
- id
- name (index in case of DataFrame)
- description
- original_name
- type_id
- type_name
Parameters:
-
(names¶list[str] | None, default:None) –List of IV Curve Diagnostic names to filter the retrieval.
If None, retrieves all IV Curve Diagnostics.
-
(types¶list[str] | None, default:None) –List of IV Curve Diagnostic Types names to filter the retrieval.
If None, retrieves all IV Curve Diagnostic Types.
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"].
By default "and"
-
(output_type¶Literal['dict', 'DataFrame'], default:'dict') –Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"
Returns:
-
dict[str, dict[str, Any]]–In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
-
DataFrame–In case output_type is "pl.DataFrame", returns a Polars DataFrame
Source code in echo_postgres/ivcurve_diagnostic_instances.py
@validate_call
def get(
self,
names: list[str] | None = None,
types: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
"""Retrieves IV Curve Diagnostics from the database.
The most useful keys/columns returned are:
- id
- name (index in case of DataFrame)
- description
- original_name
- type_id
- type_name
Parameters
----------
names : list[str] | None, optional
List of IV Curve Diagnostic names to filter the retrieval.
If None, retrieves all IV Curve Diagnostics.
types : list[str] | None, optional
List of IV Curve Diagnostic Types names to filter the retrieval.
If None, retrieves all IV Curve Diagnostic Types.
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"].
By default "and"
output_type : Literal["dict", "DataFrame"], optional
Output type of the data. Can be one of ["dict", "DataFrame"]
By default "dict"
Returns
-------
dict[str, dict[str, Any]]
In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
pl.DataFrame
In case output_type is "pl.DataFrame", returns a Polars DataFrame
"""
where_query = []
if names:
where_query.append(
sql.SQL(
"name = ANY({names})",
).format(
names=sql.Literal(names),
),
)
if types:
where_query.append(
sql.SQL(
"type_name = ANY({types})",
).format(
types=sql.Literal(types),
),
)
where_query = sql.SQL(f" {filter_type.upper()} ").join(where_query) if where_query else sql.SQL("")
if where_query != sql.SQL(""):
where_query = sql.SQL("WHERE ") + where_query
query = sql.SQL(
"SELECT * FROM performance.v_iv_curve_diagnostics {where_query} ORDER BY name",
).format(
where_query=where_query,
)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_polars(query)
if output_type == "pl.DataFrame":
return df
df = df.to_pandas(use_pyarrow_extension_array=True).set_index("name")
return df.to_dict(orient="index") if output_type == "dict" else df
insert(name, type_name, description=None, original_name=None, on_conflict='ignore')
¶
Inserts IV Curve Diagnostics into the database.
Parameters:
-
(name¶str) –Name of the IV Curve Diagnostic.
-
(type_name¶str | None) –Name of the IV Curve Diagnostic Type (eg: Huawei, Internal)
-
(description¶str | None, default:None) –Description of the IV Curve Diagnostic.
-
(original_name¶str | None, default:None) –Original name of the IV Curve Diagnostic (used to match in report)
-
(on_conflict¶Literal['ignore', 'update'], default:'ignore') –Strategy to handle conflicts when inserting data. Can be one of:
- "ignore": ignores the new data if a conflict occurs (default)
- "update": updates the existing data with the new data in case of conflict
Returns:
-
int–ID of the inserted or updated IV Curve Diagnostic. Will be None if on_conflict is "ignore" and a conflict occurs.
Source code in echo_postgres/ivcurve_diagnostic_instances.py
@validate_call
def insert(
self,
name: str,
type_name: str,
description: str | None = None,
original_name: str | None = None,
on_conflict: Literal["ignore", "update"] = "ignore",
) -> int | None:
"""Inserts IV Curve Diagnostics into the database.
Parameters
----------
name : str
Name of the IV Curve Diagnostic.
type_name : str | None, optional
Name of the IV Curve Diagnostic Type (eg: Huawei, Internal)
description : str | None, optional
Description of the IV Curve Diagnostic.
original_name : str | None, optional
Original name of the IV Curve Diagnostic (used to match in report)
on_conflict : Literal["ignore", "update"], optional
Strategy to handle conflicts when inserting data. Can be one of:
- "ignore": ignores the new data if a conflict occurs (default)
- "update": updates the existing data with the new data in case of conflict
Returns
-------
int
ID of the inserted or updated IV Curve Diagnostic. Will be None if on_conflict is "ignore" and a conflict occurs.
"""
# getting type id
type_id = self._perfdb.ivcurves.diagnostics.types.get(
names=[type_name],
output_type="dict",
)
if type_name not in type_id:
raise ValueError(f"IV Curve Diagnostic Type '{type_name}' does not exist in the database.")
type_id = type_id[type_name]["id"]
query = sql.SQL(
"""
INSERT INTO performance.iv_curve_diagnostics (name, description, original_name, iv_curve_diagnostic_type_id)
VALUES ({name}, {description}, {original_name}, {type_id})
ON CONFLICT (name) DO {conflict_action}
RETURNING id
""",
).format(
name=sql.Literal(name),
description=sql.Literal(description),
original_name=sql.Literal(original_name),
type_id=sql.Literal(type_id),
conflict_action=sql.SQL(
"NOTHING"
if on_conflict == "ignore"
else "UPDATE SET description = EXCLUDED.description, original_name = EXCLUDED.original_name, iv_curve_diagnostic_type_id = EXCLUDED.iv_curve_diagnostic_type_id",
),
)
with self._perfdb.conn.reconnect() as conn:
result = conn.execute(query)
row = result.fetchone()
diagnostic_id = row[0] if row else None
logger.debug(f"IV Curve Diagnostic '{name}' inserted/updated in the database")
return diagnostic_id