nrel.hive.util

Submodules

Package Contents

Classes

DictOps

H3Ops

DictReaderStepper

takes a DictReader and steps through it, using one specific column's values as a way to split

TupleOps

Functions

hours_to_seconds(→ Seconds)

Attributes

RequestId

VehicleId

StationId

BaseId

PowertrainId

PowercurveId

PassengerId

GeoId

LinkId

RouteStepPointer

H3Line

SimStep

KwH

J

Kw

Meters

Kilometers

Feet

Miles

Mph

Kmph

Seconds

Hours

Currency

Percentage

Ratio

HOURS_TO_SECONDS

SECONDS_IN_HOUR

SECONDS_TO_HOURS

KMPH_TO_MPH

KM_TO_MILE

WH_TO_KWH

class nrel.hive.util.DictOps[source]
K
V
classmethod iterate_vals(xs: immutables.Map[K, V], key: Callable[[V], _typeshed.SupportsRichComparison] | None = None) → Tuple[V, Ellipsis][source]

helper function for iterating on Maps in HIVE which sorts values by key unless a key function is provided. for all stateful Map collections that are being iterated on in HIVE, we need to sort them _somehow_ to guarantee deterministic runs.

if no key function is provided, the values are sorted by Map key. if a key function is provided, it takes the value type as input and returns a sortable value.

Parameters:

xs (immutables.Map[K, V]) – collection to iterate values of

Returns:

values of xs, sorted by key

Return type:

Tuple[V, …]

classmethod iterate_items(xs: immutables.Map[K, V], key: Callable[[Tuple[K, V]], _typeshed.SupportsRichComparison] | None = None) → List[Tuple[K, V]][source]

helper function for iterating on Maps in HIVE which sorts values by key unless a key function is provided. for all stateful Map collections that are being iterated on in HIVE, we need to sort them _somehow_ to guarantee deterministic runs.

Parameters:

xs (immutables.Map[K, V]) – collection to iterate values of

Returns:

values of xs, sorted by key

Return type:

Tuple[V, …]

classmethod iterate_sim_coll(collection: immutables.Map[K, V], filter_function: Callable[[V], bool] | None = None, sort_key: Callable | None = None) → Tuple[V, Ellipsis][source]

helper to iterate through a collection on the SimulationState with optional sort key function and filter function. performs filter before sort if both are provided.

Parameters:

  • collection (immutables.Map[K, V]) – collection on SimulationState

  • filter_function (Optional[Callable[[V], bool]], optional) – _description_, defaults to None

  • sort_key (Optional[Callable], optional) – _description_, defaults to None

Returns:

_description_

Return type:

Tuple[V, …]

classmethod add_to_dict(xs: immutables.Map[K, V], obj_id: K, obj: V) → immutables.Map[K, V][source]

updates Dicts for arbitrary keys and values performs a shallow copy and update, treating Dict as an immutable hash table

Parameters:

  • xs –

  • obj_id –

  • obj –

Returns:

classmethod remove_from_dict(xs: immutables.Map[K, V], obj_id: K) → immutables.Map[K, V][source]

updates Dicts for arbitrary keys and values performs a shallow copy and update, treating Dict as an immutable hash table

Parameters:

  • xs –

  • obj_id –

Returns:

classmethod merge_dicts(old: immutables.Map[K, V], new: immutables.Map[K, V]) → immutables.Map[K, V][source]

merges two Dictionaries, replacing old kv pairs with new ones

Parameters:

  • old – the old Dict

  • new – the new Dict

Returns:

a merged Dict

classmethod add_to_collection_dict(xs: immutables.Map[str, FrozenSet[V]], collection_id: str, obj_id: V) → immutables.Map[str, FrozenSet[V]][source]

updates Dicts that track collections of entities performs a shallow copy and update, treating Dict as an immutable hash table

Parameters:

  • xs –

  • collection_id –

  • obj_id –

Returns:

classmethod add_to_stack_dict(xs: immutables.Map[str, Tuple[V, Ellipsis]], collection_id: str, obj: V) → immutables.Map[str, Tuple[V, Ellipsis]][source]

updates Dicts that hold a stack of entities; note that the head of the tuple represents the top of the stack; elements always get inserted into the head; performs a shallow copy and update, treating Dict as an immutable hash table

Parameters:

  • xs –

  • collection_id –

  • obj_id –

Returns:

classmethod pop_from_stack_dict(xs: immutables.Map[str, Tuple[V, Ellipsis]], collection_id: str) → Tuple[V | None, immutables.Map[str, Tuple[V, Ellipsis]]][source]

pops an element from the stack and returns it; note that the head of the tuple represents the top of the stack; popped elements come from the tuple head; performs a shallow copy and update, treating Dict as an immutable hash table

Parameters:

  • xs –

  • collection_id –

  • obj_id –

Returns:

classmethod remove_from_collection_dict(xs: immutables.Map[str, FrozenSet[V]], collection_id: str, obj_id: str) → immutables.Map[str, FrozenSet[V]][source]

updates Dicts that track collections of entities performs a shallow copy and update, treating Dict as an immutable hash table when a geoid has no ids after a remove, it deletes that geoid, to prevent geoid Dict memory leaks

Parameters:

  • xs –

  • collection_id –

  • obj_id –

Returns:

classmethod update_entity_dictionaries(updated_entity: nrel.hive.model.entity.Entity, entities: immutables.Map[nrel.hive.util.typealiases.EntityId, nrel.hive.model.entity.Entity], locations: immutables.Map[nrel.hive.util.typealiases.GeoId, FrozenSet[nrel.hive.util.typealiases.EntityId]], search: immutables.Map[nrel.hive.util.typealiases.GeoId, FrozenSet[nrel.hive.util.typealiases.EntityId]], sim_h3_search_resolution: int) → EntityUpdateResult[source]

updates all dictionaries related to an entity

Parameters:

  • updated_entity – an entity which itself should have an “id” and a “geoid” attribute

  • entities – the dictionary containing Entities by EntityId

  • locations – the finest-resolution geoindex of this entity type

  • search – the upper-level resolution geoindex

  • sim_h3_search_resolution – the h3 resolution of the search collection

Returns:

the updated dictionaries

exception nrel.hive.util.StateOfChargeError(soc)[source]

Bases: Exception

state of charge must exist in the range [0, 1]

__str__()[source]

Return str(self).

exception nrel.hive.util.StateTransitionError(msg: str, this_state_name: str | None = None, next_state_name: str | None = None)[source]

Bases: Exception

calls out a breach in the simulation’s physics observed when a state transition’s invariants are not met.

__str__()[source]

Return str(self).

exception nrel.hive.util.SimulationStateError(msg)[source]

Bases: Exception

errors related to SimulationState operations

__str__()[source]

Return str(self).

exception nrel.hive.util.RouteStepError(msg)[source]

Bases: Exception

errors related to stepping forward along a route

__str__()[source]

Return str(self).

exception nrel.hive.util.EntityError(msg)[source]

Bases: Exception

errors related to methods on entities such as vehicles or stations.

__str__()[source]

Return str(self).

exception nrel.hive.util.UnitError(msg)[source]

Bases: Exception

errors related to units

__str__()[source]

Return str(self).

exception nrel.hive.util.H3Error(msg)[source]

Bases: Exception

errors related to H3 operations

__str__()[source]

Return str(self).

class nrel.hive.util.H3Ops[source]
classmethod nearest_entity_by_great_circle_distance(geoid: nrel.hive.util.typealiases.GeoId, entities: Iterable[nrel.hive.model.entity.Entity], entity_search: immutables.Map[nrel.hive.util.typealiases.GeoId, FrozenSet[nrel.hive.util.typealiases.EntityId]], sim_h3_search_resolution: int, is_valid: Callable[[Any], bool] = lambda x: ..., max_search_distance_km: nrel.hive.util.units.Kilometers = 10) → nrel.hive.model.entity.Entity | None[source]

returns the closest entity to the given geoid. In the case of a tie, the first entity encountered is returned. invariant: the Entity has a geoid field (Entity.geoid)

Parameters:

  • geoid – the search origin

  • entities – a collection of a certain type of entity, by Id type

  • entity_search – the location of objects of this entity type, registered at a high-level grid resolution

  • sim_h3_search_resolution – the h3 resolution of the entity_search collection

  • is_valid – a function used to filter valid search results, such as checking stations for charger_id availability

  • k – the number of concentric rings to check in the high-level search

  • max_search_distance_km – the maximum distance a result can be from the search origin

Returns:

the nearest entity, or, None if not found within the constraints

classmethod nearest_entity(geoid: nrel.hive.util.typealiases.GeoId, entities: Iterable[nrel.hive.model.entity.Entity], entity_search: immutables.Map[nrel.hive.util.typealiases.GeoId, FrozenSet[nrel.hive.util.typealiases.EntityId]], sim_h3_search_resolution: int, distance_function: Callable[[Any], float], is_valid: Callable[[Any], bool] = lambda x: ..., max_search_distance_km: nrel.hive.util.units.Kilometers = 10) → nrel.hive.model.entity.Entity | None[source]

returns the closest entity to the given geoid. In the case of a tie, the first entity encountered is returned. invariant: the Entity has a geoid field (Entity.geoid)

Parameters:

  • geoid – the search origin

  • entities – a collection of a certain type of entity, by Id type

  • entity_search – the location of objects of this entity type, registered at a high-level grid resolution

  • sim_h3_search_resolution – the h3 resolution of the entity_search collection

  • is_valid – a function used to filter valid search results, such as checking stations for charger_id availability

  • distance_function – a function used to evaluate the distance metric for selection

  • k – the number of concentric rings to check in the high-level search

  • max_search_distance_km – the maximum distance a result can be from the search origin

Returns:

the nearest entity, or, None if not found within the constraints

classmethod get_entities_at_cell(search_cell: nrel.hive.util.typealiases.GeoId, entity_search: immutables.Map[nrel.hive.util.typealiases.GeoId, FrozenSet[nrel.hive.util.typealiases.EntityId]], entities: Iterable[nrel.hive.model.entity.Entity]) → Tuple[nrel.hive.model.entity.Entity, Ellipsis][source]

gives us entities within a high-level search cell

Parameters:

  • search_cell – the search-level h3 position we are looking at

  • entity_search – the upper-level search collection for this entity type

  • entities – the actual entities

Returns:

any entities which are located at this search-level cell

classmethod nearest_entity_point_to_point(geoid: nrel.hive.util.typealiases.GeoId, entities: Dict[nrel.hive.util.typealiases.EntityId, nrel.hive.model.entity.Entity], entity_locations: immutables.Map[nrel.hive.util.typealiases.GeoId, Tuple[nrel.hive.util.typealiases.EntityId, Ellipsis]], is_valid: Callable = lambda x: ...) → nrel.hive.model.entity.Entity | None[source]

A nearest neighbor search that scans all entities and returns the one with the lowest distance to the geoid.

Parameters:

  • geoid – GeoId to match to

  • entities – Entities to search over

  • entity_locations – Location of entities

  • is_valid – Optional function to filter for valid entities

Returns:

an optional entity if found

classmethod great_circle_distance(a: nrel.hive.util.typealiases.GeoId, b: nrel.hive.util.typealiases.GeoId) → nrel.hive.util.units.Kilometers[source]

computes the distance between two geoids

Parameters:

  • a – one geoid

  • b – another geoid

Returns:

the haversine distance between the two GeoIds

finds the GeoId which is some percentage between two GeoIds along a line

Parameters:

  • available_time_seconds – the amount of time to traverse

  • link – the link we are finding a mid point along

Returns:

a GeoId along the Link

class nrel.hive.util.DictReaderStepper(dict_reader: Iterator[Dict[str, str]], file_reference: TextIO | None, step_column_name: str, initial_stop_condition: Callable = lambda x: ..., parser: Callable = lambda x: ...)[source]

takes a DictReader and steps through it, using one specific column’s values as a way to split iteration over windows (of time, or other).

read_until_value consumes the next set of rows that fall within the next upper-value for the next window.

destruction: should be explicitly closed via DictReaderStepper.close()

classmethod build(file: str | pathlib.Path, step_column_name: str, initial_stop_condition: Callable = lambda x: ..., parser: Callable = lambda x: ...) → Tuple[Exception | None, DictReaderStepper | None][source]

alternative constructor that takes a file path and returns a DictReaderStepper, or, a failure

Parameters:

  • file – the file path

  • step_column_name – the column we are comparing new bounds against

  • initial_stop_condition – the initial bounds - set low (zero) for ascending, high (inf) for descending note: descending not yet implemented

  • parser – an optional parameter for parsing the input_config value

Returns:

a new reader or an exception

classmethod from_iterator(data: Iterator[Dict[str, str]], step_column_name: str, initial_stop_condition: Callable = lambda x: ..., parser: Callable = lambda x: ...) → DictReaderStepper[source]

allows for substituting a simple Dict Iterator in place of loading from a file, allowing for programmatic data loading (for debugging, or, for dealing with default file contents)

Parameters:

  • data – a provider of row-wise data similar to a CSV

  • step_column_name – the key we are expecting in each Dict that we are comparing new bounds against

  • initial_stop_condition – the initial bounds - set low (zero) for ascending, high (inf) for descending note: descending not yet implemented

  • parser – an optional parameter for parsing the input_config value

Returns:

a new reader or an exception

read_until_stop_condition(stop_condition: Callable) → Iterator[Dict[str, str]][source]

reads rows from the DictReader as long as step_column_name is less than or equal to “value”

Parameters:

stop_condition – the condition to validate. we will read all new rows as long as each row’s value evaluates to True

Returns:

the updated DictReaderStepper and a tuple of rows, which may be empty if no new rows are consumable.

close()[source]
class nrel.hive.util.TupleOps[source]
T
classmethod is_empty(xs: Tuple[T, Ellipsis]) → bool[source]
classmethod non_empty(xs: Tuple[T, Ellipsis]) → bool[source]
classmethod remove(xs: Tuple[T, Ellipsis], value: T) → Tuple[T, Ellipsis][source]
classmethod head(xs: Tuple[T, Ellipsis]) → T[source]
classmethod head_optional(xs: Tuple[T, Ellipsis]) → T | None[source]
classmethod last(xs: Tuple[T, Ellipsis]) → T[source]
classmethod last_optional(xs: Tuple[T, Ellipsis]) → T | None[source]
classmethod head_tail(tup: Tuple[T, Ellipsis]) → Tuple[T, Tuple[T, Ellipsis]][source]
classmethod tail(tup: Tuple[T, Ellipsis]) → Tuple[T, Ellipsis][source]
classmethod partition(predicate: Callable[[T], bool], t: Tuple[T, Ellipsis]) → Tuple[Tuple[T, Ellipsis], Tuple[T, Ellipsis]][source]

partitions a tuple into two tuples where members of the first tuple match the case where the provided predicate is True

taken from https://docs.python.org/3/library/itertools.html (but result tuples reversed for readability)

Parameters:

  • predicate – tests membership in result tuples

  • t – the source tuple

Returns:

classmethod flatten(nested_tuple: Tuple[Tuple[T, Ellipsis], Ellipsis]) → Tuple[T, Ellipsis][source]

flattens a tuple of tuples

taken from https://stackoverflow.com/a/10636583/11087167

Parameters:

nested_tuple – tuple to flatten

Returns:

flattened tuple

classmethod prepend(x: T, xs: Tuple[T, Ellipsis]) → Tuple[T, Ellipsis][source]

prepends an element to a tuple

nrel.hive.util.RequestId
nrel.hive.util.VehicleId
nrel.hive.util.StationId
nrel.hive.util.BaseId
nrel.hive.util.PowertrainId
nrel.hive.util.PowercurveId
nrel.hive.util.PassengerId
nrel.hive.util.GeoId
nrel.hive.util.LinkId
nrel.hive.util.RouteStepPointer
nrel.hive.util.H3Line
nrel.hive.util.SimStep
nrel.hive.util.KwH
nrel.hive.util.J
nrel.hive.util.Kw
nrel.hive.util.Meters
nrel.hive.util.Kilometers
nrel.hive.util.Feet
nrel.hive.util.Miles
nrel.hive.util.Mph
nrel.hive.util.Kmph
nrel.hive.util.Seconds
nrel.hive.util.Hours
nrel.hive.util.Currency
nrel.hive.util.Percentage
nrel.hive.util.Ratio
nrel.hive.util.HOURS_TO_SECONDS = 3600
nrel.hive.util.hours_to_seconds(hours: Hours) → Seconds[source]
nrel.hive.util.SECONDS_IN_HOUR = 3600
nrel.hive.util.SECONDS_TO_HOURS
nrel.hive.util.KMPH_TO_MPH = 0.621371
nrel.hive.util.KM_TO_MILE = 0.621371
nrel.hive.util.WH_TO_KWH = 0.001