Graph Engine

Engine

Bases: Enum

Enum representing different types of graph engines.

Attributes:

Name	Type	Description
MEMORY		In-memory graph engine.
SQLITE		SQLite-based graph engine.

Source code in gamms/typing/graph_engine.py

class Engine(Enum):
    """
    Enum representing different types of graph engines.

    Attributes:
        MEMORY: In-memory graph engine.
        SQLITE: SQLite-based graph engine.        
    """
    MEMORY = 0
    SQLITE = 1

IGraph

Bases: ABC

Abstract base class representing a graph structure.

The graph consists of nodes and edges, allowing for addition, removal, and retrieval of these elements.

Source code in gamms/typing/graph_engine.py

class IGraph(ABC):
    """
    Abstract base class representing a graph structure.

    The graph consists of nodes and edges, allowing for addition, removal, and retrieval of these elements.
    """

    @abstractmethod
    def add_node(self, node_data: Dict[str, Any]) -> None:
        """
        Add a new node to the graph.

        Args:
            node_data (Dict[str, Any]): A dictionary containing the node's attributes. Expected keys include:
                - 'id' (int): Unique identifier for the node.
                - 'x' (float): X-coordinate of the node.
                - 'y' (float): Y-coordinate of the node.

        Raises:
            ValueError: If the node_data is missing required fields or contains invalid data.
            KeyError: If a node with the same ID already exists in the graph.
        """
        pass

    @abstractmethod
    def add_edge(self, edge_data: Dict[str, Any]) -> None:
        """
        Add a new edge to the graph.

        Args:
            edge_data (Dict[str, Any]): A dictionary containing the edge's attributes. Expected keys include:
                - 'id' (int): Unique identifier for the edge.
                - 'source' (int): ID of the source node.
                - 'target' (int): ID of the target node.
                - 'length' (float): Length of the edge.
                - 'linestring' (List[Tuple[float, float]], optional): Geometry of the edge.

        Raises:
            ValueError: If the edge_data is missing required fields, contains invalid data, or references non-existent nodes.
            KeyError: If an edge with the same ID already exists in the graph.
            KeyError: If source or target nodes do not exist in the graph.
        """
        pass

    @abstractmethod
    @overload
    def get_nodes(self) -> Iterator[int]:
        """
        Creates an iterator of node IDs in the graph.

        Returns:
            Iterator[int]: An iterator that yields node IDs.
        """
        pass

    @abstractmethod
    @overload
    def get_nodes(self, d: float, x: float, y: float) -> Iterator[int]:
        """
        Creates an iterator of node IDs in the graph.

        If d is non-negative, it returns nodes within a distance d from the point (x, y).
        May return nodes that are farther than d but will always return nodes that are within d.

        Returns:
            Iterator[int]: An iterator that yields node IDs.
        """
        pass

    @abstractmethod
    @overload
    def get_edges(self) -> Iterator[int]:
        """
        Creates an iterator of edge IDs in the graph.

        Returns:
            Iterator[int]: An iterator that yields edge IDs.
        """
        pass

    @abstractmethod
    @overload
    def get_edges(self, d: float, x: float, y: float) -> Iterator[int]:
        """
        Creates an iterator of edge IDs in the graph.
        If d is non-negative, it returns edges within a distance d from the point (x, y).
        May return edges that are farther than d but will always return edges that are within d.

        "Within" means that atleast one of the edge's nodes is within distance d from the point (x, y).

        Returns:
            Iterator[int]: An iterator that yields edge IDs.
        """
        pass

    @abstractmethod
    def update_node(self, node_data: Dict[str, Any]) -> None:
        """
        Update an existing node's attributes.

        Args:
            node_data (Dict[str, Any]): A dictionary containing the node's updated attributes. Must include:
                - 'id' (int): Unique identifier for the node.
                - Any other attributes to be updated (e.g., 'x', 'y').

        Raises:
            KeyError: If the node with the specified ID does not exist.
            ValueError: If the provided data is invalid.
        """
        pass

    @abstractmethod
    def update_edge(self, edge_data: Dict[str, Any]) -> None:
        """
        Update an existing edge's attributes.

        Args:
            edge_data (Dict[str, Any]): A dictionary containing the edge's updated attributes. Must include:
                - 'id' (int): Unique identifier for the edge.
                - Any other attributes to be updated (e.g., 'source', 'target', 'length', 'linestring').

        Raises:
            KeyError: If the edge with the specified ID does not exist.
            ValueError: If the provided data is invalid or references non-existent nodes.
        """
        pass

    @abstractmethod
    def remove_node(self, node_id: int) -> None:
        """
        Remove a node from the graph. Removing a node will also remove all edges connected to it.

        Args:
            node_id (int): The unique identifier of the node to be removed.
        """
        pass

    @abstractmethod
    def remove_edge(self, edge_id: int) -> None:
        """
        Remove an edge from the graph.

        Args:
            edge_id (int): The unique identifier of the edge to be removed.
        """
        pass

    @abstractmethod
    def get_node(self, node_id: int) -> Node:
        """
        Retrieve the attributes of a specific node.

        Args:
            node_id (int): The unique identifier of the node to retrieve.

        Returns:
            Dict[str, Any]: A dictionary containing the node's attributes.

        Raises:
            KeyError: If the node with the specified ID does not exist.
        """
        pass

    @abstractmethod
    def get_edge(self, edge_id: int) -> OSMEdge:
        """
        Retrieve the attributes of a specific edge.

        Args:
            edge_id (int): The unique identifier of the edge to retrieve.

        Returns:
            OSMEdge: A dictionary containing the edge's attributes.

        Raises:
            KeyError: If the edge with the specified ID does not exist.
        """
        pass

    @abstractmethod
    def get_neighbors(self, node_id: int) -> Iterator[int]:
        """
        Get the neighbors of a specific node.

        Args:
            node_id (int): The unique identifier of the node whose neighbors are to be retrieved.

        Returns:
            Iterator[int]: An iterator that yields the IDs of neighboring nodes.

        Raises:
            KeyError: If the node with the specified ID does not exist.
        """
        pass

add_edge(edge_data) abstractmethod

Add a new edge to the graph.

Parameters:

Name	Type	Description	Default
edge_data	Dict[str, Any]	A dictionary containing the edge's attributes. Expected keys include: - 'id' (int): Unique identifier for the edge. - 'source' (int): ID of the source node. - 'target' (int): ID of the target node. - 'length' (float): Length of the edge. - 'linestring' (List[Tuple[float, float]], optional): Geometry of the edge.	required

Raises:

Type	Description
ValueError	If the edge_data is missing required fields, contains invalid data, or references non-existent nodes.
KeyError	If an edge with the same ID already exists in the graph.
KeyError	If source or target nodes do not exist in the graph.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def add_edge(self, edge_data: Dict[str, Any]) -> None:
    """
    Add a new edge to the graph.

    Args:
        edge_data (Dict[str, Any]): A dictionary containing the edge's attributes. Expected keys include:
            - 'id' (int): Unique identifier for the edge.
            - 'source' (int): ID of the source node.
            - 'target' (int): ID of the target node.
            - 'length' (float): Length of the edge.
            - 'linestring' (List[Tuple[float, float]], optional): Geometry of the edge.

    Raises:
        ValueError: If the edge_data is missing required fields, contains invalid data, or references non-existent nodes.
        KeyError: If an edge with the same ID already exists in the graph.
        KeyError: If source or target nodes do not exist in the graph.
    """
    pass

add_node(node_data) abstractmethod

Add a new node to the graph.

Parameters:

Name	Type	Description	Default
node_data	Dict[str, Any]	A dictionary containing the node's attributes. Expected keys include: - 'id' (int): Unique identifier for the node. - 'x' (float): X-coordinate of the node. - 'y' (float): Y-coordinate of the node.	required

Raises:

Type	Description
ValueError	If the node_data is missing required fields or contains invalid data.
KeyError	If a node with the same ID already exists in the graph.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def add_node(self, node_data: Dict[str, Any]) -> None:
    """
    Add a new node to the graph.

    Args:
        node_data (Dict[str, Any]): A dictionary containing the node's attributes. Expected keys include:
            - 'id' (int): Unique identifier for the node.
            - 'x' (float): X-coordinate of the node.
            - 'y' (float): Y-coordinate of the node.

    Raises:
        ValueError: If the node_data is missing required fields or contains invalid data.
        KeyError: If a node with the same ID already exists in the graph.
    """
    pass

get_edge(edge_id) abstractmethod

Retrieve the attributes of a specific edge.

Parameters:

Name	Type	Description	Default
edge_id	int	The unique identifier of the edge to retrieve.	required

Returns:

Name	Type	Description
OSMEdge	OSMEdge	A dictionary containing the edge's attributes.

Raises:

Type	Description
KeyError	If the edge with the specified ID does not exist.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def get_edge(self, edge_id: int) -> OSMEdge:
    """
    Retrieve the attributes of a specific edge.

    Args:
        edge_id (int): The unique identifier of the edge to retrieve.

    Returns:
        OSMEdge: A dictionary containing the edge's attributes.

    Raises:
        KeyError: If the edge with the specified ID does not exist.
    """
    pass

get_neighbors(node_id) abstractmethod

Get the neighbors of a specific node.

Parameters:

Name	Type	Description	Default
node_id	int	The unique identifier of the node whose neighbors are to be retrieved.	required

Returns:

Type	Description
Iterator[int]	Iterator[int]: An iterator that yields the IDs of neighboring nodes.

Raises:

Type	Description
KeyError	If the node with the specified ID does not exist.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def get_neighbors(self, node_id: int) -> Iterator[int]:
    """
    Get the neighbors of a specific node.

    Args:
        node_id (int): The unique identifier of the node whose neighbors are to be retrieved.

    Returns:
        Iterator[int]: An iterator that yields the IDs of neighboring nodes.

    Raises:
        KeyError: If the node with the specified ID does not exist.
    """
    pass

get_node(node_id) abstractmethod

Retrieve the attributes of a specific node.

Parameters:

Name	Type	Description	Default
node_id	int	The unique identifier of the node to retrieve.	required

Returns:

Type	Description
Node	Dict[str, Any]: A dictionary containing the node's attributes.

Raises:

Type	Description
KeyError	If the node with the specified ID does not exist.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def get_node(self, node_id: int) -> Node:
    """
    Retrieve the attributes of a specific node.

    Args:
        node_id (int): The unique identifier of the node to retrieve.

    Returns:
        Dict[str, Any]: A dictionary containing the node's attributes.

    Raises:
        KeyError: If the node with the specified ID does not exist.
    """
    pass

remove_edge(edge_id) abstractmethod

Remove an edge from the graph.

Parameters:

Name	Type	Description	Default
edge_id	int	The unique identifier of the edge to be removed.	required

Source code in gamms/typing/graph_engine.py

@abstractmethod
def remove_edge(self, edge_id: int) -> None:
    """
    Remove an edge from the graph.

    Args:
        edge_id (int): The unique identifier of the edge to be removed.
    """
    pass

remove_node(node_id) abstractmethod

Remove a node from the graph. Removing a node will also remove all edges connected to it.

Parameters:

Name	Type	Description	Default
node_id	int	The unique identifier of the node to be removed.	required

Source code in gamms/typing/graph_engine.py

@abstractmethod
def remove_node(self, node_id: int) -> None:
    """
    Remove a node from the graph. Removing a node will also remove all edges connected to it.

    Args:
        node_id (int): The unique identifier of the node to be removed.
    """
    pass

update_edge(edge_data) abstractmethod

Update an existing edge's attributes.

Parameters:

Name	Type	Description	Default
edge_data	Dict[str, Any]	A dictionary containing the edge's updated attributes. Must include: - 'id' (int): Unique identifier for the edge. - Any other attributes to be updated (e.g., 'source', 'target', 'length', 'linestring').	required

Raises:

Type	Description
KeyError	If the edge with the specified ID does not exist.
ValueError	If the provided data is invalid or references non-existent nodes.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def update_edge(self, edge_data: Dict[str, Any]) -> None:
    """
    Update an existing edge's attributes.

    Args:
        edge_data (Dict[str, Any]): A dictionary containing the edge's updated attributes. Must include:
            - 'id' (int): Unique identifier for the edge.
            - Any other attributes to be updated (e.g., 'source', 'target', 'length', 'linestring').

    Raises:
        KeyError: If the edge with the specified ID does not exist.
        ValueError: If the provided data is invalid or references non-existent nodes.
    """
    pass

update_node(node_data) abstractmethod

Update an existing node's attributes.

Parameters:

Name	Type	Description	Default
node_data	Dict[str, Any]	A dictionary containing the node's updated attributes. Must include: - 'id' (int): Unique identifier for the node. - Any other attributes to be updated (e.g., 'x', 'y').	required

Raises:

Type	Description
KeyError	If the node with the specified ID does not exist.
ValueError	If the provided data is invalid.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def update_node(self, node_data: Dict[str, Any]) -> None:
    """
    Update an existing node's attributes.

    Args:
        node_data (Dict[str, Any]): A dictionary containing the node's updated attributes. Must include:
            - 'id' (int): Unique identifier for the node.
            - Any other attributes to be updated (e.g., 'x', 'y').

    Raises:
        KeyError: If the node with the specified ID does not exist.
        ValueError: If the provided data is invalid.
    """
    pass

IGraphEngine

Bases: ABC

Abstract base class representing the graph engine.

The graph engine manages the underlying graph structure, providing access and control over graph operations.

Source code in gamms/typing/graph_engine.py

class IGraphEngine(ABC):
    """
    Abstract base class representing the graph engine.

    The graph engine manages the underlying graph structure, providing access and control over graph operations.
    """

    @property
    @abstractmethod
    def graph(self) -> IGraph:
        """
        Get the current graph managed by the engine.

        Returns:
            IGraph: The graph instance being managed.

        Raises:
            RuntimeError: If the graph has not been initialized.
        """
        pass

    @abstractmethod
    def attach_networkx_graph(self, G: nx.Graph) -> IGraph:
        """
        Attach a NetworkX graph to the graph engine.

        Args:
            G (nx.Graph): The NetworkX graph to attach.

        Returns:
            IGraph: The graph instance created from the NetworkX graph.

        Raises:
            ValueError: If the provided graph is invalid or cannot be attached.
        """
        pass

    @abstractmethod
    def terminate(self) -> None:
        """
        Terminate the graph engine and perform necessary cleanup operations.

        This method should ensure that all resources allocated to the graph engine are properly released
        and that any ongoing operations are gracefully stopped.
        """
        pass

graph abstractmethod property

Get the current graph managed by the engine.

Returns:

Name	Type	Description
IGraph	IGraph	The graph instance being managed.

Raises:

Type	Description
RuntimeError	If the graph has not been initialized.

attach_networkx_graph(G) abstractmethod

Attach a NetworkX graph to the graph engine.

Parameters:

Name	Type	Description	Default
G	Graph	The NetworkX graph to attach.	required

Returns:

Name	Type	Description
IGraph	IGraph	The graph instance created from the NetworkX graph.

Raises:

Type	Description
ValueError	If the provided graph is invalid or cannot be attached.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def attach_networkx_graph(self, G: nx.Graph) -> IGraph:
    """
    Attach a NetworkX graph to the graph engine.

    Args:
        G (nx.Graph): The NetworkX graph to attach.

    Returns:
        IGraph: The graph instance created from the NetworkX graph.

    Raises:
        ValueError: If the provided graph is invalid or cannot be attached.
    """
    pass

terminate() abstractmethod

Terminate the graph engine and perform necessary cleanup operations.

This method should ensure that all resources allocated to the graph engine are properly released and that any ongoing operations are gracefully stopped.

Source code in gamms/typing/graph_engine.py

@abstractmethod
def terminate(self) -> None:
    """
    Terminate the graph engine and perform necessary cleanup operations.

    This method should ensure that all resources allocated to the graph engine are properly released
    and that any ongoing operations are gracefully stopped.
    """
    pass

Node

Represents a node within a graph.

Attributes:

Name	Type	Description
id	int	The unique identifier for the node.
x	float	The x-coordinate of the node's position.
y	float	The y-coordinate of the node's position.

Source code in gamms/typing/graph_engine.py

class Node:
    """
    Represents a node within a graph.

    Attributes:
        id (int): The unique identifier for the node.
        x (float): The x-coordinate of the node's position.
        y (float): The y-coordinate of the node's position.
    """
    id: int
    x: float
    y: float

OSMEdge

Represents an OpenStreetMap (OSM) edge within a graph.

Attributes:

Name	Type	Description
id	int	The unique identifier for the edge.
source	int	The ID of the source node.
target	int	The ID of the target node.
length	float	The length of the edge.
linestring	LineString	The geometry of the edge represented as a LineString. Defaults to None.

Source code in gamms/typing/graph_engine.py

class OSMEdge:
    """
    Represents an OpenStreetMap (OSM) edge within a graph.

    Attributes:
        id (int): The unique identifier for the edge.
        source (int): The ID of the source node.
        target (int): The ID of the target node.
        length (float): The length of the edge.
        linestring (LineString): The geometry of the edge represented as a LineString.
            Defaults to None.
    """
    id: int
    source: int  # Node ID
    target: int  # Node ID
    length: float
    linestring: LineString