Package Documentation

Vessel Emissions API Package.

Classes:

Name	Description
VesselEmissionsAPI	Represents Signal's Vessel Emissions API.
EmissionsEstimation	Represents Emissions Estimation for a single Voyage.
Metrics	Represents Emissions Metrics for a Vessel.
VesselClassEmissions	Represents Emissions Estimation
VesselClassMetrics	Represents Emissions Metrics

EmissionsEstimation dataclass

Contains info about the emissions estimation.

Attributes:

Name	Type	Description
id	str	Voyage ID
imo	int	Vessel IMO
vessel_name	str	Vessel Name
voyage_number	int	Voyage Number
vessel_type_id	int	Vessel type id
vessel_type	Optional[str]	Vessel type
vessel_class_id	int	Vessel class id
vessel_class	Optional[str]	Vessel class
start_date	str	Start date of the Voyage
end_date	str	End date of the Voyage
quantity	Optional[float]	Cargo quantity of the voyage
deadweight	int	Vessel's deadweight
transport_work_in_million_tonne_miles	Optional[float]	Transport
transport_work_in_million_dwt_miles	Optional[float]	Transport work
contains_eu_emissions	Optional[bool]	Declares whether
emissions	EmissionsBreakdown	Emissions breakdown for the voyage
consumptions	Optional[ConsumptionsBreakdown]	Consumptions breakdown for the voyage
seagoing_speed_statistics	Optional[SeagoingSpeedStatisticsBreakdown]	Seagoing
duration	Optional[DurationBreakdown]	Duration breakdown for the voyage
distances	Optional[DistancesBreakdown]	Distance travelled breakdown for the voyage
efficiency_metrics	Optional[Metrics]	Emissions Efficiency metrics for the voyage
european_union_regulated	Optional[EmissionsEssentialStatistics]	European Union Related Emissions

Source code in signal_ocean/vessel_emissions/models.py

@dataclass(frozen=True)
class EmissionsEstimation:
    """Contains info about the emissions estimation.

    Attributes:
        id: Voyage ID
        imo: Vessel IMO
        vessel_name: Vessel Name
        voyage_number: Voyage Number
        vessel_type_id: Vessel type id
        vessel_type: Vessel type
        vessel_class_id: Vessel class id
        vessel_class: Vessel class
        start_date: Start date of the Voyage
        end_date: End date of the Voyage
        quantity: Cargo quantity of the voyage
        deadweight: Vessel's deadweight
        transport_work_in_million_tonne_miles: Transport
        work measured in millions tonnes of cargo carried
        times the travelled distance in miles of the current voyage
        transport_work_in_million_dwt_miles: Transport work
        measured in millions tonnes when the cargo
        curried equals the DWT times the travelled
        distance in miles of the current voyage
        contains_eu_emissions: Declares whether
        voyage has European Union Related Emissions
        emissions: Emissions breakdown for the voyage
        consumptions: Consumptions breakdown for the voyage
        seagoing_speed_statistics: Seagoing
        Speed Statistics breakdown for the voyage
        duration: Duration breakdown for the voyage
        distances: Distance travelled breakdown for the voyage
        efficiency_metrics: Emissions Efficiency metrics for the voyage
        european_union_regulated: European Union Related Emissions

    """
    id: str
    imo: int
    vessel_name: str
    voyage_number: int
    vessel_type_id: int
    vessel_class_id: int
    start_date: str
    end_date: str
    deadweight: int
    emissions: EmissionsBreakdown
    vessel_type: Optional[str] = None
    vessel_class: Optional[str] = None
    quantity: Optional[float] = None
    transport_work_in_million_tonne_miles: Optional[float] = None
    transport_work_in_million_dwt_miles: Optional[float] = None
    contains_eu_emissions: Optional[bool] = None
    consumptions: Optional[ConsumptionsBreakdown] = None
    seagoing_speed_statistics: \
        Optional[SeagoingSpeedStatisticsBreakdown] = None
    duration: Optional[DurationBreakdown] = None
    distances: Optional[DistancesBreakdown] = None
    efficiency_metrics: Optional[Metrics] = None
    european_union_regulated: Optional[EmissionsEssentialStatistics] = None

    def __repr__(self) -> str:
        """Override of the default __repr__ function.

        Returns:
            Object string representation omitting None attributes

        """
        nodef_f_vals = (
            (f.name, attrgetter(f.name)(self))
            for f in dataclasses.fields(self)
            if attrgetter(f.name)(self) != f.default
        )

        nodef_f_repr = ", ".join(f"{name}={value}"
                                 for name, value
                                 in nodef_f_vals)
        return f"{self.__class__.__name__}({nodef_f_repr})"

    def to_dict(self) -> Dict[Any, Any]:
        """Cast EmissionsEstimation object to dict.

        Returns:
            Dict representation of EmissionsEstimation model

        """
        return dataclasses.asdict(
            self,
            dict_factory=lambda x: {
                _to_camel_case_with_special_keywords(k): v
                for (k, v) in x if v is not None
            })

__repr__()

Override of the default repr function.

Returns:

Type	Description
str	Object string representation omitting None attributes

Source code in signal_ocean/vessel_emissions/models.py

def __repr__(self) -> str:
    """Override of the default __repr__ function.

    Returns:
        Object string representation omitting None attributes

    """
    nodef_f_vals = (
        (f.name, attrgetter(f.name)(self))
        for f in dataclasses.fields(self)
        if attrgetter(f.name)(self) != f.default
    )

    nodef_f_repr = ", ".join(f"{name}={value}"
                             for name, value
                             in nodef_f_vals)
    return f"{self.__class__.__name__}({nodef_f_repr})"

to_dict()

Cast EmissionsEstimation object to dict.

Returns:

Type	Description
Dict[Any, Any]	Dict representation of EmissionsEstimation model

Source code in signal_ocean/vessel_emissions/models.py

def to_dict(self) -> Dict[Any, Any]:
    """Cast EmissionsEstimation object to dict.

    Returns:
        Dict representation of EmissionsEstimation model

    """
    return dataclasses.asdict(
        self,
        dict_factory=lambda x: {
            _to_camel_case_with_special_keywords(k): v
            for (k, v) in x if v is not None
        })

Metrics dataclass

Vessel Metrics estimation.

Attributes:

Name	Type	Description
voyage_cii	Optional[float]	Carbon Intensity Indicator
voyage_cii_unit	Optional[str]	The unit of the voyage
voyage_cii_rating	Optional[str]	The rating of the vessel
voyage_cii_target	Optional[float]	The alignment target for the
voyage_cii_target_year	Optional[int]	The year that has taken
capacity_eeoi	Optional[float]	The EEOI efficiency metric when
capacity_eeoi_unit	Optional[str]	The unit of the Capacity
capacity_eeoi_sea_cargo_charter_year_target	Optional[float]
capacity_eeoi_sea_cargo_charter_class	Optional[str]
capacity_eeoi_sea_cargo_charter_alignment_in_percentage	Optional[float]
eeoi	Optional[float]	The EEOI efficiency metrics calculated for the voyage
eeoi_unit	Optional[str]	The unit of the EEOI efficiency metrics
eeoi_sea_cargo_charter_year_target	Optional[float]	The alignment
eeoi_sea_cargo_charter_class	Optional[str]	The Sea Cargo Charter
eeoi_sea_cargo_charter_alignment_in_percentage	Optional[float]
kg_co2_per_tonne_cargo	Optional[float]	Kg of CO2 per tonne of cargo carried
kg_co2_per_tonne_dwt	Optional[float]	Kg of CO2 per tonne of the vessel's deadweight

Source code in signal_ocean/vessel_emissions/models.py

@dataclass(frozen=True)
class Metrics:
    """Vessel Metrics estimation.

    Attributes:
        voyage_cii: Carbon Intensity Indicator
        (VoyageCII)
        efficiency metrics for the voyage
        voyage_cii_unit: The unit of the voyage
        Carbon Intensity Indicator (VoyageCII)
        voyage_cii_rating: The rating of the vessel
        based on the VoyageCII
        voyage_cii_target: The alignment target for the
        Voyage CII efficiency metric of the current year
        voyage_cii_target_year: The year that has taken
        into account for calculating the alignment target
        for the VoyageCII efficiency metric
        capacity_eeoi: The EEOI efficiency metric when
        the quantity equals with the capacity of
        the vessel (CapacityEEOI)
        e.g. the deadweight of the vessel
        capacity_eeoi_unit: The unit of the Capacity
        EEOI efficiency metric
        capacity_eeoi_sea_cargo_charter_year_target:
        The alignment target for the Capacity EEOI
        efficiency metric based on the Sea Cargo
        Charter guidelines for the year of the voyage
        capacity_eeoi_sea_cargo_charter_class:
        The Sea Cargo Charter defined class
        that the vessel belongs to based on it's size
        capacity_eeoi_sea_cargo_charter_alignment_in_percentage:
        The alignment delta in percentage, based on the acquired
        Capacity EEOI and the given alignment target
        from Sea Cargo Charter. Negative is aligned,
        positive is misaligned
        eeoi: The EEOI efficiency metrics calculated for the voyage
        eeoi_unit: The unit of the EEOI efficiency metrics
        eeoi_sea_cargo_charter_year_target: The alignment
        target for the EEOI efficiency metric based on the Sea Cargo
        Charter guidelines for the year of the voyage
        eeoi_sea_cargo_charter_class: The Sea Cargo Charter
        defined class that the vessel belongs to based on it's size
        eeoi_sea_cargo_charter_alignment_in_percentage:
        The alignment delta in percentage, based on the acquired
        EOI and the given alignment target from Sea Cargo Charter.
        Negative is aligned, positive is misaligned.
        kg_co2_per_tonne_cargo: Kg of CO2 per tonne of cargo carried
        kg_co2_per_tonne_dwt: Kg of CO2 per tonne of the vessel's deadweight


    """
    voyage_cii: Optional[float] = None
    voyage_cii_unit: Optional[str] = None
    voyage_cii_rating: Optional[str] = None
    voyage_cii_target: Optional[float] = None
    voyage_cii_target_year: Optional[int] = None
    capacity_eeoi: Optional[float] = None
    capacity_eeoi_unit: Optional[str] = None
    capacity_eeoi_sea_cargo_charter_year_target: \
        Optional[float] = None
    capacity_eeoi_sea_cargo_charter_class: Optional[str] = None
    capacity_eeoi_sea_cargo_charter_alignment_in_percentage: \
        Optional[float] = None
    eeoi: Optional[float] = None
    eeoi_unit: Optional[str] = None
    eeoi_sea_cargo_charter_year_target: Optional[float] = None
    eeoi_sea_cargo_charter_class: Optional[str] = None
    eeoi_sea_cargo_charter_alignment_in_percentage: \
        Optional[float] = None
    kg_co2_per_tonne_cargo: Optional[float] = None
    kg_co2_per_tonne_dwt: Optional[float] = None

VesselClassEmissions dataclass

Contains Vessel Class Emissions.

Attributes:

Name	Type	Description
next_page_token	str	String. The key to retrieve the next page
data	List[EmissionsEstimation]	List of Vessel Emissions for available vessels of this class

Source code in signal_ocean/vessel_emissions/models.py

@dataclass(frozen=True)
class VesselClassEmissions:
    """Contains Vessel Class Emissions.

    Attributes:
        next_page_token: String. The key  to retrieve the next page
        data: List of Vessel Emissions for available vessels of this class

    """
    next_page_token: str
    data: List[EmissionsEstimation]

    def to_dict(self) -> Dict[Any, Any]:
        """Cast VesselClassEmissions object to dict.

        Returns:
            Dict representation of VesselClassEmissions object

        """
        return {
            "NextPageToken": self.next_page_token,
            "Data": [
                vessel_emissions.to_dict()
                for vessel_emissions in self.data
            ]
        }

to_dict()

Cast VesselClassEmissions object to dict.

Returns:

Type	Description
Dict[Any, Any]	Dict representation of VesselClassEmissions object

Source code in signal_ocean/vessel_emissions/models.py

def to_dict(self) -> Dict[Any, Any]:
    """Cast VesselClassEmissions object to dict.

    Returns:
        Dict representation of VesselClassEmissions object

    """
    return {
        "NextPageToken": self.next_page_token,
        "Data": [
            vessel_emissions.to_dict()
            for vessel_emissions in self.data
        ]
    }

VesselClassMetrics dataclass

Contains Vessel Class Metrics.

Attributes:

Name	Type	Description
next_page_token	str	String. The key that should be used
data	List[VesselMetrics]	List of Vessel Metrics for available vessels of this class

Source code in signal_ocean/vessel_emissions/models.py

@dataclass(frozen=True)
class VesselClassMetrics:
    """Contains Vessel Class Metrics.

    Attributes:
        next_page_token: String. The key that should be used
        as a parameter of the token to retrieve the next page
        data: List of Vessel Metrics for available vessels of this class

    """
    next_page_token: str
    data: List[VesselMetrics]

    def to_dict(self) -> Dict[Any, Any]:
        """Cast VesselClassMetrics object to dict.

        Returns:
            Dict representation of VesselClassMetrics object

        """
        return {
            "NextPageToken": self.next_page_token,
            "Data": [
                vessel_metrics.to_dict()
                for vessel_metrics in self.data
            ]
        }

to_dict()

Cast VesselClassMetrics object to dict.

Returns:

Type	Description
Dict[Any, Any]	Dict representation of VesselClassMetrics object

Source code in signal_ocean/vessel_emissions/models.py

def to_dict(self) -> Dict[Any, Any]:
    """Cast VesselClassMetrics object to dict.

    Returns:
        Dict representation of VesselClassMetrics object

    """
    return {
        "NextPageToken": self.next_page_token,
        "Data": [
            vessel_metrics.to_dict()
            for vessel_metrics in self.data
        ]
    }

VesselEmissionsAPI

Represents Signal's Vessel Emissions API.

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

class VesselEmissionsAPI:
    """Represents Signal's Vessel Emissions API."""

    relative_url = "vessel-emissions-api/public/v2/"
    default_pit = str(date.today())

    def __init__(self, connection: Optional[Connection] = None):
        """Initializes VesselEmissionsAPI.

        Args:
            connection: API connection configuration. If not provided, the
                default connection method is used.
        """
        if connection is not None:
            em_connection = copy.deepcopy(connection)
            func_type = type(
                em_connection._Connection__get_headers  # type: ignore
            )
            em_connection._Connection__get_headers = func_type(  # type: ignore
                custom_headers, em_connection
            )
            self.__connection = em_connection
        else:
            connection = Connection()
            func_type = type(
                connection._Connection__get_headers  # type: ignore
            )
            connection._Connection__get_headers = func_type(  # type: ignore
                custom_headers, connection
            )
            self.__connection = connection

    def construct_url_parameters(
            self,
            quantity: Union[int, None] = None,
            token: Optional[str] = None,
            include_consumptions: bool = False,
            include_efficiency_metrics: bool = False,
            include_distances: bool = False,
            include_durations: bool = False,
            include_speed_statistics: bool = False,
            include_eu_emissions: bool = False,
            sulphur_content_hfo: Optional[Union[float, None]] = None,
            sulphur_content_lfo: Optional[Union[float, None]] = None,
            sulphur_content_mgo: Optional[Union[float, None]] = None,
            sulphur_content_lng: Optional[Union[float, None]] = None
    ) -> Dict[Any, str]:
        """Construct the request parameters based on the user's input.

        Args:
        quantity: Cargo quantity of the voyage
        token: Next page token
        include_consumptions: Include consumption data in response
        include_efficiency_metrics: Include efficiency metrics in response
        include_distances: Include distances data in the response
        include_durations: Include duration data in the response
        include_speed_statistics: Include speed statistics data in response
        include_eu_emissions: Include European Union related
        emissions in response
        sulphur_content_hfo: Sulphur Content of HFO fuel type
        sulphur_content_lfo: Sulphur Content of LFO fuel type
        sulphur_content_mgo: Sulphur Content of MGO fuel type
        sulphur_content_lng: Sulphur Content of LNG fuel type

        Returns:
            The parameters dictionary of Request URL
        """
        params = {}
        if quantity is not None:
            params['quantity'] = str(quantity)
        if token is not None:
            params['token'] = token
        params['include_consumptions'] = str(include_consumptions)
        params['include_efficiency_metrics'] = str(include_efficiency_metrics)
        params['include_distances'] = str(include_distances)
        params['include_durations'] = str(include_durations)
        params['include_speed_statistics'] = str(include_speed_statistics)
        params['include_eu_emissions'] = str(include_eu_emissions)
        if sulphur_content_hfo is not None:
            params['sulphur_content_hfo'] = str(sulphur_content_hfo)
        if sulphur_content_lfo is not None:
            params['sulphur_content_lfo'] = str(sulphur_content_lfo)
        if sulphur_content_mgo is not None:
            params['sulphur_content_mgo'] = str(sulphur_content_mgo)
        if sulphur_content_lng is not None:
            params['sulphur_content_lng'] = str(sulphur_content_lng)
        return params

    def get_emissions_by_imo_and_voyage_number(
            self,
            imo: int,
            voyage_number: int,
            quantity: Union[int, None] = None,
            include_consumptions: bool = False,
            include_efficiency_metrics: bool = False,
            include_distances: bool = False,
            include_durations: bool = False,
            include_speed_statistics: bool = False,
            include_eu_emissions: bool = False,
            sulphur_content_hfo: Union[float, None] = None,
            sulphur_content_lfo: Union[float, None] = None,
            sulphur_content_mgo: Union[float, None] = None,
            sulphur_content_lng: Union[float, None] = None
    ) -> Optional[EmissionsEstimation]:
        """Retrieves voyage emissions for a vessel by its IMO and Voyage Number.

        Args:
            imo: Vessel IMO to retrieve.
            voyage_number: Voyage Number to retrieve.
            quantity: User defined transported quantity for voyage.
            include_consumptions: Include consumption data in the response.
            include_efficiency_metrics: Include efficiency metrics in response.
            include_distances: Include distances data in the response.
            include_durations: Include duration data in the response.
            include_speed_statistics: Include speed statistics in response.
            include_eu_emissions: Include European Union related
            emissions in response.
            sulphur_content_hfo: Sulphur Content of HFO fuel type.
            sulphur_content_lfo: Sulphur Content of LFO fuel type.
            sulphur_content_mgo: Sulphur Content of MGO fuel type.
            sulphur_content_lng: Sulphur Content of LNG fuel type.

        Returns:
            EmissionsEstimation if no vessel with
            the specified IMO or Voyage Number has been found.
        """
        params_dict = self.construct_url_parameters(
            quantity=quantity,
            include_consumptions=include_consumptions,
            include_efficiency_metrics=include_efficiency_metrics,
            include_distances=include_distances,
            include_durations=include_durations,
            include_speed_statistics=include_speed_statistics,
            include_eu_emissions=include_eu_emissions,
            sulphur_content_hfo=sulphur_content_hfo,
            sulphur_content_lfo=sulphur_content_lfo,
            sulphur_content_mgo=sulphur_content_mgo,
            sulphur_content_lng=sulphur_content_lng)
        query_url = make_url('emissions',
                             'imo', imo,
                             'voyage_number', voyage_number,
                             **params_dict)
        url = urljoin(VesselEmissionsAPI.relative_url, query_url)
        return get_single(self.__connection, url, EmissionsEstimation)

    def get_emissions_by_imo(
            self,
            imo: int,
            include_consumptions: bool = False,
            include_efficiency_metrics: bool = False,
            include_distances: bool = False,
            include_durations: bool = False,
            include_speed_statistics: bool = False,
            include_eu_emissions: bool = False,
            sulphur_content_hfo: Union[float, None] = None,
            sulphur_content_lfo: Union[float, None] = None,
            sulphur_content_mgo: Union[float, None] = None,
            sulphur_content_lng: Union[float, None] = None
    ) -> List[EmissionsEstimation]:
        """Retrieves a list of vessel emissions by its IMO.

        Args:
            imo: IMO of the vessel to retrieve emissions.
            include_consumptions: Include consumption data in the response.
            include_efficiency_metrics: Include efficiency metrics
             data in the response.
            include_distances: Include distances data in the response.
            include_durations: Include duration data in the response.
            include_speed_statistics: Include speed statistics
             data in the response.
            include_eu_emissions: Include European Union related
             emissions data in the response.
            sulphur_content_hfo: Sulphur Content of HFO fuel type.
            sulphur_content_lfo: Sulphur Content of LFO fuel type.
            sulphur_content_mgo: Sulphur Content of MGO fuel type.
            sulphur_content_lng: Sulphur Content of LNG fuel type.

        Returns:
            A list of vessel emissions or None if no vessel
             with the specified IMO has been found.
        """
        params_dict = self.construct_url_parameters(
            include_consumptions=include_consumptions,
            include_efficiency_metrics=include_efficiency_metrics,
            include_distances=include_distances,
            include_durations=include_durations,
            include_speed_statistics=include_speed_statistics,
            include_eu_emissions=include_eu_emissions,
            sulphur_content_hfo=sulphur_content_hfo,
            sulphur_content_lfo=sulphur_content_lfo,
            sulphur_content_mgo=sulphur_content_mgo,
            sulphur_content_lng=sulphur_content_lng
        )
        query_url = make_url('emissions',
                             'imo', imo,
                             **params_dict)
        url = urljoin(VesselEmissionsAPI.relative_url, query_url)
        return [i for i in get_multiple(self.__connection,
                                        url,
                                        EmissionsEstimation)]

    def get_metrics_by_imo(
            self,
            imo: int,
            year: Union[int, None] = None
    ) -> List[VesselMetrics]:
        """Get vessel metrics.

        Args:
            imo: Vessel IMO to retrieve
            year: The year for the annual metrics

        Returns:
            VesselMetrics for the requested IMO

        """
        url = urljoin(VesselEmissionsAPI.relative_url,
                      f"emissions/metrics/imo/{imo}")
        if year is not None:
            url = urljoin(url, f"?year={year}")
        return [i for i in get_multiple(self.__connection,
                                        url,
                                        VesselMetrics)]

    def get_emissions_by_vessel_class_id(
            self,
            vessel_class_id: int,
            token: Union[str, None] = None,
            include_consumptions: bool = False,
            include_efficiency_metrics: bool = False,
            include_distances: bool = False,
            include_durations: bool = False,
            include_speed_statistics: bool = False,
            include_eu_emissions: bool = False
    ) -> Optional[VesselClassEmissions]:
        """Get emissions estimations for a vessel class (supports incremental updates).

        Args:
            vessel_class_id: The vessel class to retrieve
            token: Next page token
            include_consumptions: Include consumption data in the response.
            include_efficiency_metrics: Include efficiency metrics
             data in the response.
            include_distances: Include distances data in the response.
            include_durations: Include duration data in the response.
            include_speed_statistics: Include speed statistics
             data in the response.
            include_eu_emissions: Include European Union related emissions
             data in the response.

        Returns:
            List of emissions estimation for all
             available voyages of a vessel class.

        """
        params_dict = self.construct_url_parameters(
            token=token,
            include_consumptions=include_consumptions,
            include_efficiency_metrics=include_efficiency_metrics,
            include_distances=include_distances,
            include_durations=include_durations,
            include_speed_statistics=include_speed_statistics,
            include_eu_emissions=include_eu_emissions
        )
        query_url = make_url('emissions',
                             'class', vessel_class_id,
                             **params_dict)

        url = urljoin(VesselEmissionsAPI.relative_url, query_url)
        return get_single(self.__connection,
                          url,
                          VesselClassEmissions)

    def get_metrics_by_vessel_class_id(self,
                                       vessel_class_id: int,
                                       year: Union[int, None] = None,
                                       token: Union[str, None] = None
                                       ) -> Optional[VesselClassMetrics]:
        """Get vessel class metrics.

        Args:
            vessel_class_id: The vessel class to retrieve
            year: The year for the annual metrics
            token: Next page token

        Returns:
            VesselClassMetrics for the requested Class

        """
        if year is None and token is None:
            query_url = ""
        else:
            query_url = "?"
            if year is not None:
                query_url += f"year={year}"
                if token is not None:
                    query_url += f"&token={token}"
            else:
                if token is not None:
                    query_url += f"token={token}"
        query_url = urljoin(f"emissions/metrics/class/"
                            f"{vessel_class_id}",
                            query_url)
        url = urljoin(VesselEmissionsAPI.relative_url, query_url)
        return get_single(self.__connection, url, VesselClassMetrics)

__init__(connection=None)

Initializes VesselEmissionsAPI.

Parameters:

Name	Type	Description	Default
connection	Optional[Connection]	API connection configuration. If not provided, the default connection method is used.	None

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

def __init__(self, connection: Optional[Connection] = None):
    """Initializes VesselEmissionsAPI.

    Args:
        connection: API connection configuration. If not provided, the
            default connection method is used.
    """
    if connection is not None:
        em_connection = copy.deepcopy(connection)
        func_type = type(
            em_connection._Connection__get_headers  # type: ignore
        )
        em_connection._Connection__get_headers = func_type(  # type: ignore
            custom_headers, em_connection
        )
        self.__connection = em_connection
    else:
        connection = Connection()
        func_type = type(
            connection._Connection__get_headers  # type: ignore
        )
        connection._Connection__get_headers = func_type(  # type: ignore
            custom_headers, connection
        )
        self.__connection = connection

construct_url_parameters(quantity=None, token=None, include_consumptions=False, include_efficiency_metrics=False, include_distances=False, include_durations=False, include_speed_statistics=False, include_eu_emissions=False, sulphur_content_hfo=None, sulphur_content_lfo=None, sulphur_content_mgo=None, sulphur_content_lng=None)

Construct the request parameters based on the user's input.

Args: quantity: Cargo quantity of the voyage token: Next page token include_consumptions: Include consumption data in response include_efficiency_metrics: Include efficiency metrics in response include_distances: Include distances data in the response include_durations: Include duration data in the response include_speed_statistics: Include speed statistics data in response include_eu_emissions: Include European Union related emissions in response sulphur_content_hfo: Sulphur Content of HFO fuel type sulphur_content_lfo: Sulphur Content of LFO fuel type sulphur_content_mgo: Sulphur Content of MGO fuel type sulphur_content_lng: Sulphur Content of LNG fuel type

Returns:

Type	Description
Dict[Any, str]	The parameters dictionary of Request URL

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

def construct_url_parameters(
        self,
        quantity: Union[int, None] = None,
        token: Optional[str] = None,
        include_consumptions: bool = False,
        include_efficiency_metrics: bool = False,
        include_distances: bool = False,
        include_durations: bool = False,
        include_speed_statistics: bool = False,
        include_eu_emissions: bool = False,
        sulphur_content_hfo: Optional[Union[float, None]] = None,
        sulphur_content_lfo: Optional[Union[float, None]] = None,
        sulphur_content_mgo: Optional[Union[float, None]] = None,
        sulphur_content_lng: Optional[Union[float, None]] = None
) -> Dict[Any, str]:
    """Construct the request parameters based on the user's input.

    Args:
    quantity: Cargo quantity of the voyage
    token: Next page token
    include_consumptions: Include consumption data in response
    include_efficiency_metrics: Include efficiency metrics in response
    include_distances: Include distances data in the response
    include_durations: Include duration data in the response
    include_speed_statistics: Include speed statistics data in response
    include_eu_emissions: Include European Union related
    emissions in response
    sulphur_content_hfo: Sulphur Content of HFO fuel type
    sulphur_content_lfo: Sulphur Content of LFO fuel type
    sulphur_content_mgo: Sulphur Content of MGO fuel type
    sulphur_content_lng: Sulphur Content of LNG fuel type

    Returns:
        The parameters dictionary of Request URL
    """
    params = {}
    if quantity is not None:
        params['quantity'] = str(quantity)
    if token is not None:
        params['token'] = token
    params['include_consumptions'] = str(include_consumptions)
    params['include_efficiency_metrics'] = str(include_efficiency_metrics)
    params['include_distances'] = str(include_distances)
    params['include_durations'] = str(include_durations)
    params['include_speed_statistics'] = str(include_speed_statistics)
    params['include_eu_emissions'] = str(include_eu_emissions)
    if sulphur_content_hfo is not None:
        params['sulphur_content_hfo'] = str(sulphur_content_hfo)
    if sulphur_content_lfo is not None:
        params['sulphur_content_lfo'] = str(sulphur_content_lfo)
    if sulphur_content_mgo is not None:
        params['sulphur_content_mgo'] = str(sulphur_content_mgo)
    if sulphur_content_lng is not None:
        params['sulphur_content_lng'] = str(sulphur_content_lng)
    return params

get_emissions_by_imo(imo, include_consumptions=False, include_efficiency_metrics=False, include_distances=False, include_durations=False, include_speed_statistics=False, include_eu_emissions=False, sulphur_content_hfo=None, sulphur_content_lfo=None, sulphur_content_mgo=None, sulphur_content_lng=None)

Retrieves a list of vessel emissions by its IMO.

Parameters:

Name	Type	Description	Default
imo	int	IMO of the vessel to retrieve emissions.	required
include_consumptions	bool	Include consumption data in the response.	False
include_efficiency_metrics	bool	Include efficiency metrics data in the response.	False
include_distances	bool	Include distances data in the response.	False
include_durations	bool	Include duration data in the response.	False
include_speed_statistics	bool	Include speed statistics data in the response.	False
include_eu_emissions	bool	Include European Union related emissions data in the response.	False
sulphur_content_hfo	Union[float, None]	Sulphur Content of HFO fuel type.	None
sulphur_content_lfo	Union[float, None]	Sulphur Content of LFO fuel type.	None
sulphur_content_mgo	Union[float, None]	Sulphur Content of MGO fuel type.	None
sulphur_content_lng	Union[float, None]	Sulphur Content of LNG fuel type.	None

Returns:

Type	Description
List[EmissionsEstimation]	A list of vessel emissions or None if no vessel with the specified IMO has been found.

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

def get_emissions_by_imo(
        self,
        imo: int,
        include_consumptions: bool = False,
        include_efficiency_metrics: bool = False,
        include_distances: bool = False,
        include_durations: bool = False,
        include_speed_statistics: bool = False,
        include_eu_emissions: bool = False,
        sulphur_content_hfo: Union[float, None] = None,
        sulphur_content_lfo: Union[float, None] = None,
        sulphur_content_mgo: Union[float, None] = None,
        sulphur_content_lng: Union[float, None] = None
) -> List[EmissionsEstimation]:
    """Retrieves a list of vessel emissions by its IMO.

    Args:
        imo: IMO of the vessel to retrieve emissions.
        include_consumptions: Include consumption data in the response.
        include_efficiency_metrics: Include efficiency metrics
         data in the response.
        include_distances: Include distances data in the response.
        include_durations: Include duration data in the response.
        include_speed_statistics: Include speed statistics
         data in the response.
        include_eu_emissions: Include European Union related
         emissions data in the response.
        sulphur_content_hfo: Sulphur Content of HFO fuel type.
        sulphur_content_lfo: Sulphur Content of LFO fuel type.
        sulphur_content_mgo: Sulphur Content of MGO fuel type.
        sulphur_content_lng: Sulphur Content of LNG fuel type.

    Returns:
        A list of vessel emissions or None if no vessel
         with the specified IMO has been found.
    """
    params_dict = self.construct_url_parameters(
        include_consumptions=include_consumptions,
        include_efficiency_metrics=include_efficiency_metrics,
        include_distances=include_distances,
        include_durations=include_durations,
        include_speed_statistics=include_speed_statistics,
        include_eu_emissions=include_eu_emissions,
        sulphur_content_hfo=sulphur_content_hfo,
        sulphur_content_lfo=sulphur_content_lfo,
        sulphur_content_mgo=sulphur_content_mgo,
        sulphur_content_lng=sulphur_content_lng
    )
    query_url = make_url('emissions',
                         'imo', imo,
                         **params_dict)
    url = urljoin(VesselEmissionsAPI.relative_url, query_url)
    return [i for i in get_multiple(self.__connection,
                                    url,
                                    EmissionsEstimation)]

get_emissions_by_imo_and_voyage_number(imo, voyage_number, quantity=None, include_consumptions=False, include_efficiency_metrics=False, include_distances=False, include_durations=False, include_speed_statistics=False, include_eu_emissions=False, sulphur_content_hfo=None, sulphur_content_lfo=None, sulphur_content_mgo=None, sulphur_content_lng=None)

Retrieves voyage emissions for a vessel by its IMO and Voyage Number.

Parameters:

Name	Type	Description	Default
imo	int	Vessel IMO to retrieve.	required
voyage_number	int	Voyage Number to retrieve.	required
quantity	Union[int, None]	User defined transported quantity for voyage.	None
include_consumptions	bool	Include consumption data in the response.	False
include_efficiency_metrics	bool	Include efficiency metrics in response.	False
include_distances	bool	Include distances data in the response.	False
include_durations	bool	Include duration data in the response.	False
include_speed_statistics	bool	Include speed statistics in response.	False
include_eu_emissions	bool	Include European Union related	False
sulphur_content_hfo	Union[float, None]	Sulphur Content of HFO fuel type.	None
sulphur_content_lfo	Union[float, None]	Sulphur Content of LFO fuel type.	None
sulphur_content_mgo	Union[float, None]	Sulphur Content of MGO fuel type.	None
sulphur_content_lng	Union[float, None]	Sulphur Content of LNG fuel type.	None

Returns:

Type	Description
Optional[EmissionsEstimation]	EmissionsEstimation if no vessel with
Optional[EmissionsEstimation]	the specified IMO or Voyage Number has been found.

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

def get_emissions_by_imo_and_voyage_number(
        self,
        imo: int,
        voyage_number: int,
        quantity: Union[int, None] = None,
        include_consumptions: bool = False,
        include_efficiency_metrics: bool = False,
        include_distances: bool = False,
        include_durations: bool = False,
        include_speed_statistics: bool = False,
        include_eu_emissions: bool = False,
        sulphur_content_hfo: Union[float, None] = None,
        sulphur_content_lfo: Union[float, None] = None,
        sulphur_content_mgo: Union[float, None] = None,
        sulphur_content_lng: Union[float, None] = None
) -> Optional[EmissionsEstimation]:
    """Retrieves voyage emissions for a vessel by its IMO and Voyage Number.

    Args:
        imo: Vessel IMO to retrieve.
        voyage_number: Voyage Number to retrieve.
        quantity: User defined transported quantity for voyage.
        include_consumptions: Include consumption data in the response.
        include_efficiency_metrics: Include efficiency metrics in response.
        include_distances: Include distances data in the response.
        include_durations: Include duration data in the response.
        include_speed_statistics: Include speed statistics in response.
        include_eu_emissions: Include European Union related
        emissions in response.
        sulphur_content_hfo: Sulphur Content of HFO fuel type.
        sulphur_content_lfo: Sulphur Content of LFO fuel type.
        sulphur_content_mgo: Sulphur Content of MGO fuel type.
        sulphur_content_lng: Sulphur Content of LNG fuel type.

    Returns:
        EmissionsEstimation if no vessel with
        the specified IMO or Voyage Number has been found.
    """
    params_dict = self.construct_url_parameters(
        quantity=quantity,
        include_consumptions=include_consumptions,
        include_efficiency_metrics=include_efficiency_metrics,
        include_distances=include_distances,
        include_durations=include_durations,
        include_speed_statistics=include_speed_statistics,
        include_eu_emissions=include_eu_emissions,
        sulphur_content_hfo=sulphur_content_hfo,
        sulphur_content_lfo=sulphur_content_lfo,
        sulphur_content_mgo=sulphur_content_mgo,
        sulphur_content_lng=sulphur_content_lng)
    query_url = make_url('emissions',
                         'imo', imo,
                         'voyage_number', voyage_number,
                         **params_dict)
    url = urljoin(VesselEmissionsAPI.relative_url, query_url)
    return get_single(self.__connection, url, EmissionsEstimation)

get_emissions_by_vessel_class_id(vessel_class_id, token=None, include_consumptions=False, include_efficiency_metrics=False, include_distances=False, include_durations=False, include_speed_statistics=False, include_eu_emissions=False)

Get emissions estimations for a vessel class (supports incremental updates).

Parameters:

Name	Type	Description	Default
vessel_class_id	int	The vessel class to retrieve	required
token	Union[str, None]	Next page token	None
include_consumptions	bool	Include consumption data in the response.	False
include_efficiency_metrics	bool	Include efficiency metrics data in the response.	False
include_distances	bool	Include distances data in the response.	False
include_durations	bool	Include duration data in the response.	False
include_speed_statistics	bool	Include speed statistics data in the response.	False
include_eu_emissions	bool	Include European Union related emissions data in the response.	False

Returns:

Type	Description
Optional[VesselClassEmissions]	List of emissions estimation for all available voyages of a vessel class.

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

def get_emissions_by_vessel_class_id(
        self,
        vessel_class_id: int,
        token: Union[str, None] = None,
        include_consumptions: bool = False,
        include_efficiency_metrics: bool = False,
        include_distances: bool = False,
        include_durations: bool = False,
        include_speed_statistics: bool = False,
        include_eu_emissions: bool = False
) -> Optional[VesselClassEmissions]:
    """Get emissions estimations for a vessel class (supports incremental updates).

    Args:
        vessel_class_id: The vessel class to retrieve
        token: Next page token
        include_consumptions: Include consumption data in the response.
        include_efficiency_metrics: Include efficiency metrics
         data in the response.
        include_distances: Include distances data in the response.
        include_durations: Include duration data in the response.
        include_speed_statistics: Include speed statistics
         data in the response.
        include_eu_emissions: Include European Union related emissions
         data in the response.

    Returns:
        List of emissions estimation for all
         available voyages of a vessel class.

    """
    params_dict = self.construct_url_parameters(
        token=token,
        include_consumptions=include_consumptions,
        include_efficiency_metrics=include_efficiency_metrics,
        include_distances=include_distances,
        include_durations=include_durations,
        include_speed_statistics=include_speed_statistics,
        include_eu_emissions=include_eu_emissions
    )
    query_url = make_url('emissions',
                         'class', vessel_class_id,
                         **params_dict)

    url = urljoin(VesselEmissionsAPI.relative_url, query_url)
    return get_single(self.__connection,
                      url,
                      VesselClassEmissions)

get_metrics_by_imo(imo, year=None)

Get vessel metrics.

Parameters:

Name	Type	Description	Default
imo	int	Vessel IMO to retrieve	required
year	Union[int, None]	The year for the annual metrics	None

Returns:

Type	Description
List[VesselMetrics]	VesselMetrics for the requested IMO

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

def get_metrics_by_imo(
        self,
        imo: int,
        year: Union[int, None] = None
) -> List[VesselMetrics]:
    """Get vessel metrics.

    Args:
        imo: Vessel IMO to retrieve
        year: The year for the annual metrics

    Returns:
        VesselMetrics for the requested IMO

    """
    url = urljoin(VesselEmissionsAPI.relative_url,
                  f"emissions/metrics/imo/{imo}")
    if year is not None:
        url = urljoin(url, f"?year={year}")
    return [i for i in get_multiple(self.__connection,
                                    url,
                                    VesselMetrics)]

get_metrics_by_vessel_class_id(vessel_class_id, year=None, token=None)

Get vessel class metrics.

Parameters:

Name	Type	Description	Default
vessel_class_id	int	The vessel class to retrieve	required
year	Union[int, None]	The year for the annual metrics	None
token	Union[str, None]	Next page token	None

Returns:

Type	Description
Optional[VesselClassMetrics]	VesselClassMetrics for the requested Class

Source code in signal_ocean/vessel_emissions/vessel_emissions_api.py

def get_metrics_by_vessel_class_id(self,
                                   vessel_class_id: int,
                                   year: Union[int, None] = None,
                                   token: Union[str, None] = None
                                   ) -> Optional[VesselClassMetrics]:
    """Get vessel class metrics.

    Args:
        vessel_class_id: The vessel class to retrieve
        year: The year for the annual metrics
        token: Next page token

    Returns:
        VesselClassMetrics for the requested Class

    """
    if year is None and token is None:
        query_url = ""
    else:
        query_url = "?"
        if year is not None:
            query_url += f"year={year}"
            if token is not None:
                query_url += f"&token={token}"
        else:
            if token is not None:
                query_url += f"token={token}"
    query_url = urljoin(f"emissions/metrics/class/"
                        f"{vessel_class_id}",
                        query_url)
    url = urljoin(VesselEmissionsAPI.relative_url, query_url)
    return get_single(self.__connection, url, VesselClassMetrics)