Skip to content

annual_exports

annual_exports

Annual export data models and configurations.

Classes

AnnualExportError

Bases: Exception

Base annual export error

AnnualExport 

AnnualExport(**data)

Bases: Base

Single annual export.

Source code in src/pyetm/models/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(self, **data: Any) -> None:
    """
    Initialize the model, converting validation errors to warnings.
    """
    super(BaseModel, self).__setattr__("__pydantic_private__", {})

    # Initialize all private attributes with their defaults
    private_dict: Dict[str, Any] = self.__pydantic_private__  # type: ignore[assignment]
    for attr_name, attr_info in self.__class__.__private_attributes__.items():
        if (
            hasattr(attr_info, "default_factory")
            and attr_info.default_factory is not None
        ):
            # Call factory - signature varies between pydantic versions
            private_dict[attr_name] = attr_info.default_factory()
        elif hasattr(attr_info, "default"):
            private_dict[attr_name] = attr_info.default
        else:
            private_dict[attr_name] = None

    try:
        super().__init__(**data)
    except ValidationError as e:
        # Check if data is None or empty - this indicates API error
        if not data or data is None:
            # Re-raise with clearer message
            raise ValueError(
                f"Cannot create {self.__class__.__name__} with empty data. "
                "This usually indicates an authentication or API error."
            ) from e

        # If validation fails, create model without validation and collect warnings
        # Use model_construct to bypass validation
        temp_instance = self.__class__.model_construct(**data)

        # Copy the constructed data to this instance
        for field_name, field_value in temp_instance.__dict__.items():
            if not field_name.startswith("_"):
                object.__setattr__(self, field_name, field_value)

        # Ensure required Pydantic slot attributes exist to prevent AttributeError
        for slot in ("__pydantic_fields_set__", "__pydantic_extra__"):
            try:
                value = object.__getattribute__(temp_instance, slot)
                object.__setattr__(self, slot, value)
            except AttributeError:
                # Initialize missing slot attributes with defaults
                if slot == "__pydantic_extra__":
                    object.__setattr__(self, slot, {})
                elif slot == "__pydantic_fields_set__":
                    object.__setattr__(self, slot, set())

        # Convert validation errors to warnings
        for error in e.errors():
            field_path = ".".join(str(part) for part in error.get("loc", []))
            message = error.get("msg", "Validation failed")
            self._warning_collector.add(field_path, message, "error")
Functions
available 
available()

Check if export data has been retrieved

Source code in src/pyetm/models/annual_exports.py 
40
41
42
def available(self) -> bool:
    """Check if export data has been retrieved"""
    return self.data is not None
retrieve 
retrieve(client, scenario, force_refresh=False)

Fetch export data from API

Source code in src/pyetm/models/annual_exports.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def retrieve(
    self, client: BaseClient, scenario: Any, force_refresh: bool = False
) -> Optional[pd.DataFrame]:
    """Fetch export data from API"""
    # Return cached data unless explicitly refreshing
    if not force_refresh and self.available():
        return self.data

    try:
        result = DownloadAnnualExportRunner.run(client, scenario, self.name)
        if result.success and result.data:
            try:
                result.data.seek(0)
                df = pd.read_csv(result.data, index_col=0)

                if df.notna().sum().sum() == 0 and len(df) > 0:
                    self.add_warning(
                        "data",
                        f"CSV parsing resulted in all NaN values for {self.name}.",
                    )
                    return None

                # Clean up any completely empty rows
                df_clean = df.dropna(how="all")
                self.data = df_clean
                return df_clean
            except Exception as e:
                self.add_warning(
                    "data", f"Failed to process export data for {self.name}: {e}"
                )
                return None
        else:
            for error in result.errors or []:
                self.add_warning("api", f"API error: {error}")
            return None
    except Exception as e:
        self.add_warning(
            "base", f"Unexpected error retrieving export {self.name}: {e}"
        )
        return None
contents 
contents()

Get export contents (returns cached data if available)

Source code in src/pyetm/models/annual_exports.py 
85
86
87
88
89
90
91
92
def contents(self) -> Optional[pd.DataFrame]:
    """Get export contents (returns cached data if available)"""
    if not self.available():
        self.add_warning(
            "data", f"Export {self.name} not available - data not yet retrieved"
        )
        return None
    return self.data
from_name classmethod 
from_name(name)

Create an AnnualExport instance for a given export type

Source code in src/pyetm/models/annual_exports.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def from_name(cls, name: str) -> "AnnualExport":
    """Create an AnnualExport instance for a given export type"""
    if name not in ANNUAL_EXPORT_TYPES:
        export = cls(name=name)
        valid_types = ", ".join(ANNUAL_EXPORT_TYPES)
        export.add_warning(
            "name",
            f"Unknown export type '{name}'. Valid types: {valid_types}",
        )
        return export
    return cls(name=name)

AnnualExports 

AnnualExports(**data)

Bases: Base

Collection of Annual Exports.

Source code in src/pyetm/models/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(self, **data: Any) -> None:
    """
    Initialize the model, converting validation errors to warnings.
    """
    super(BaseModel, self).__setattr__("__pydantic_private__", {})

    # Initialize all private attributes with their defaults
    private_dict: Dict[str, Any] = self.__pydantic_private__  # type: ignore[assignment]
    for attr_name, attr_info in self.__class__.__private_attributes__.items():
        if (
            hasattr(attr_info, "default_factory")
            and attr_info.default_factory is not None
        ):
            # Call factory - signature varies between pydantic versions
            private_dict[attr_name] = attr_info.default_factory()
        elif hasattr(attr_info, "default"):
            private_dict[attr_name] = attr_info.default
        else:
            private_dict[attr_name] = None

    try:
        super().__init__(**data)
    except ValidationError as e:
        # Check if data is None or empty - this indicates API error
        if not data or data is None:
            # Re-raise with clearer message
            raise ValueError(
                f"Cannot create {self.__class__.__name__} with empty data. "
                "This usually indicates an authentication or API error."
            ) from e

        # If validation fails, create model without validation and collect warnings
        # Use model_construct to bypass validation
        temp_instance = self.__class__.model_construct(**data)

        # Copy the constructed data to this instance
        for field_name, field_value in temp_instance.__dict__.items():
            if not field_name.startswith("_"):
                object.__setattr__(self, field_name, field_value)

        # Ensure required Pydantic slot attributes exist to prevent AttributeError
        for slot in ("__pydantic_fields_set__", "__pydantic_extra__"):
            try:
                value = object.__getattribute__(temp_instance, slot)
                object.__setattr__(self, slot, value)
            except AttributeError:
                # Initialize missing slot attributes with defaults
                if slot == "__pydantic_extra__":
                    object.__setattr__(self, slot, {})
                elif slot == "__pydantic_fields_set__":
                    object.__setattr__(self, slot, set())

        # Convert validation errors to warnings
        for error in e.errors():
            field_path = ".".join(str(part) for part in error.get("loc", []))
            message = error.get("msg", "Validation failed")
            self._warning_collector.add(field_path, message, "error")
Attributes
names property 
names

Get list of all export names in collection

Functions
__iter__ 
__iter__()

Iterate over export objects

Source code in src/pyetm/models/annual_exports.py 
118
119
120
def __iter__(self) -> Any:
    """Iterate over export objects"""
    yield from self.exports.values()
__contains__ 
__contains__(name)

Check if export name exists in collection

Source code in src/pyetm/models/annual_exports.py 
122
123
124
def __contains__(self, name: str) -> bool:
    """Check if export name exists in collection"""
    return name in self.exports
get 
get(name)

Get contents of a specific export by name. If not yet retrieved, returns None.

Source code in src/pyetm/models/annual_exports.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
def get(self, name: str) -> Optional[pd.DataFrame]:
    """
    Get contents of a specific export by name.
    If not yet retrieved, returns None.
    """
    if name not in self.exports:
        self.add_warning("exports", f"Export {name} not found in collection")
        return None

    export = self.exports[name]
    contents = export.contents()
    self._merge_submodel_warnings(export, key_attr="name")
    return contents
retrieve 
retrieve(client, scenario, name, force_refresh=False)

Retrieve a specific export by name.

Source code in src/pyetm/models/annual_exports.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
def retrieve(
    self, client: BaseClient, scenario: Any, name: str, force_refresh: bool = False
) -> Optional[pd.DataFrame]:
    """
    Retrieve a specific export by name.
    """
    if name not in self.exports:
        self.exports[name] = AnnualExport.from_name(name)

    export = self.exports[name]
    result = export.retrieve(client, scenario, force_refresh)
    self._merge_submodel_warnings(export, key_attr="name")
    return result
retrieve_multiple 
retrieve_multiple(client, scenario, names, force_refresh=False)

Retrieve multiple exports by name.

Source code in src/pyetm/models/annual_exports.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def retrieve_multiple(
    self,
    client: BaseClient,
    scenario: Any,
    names: list[str],
    force_refresh: bool = False,
) -> dict[str, pd.DataFrame]:
    """
    Retrieve multiple exports by name.
    """
    results = {}
    for name in names:
        data = self.retrieve(client, scenario, name, force_refresh)
        if data is not None:
            results[name] = data
    return results