IV Curve Report Details¶
IVCurveReportDetails(perfdb)
¶
Class used for handling IV Curve Report Details data. Can be accessed via perfdb.ivcurves.report.details.
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(iv_curve_report_ids=None, iv_curve_report_detail_ids=None, timestamps=None, period=None, ts_names=None, inverter_names=None, string_numbers=None, source_diagnostics=None, echo_diagnostics=None, include_null_diagnostics=False, service_order_names=None, service_order_sap_ids=None, service_order_note_sap_ids=None, service_order_status_names=None, service_order_note_status_names=None, filter_type='and')
¶
Deletes IV Curve Report Details from the database based on provided filters.
Be cautious when using this method, as it will permanently remove data from the database.
Parameters:
-
(iv_curve_report_ids¶list[int] | None, default:None) –List of IV Curve Report IDs to filter the retrieval.
If None, no filtering is applied.
-
(iv_curve_report_detail_ids¶list[int] | None, default:None) –List of IV Curve Report Detail IDs to filter the retrieval.
If None, no filtering is applied.
-
(timestamps¶list[datetime] | None, default:None) –List of timestamps to filter the retrieval.
If None, no filtering is applied.
-
(period¶DateTimeRange | None, default:None) –DateTimeRange to filter the retrieval based on timestamp. Different from previous
timestampsparameter which is a list of specific timestamps this parameter filters a range of timestamps based on start and end datetime.If None, no filtering is applied.
-
(ts_names¶list[str] | None, default:None) –List of Transformer Station names to filter the retrieval.
If None, no filtering is applied.
-
(inverter_names¶list[str] | None, default:None) –List of Inverter names to filter the retrieval.
If None, no filtering is applied.
-
(string_numbers¶list[int] | None, default:None) –List of string numbers to filter the retrieval.
If None, no filtering is applied.
-
(source_diagnostics¶list[str] | None, default:None) –List of source diagnostics to filter the retrieval.
If None, no filtering is applied.
-
(echo_diagnostics¶list[str] | None, default:None) –List of echo diagnostics to filter the retrieval.
If None, no filtering is applied.
-
(include_null_diagnostics¶bool, default:False) –Whether to include records with null diagnostics in the retrieval. By default False.
-
(service_order_names¶list[str] | None, default:None) –List of service order names to filter the retrieval.
If None, no filtering is applied.
-
(service_order_sap_ids¶list[str] | None, default:None) –List of service order SAP IDs to filter the retrieval.
If None, no filtering is applied.
-
(service_order_note_sap_ids¶list[str] | None, default:None) –List of service order note SAP IDs to filter the retrieval.
If None, no filtering is applied.
-
(service_order_status_names¶list[str] | None, default:None) –List of service order status names to filter the retrieval.
If None, no filtering is applied.
-
(service_order_note_status_names¶list[str] | None, default:None) –List of service order note status names to filter the retrieval.
If None, no filtering is applied.
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"]. By default "and".
Source code in echo_postgres/ivcurve_report_details.py
@validate_call
def delete(
self,
iv_curve_report_ids: list[int] | None = None,
iv_curve_report_detail_ids: list[int] | None = None,
timestamps: list[datetime] | None = None,
period: DateTimeRange | None = None,
ts_names: list[str] | None = None,
inverter_names: list[str] | None = None,
string_numbers: list[int] | None = None,
source_diagnostics: list[str] | None = None,
echo_diagnostics: list[str] | None = None,
include_null_diagnostics: bool = False,
service_order_names: list[str] | None = None,
service_order_sap_ids: list[str] | None = None,
service_order_note_sap_ids: list[str] | None = None,
service_order_status_names: list[str] | None = None,
service_order_note_status_names: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
) -> None:
"""Deletes IV Curve Report Details from the database based on provided filters.
Be cautious when using this method, as it will permanently remove data from the database.
Parameters
----------
iv_curve_report_ids : list[int] | None, optional
List of IV Curve Report IDs to filter the retrieval.
If None, no filtering is applied.
iv_curve_report_detail_ids : list[int] | None, optional
List of IV Curve Report Detail IDs to filter the retrieval.
If None, no filtering is applied.
timestamps : list[datetime] | None, optional
List of timestamps to filter the retrieval.
If None, no filtering is applied.
period: DateTimeRange | None, optional
DateTimeRange to filter the retrieval based on timestamp. Different from previous `timestamps` parameter which is a list of specific timestamps this parameter filters a range of timestamps based on start and end datetime.
If None, no filtering is applied.
ts_names : list[str] | None, optional
List of Transformer Station names to filter the retrieval.
If None, no filtering is applied.
inverter_names : list[str] | None, optional
List of Inverter names to filter the retrieval.
If None, no filtering is applied.
string_numbers : list[int] | None, optional
List of string numbers to filter the retrieval.
If None, no filtering is applied.
source_diagnostics : list[str] | None, optional
List of source diagnostics to filter the retrieval.
If None, no filtering is applied.
echo_diagnostics : list[str] | None, optional
List of echo diagnostics to filter the retrieval.
If None, no filtering is applied.
include_null_diagnostics : bool, optional
Whether to include records with null diagnostics in the retrieval. By default False.
service_order_names : list[str] | None, optional
List of service order names to filter the retrieval.
If None, no filtering is applied.
service_order_sap_ids : list[str] | None, optional
List of service order SAP IDs to filter the retrieval.
If None, no filtering is applied.
service_order_note_sap_ids : list[str] | None, optional
List of service order note SAP IDs to filter the retrieval.
If None, no filtering is applied.
service_order_status_names : list[str] | None, optional
List of service order status names to filter the retrieval.
If None, no filtering is applied.
service_order_note_status_names : list[str] | None, optional
List of service order note status names to filter the retrieval.
If None, no filtering is applied.
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"].
By default "and".
"""
# checking if at least one filter is provided to avoid deleting all data
if not any(
[
iv_curve_report_ids,
iv_curve_report_detail_ids,
timestamps,
period,
ts_names,
inverter_names,
string_numbers,
source_diagnostics,
echo_diagnostics,
service_order_names,
service_order_sap_ids,
service_order_note_sap_ids,
service_order_status_names,
service_order_note_status_names,
],
):
raise ValueError("At least one filter must be provided to delete IV Curve Report Details.")
# getting ids to delete
ids_to_delete = self.get_ids(
iv_curve_report_ids=iv_curve_report_ids,
iv_curve_report_detail_ids=iv_curve_report_detail_ids,
timestamps=timestamps,
period=period,
ts_names=ts_names,
inverter_names=inverter_names,
string_numbers=string_numbers,
source_diagnostics=source_diagnostics,
echo_diagnostics=echo_diagnostics,
include_null_diagnostics=include_null_diagnostics,
service_order_names=service_order_names,
service_order_sap_ids=service_order_sap_ids,
service_order_note_sap_ids=service_order_note_sap_ids,
service_order_status_names=service_order_status_names,
service_order_note_status_names=service_order_note_status_names,
filter_type=filter_type,
)
if not ids_to_delete:
logger.debug("No IV Curve Report Details found matching the provided filters. No data deleted.")
return
query = sql.SQL(
"DELETE FROM performance.iv_curve_report_details WHERE id = ANY({ids_to_delete})",
).format(
ids_to_delete=sql.Literal([id_ for ids in ids_to_delete.values() for id_ in ids]),
)
with self._perfdb.conn.reconnect() as conn:
conn.execute(query)
logger.debug(
f"Deleted {conn.rowcount} IV Curve Report Details from the database matching the provided filters.",
)
get(iv_curve_report_ids=None, iv_curve_report_detail_ids=None, timestamps=None, period=None, ts_names=None, inverter_names=None, string_numbers=None, source_diagnostics=None, echo_diagnostics=None, include_null_diagnostics=False, service_order_names=None, service_order_sap_ids=None, service_order_note_sap_ids=None, service_order_status_names=None, service_order_note_status_names=None, get_curve_array=False, filter_type='and', output_type='dict')
¶
Retrieves IV Curve Details information from the database.
The most useful keys/columns returned are:
- id (will be the index in case of DataFrame)
- timestamp
- iv_curve_report_id
- document_id
- ts_id
- ts_name
- inverter_id
- inverter_name
- string_number
- irradiance
- module_temperature
- v_oc
- i_sc
- source_diagnostic_id
- source_diagnostic
- echo_diagnostic_id
- echo_diagnostic
- service_order_id
- service_order_name
- service_order_sap_id
- service_order_status_name
- service_order_note_id
- service_order_note_sap_id
- service_order_note_status_name
- iv_curve_array (if get_curve_array is True)
Parameters:
-
(iv_curve_report_ids¶list[int] | None, default:None) –List of IV Curve Report IDs to filter the retrieval.
If None, no filtering is applied.
-
(iv_curve_report_detail_ids¶list[int] | None, default:None) –List of IV Curve Report Detail IDs to filter the retrieval.
If None, no filtering is applied.
-
(timestamps¶list[datetime] | None, default:None) –List of timestamps to filter the retrieval.
If None, no filtering is applied.
-
(period¶DateTimeRange, default:None) –DateTimeRange to filter the retrieval based on timestamp. Different from previous
timestampsparameter which is a list of specific timestamps this parameter filters a range of timestamps based on start and end datetime.If None, no filtering is applied.
-
(ts_names¶list[str] | None, default:None) –List of Transformer Station names to filter the retrieval.
If None, no filtering is applied.
-
(inverter_names¶list[str] | None, default:None) –List of Inverter names to filter the retrieval.
If None, no filtering is applied.
-
(string_numbers¶list[int] | None, default:None) –List of string numbers to filter the retrieval.
If None, no filtering is applied.
-
(source_diagnostics¶list[str] | None, default:None) –List of source diagnostics to filter the retrieval.
If None, no filtering is applied.
-
(echo_diagnostics¶list[str] | None, default:None) –List of echo diagnostics to filter the retrieval.
If None, no filtering is applied.
-
(include_null_diagnostics¶bool, default:False) –Whether to include records with null diagnostics in the retrieval. By default False.
-
(service_order_names¶list[str] | None, default:None) –List of service order names to filter the retrieval.
If None, no filtering is applied.
-
(service_order_sap_ids¶list[str] | None, default:None) –List of service order SAP IDs to filter the retrieval.
If None, no filtering is applied.
-
(service_order_note_sap_ids¶list[str] | None, default:None) –List of service order note SAP IDs to filter the retrieval.
If None, no filtering is applied.
-
(service_order_status_names¶list[str] | None, default:None) –List of service order status names to filter the retrieval.
If None, no filtering is applied.
-
(service_order_note_status_names¶list[str] | None, default:None) –List of service order note status names to filter the retrieval.
If None, no filtering is applied.
-
(get_curve_array¶bool, default:False) –Whether to include the IV curve array in the output. By default False.
-
(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', 'pl.DataFrame'], default:'dict') –Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"] By default "dict"
Returns:
-
dict[int, dict[str, Any]]–In case output_type is "dict", returns a dictionary in the format {id: {attribute: value, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = id, columns = [attribute, ...]
-
DataFrame–In case output_type is "pl.DataFrame", returns a Polars DataFrame
Source code in echo_postgres/ivcurve_report_details.py
@validate_call
def get(
self,
iv_curve_report_ids: list[int] | None = None,
iv_curve_report_detail_ids: list[int] | None = None,
timestamps: list[datetime] | None = None,
period: DateTimeRange | None = None,
ts_names: list[str] | None = None,
inverter_names: list[str] | None = None,
string_numbers: list[int] | None = None,
source_diagnostics: list[str] | None = None,
echo_diagnostics: list[str] | None = None,
include_null_diagnostics: bool = False,
service_order_names: list[str] | None = None,
service_order_sap_ids: list[str] | None = None,
service_order_note_sap_ids: list[str] | None = None,
service_order_status_names: list[str] | None = None,
service_order_note_status_names: list[str] | None = None,
get_curve_array: bool = False,
filter_type: Literal["and", "or"] = "and",
output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[int, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
"""Retrieves IV Curve Details information from the database.
The most useful keys/columns returned are:
- id (will be the index in case of DataFrame)
- timestamp
- iv_curve_report_id
- document_id
- ts_id
- ts_name
- inverter_id
- inverter_name
- string_number
- irradiance
- module_temperature
- v_oc
- i_sc
- source_diagnostic_id
- source_diagnostic
- echo_diagnostic_id
- echo_diagnostic
- service_order_id
- service_order_name
- service_order_sap_id
- service_order_status_name
- service_order_note_id
- service_order_note_sap_id
- service_order_note_status_name
- iv_curve_array (if get_curve_array is True)
Parameters
----------
iv_curve_report_ids : list[int] | None, optional
List of IV Curve Report IDs to filter the retrieval.
If None, no filtering is applied.
iv_curve_report_detail_ids : list[int] | None, optional
List of IV Curve Report Detail IDs to filter the retrieval.
If None, no filtering is applied.
timestamps : list[datetime] | None, optional
List of timestamps to filter the retrieval.
If None, no filtering is applied.
period : DateTimeRange
DateTimeRange to filter the retrieval based on timestamp. Different from previous `timestamps` parameter which is a list of specific timestamps this parameter filters a range of timestamps based on start and end datetime.
If None, no filtering is applied.
ts_names : list[str] | None, optional
List of Transformer Station names to filter the retrieval.
If None, no filtering is applied.
inverter_names : list[str] | None, optional
List of Inverter names to filter the retrieval.
If None, no filtering is applied.
string_numbers : list[int] | None, optional
List of string numbers to filter the retrieval.
If None, no filtering is applied.
source_diagnostics : list[str] | None, optional
List of source diagnostics to filter the retrieval.
If None, no filtering is applied.
echo_diagnostics : list[str] | None, optional
List of echo diagnostics to filter the retrieval.
If None, no filtering is applied.
include_null_diagnostics : bool, optional
Whether to include records with null diagnostics in the retrieval. By default False.
service_order_names : list[str] | None, optional
List of service order names to filter the retrieval.
If None, no filtering is applied.
service_order_sap_ids : list[str] | None, optional
List of service order SAP IDs to filter the retrieval.
If None, no filtering is applied.
service_order_note_sap_ids : list[str] | None, optional
List of service order note SAP IDs to filter the retrieval.
If None, no filtering is applied.
service_order_status_names : list[str] | None, optional
List of service order status names to filter the retrieval.
If None, no filtering is applied.
service_order_note_status_names : list[str] | None, optional
List of service order note status names to filter the retrieval.
If None, no filtering is applied.
get_curve_array : bool, optional
Whether to include the IV curve array in the output. By default False.
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", "pl.DataFrame"], optional
Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"]
By default "dict"
Returns
-------
dict[int, dict[str, Any]]
In case output_type is "dict", returns a dictionary in the format {id: {attribute: value, ...}, ...}
pd.DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = id, columns = [attribute, ...]
pl.DataFrame
In case output_type is "pl.DataFrame", returns a Polars DataFrame
"""
where_query = self._check_get_args(
iv_curve_report_ids=iv_curve_report_ids,
iv_curve_report_detail_ids=iv_curve_report_detail_ids,
ts_names=ts_names,
inverter_names=inverter_names,
timestamps=timestamps,
period=period,
string_numbers=string_numbers,
source_diagnostics=source_diagnostics,
include_null_diagnostics=include_null_diagnostics,
echo_diagnostics=echo_diagnostics,
service_order_names=service_order_names,
service_order_sap_ids=service_order_sap_ids,
service_order_note_sap_ids=service_order_note_sap_ids,
service_order_status_names=service_order_status_names,
service_order_note_status_names=service_order_note_status_names,
filter_type=filter_type,
)
query = sql.SQL(
"""SELECT
id,
"timestamp" :: TIMESTAMP,
iv_curve_report_id,
document_id,
ts_id,
ts_name,
inverter_id,
inverter_name,
string_number,
irradiance,
module_temperature,
v_oc,
i_sc,
source_diagnostic_id,
source_diagnostic,
echo_diagnostic_id,
echo_diagnostic,
service_order_id,
service_order_name,
service_order_sap_id,
service_order_status_name,
service_order_note_id,
service_order_note_sap_id,
service_order_note_status_name,
{iv_curve_array}
modified_date
FROM performance.v_iv_curve_report_details {where_query} ORDER BY iv_curve_report_id, ts_id, inverter_id, string_number""",
).format(
where_query=where_query,
iv_curve_array=sql.SQL("iv_curve_array,") if get_curve_array else sql.SQL(""),
)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_polars(
query,
# status names might be all null, so lets force them to Utf8
schema_overrides={"service_order_status_name": pl.Utf8, "service_order_note_status_name": pl.Utf8},
)
if output_type == "pl.DataFrame":
return df
df = df.to_pandas(use_pyarrow_extension_array=True)
df = df.set_index("id")
if output_type == "DataFrame":
return df
return df.to_dict(orient="index")
get_ids(iv_curve_report_ids=None, iv_curve_report_detail_ids=None, timestamps=None, period=None, ts_names=None, inverter_names=None, string_numbers=None, source_diagnostics=None, echo_diagnostics=None, include_null_diagnostics=False, service_order_names=None, service_order_sap_ids=None, service_order_note_sap_ids=None, service_order_status_names=None, service_order_note_status_names=None, filter_type='and')
¶
Retrieves IV Curve Report Detail IDs from the database based on provided filters.
Parameters:
-
(iv_curve_report_ids¶list[int] | None, default:None) –List of IV Curve Report IDs to filter the retrieval.
If None, no filtering is applied.
-
(iv_curve_report_detail_ids¶list[int] | None, default:None) –List of IV Curve Report Detail IDs to filter the retrieval.
If None, no filtering is applied.
-
(timestamps¶list[datetime] | None, default:None) –List of timestamps to filter the retrieval.
If None, no filtering is applied.
-
(ts_names¶list[str] | None, default:None) –List of Transformer Station names to filter the retrieval.
If None, no filtering is applied.
-
(period¶DateTimeRange | None, default:None) –DateTimeRange to filter the retrieval based on timestamp. Different from previous
timestampsparameter which is a list of specific timestamps this parameter filters a range of timestamps based on start and end datetime.If None, no filtering is applied.
-
(inverter_names¶list[str] | None, default:None) –List of Inverter names to filter the retrieval.
If None, no filtering is applied.
-
(string_numbers¶list[int] | None, default:None) –List of string numbers to filter the retrieval.
If None, no filtering is applied.
-
(source_diagnostics¶list[str] | None, default:None) –List of source diagnostics to filter the retrieval.
If None, no filtering is applied.
-
(echo_diagnostics¶list[str] | None, default:None) –List of echo diagnostics to filter the retrieval.
If None, no filtering is applied.
-
(include_null_diagnostics¶bool, default:False) –Whether to include records with null diagnostics in the retrieval. By default False.
-
(service_order_names¶list[str] | None, default:None) –List of service order names to filter the retrieval.
If None, no filtering is applied.
-
(service_order_sap_ids¶list[str] | None, default:None) –List of service order SAP IDs to filter the retrieval.
If None, no filtering is applied.
-
(service_order_note_sap_ids¶list[str] | None, default:None) –List of service order note SAP IDs to filter the retrieval.
If None, no filtering is applied.
-
(service_order_status_names¶list[str] | None, default:None) –List of service order status names to filter the retrieval.
If None, no filtering is applied.
-
(service_order_note_status_names¶list[str] | None, default:None) –List of service order note status names to filter the retrieval.
If None, no filtering is applied.
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"]. By default "and".
Returns:
-
dict[int, list[int]]–Dictionary in the format {iv_curve_report_id: [detail_id1, detail_id2, ...], ...}
Source code in echo_postgres/ivcurve_report_details.py
@validate_call
def get_ids(
self,
iv_curve_report_ids: list[int] | None = None,
iv_curve_report_detail_ids: list[int] | None = None,
timestamps: list[datetime] | None = None,
period: DateTimeRange | None = None,
ts_names: list[str] | None = None,
inverter_names: list[str] | None = None,
string_numbers: list[int] | None = None,
source_diagnostics: list[str] | None = None,
echo_diagnostics: list[str] | None = None,
include_null_diagnostics: bool = False,
service_order_names: list[str] | None = None,
service_order_sap_ids: list[str] | None = None,
service_order_note_sap_ids: list[str] | None = None,
service_order_status_names: list[str] | None = None,
service_order_note_status_names: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
) -> dict[int, list[int]]:
"""Retrieves IV Curve Report Detail IDs from the database based on provided filters.
Parameters
----------
iv_curve_report_ids : list[int] | None, optional
List of IV Curve Report IDs to filter the retrieval.
If None, no filtering is applied.
iv_curve_report_detail_ids : list[int] | None, optional
List of IV Curve Report Detail IDs to filter the retrieval.
If None, no filtering is applied.
timestamps : list[datetime] | None, optional
List of timestamps to filter the retrieval.
If None, no filtering is applied.
ts_names : list[str] | None, optional
List of Transformer Station names to filter the retrieval.
If None, no filtering is applied.
period: DateTimeRange | None, optional
DateTimeRange to filter the retrieval based on timestamp. Different from previous `timestamps` parameter which is a list of specific timestamps this parameter filters a range of timestamps based on start and end datetime.
If None, no filtering is applied.
inverter_names : list[str] | None, optional
List of Inverter names to filter the retrieval.
If None, no filtering is applied.
string_numbers : list[int] | None, optional
List of string numbers to filter the retrieval.
If None, no filtering is applied.
source_diagnostics : list[str] | None, optional
List of source diagnostics to filter the retrieval.
If None, no filtering is applied.
echo_diagnostics : list[str] | None, optional
List of echo diagnostics to filter the retrieval.
If None, no filtering is applied.
include_null_diagnostics : bool, optional
Whether to include records with null diagnostics in the retrieval. By default False.
service_order_names : list[str] | None, optional
List of service order names to filter the retrieval.
If None, no filtering is applied.
service_order_sap_ids : list[str] | None, optional
List of service order SAP IDs to filter the retrieval.
If None, no filtering is applied.
service_order_note_sap_ids : list[str] | None, optional
List of service order note SAP IDs to filter the retrieval.
If None, no filtering is applied.
service_order_status_names : list[str] | None, optional
List of service order status names to filter the retrieval.
If None, no filtering is applied.
service_order_note_status_names : list[str] | None, optional
List of service order note status names to filter the retrieval.
If None, no filtering is applied.
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"].
By default "and".
Returns
-------
dict[int, list[int]]
Dictionary in the format {iv_curve_report_id: [detail_id1, detail_id2, ...], ...}
"""
where_query = self._check_get_args(
iv_curve_report_ids=iv_curve_report_ids,
iv_curve_report_detail_ids=iv_curve_report_detail_ids,
ts_names=ts_names,
inverter_names=inverter_names,
timestamps=timestamps,
period=period,
string_numbers=string_numbers,
source_diagnostics=source_diagnostics,
include_null_diagnostics=include_null_diagnostics,
echo_diagnostics=echo_diagnostics,
service_order_names=service_order_names,
service_order_sap_ids=service_order_sap_ids,
service_order_note_sap_ids=service_order_note_sap_ids,
service_order_status_names=service_order_status_names,
service_order_note_status_names=service_order_note_status_names,
filter_type=filter_type,
)
query = sql.SQL(
"SELECT iv_curve_report_id, id FROM performance.v_iv_curve_report_details {where_query} ORDER BY iv_curve_report_id, ts_id, inverter_id, string_number",
).format(
where_query=where_query,
)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_polars(query)
df = df.group_by("iv_curve_report_id").agg(pl.col("id")).sort("iv_curve_report_id")
result = dict(zip(df["iv_curve_report_id"].to_list(), df["id"].to_list(), strict=False))
return result
insert(iv_curve_report_id=None, inverter_name=None, string_number=None, service_order_note_sap_id=None, v_oc=None, i_sc=None, iv_curve_array=None, source_diagnostic=None, echo_diagnostic=None, data_df=None, create_diagnostic=False, on_conflict='ignore')
¶
Inserts or updates an IV Curve Report Detail into the database.
Fields left as None will not be inserted/updated.
Parameters:
-
(iv_curve_report_id¶int | None, default:None) –ID of the IV Curve Report to which this detail belongs.
-
(inverter_name¶str | None, default:None) –Name of the inverter associated with this IV Curve Report Detail.
-
(string_number¶int | None, default:None) –String number associated with this IV Curve Report Detail.
-
(service_order_note_sap_id¶int | None, default:None) –SAP ID of the service order note associated with this IV Curve Report Detail.
-
(v_oc¶float | None, default:None) –Open-circuit voltage (Voc) value.
-
(i_sc¶float | None, default:None) –Short-circuit current (Isc) value.
-
(iv_curve_array¶list[list[float]] | None, default:None) –IV curve data as a list of [voltage, current] pairs.
-
(source_diagnostic¶str | None, default:None) –Source diagnostic description.
-
(echo_diagnostic¶str | None, default:None) –Echo diagnostic description.
-
(data_df¶DataFrame | None, default:None) –Polars DataFrame containing multiple IV Curve Report Details to insert.
The needed columns are: - iv_curve_report_id - inverter_name - string_number - service_order_note_sap_id (optional) - v_oc (optional) - i_sc (optional) - iv_curve_array (optional) - source_diagnostic (optional) - echo_diagnostic (optional)
If data_df is provided, all other parameters except create_diagnostic and on_conflict will be ignored.
-
(create_diagnostic¶bool, default:False) –Whether to create the source diagnostic in the database if it does not exist. Defaults to False.
-
(on_conflict¶Literal['ignore', 'update'], default:'ignore') –What to do in case of conflict. Can be one of ["ignore", "update"]. By default "ignore"
Conflicts will be evaluated against iv_curve_array_id, inverter_id and string_number.
Returns:
-
list[int]–IDs of the inserted or updated IV Curve Report Detail.
Source code in echo_postgres/ivcurve_report_details.py
@validate_call
def insert(
self,
iv_curve_report_id: int | None = None,
inverter_name: str | None = None,
string_number: int | None = None,
service_order_note_sap_id: int | None = None,
v_oc: float | None = None,
i_sc: float | None = None,
iv_curve_array: list[list[float]] | None = None,
source_diagnostic: str | None = None,
echo_diagnostic: str | None = None,
data_df: pl.DataFrame | None = None,
create_diagnostic: bool = False,
on_conflict: Literal["ignore", "update"] = "ignore",
) -> list[int]:
"""Inserts or updates an IV Curve Report Detail into the database.
Fields left as None will not be inserted/updated.
Parameters
----------
iv_curve_report_id : int | None, optional
ID of the IV Curve Report to which this detail belongs.
inverter_name : str | None, optional
Name of the inverter associated with this IV Curve Report Detail.
string_number : int | None, optional
String number associated with this IV Curve Report Detail.
service_order_note_sap_id : int | None, optional
SAP ID of the service order note associated with this IV Curve Report Detail.
v_oc : float | None, optional
Open-circuit voltage (Voc) value.
i_sc : float | None, optional
Short-circuit current (Isc) value.
iv_curve_array : list[list[float]] | None, optional
IV curve data as a list of [voltage, current] pairs.
source_diagnostic : str | None, optional
Source diagnostic description.
echo_diagnostic : str | None, optional
Echo diagnostic description.
data_df : pl.DataFrame | None, optional
Polars DataFrame containing multiple IV Curve Report Details to insert.
The needed columns are:
- iv_curve_report_id
- inverter_name
- string_number
- service_order_note_sap_id (optional)
- v_oc (optional)
- i_sc (optional)
- iv_curve_array (optional)
- source_diagnostic (optional)
- echo_diagnostic (optional)
If data_df is provided, all other parameters except create_diagnostic and on_conflict will be ignored.
create_diagnostic : bool, optional
Whether to create the source diagnostic in the database if it does not exist. Defaults to False.
on_conflict : Literal["ignore", "update"], optional
What to do in case of conflict. Can be one of ["ignore", "update"]. By default "ignore"
Conflicts will be evaluated against iv_curve_array_id, inverter_id and string_number.
Returns
-------
list[int]
IDs of the inserted or updated IV Curve Report Detail.
"""
# if a single record is being inserted, lets convert it to a DataFrame
df_schema = {
"iv_curve_report_id": pl.Int64,
"inverter_name": pl.Utf8,
"string_number": pl.Int64,
"service_order_note_sap_id": pl.Int64,
"v_oc": pl.Float64,
"i_sc": pl.Float64,
"iv_curve_array": pl.List(pl.List(pl.Float64)),
"source_diagnostic": pl.Utf8,
"echo_diagnostic": pl.Utf8,
}
if data_df is None:
single_insert = True
data_df = pl.DataFrame(
{
"iv_curve_report_id": [iv_curve_report_id],
"inverter_name": [inverter_name],
"string_number": [string_number],
"service_order_note_sap_id": [service_order_note_sap_id],
"v_oc": [v_oc],
"i_sc": [i_sc],
"iv_curve_array": [iv_curve_array],
"source_diagnostic": [source_diagnostic],
"echo_diagnostic": [echo_diagnostic],
},
schema=df_schema,
)
else:
single_insert = False
# validating if data_df has the correct schema
required_cols = ["iv_curve_report_id", "inverter_name", "string_number"]
for col, dtype in df_schema.items():
if col not in data_df.columns and col in required_cols:
raise ValueError(f"data_df is missing required column '{col}'.")
if col in data_df.columns:
if not data_df[col].dtype == dtype:
raise ValueError(f"data_df column '{col}' has incorrect dtype. Expected {dtype}, got {data_df[col].dtype}.")
else:
data_df = data_df.with_columns(pl.lit(None).cast(dtype).alias(col))
# validating if the iv_curve_report_id exists
existing_ids = self._perfdb.ivcurves.reports.get_ids(ids=data_df["iv_curve_report_id"].to_list())
if wrong_ids := set(data_df["iv_curve_report_id"].to_list()) - set(existing_ids):
raise ValueError(f"IV Curve Report IDs {wrong_ids} do not exist in the database.")
# getting the inverter_id from inverter_name
inverter_ids = self._perfdb.objects.instances.get_ids(object_names=data_df["inverter_name"].to_list())
if wrong_inverters := set(data_df["inverter_name"].to_list()) - set(inverter_ids):
raise ValueError(f"Inverter names {wrong_inverters} do not exist in the database.")
# replacing inverter names with IDs
data_df = data_df.with_columns(
pl.col("inverter_name").replace_strict(inverter_ids, return_dtype=pl.Int64).alias("inverter_id"),
)
# validating if the service_order_note_sap_id exists
if len(data_df.filter(pl.col("service_order_note_sap_id").is_not_null())) > 0:
service_order_note_ids = self._perfdb.service_orders.notes.get_ids(
note_sap_ids=data_df["service_order_note_sap_id"].drop_nulls().unique().to_list(),
)
if wrong_ids := set(data_df["service_order_note_sap_id"].drop_nulls().unique().to_list()) - set(service_order_note_ids):
raise ValueError(f"Service Order Note SAP IDs {wrong_ids} do not exist in the database.")
else:
service_order_note_ids = None
# replacing service_order_note_sap_id with IDs
if service_order_note_ids:
data_df = data_df.with_columns(
pl.col("service_order_note_sap_id")
.replace_strict(service_order_note_ids, return_dtype=pl.Int64, default=None)
.alias("service_order_note_id"),
)
# validating if iv_curve_array is in the correct format
for array in data_df["iv_curve_array"].to_list():
if array is not None:
self._validate_iv_curve_array(iv_curve_array=array)
# validating and getting diagnostic IDs
diagnostic_ids = self._validate_and_get_diagnostic_ids(
source_diagnostics=data_df["source_diagnostic"].to_list(),
echo_diagnostics=data_df["echo_diagnostic"].to_list(),
create_diagnostic=create_diagnostic,
)
# adjust diagnostic_id in data_df
data_df = data_df.with_columns(
pl.Series(diagnostic_ids["source_diagnostic_id"]).cast(pl.Int64).alias("source_diagnostic_id"),
pl.Series(diagnostic_ids["echo_diagnostic_id"]).cast(pl.Int64).alias("echo_diagnostic_id"),
)
# inserting
detail_ids: pl.DataFrame = self._perfdb.conn.polars_to_sql(
df=data_df.drop(["inverter_name", "source_diagnostic", "echo_diagnostic", "service_order_note_sap_id"]),
table_name="iv_curve_report_details",
schema="performance",
conflict_cols=["iv_curve_report_id", "inverter_id", "string_number"],
return_cols=["id"],
if_exists="append" if on_conflict == "ignore" else on_conflict,
ignore_null_cols=single_insert,
)
logger.debug(f"{len(detail_ids)} IV Curve Report Details inserted/updated in the database: {detail_ids['id'].to_list()}")
return detail_ids["id"].to_list()
insert_by_pdf(pdf_file_path)
¶
Inserts IV Curve Diagnostic data from a PDF file into the database.
Parameters:
-
(pdf_file_path¶str) –Path to the PDF file containing IV Curve Diagnostic data.
Source code in echo_postgres/ivcurve_report_details.py
@validate_call
def insert_by_pdf(self, pdf_file_path: str) -> None:
"""Inserts IV Curve Diagnostic data from a PDF file into the database.
Parameters
----------
pdf_file_path : str
Path to the PDF file containing IV Curve Diagnostic data.
"""
iv_curve_diagnostic_df = self._data_treatment(pdf_file_path)
self.insert(iv_curve_diagnostic_df=iv_curve_diagnostic_df, on_conflict="update")
update(detail_id=None, iv_curve_report_id=None, inverter_name=None, string_number=None, service_order_note_sap_id=None, v_oc=None, i_sc=None, iv_curve_array=None, source_diagnostic=None, echo_diagnostic=None, data_df=None, create_diagnostic=False, consider_null_values=False)
¶
Updates an existing IV Curve Report Detail in the database.
Only non-None fields will be updated.
Parameters:
-
(detail_id¶int, default:None) –ID of the IV Curve Report Detail to update.
-
(iv_curve_report_id¶int | None, default:None) –ID of the IV Curve Report to which this detail belongs.
-
(inverter_name¶str | None, default:None) –Name of the inverter associated with this IV Curve Report Detail.
-
(string_number¶int | None, default:None) –String number associated with this IV Curve Report Detail.
-
(service_order_note_sap_id¶int | None, default:None) –SAP ID of the service order note associated with this IV Curve Report Detail.
-
(v_oc¶float | None, default:None) –Open-circuit voltage (Voc) value.
-
(i_sc¶float | None, default:None) –Short-circuit current (Isc) value.
-
(iv_curve_array¶list[list[float]] | None, default:None) –IV curve data as a list of [voltage, current] pairs.
-
(source_diagnostic¶str | None, default:None) –Source diagnostic description.
-
(echo_diagnostic¶str | None, default:None) –Echo diagnostic description.
-
(data_df¶DataFrame | None, default:None) –Polars DataFrame containing multiple IV Curve Report Details to update.
The needed columns are: - detail_id - iv_curve_report_id (optional) - inverter_name (optional) - string_number (optional) - service_order_note_sap_id (optional) - v_oc (optional) - i_sc (optional) - iv_curve_array (optional) - source_diagnostic (optional) - echo_diagnostic (optional)
If data_df is provided, all other parameters except create_diagnostic will be ignored.
-
(create_diagnostic¶bool, default:False) –Whether to create the diagnostic in the database if it does not exist. Defaults to False.
-
(consider_null_values¶bool, default:False) –If set to True, when applicable, Null values in data_df or parameters will be used to update the database, setting the respective fields to Null.
Source code in echo_postgres/ivcurve_report_details.py
@validate_call
def update(
self,
detail_id: int | None = None,
iv_curve_report_id: int | None = None,
inverter_name: str | None = None,
string_number: int | None = None,
service_order_note_sap_id: int | None = None,
v_oc: float | None = None,
i_sc: float | None = None,
iv_curve_array: list[list[float]] | None = None,
source_diagnostic: str | None = None,
echo_diagnostic: str | None = None,
data_df: pl.DataFrame | None = None,
create_diagnostic: bool = False,
consider_null_values: bool = False,
) -> None:
"""Updates an existing IV Curve Report Detail in the database.
Only non-None fields will be updated.
Parameters
----------
detail_id : int
ID of the IV Curve Report Detail to update.
iv_curve_report_id : int | None, optional
ID of the IV Curve Report to which this detail belongs.
inverter_name : str | None, optional
Name of the inverter associated with this IV Curve Report Detail.
string_number : int | None, optional
String number associated with this IV Curve Report Detail.
service_order_note_sap_id : int | None, optional
SAP ID of the service order note associated with this IV Curve Report Detail.
v_oc : float | None, optional
Open-circuit voltage (Voc) value.
i_sc : float | None, optional
Short-circuit current (Isc) value.
iv_curve_array : list[list[float]] | None, optional
IV curve data as a list of [voltage, current] pairs.
source_diagnostic : str | None, optional
Source diagnostic description.
echo_diagnostic : str | None, optional
Echo diagnostic description.
data_df : pl.DataFrame | None, optional
Polars DataFrame containing multiple IV Curve Report Details to update.
The needed columns are:
- detail_id
- iv_curve_report_id (optional)
- inverter_name (optional)
- string_number (optional)
- service_order_note_sap_id (optional)
- v_oc (optional)
- i_sc (optional)
- iv_curve_array (optional)
- source_diagnostic (optional)
- echo_diagnostic (optional)
If data_df is provided, all other parameters except create_diagnostic will be ignored.
create_diagnostic : bool, optional
Whether to create the diagnostic in the database if it does not exist. Defaults to False.
consider_null_values : bool, optional
If set to True, when applicable, Null values in data_df or parameters will be used to update the database, setting the respective fields to Null.
"""
# if a single record is being inserted, lets convert it to a DataFrame
df_schema = {
"detail_id": pl.Int64,
"iv_curve_report_id": pl.Int64,
"inverter_name": pl.Utf8,
"string_number": pl.Int64,
"service_order_note_sap_id": pl.Int64,
"v_oc": pl.Float64,
"i_sc": pl.Float64,
"iv_curve_array": pl.List(pl.List(pl.Float64)),
"source_diagnostic": pl.Utf8,
"echo_diagnostic": pl.Utf8,
}
if data_df is None:
single_update = True
data_df = pl.DataFrame(
{
"detail_id": [detail_id],
"iv_curve_report_id": [iv_curve_report_id],
"inverter_name": [inverter_name],
"string_number": [string_number],
"service_order_note_sap_id": [service_order_note_sap_id],
"v_oc": [v_oc],
"i_sc": [i_sc],
"iv_curve_array": [iv_curve_array],
"source_diagnostic": [source_diagnostic],
"echo_diagnostic": [echo_diagnostic],
},
schema=df_schema,
)
else:
single_update = False
# validating if data_df has the correct schema
required_cols = ["detail_id"]
for col, dtype in df_schema.items():
if col not in data_df.columns and col in required_cols:
raise ValueError(f"data_df is missing required column '{col}'.")
if col in data_df.columns:
if not data_df[col].dtype == dtype:
raise ValueError(f"data_df column '{col}' has incorrect dtype. Expected {dtype}, got {data_df[col].dtype}.")
else:
data_df = data_df.with_columns(pl.lit(None).cast(dtype).alias(col))
# making sure all rows have at least one column other than detail_id to update
# creating non_null_count column
if not consider_null_values:
data_df = data_df.with_columns(
sum(pl.col(col).is_not_null().cast(pl.Int64) for col in df_schema if col != "detail_id").alias("non_null_count"),
)
if (invalid_rows := data_df.filter(pl.col("non_null_count") == 0)).height > 0:
raise ValueError(
f"The following detail_id(s) have no fields to update: {invalid_rows['detail_id'].to_list()}",
)
# dropping non_null_count column
data_df = data_df.drop("non_null_count")
# Check if detail_id exists
existing_query = sql.SQL("""SELECT id FROM performance.iv_curve_report_details WHERE id = ANY({detail_ids})""").format(
detail_ids=sql.Literal(data_df["detail_id"].to_list()),
)
existing_df = self._perfdb.conn.read_to_polars(existing_query)
if wrong_ids := set(data_df["detail_id"].to_list()) - set(existing_df["id"].to_list()):
raise ValueError(f"IV Curve Report Detail IDs {wrong_ids} do not exist in the database.")
# validating if the iv_curve_report_id exists
if len(data_df.filter(pl.col("iv_curve_report_id").is_not_null())) > 0:
existing_ids = self._perfdb.ivcurves.reports.get_ids(ids=data_df["iv_curve_report_id"].to_list())
if wrong_ids := set(data_df["iv_curve_report_id"].to_list()) - set(existing_ids):
raise ValueError(f"IV Curve Report IDs {wrong_ids} do not exist in the database.")
# getting the inverter_id from inverter_name
if len(data_df.filter(pl.col("inverter_name").is_not_null())) > 0:
inverter_ids = self._perfdb.objects.instances.get_ids(object_names=data_df["inverter_name"].to_list())
if wrong_inverters := set(data_df["inverter_name"].to_list()) - set(inverter_ids):
raise ValueError(f"Inverter names {wrong_inverters} do not exist in the database.")
# replacing inverter names with IDs
data_df = data_df.with_columns(
pl.col("inverter_name").replace_strict(inverter_ids, return_dtype=pl.Int64).alias("inverter_id"),
)
else:
data_df = data_df.with_columns(pl.lit(None).cast(pl.Int64).alias("inverter_id"))
# validating if the service_order_note_sap_id exists
if len(data_df.filter(pl.col("service_order_note_sap_id").is_not_null())) > 0:
service_order_note_ids = self._perfdb.service_orders.notes.get_ids(
note_sap_ids=data_df["service_order_note_sap_id"].drop_nulls().unique().to_list(),
)
if wrong_ids := set(data_df["service_order_note_sap_id"].drop_nulls().unique().to_list()) - set(service_order_note_ids):
raise ValueError(f"Service Order Note SAP IDs {wrong_ids} do not exist in the database.")
# replacing service_order_note_sap_id with IDs
data_df = data_df.with_columns(
pl.col("service_order_note_sap_id")
.replace_strict(service_order_note_ids, return_dtype=pl.Int64)
.alias("service_order_note_id"),
)
else:
data_df = data_df.with_columns(pl.lit(None).cast(pl.Int64).alias("service_order_note_id"))
for array in data_df["iv_curve_array"].to_list():
if array is not None:
self._validate_iv_curve_array(iv_curve_array=array)
# validating and getting diagnostic IDs
diagnostic_ids = self._validate_and_get_diagnostic_ids(
source_diagnostics=data_df["source_diagnostic"].to_list(),
echo_diagnostics=data_df["echo_diagnostic"].to_list(),
create_diagnostic=create_diagnostic,
)
# adjust diagnostic_id in data_df
data_df = data_df.with_columns(
pl.Series(diagnostic_ids["source_diagnostic_id"]).cast(pl.Int64).alias("source_diagnostic_id"),
pl.Series(diagnostic_ids["echo_diagnostic_id"]).cast(pl.Int64).alias("echo_diagnostic_id"),
)
# rename detail_id to id for updating
data_df = data_df.rename({"detail_id": "id"})
# updating
self._perfdb.conn.polars_to_sql(
df=data_df.drop(["inverter_name", "source_diagnostic", "echo_diagnostic", "service_order_note_sap_id"]),
table_name="iv_curve_report_details",
schema="performance",
conflict_cols=["id"],
if_exists="update_only",
ignore_null_cols=single_update if not consider_null_values else False,
)