dataio.utils package

Subpackages

• dataio.utils.hsut package
  • Submodules
  • dataio.utils.hsut.io module
    • DataReader
      • DataReader.read()
    • DataWriter
      • DataWriter.write()
    • TableColumn
      • TableColumn.CODE
      • TableColumn.DESCRIPTION
    • TableType
      • TableType.DIMENSION
      • TableType.FACT
  • dataio.utils.hsut.resources_hsut module
    • CSVResourceRepository
      • CSVResourceRepository.add()
      • CSVResourceRepository.add_from_dataframe()
      • CSVResourceRepository.add_or_update()
      • CSVResourceRepository.comment_res()
      • CSVResourceRepository.get()
      • CSVResourceRepository.get_dataframe_for_task()
      • CSVResourceRepository.list()
      • CSVResourceRepository.update()
    • Resource
      • Resource.append_comment()
      • Resource.from_dataframe()
      • Resource.to_dataframe()
    • ResourceRepository
      • ResourceRepository.add()
      • ResourceRepository.add_or_update()
      • ResourceRepository.get()
      • ResourceRepository.get_dataframe_for_task()
      • ResourceRepository.list()
      • ResourceRepository.update()
    • SQLiteResourceRepository
      • SQLiteResourceRepository.add()
      • SQLiteResourceRepository.add_from_dataframe()
      • SQLiteResourceRepository.add_or_update()
      • SQLiteResourceRepository.connect()
      • SQLiteResourceRepository.get()
      • SQLiteResourceRepository.get_dataframe_for_task()
      • SQLiteResourceRepository.list()
      • SQLiteResourceRepository.to_csv()
      • SQLiteResourceRepository.update()
  • Module contents

Submodules

dataio.utils.accounts module

class dataio.utils.accounts.Account(name: str, abbreviation: str)

Bases: object

abbreviation: str

name: str

class dataio.utils.accounts.AccountRepository(path: Path = None)

Bases: object

add_account(name: str) → Account

generate_abbreviation(name: str) → str

get_account(name: str) → Account

get_or_create_account(name: str) → Account

load_accounts()

save_accounts()

dataio.utils.connectors module

dataio.utils.path_manager module

class dataio.utils.path_manager.BasePaths(root_path: str | Path)

Bases: object

property classification

property correspondence: Path

property data_clean: Path

property data_collection: Path

property exiobase4: Path

property merged_collected_data: Path

class dataio.utils.path_manager.PathBuilder(root_path: str | Path, version_source: Path | str | None = None, account_repository: AccountRepository = None)

Bases: object

PathBuilder provides shorthand paths for files on a cloud storage specified by root_path to be used by the algorithms to create the full exiobase database.

You can provide a version_source so that you can link to files of previous versions. Please resort to default_version for the structure of the version file.

for entries that are not specified in the dictionary, the path to the latest available version is generated.

Additionally, it is possible to provide the path to an alternative version file that is used instead of the standard one.

property b2_version

property balance_raw

property cement

property cement_data

property cleaned_exio_3

property cleaned_fao: Path

property cleaned_forestry

compose(path: Path, account_name: str | None = None, version_date: str | None = 'latest') → Path

Ensures a path exists by creating it if necessary.

This method constructs a path by appending optional account and version date components. If the account name is provided, it retrieves the account, adding it if it doesn’t exist. The version date can be specified or default to the latest version. The final path is created if it does not already exist.

pathPath

The base path to compose.

account_nameOptional[str], optional

The account name to include in the path.

If the account does not exist, it will be added. Defaults to None. version_date : Optional[str], optional

The version date to include in the path.

It can be “latest” to use the latest version date, a specific date in “YYYY-MM-DD” format, or None to exclude the version date. Defaults to “latest”.

Path

The composed path, created if it didn’t already exist.

KeyError: If the account name is not found in the account repository. ValueError: If the provided version date is invalid.

property corresp_fao

property dm_coeff

property emissions_coeff

property emissions_intermediate

property emissions_raw

property fao_collection

property fao_processed

property fao_raw

property fao_store

property fert_interm

property fertilisers

property ferts_collection

property fish_markets

property gams_inputs

property heat_markets

property hiot_interm

property hiot_raw

property hiot_with_capital

property hiot_with_iluc

property hiot_with_marg_elect

property iLUC_interm

property iLUC_param

property iLUC_raw

property iea_clean

property iea_interm

property iea_raw_exio

property ipcc_param

property land_use

property lci_act_generic

property lci_cleaned

property lci_concito

property lci_country

property lci_exio4

property lci_fish

property lci_prod_generic

property lci_products

property lci_raw

property lci_vehicles

list_path_attributes()

load_version_dates(source: str | Path | None = None) → VersionCollection

property matrix_of_invest

property monetary_tables

property natural_resource: Path

property outlier

property prices

property prod_markets

property property_param

property property_values

property simapro: Path

property supply_intermediate

property supply_raw

property trade_intermediate

property trade_raw

property trade_route

property un_data

property un_data_elab

property use_intermediate

property use_raw

property value_added

property waste_accounts

property waste_markets

dataio.utils.schema_enums module

class dataio.utils.schema_enums.APIEndpoints(value)

Bases: Enum

ACTIVITIES = 'activity-names/'

ACTIVITY_CORR = 'activity-corr/'

BASE_URL = 'https://lca.aau.dk/api/'

FOOTPRINT = 'footprint/'

LOCATIONS = 'locations/'

LOCATION_CORR = 'location-corr/'

METADATA = 'datasets/'

PRODUCT = 'product/'

PRODUCTS = 'products/'

PRODUCT_CORR = 'product-corr/'

PROPERTIES = 'properties/'

RECIPES = 'recipes/'

SUPPLY = 'supply/'

TOKEN = 'user/token/'

USE = 'use/'

classmethod get_url(name: str) → str

Return the full URL for a given endpoint name.

Parameters: - name: The name of the endpoint to retrieve.

Returns: - The full URL of the requested endpoint.

class dataio.utils.schema_enums.EmissCompartment(value)

Bases: Enum

AIR = 'Air'

SOIL = 'Soil'

WATER = 'Water'

class dataio.utils.schema_enums.Exio_fert_nutrients

Bases: object

market_dict = {'K2O': 'M_K2Ofert', 'N': 'M_Nfert', 'P2O5': 'M_P2O5fert'}

nutrients = ['N', 'P2O5', 'K2O']

class dataio.utils.schema_enums.IndexNames(value)

Bases: Enum

ACT_CODE = 'Exio act code'

AGGREGATION = 'Aggregation'

AGRI_SYSTEM = 'system'

ANIMAL_CATEGORY = 'animal categ'

ASSOCIATED_TREAT = 'associated treatment service'

CLIMATE_METRIC = 'climate metric'

COEFFICIENT = 'coeff'

COUNTRY_CODE = 'Exio country code'

DESCRIP = 'description'

DESTIN_COUNTRY = 'Exio destin. country code'

DESTIN_COUNTRY_EDGE = 'Exio destin. country code edge'

EDGE = 'edge'

EMIS_COMPARTMENT = 'Exio compartment'

EMIS_SUBSTANCE = 'Exio substance'

EXIO3 = 'nace-related code'

EXIO3_ACT = 'exio_v3'

EXIO_ACCOUNT = 'Account'

EXIO_ACT = 'Exiobase activity'

EXIO_CNT = 'Exiobase country'

EXIO_CNT_acron = 'Exiobase cnt'

EXIO_CODE = 'Exiobase code'

EXIO_PRD = 'Exiobase product'

FACTOR = 'factor'

FLAG = 'flag'

GLOBAL_AREA = 'Global'

INPUT_PROD = 'Exio prod code (input)'

LCI_FLAG = 'flag_lci'

LCI_UNIT = 'unit_lci'

LCI_VALUE = 'Value_lci'

MARKET = 'market'

NUTRIENT_CONT = 'nutrient content'

OJBECT_CODE = 'object_code'

ORIGIN_COUNTRY = 'Exio origin country code'

ORIGIN_COUNTRY_EDGE = 'Exio origin country code edge'

PACKAGED = 'Packaged product'

PACK_CODE = 'exio4_pack_id'

PACK_MARKET = 'Exio packaging market'

PACK_PROD = 'Exio packaging prod code'

PARENT_CLASS = 'parent_class'

PARENT_CODE = 'parent_code'

PARENT_LINK = 'parent_link'

PARENT_NAME = 'parent_name'

PERIOD = 'time'

PERIOD_DELAY = 'time delay'

POSITION = 'position'

PRODUCT = 'product'

PRODUCTION = 'production'

PROD_CODE = 'Exio prod code'

REF_PROD_CODE = 'Exio ref. prod code'

REPLACED_MRK = 'Replaced market'

REPLACED_PRODUCT = 'replaced product'

REPL_FACTOR = 'repl factor'

RESOURCE = 'Exio resource'

SCENARIO = 'scenario'

SHARE = 'share'

SOURCE = 'source'

SOURCE_CLASS = 'source_class'

SOURCE_CODE = 'source_code'

SOURCE_LINK = 'source_link'

SOURCE_NAME = 'source_name'

SUBSTITUTION_FACTOR = 'substitution factor'

TARGET_CLASS = 'target_class'

TARGET_CODE = 'target_code'

TARGET_LINK = 'target_link'

TARGET_NAME = 'target_name'

TRADE_ROUTE_ID = 'trade route ID'

TRANSPORT_MODE = 'transport mode'

TRANSPORT_MODE_EDGE = 'transport mode edge'

UNIT = 'unit'

UNIT_DESTIN = 'unit (source)'

UNIT_FOOTPRINT = 'unit (footprint)'

UNIT_SOURCE = 'unit (input)'

VALUE = 'Value'

VALUE_FOOTPRINT = 'Value (footprint)'

VALUE_IN = 'Value (input)'

WASTE_FRACTION = 'Exio waste'

WASTE_MARKET = 'Exio waste market'

class dataio.utils.schema_enums.Property(name: str, value: float, unit: str, upperbound: float | None = None, lowerbound: float | None = None)

Bases: object

lowerbound: float | None = None

name: str

unit: str

upperbound: float | None = None

value: float

class dataio.utils.schema_enums.PropertyEnum(value)

Bases: Enum

CURRENCY = ('Meuro', <class 'dataio.utils.schema_enums.Property'>)

ENERGY = ('TJ', <class 'dataio.utils.schema_enums.Property'>)

FREIGHT = ('tkm', <class 'dataio.utils.schema_enums.Property'>)

ITEM_COUNT = ('items', <class 'dataio.utils.schema_enums.Property'>)

LAND = ('ha-weighted', <class 'dataio.utils.schema_enums.Property'>)

MASS = ('tonnes', <class 'dataio.utils.schema_enums.Property'>)

NA = ('undefined', <class 'dataio.utils.schema_enums.Property'>)

WASTE_SERVICE = ('tonnes (service)', <class 'dataio.utils.schema_enums.Property'>)

create_property(value, upperbound=inf, lowerbound=-inf) → Property

class dataio.utils.schema_enums.animal_system(value)

Bases: Enum

MEAT = 'meat'

MILK = 'milk'

class dataio.utils.schema_enums.data_index_categ

Bases: object

balance_categ = ['Exio origin country code', 'Exio prod code', 'Exio destin. country code', 'Exio act code']

balance_columns = ['Value', 'flag', 'unit']

column_categ = ['Value', 'flag']

emiss_categ = ['Exio country code', 'Exio substance', 'Exio compartment', 'Exio act code', 'unit']

fao_animal_system_index = ['Exio country code', 'Item', 'Element', 'system', 'animal categ', 'unit', 'time']

fao_clean_index = ['Exio country code', 'Item', 'Element', 'Unit']

general_categ = ['Exio origin country code', 'Exio prod code', 'Exio destin. country code', 'Exio act code', 'unit']

pack_index = ['Exio origin country code', 'Exio prod code', 'Exio destin. country code', 'Exio act code', 'exio4_pack_id']

trade_categ = ['Exio origin country code', 'Exio prod code', 'Exio destin. country code', 'unit']

waste_sup_col = ['Value', 'flag', 'unit']

waste_sup_index = ['Exio origin country code', 'Exio prod code', 'Exio destin. country code', 'Exio act code', 'Exio waste']

class dataio.utils.schema_enums.fao_categ(value)

Bases: Enum

AREA = 'Area'

ELEM = 'Element'

ITEM = 'Item'

UNIT = 'Unit'

class dataio.utils.schema_enums.global_land_categ(value)

Bases: Enum

arable = 'Arable Land'

forest = 'Forest Land'

grass = 'Grassland'

class dataio.utils.schema_enums.ipcc_categ(value)

Bases: Enum

ef4_dry = 'Dry climate'

ef4_wet = 'Wet climate'

fract_leach_wet = 'Wet climate'

class dataio.utils.schema_enums.land_use_categ(value)

Bases: Enum

gr_to_ps = 'grassland to pasture'

int_cp = 'intensification of cropland'

int_ps = 'intensification of pasture'

pf_to_mf = 'primary forest to managed forest'

sf_to_cp = 'secondary forest to crop'

sf_to_mf = 'secondary forest to managed forest'

unit_land = 'ha-weighted'

dataio.utils.versions module

class dataio.utils.versions.APIVersionSource(name: str = 'METADATA')

Bases: object

load() → VersionCollection

save(versions: list[Version])

class dataio.utils.versions.TXTVersionSource(path: Path)

Bases: object

load()

read the version-date of a previously produced dataset

Here the list of datasets (in key) and the acronym inserted in the version.txt:

Parameters:

path (path of the version file.)

save(versions: list[Version])

class dataio.utils.versions.Version(name: 'str', acronym: 'str', version_date: 'str', version_semantic: 'str' = None, is_latest: 'bool' = True)

Bases: object

acronym: str

has_valid_version_date_format(expected_format: str = '%Y-%m-%d') → bool

Checks if the version_date is in the expected format.

is_latest: bool = True

name: str

parse_date(from_format: str = '%Y-%m-%d', to_format: str = '%Y%m%d')

Parses the version_date from ‘YYYY-MM-DD’ to ‘YYYYMMDD’ format.

to_json()

update_version_date(version_date: str)

Updates the version_date to a new date provided in ‘YYYY-MM-DD’ format.

version_date: str

version_semantic: str = None

class dataio.utils.versions.VersionCollection(versions: List[Version], source: str | Path = None, account_repo: AccountRepository = None)

Bases: object

add_account_version(account_name: str, version_date: str, is_latest: bool = True, version_semantic: str = None) → Version

Adds a new version for an account. If the account does not exist, it adds the account.

get_account_versions(account: str) → List[Version]

get_latest_version(account: str) → Version

classmethod load_from(source: Path | str)

save()

to_api(url: str = 'https://lca.aau.dk/api/datasets/')

to_txt(path: Path)

update_account_version_date(account_name: str, version_date: str)

Replace the version_date of account in version collection with a new version_date

class dataio.utils.versions.VersionSource

Bases: ABC

abstractmethod load() → VersionCollection

abstractmethod save(versions: list[Version])

Module contents