EPTR2 Class¶

The main class for interacting with the EPIAS Transparency Platform API.

Overview¶

The EPTR2 class handles authentication, API calls, and response processing for Turkish electricity market data.

Basic Usage¶

from eptr2 import EPTR2

# Using .env file for credentials
eptr = EPTR2(use_dotenv=True, recycle_tgt=True)

# Direct credentials
eptr = EPTR2(username="email@example.com", password="password")

# Make API calls
df = eptr.call("mcp", start_date="2024-07-29", end_date="2024-07-29")

Class Reference¶

EPTR2 ¶

EPTR2(username: str = None, password: str = None, recycle_tgt: bool = True, use_dotenv: bool = True, dotenv_path: str = '.env', **kwargs)

Source code in src/eptr2/main.py

def __init__(
    self,
    username: str = None,
    password: str = None,
    recycle_tgt: bool = True,
    use_dotenv: bool = True,
    dotenv_path: str = ".env",
    **kwargs,
) -> None:
    ## kwargs are
    ### map_param_labels: bool
    ### secure: bool
    ### query_parameters: dict
    ### just_call_phrase: bool
    ### root_phrase: str
    self.ssl_verify = kwargs.get("ssl_verify", True)
    self.check_postprocess(postprocess=kwargs.get("postprocess", True))
    self.get_raw_response = kwargs.get("get_raw_response", False)

    ### Credentials and Login
    self.username = username
    self.password = password
    self.is_test = kwargs.get("is_test", False)  ## Currently not being used

    self.temp_new_login_method = kwargs.get("new_login_method", False)

    ### Credentials file path is being deprecated in favor of dotenv usage
    self.credentials_file_path = kwargs.get("credentials_file_path", None)

    ### Dotenv usage. Get EPTR_USERNAME and EPTR_PASSWORD from .env file if use_dotenv is True and environment file exists and credentials are recorded there.
    self.use_dotenv = use_dotenv
    if self.use_dotenv:
        env_check_d = self.check_dotenv(dotenv_path=dotenv_path)
    else:
        env_check_d = None

    self.upass_check(
        env_check_d=env_check_d, custom_root_phrase=kwargs.get("root_phrase", None)
    )

    ### Options to recycle tgt
    self.recycle_tgt = recycle_tgt
    self.tgt_dir_path = kwargs.get("tgt_path", ".")

    input_tgt_d = kwargs.get("tgt_d", None)
    self.import_tgt_info(input_tgt_d)

    self.check_renew_tgt(**kwargs)

    ## Path map keys and custom aliases
    self.path_map_keys = get_path_map(just_call_keys=True)
    self.custom_aliases = kwargs.get("custom_aliases", {})

call ¶

call(key: str, **kwargs)

Main call function for the API. This function is used to process parameters and make calls to EPIAS Transparency API.

Source code in src/eptr2/main.py

def call(self, key: str, **kwargs):
    """
    Main call function for the API. This function is used to process parameters and make calls to EPIAS Transparency API.
    """

    self.check_renew_tgt(**kwargs)
    raw_key = key
    key = alias_to_path(alias=key, custom_aliases=self.custom_aliases)

    if key not in self.path_map_keys:
        if raw_key == key:
            raise Exception(
                f"This call {raw_key} is not yet defined in calls or aliases. Call 'get_available_calls' method to see the available calls."
            )
        else:
            raise Exception(
                f"This alias {raw_key} forwarded to key {key} is not yet defined in calls. Call 'get_available_calls' method to see the available calls."
            )

    call_path = get_total_path(key)
    call_method = get_call_method(key)
    required_body_params = get_required_parameters(key)
    call_body_raw = kwargs.pop("call_body", kwargs)

    ### There are some calls requiring special handling, they have a special function to process them
    call_body_raw = process_special_calls(key, call_body_raw)

    optional_body_params = get_optional_parameters(key)
    all_params = required_body_params + optional_body_params

    call_body = {
        k: preprocess_parameter(k, v)
        for k, v in call_body_raw.items()  ## If there is a call_body parameter in kwargsi, use it, else use kwargs
        if k in all_params
    }

    for body_key in required_body_params:
        if body_key not in call_body.keys():
            call_body[body_key] = preprocess_parameter(body_key, None)
        kwargs.pop(body_key, None)

    if call_body is not None:
        if not all([x in call_body.keys() for x in required_body_params]):
            raise Exception("Some required parameters are missing in call body.")

        if kwargs.get("map_param_labels", True):
            cb2 = {}
            for k, v in call_body.items():
                ## Updated this part to handle multiple labels originating from a single label
                label = get_param_label(k)["label"]
                value = v
                if isinstance(label, list):
                    for l in label:  # noqa: E741
                        cb2[l] = value
                else:
                    cb2[label] = value
            call_body = copy.deepcopy(cb2)

    elif len(required_body_params) > 0:
        raise Exception("Required parameters are missing in call body.")

    ## If the call is uevm, change powerPlantId to powerplantId due to only non-standard naming in powerPlantId
    if key == "uevm" and "powerPlantId" in call_body.keys():
        call_body["powerplantId"] = call_body.pop("powerPlantId")

    res = transparency_call(
        call_path=call_path,
        call_method=call_method,
        call_body=call_body,
        root_phrase=self.root_phrase,
        ssl_verify=self.ssl_verify,
        is_test=self.is_test,
        tgt=self.tgt,
        **kwargs,
    )

    ## Set soft timeout for tgt renewal
    self.tgt_exp_0 = min(
        self.tgt_exp,
        datetime.now().timestamp() + 60 * 90,
    )

    if self.recycle_tgt:
        self.export_tgt_info()

    if kwargs.get("get_raw_response", self.get_raw_response):
        return res

    res = json.loads(res.data.decode("utf-8"))
    if kwargs.get("postprocess", self.postprocess):
        from eptr2.mapping.processing import get_postprocess_function

        df = get_postprocess_function(key)(res, key=key)
        return df

    return res

get_available_calls ¶

get_available_calls(include_aliases: bool = False)

Gets all the available calls of eptr2 package. As a reminder, number of calls at eptr2 package might be lower than the actual calls. If include_aliases is set to True, it also includes the aliases.

Source code in src/eptr2/main.py

def get_available_calls(self, include_aliases: bool = False):
    """
    Gets all the available calls of eptr2 package. As a reminder, number of calls at eptr2 package might be lower than the actual calls. If include_aliases is set to True, it also includes the aliases.
    """

    if include_aliases:
        return {
            "keys": self.path_map_keys,
            "default_aliases": get_alias_map(),
            "custom_aliases": self.custom_aliases,
        }

    return self.path_map_keys

get_number_of_calls ¶

get_number_of_calls()

List the number of calls in the package. It also lists the number of derived calls and API calls (excluding derived calls).

Source code in src/eptr2/main.py

def get_number_of_calls(self):
    """
    List the number of calls in the package. It also lists the number of derived calls and API calls (excluding derived calls).
    """

    d = {
        "all_calls": self.path_map_keys,
        "derived_calls": get_derived_calls(),
    }
    d["n_total_calls"] = len(d["all_calls"])
    d["n_derived_calls"] = len(d["derived_calls"])
    d["n_api_calls"] = len(d["all_calls"]) - len(d["derived_calls"])

    return d

get_aliases ¶

get_aliases(include_custom_aliases: bool = False)

Gets only the aliases. If include_custom_aliases is set to True, it also includes the user defined aliases (custom_aliases).

Source code in src/eptr2/main.py

def get_aliases(self, include_custom_aliases: bool = False):
    """
    Gets only the aliases. If include_custom_aliases is set to True, it also includes the user defined aliases (custom_aliases).
    """

    alias_d = get_alias_map()
    if include_custom_aliases:
        alias_d.update(self.custom_aliases)

    return alias_d

get_tgt ¶

get_tgt(**kwargs)

Source code in src/eptr2/main.py

def get_tgt(self, **kwargs):
    if self.username is None or self.password is None:
        raise Exception("Username and password must be provided for tgt renewal.")

    test_suffix = "-prp" if self.is_test else ""
    login_url = f"""https://giris{test_suffix}.epias.com.tr/cas/v1/tickets"""

    body_str = f"username={quote(self.username)}&password={quote(self.password)}"

    http = urllib3.PoolManager(
        cert_reqs="CERT_REQUIRED" if self.ssl_verify else "CERT_NONE"
    )

    res = http.request(
        method="POST",
        url=login_url,
        headers={
            "Content-Type": "application/x-www-form-urlencoded",
            "Accept": "text/plain",
        },
        body=body_str,
        **kwargs.get("request_kwargs", {"timeout": 10}),
    )
    if res.status not in [200, 201]:
        raise Exception(
            "Request failed with status code: "
            + str(res.status)
            + ": "
            + res.data.decode("utf-8")
        )

    if isinstance(res.data, bytes):
        res_data = res.data.decode("utf-8")
    else:
        res_data = res.data

    if res_data.startswith("TGT-"):
        tgt = res_data
    else:
        raise Exception("Login failed. TGT not found in response: " + res_data)

    self.tgt = tgt
    tgt_start_time = datetime.now()

    ## Hard timeout
    self.tgt_exp = (tgt_start_time + timedelta(hours=1, minutes=45)).timestamp()

    ## Soft timeout
    self.tgt_exp_0 = min(
        self.tgt_exp,
        (tgt_start_time + timedelta(hours=1, minutes=45)).timestamp(),
    )

    if self.recycle_tgt:
        self.export_tgt_info()

check_renew_tgt ¶

check_renew_tgt(**kwargs)

Source code in src/eptr2/main.py

def check_renew_tgt(self, **kwargs):
    force_renew_tgt = kwargs.get("force_renew_tgt", False)
    if (
        self.tgt is None
        or self.tgt_exp_0 < datetime.now().timestamp()
        or force_renew_tgt
    ):
        self.get_tgt(**kwargs)

Constructor Parameters¶

Parameter	Type	Default	Description
`username`	`str`	`None`	EPIAS platform username (email)
`password`	`str`	`None`	EPIAS platform password
`recycle_tgt`	`bool`	`True`	Reuse authentication tickets
`use_dotenv`	`bool`	`True`	Load credentials from `.env` file
`dotenv_path`	`str`	`".env"`	Path to the `.env` file
`ssl_verify`	`bool`	`True`	Verify SSL certificates
`postprocess`	`bool`	`True`	Return DataFrame instead of raw response
`get_raw_response`	`bool`	`False`	Return raw HTTP response

The `call` Method¶

The primary method for making API requests:

result = eptr.call(
    key="mcp",                      # API endpoint key
    start_date="2024-07-29",        # Start date (YYYY-MM-DD)
    end_date="2024-07-29",          # End date (YYYY-MM-DD)
    postprocess=True,               # Return DataFrame
    request_kwargs={"timeout": 10}  # urllib3 request options
)

Parameters¶

Parameter	Type	Required	Description
`key`	`str`	Yes	API endpoint name or alias
`start_date`	`str`	Usually	Start date in YYYY-MM-DD format
`end_date`	`str`	Usually	End date in YYYY-MM-DD format
`postprocess`	`bool`	No	Return DataFrame (default: True)
`request_kwargs`	`dict`	No	Additional request parameters
`**kwargs`	-	No	Additional endpoint-specific parameters

Returns¶

DataFrame: When postprocess=True (default)
dict: When postprocess=False
urllib3.HTTPResponse: When get_raw_response=True on EPTR2 instance

Discovery Methods¶

get_available_calls¶

Returns list of all available API endpoint keys:

calls = eptr.get_available_calls()
print(len(calls))  # 213+

get_number_of_calls¶

Returns count of available calls:

count = eptr.get_number_of_calls()
print(f"{count} calls available")

get_aliases¶

Returns dictionary of call aliases:

aliases = eptr.get_aliases()
# {'mcp': 'ptf', 'smp': 'smf', ...}

Authentication Methods¶

get_tgt¶

Manually request a new TGT:

eptr.get_tgt()

check_renew_tgt¶

Check and renew TGT if expired:

eptr.check_renew_tgt(force_renew_tgt=True)

Examples¶

Basic Price Query¶

from eptr2 import EPTR2

eptr = EPTR2(use_dotenv=True, recycle_tgt=True)

# Market Clearing Price
mcp = eptr.call("mcp", start_date="2024-07-29", end_date="2024-07-29")
print(mcp.head())

Multiple Queries¶

# Reuse instance for multiple queries
eptr = EPTR2(use_dotenv=True, recycle_tgt=True)

mcp = eptr.call("mcp", start_date="2024-07-01", end_date="2024-07-31")
smp = eptr.call("smp", start_date="2024-07-01", end_date="2024-07-31")
consumption = eptr.call("rt-cons", start_date="2024-07-01", end_date="2024-07-31")

With Custom Timeout¶

df = eptr.call(
    "mcp",
    start_date="2024-07-01",
    end_date="2024-07-31",
    request_kwargs={"timeout": 30}
)

Raw JSON Response¶

raw = eptr.call(
    "mcp",
    start_date="2024-07-29",
    end_date="2024-07-29",
    postprocess=False
)
print(type(raw))  # dict

With Organization ID¶

df = eptr.call(
    "bilateral-contracts-org",
    start_date="2024-07-29",
    end_date="2024-07-29",
    org_id=123
)

Error Handling¶

from eptr2 import EPTR2

eptr = EPTR2(use_dotenv=True)

try:
    df = eptr.call("mcp", start_date="2024-07-29", end_date="2024-07-29")
except Exception as e:
    print(f"Error: {e}")

EPTR2 Class¶

Overview¶

Basic Usage¶

Class Reference¶

EPTR2 ¶

call ¶

get_available_calls ¶

get_number_of_calls ¶

get_aliases ¶

get_tgt ¶

check_renew_tgt ¶

Constructor Parameters¶

The call Method¶

Parameters¶

Returns¶

Discovery Methods¶

get_available_calls¶

get_number_of_calls¶

get_aliases¶

Authentication Methods¶

get_tgt¶

check_renew_tgt¶

Examples¶

Basic Price Query¶

Multiple Queries¶

With Custom Timeout¶

Raw JSON Response¶

With Organization ID¶

Error Handling¶

See Also¶

The `call` Method¶