RUFAS.EEE.emissions module#

RUFAS.EEE.emissions.HOMEGROWN_FEEDS_AND_FERTILIZERS_FILTERS: dict[str, dict[str, Any]] = {'fertilizer applications filter': {'date_fields': ('year', 'day'), 'description': 'Collects all synthetic fertilizer applications that occurred in the simulation.', 'filters': ["Field._record_fertilizer_application\\.fertilizer_application\\.field='.*'"], 'name': 'Fertilizer Applications', 'variables': ['.*']}, 'homegrown feeds filter': {'date_fields': ('harvest_year', 'harvest_day'), 'description': 'Collects all crop harvests that occurred in the simulation.', 'filters': ["CropManagement._record_yield.harvest_yield.field='.*'"], 'name': 'Homegrown Feeds', 'variables': ['.*']}, 'manure applications filter': {'date_fields': ('year', 'day'), 'description': 'Collects all manure applications that occurred in the simulation.', 'filters': ["Field._record_manure_application\\.manure_application\\.field='.*'"], 'name': 'Manure Applications', 'variables': ['.*']}, 'manure requests filter': {'date_fields': ('year', 'day'), 'description': 'Collects all manure requests that occurred in the simulation.', 'filters': ["Field._record_manure_application\\.manure_request\\.field='.*'"], 'name': 'Manure Requests', 'variables': ['.*']}}#: Patterns for matching purchased feed storage and totals in the output variables.

RUFAS.EEE.emissions.PURCHASED_FEED_PATTERN = '_(\\d+)_fed_to_date.*'#: These are constants for calculating the embedded emissions of synthetic nitrogen and phosphorus fertilizer. Their units are in kg CO2e / kg N and kg CO2e / kg P, respectively. The nitrogen and phosphorus constants reference IPCC 2021, GWP 100, the potassium constant references BASF’s Eco-efficiency analysis tool.

class RUFAS.EEE.emissions.EmissionsEstimator#

Bases: object

Estimates emissions associated with purchased feeds used for animals.

Attributes#

imInputManager: An instance of the InputManager class.
omOutputManager: An instance of the OutputManager class.
crop_species_to_purchased_feed_iddict[str, list[str]]: A dictionary mapping crop species to their corresponding RuFaS feed IDs.
purchased_feed_emissions_by_locationdict[str, float]: A dictionary mapping RuFaS feed IDs to their emissions factors (kg CO2e / kg dry matter) for the location of the simulation.
land_use_change_emissions_by_locationdict[str, float]: A dictionary mapping RuFaS feed IDs to their land use change emissions factors (kg CO2e / kg dry matter) for the location of the simulation.
_missing_purchased_idsset[str]: A set of RuFaS feed IDs that were used in the simulation but do not have purchased feed emissions data.
_missing_land_use_idsset[str]: A set of RuFaS feed IDs that were used in the simulation but do not have land use change emissions data.

__init__() → None#

estimate_emissions() → None#: Estimates emissions associated with farmgrown feeds.

check_available_purchased_feed_data(available_feed_ids: list[int]) → None#: Checks that all purchased feed IDs used in the simulation have emissions data available for them.

_gather_homegrown_feeds_and_fertilizer_apps() → dict[str, list[dict[str, Any]]]#: Gathers the yields that were harvested and fertilizer applications that were applied in the last 365 days of the simulation.

_transform_outputs_to_list_of_dicts(data: dict[str, Any]) → list[dict[str, Any]]#

Transforms dictionary of lists collected from the Output Manager into list of dictionaries.

Examples#

>>> a = {'one': {'values': [1, 2, 3]}, 'two': {'values': [4, 5, 6]}}
>>> _transform_outputs_to_list_of_dicts(a)
[{'one': 1, 'two': 4}, {'one': 2, 'two': 5}, {'one': 3, 'two': 6}]

_calculate_total_homegrown_feed_amounts_by_crop_type(homegrown_feeds: list[dict[str, Any]]) → dict[str, float]#: Calculates the total amount of each crop species grown on the farm.

calculate_emissions(purchased_feeds: dict[int, float]) → None#: Calculates the emissions from purchased feeds and land use changes. If there are feed IDs with missing emissions factor data, they will be omitted from the calculations and not reported.

_get_feed_emissions_data(county_code: int, feed_emissions_data: dict[str, list[float]]) → dict[str, float]#: Grabs the appropriate list of emissions for purchased feeds for the location of the simulation.

_calculate_homegrown_feed_emissions(homegrown_feeds: list[dict[str, Any]], fertilizer_applications: list[dict[str, Any]], manure_applications: list[dict[str, Any]], manure_requests: list[dict[str, Any]]) → None#: Calculates the emissions associated with feeds grown on the farm.

_collect_target_soil_characteristics(field_names: list[str]) → dict[str, dict[str, Any]]#: Collects the emissions and soil carbon characteristics used to calculate farm-grown feed emissions.

_calculate_emissions_by_field(field_name: str, feeds_grown: list[dict[str, Any]], field_emissions: dict[str, float], manure_applications: dict[str, float], manure_requests: dict[str, float], fertilizer_applications_data: list[dict[str, Any]]) → list[dict[str, Any]]#: Partitions emissions from the field where crops/feeds were grown to those crops.

_partition_applied_crop_fertilizer_emissions(fertilizer_application: dict[str, float], applied_crops: list[dict[str, Any]]) → None#: Partitions synthetic emissions from fertilizer applications to crop(s) whose planting and harvesting dates encompass the fertilizer application date.

_filter_results(filters: dict[str, Any], date_cutoff: date, year_key: str, day_key: str) → list[dict[str, Any]]#

This method help with filtering data based on the timeframe provided.

Parameters#

filters: dict[str, Any]: A filter to collect the desired data from OutputManager.
date_cutoff: datetime: Date before which all data collected from the Output Manager is filtered out.
year_key: str: How the key is named for the year data retrieved from OM.
day_key: str: How the key is named for the day data retrieved from OM.

Returns#

list[dict[str, Any]]: Filtered result of the retrieved data.

_aggregate_data(operations: list[dict[str, Any]], all_fields: list[str], nutrients: list[str]) → dict[str, dict[str, float]]#

Aggregate nutrient data for different types of applications (fertilizer, manure, etc.).

Parameters#

operationslist[dict[str, Any]]: A list of application dictionaries. Each dictionary contains the field name and nutrient amounts. Example: {“field_name”: “Field1”, “nitrogen”: 50.0, “phosphorus”: 20.0}
all_fieldslist[str]: A list of field names to aggregate data for, usually combining fields with crops and fields with applications.
nutrientslist[str]: A list of nutrient keys to aggregate (e.g., [“nitrogen”, “phosphorus”, “potassium”]).

Returns#

dict: A dictionary where keys are field names and values are dictionaries of aggregated nutrient values.

_apply_fertilizer_to_next_crop(fertilizer_application: dict[str, Any], sorted_crops: list[dict[str, Any]], fertilizer_application_date: date) → bool#: Applies fertilizer to the next available crop after harvest and before the next planting. Returns True if fertilizer was applied, False otherwise.

_extract_applied_crops(sorted_crops: list[dict[str, Any]], fertilizer_application_date: date) → list[dict[str, Any]]#

Extracts a list of crops that had fertilizer applied to them between their planting and harvesting dates.

Parameters#

sorted_cropslist[dict[str, Any]]: The list of crops in the field sorted by planting date.
fertilizer_application_datedate: The date of the fertilizer applciation.

Returns#

list[dict[str, Any]]: A list of the crops whose planting and harvesting dates encompass the fertilizer application date.

_soil_data_update(filters: dict[str, dict[str, Any]]) → dict[str, int | Literal[0]]#

This method will update the soil data update according to data extracted using the filters.

Parameters#

filters: dict[str, dict[str, Any]]: A dictionary of filters to retrieve data from the OM.

Return#

dict[str, int | Literal[0]]: The updated soil data.