impuls.tasks.generate_shapes¶

class impuls.tasks.generate_shapes.GenerateShapes¶: See impuls.tasks.GenerateShapes.

class impuls.tasks.generate_shapes.AbstractGenerateShapes(*, routes: Routes, id_prefix: str, overwrite: bool = False, progress_report_step: int = 100, task_name: dataclasses.InitVar[str | None] = None)¶

Bases: Task, Generic[GraphT]

Base abstract class for generating shapes, agnostic to any “routing graphs”.

This class is mostly responsible for negotiating with the database - querying for trips, inserting shapes and updating shape_dist_traveled, but it also generates the shapes using the quasi-private ShapeBuilder.

Responsibility for providing the routing graph and other required lays on the programmer. Implementing shape generation involves injecting dependencies by overriding the create_routing_graph(), create_leg_router(), create_stop_snapper() and possibly other create_xxx methods.

AbstractGenerateShapes is a kw-only dataclass, in order to make adding new options/settings as easy as possible, without having to write out enormous __init__ methods manually. All subclasses are automatically wrapped in @dataclass though a __init_subclass__ hook. To add new options, simply do:

class MyGenerateShapes(AbstractGenerateShapes[...]):
    custom_knob: float
    another_parameter_with_default: int = 42

And those parameters will automatically be added as parameters to the __init__ method of MyGenerateShapes. InitVar is more convoluted, as the base class uses one for the task name:

class MyGenerateShapes(AbstractGenerateShapes[...]):
    custom_knob: InitVar[float | None] = None
    custom_knob_resolved: float = field(init=False)

    def __post_init__(self, task_name: str | None, custom_knob: float | None) -> None:
        super().__post_init__(task_name)
        self.custom_knob_resolved = custom_knob or fallback_custom_knob()

Of course, if you don’t want to add new options, you don’t need to worry about all that.

assign_shape(db: DBConnection, shape_id: str, shape: GeneratedShape, trip_ids: Sequence[str]) → None¶

Inserts the provided shape and assigns it to the trips.

This amounts to 4 different operations, wrapped in a transaction:

INSERT INTO shapes,
INSERT INTO shape_points,
UPDATE trips SET shape_id,
UPDATE stop_times SET shape_dist_traveled.

create_error_observer(r: TaskRuntime, graph: GraphT) → ErrorObserver¶

Instantiates an ErrorObserver, a callback when shape segment fails to generate.

The default implementation logs failures as warnings.

create_fallback_policy(r: TaskRuntime, graph: GraphT) → FallbackPolicy¶

Instantiates a FallbackPolicy, used to determine what to do when a shape segment fails to generate.

The default implementation aborts the entire shape.

create_leg_planner(r: TaskRuntime, graph: GraphT) → LegPlanner¶

Instantiates a LegPlanner, used to generate all waypoints to traverse between two stops.

The default implementation simply returns no waypoints; converting the stop pair to a single leg.

abstract create_leg_router(r: TaskRuntime, graph: GraphT) → LegRouter¶: Instantiates a LegRouter, used to generate shape segments between two Waypoints.

abstract create_routing_graph(r: TaskRuntime) → GraphT¶: Instantiates a routing graph, which will be passed along to other shape-building classes.

create_shape_builder(r: TaskRuntime, graph: GraphT) → ShapeBuilder¶

Instantiates a ShapeBuilder, by calling all other create_xxx methods.

This should not really by overridden - if there’s a need for that, file an issue.

create_shape_validator(r: TaskRuntime, graph: GraphT) → ShapeValidator¶

Instantiates a ShapeValidator, used to determine if a shape segment looks correct.

The default implementation allows any shapes.

abstract create_stop_snapper(r: TaskRuntime, graph: GraphT) → StopSnapper¶: Instantiates a StopSnapper, used to match Waypoints with nodes in the routing graph.

execute(r: TaskRuntime) → None¶

Generates shapes by:

gathering trips to process,
creating the routing graph,
creating the shape builder,
generating shapes and inserting them into the database.

generate_shapes(db: DBConnection, grouped_trips: Mapping[tuple[tuple[str, int], ...], Sequence[str]], builder: ShapeBuilder) → None¶: Generates and assigns shapes for the provided, already grouped trips.

get_stop_locations(r: TaskRuntime) → dict[str, tuple[float, float]]¶

Gets positions of all stops, for snapping.

The default implementation simply calls SELECT stop_id, lat, lon FROM stops, as that’s significantly easier than actually filtering out used stops, without increasing memory that much.

get_trip_stops(db: DBConnection, trip_id: str) → tuple[tuple[str, int], ...]¶

Retrieves all stops of a trip - its ShapeKey.

The default implementation simply executes:

SELECT stop_id, stop_sequence
FROM stop_times
WHERE trip_id = ?
ORDER BY stop_sequence ASC

get_trips_to_process(db: DBConnection) → Sequence[str]¶

Gets identifiers of all trips to generate shapes for.

This returns all trips of routes selected by the provided selector; that don’t already have shapes (shape_id IS NULL) unless overwrite is set.

group_trips(db: DBConnection, trip_ids: Iterable[str]) → Mapping[tuple[tuple[str, int], ...], Sequence[str]]¶

Groups the provided trips by the same ShapeKey - sequence of stops.

Trips’ shape keys are retrieved by calling get_trip_stops().

remove_existing_shapes(db: DBConnection, trip_ids: Iterable[str]) → None¶: Removes shapes of the provided trips, but only if Options.overwrite is set.

id_prefix: str¶

Prefix for generated shape IDs; which are consecutive integers.

Users must ensure the prefix guarantees shape ID uniqueness, particularly if using generate shapes multiple times, or if the database already has shapes.

overwrite: bool = False¶

Should any existing shapes (in selected routes) be overwritten?

Note that this only replaces Trip.shape_id, without removing shapes; as those shapes may be used by unselected trips. Vacuum unused shapes afterwards with an ExecuteSQL or a RemoveUnusedEntities task.

Defaults to False.

progress_report_step: int = 100¶

How often should an info message be logged with shape generation progress?

Defaults to 100 - every 100th shape causes an info message to be logged. All other generated shapes cause a debug logging message.

routes: Routes¶: Selector for which routes shapes should be generated.

task_name: dataclasses.InitVar[str | None] = None¶

class impuls.tasks.generate_shapes.CachedStopSnapper(snapper: StopSnapper)¶

Bases: StopSnapper

A decorator StopSnapper caching the results of matching.

cached() → Self¶: Returns this snapper unmodified.

snap(pt: Waypoint) → int¶

Snaps a waypoint to a corresponding node in the routing graph, and returns that node’s id. If there is no matched node, returns 0.

This method should ignore any existing Waypoint.node and is not required to set that attribute back.

cache: dict[str, int]¶: Cache from a Waypoint key to matched node.

snapper: StopSnapper¶: Main StopSnapper, doing the actual matching.

class impuls.tasks.generate_shapes.DistanceLimitedStopSnapper(snapper: StopSnapper, get_node_location: Callable[[int], tuple[float, float]], max_distance: float = 500.0)¶

Bases: StopSnapper

A decorator StopSnapper limiting snapped nodes to a maximum distance.

distance(pt_lat: float, pt_lon: float, node_lat: float, node_lon: float) → float¶

Calculates the distance between two points, for comparing against max_distance.

Defaults to meters, as calculated by impuls.tools.geo.earth_distance_m().

distance_limited(get_node_location: Callable[[int], tuple[float, float]], max_distance_m: float = 500) → Self¶: Modifies attributes of this distance-limiting decorator.

snap(pt: Waypoint) → int¶

Snaps a waypoint to a corresponding node in the routing graph, and returns that node’s id. If there is no matched node, returns 0.

This method should ignore any existing Waypoint.node and is not required to set that attribute back.

get_node_location: Callable[[int], tuple[float, float]]¶: Callback for retrieving the location of a node by its id.

max_distance: float¶: Max allowed distance to a node to allow a match. In meters, unless distance() is overridden.

snapper: StopSnapper¶: Main StopSnapper, doing the actual matching.

class impuls.tasks.generate_shapes.ErrorLogger(logger: Logger | None = None)¶

Bases: ErrorObserver

ErrorObserver implementation which logs all errors as warnings.

on_error(ctx: LegContext, error: Any, points: Sequence[tuple[float, float, float]]) → None¶: Callback on an error.

logger: Logger¶

class impuls.tasks.generate_shapes.ErrorObserver¶

Bases: object

Base class for error observers - objects collecting information on errors encountered during shape generation.

The default implementation does nothing.

on_error(ctx: LegContext, error: Any, points: Sequence[tuple[float, float, float]]) → None¶: Callback on an error.

class impuls.tasks.generate_shapes.FallbackPolicy¶

Bases: object

Base class for a fallback policy - objects which decide what to do when a shape’s leg fails to be generated.

The default implementation returns no substitutions, aborting entire shapes.

substitute_leg(ctx: LegContext, error: Any) → Sequence[tuple[float, float, float]]¶: Return a substitute shape for the provided leg; or an empty sequence to completely abandon the entire shape.

class impuls.tasks.generate_shapes.GeneratedShape(points: list[tuple[float, float, float]] = <factory>, distances: dict[int, float] = <factory>)¶

Bases: object

Full shape, generated for a ShapeKey.

append_leg(ctx: LegContext, points: Iterable[tuple[float, float, float]]) → None¶: Adds a leg to this generated shape.

round(coord_precision: int = 6, dist_precision: int = 3) → None¶: Rounds all stored coordinates and distances to the provided number of decimal places.

distances: dict[int, float]¶: Mapping from stop_sequence to a cumulative distance along a shape to that stop.

points: list[tuple[float, float, float]]¶: All points of this shape.

class impuls.tasks.generate_shapes.GeoJSONErrorWriter(directory: str | PathLike[str], /, clear: bool = False)¶

Bases: ErrorObserver

ErrorObserver implementation which dumps all shape generation errors as GeoJSON to the provided directory.

All error reasons must be serializable as JSON.

context_to_filename(ctx: LegContext, error: Any) → str¶

on_error(ctx: LegContext, error: Any, points: Sequence[tuple[float, float, float]]) → None¶: Callback on an error.

directory: Path¶

class impuls.tasks.generate_shapes.LegContext(start: Waypoint, end: Waypoint)¶

Bases: object

All data necessary when generating a shape’s leg between two waypoints.

end: Waypoint¶: End of the leg.

start: Waypoint¶: Beginning of the leg.

class impuls.tasks.generate_shapes.LegPlanner¶

Bases: object

Base class for leg planners - objects which decide how a route between two waypoints is generated; in particular if any extra waypoints should be inserted.

The root planner (the one returned by AbstractGenerateShapes.create_leg_planner()) is only called with both waypoints representing stops; but that might not hold if planners are composed/nested.

The default implementation doesn’t add any waypoints, and simply returns a single LegContext(start, end).

If the LegPlanner knows the exact nodes corresponding to waypoints, it may fill in their Waypoint.node attributes. This will in turn bypass stop snapping for those waypoints.

plan_waypoints(start: Waypoint, end: Waypoint) → Iterable[LegContext]¶

class impuls.tasks.generate_shapes.LegRouter¶

Bases: ABC

Abstract base class (interface) for a router - an object generating routes between two points.

abstract generate_leg(ctx: LegContext) → Sequence[tuple[float, float, float]]¶

Generates the shape between the two points in the provided context.

If it’s not possible to reach end from start, or the route can’t otherwise be generated (and the issue is to be suppressed), returns an empty sequence ([]).

However, if the start and end were snapped to the same node, a sequence of length one (representing that node) should be returned.

ShapeBuilder ensures this method is not called with any nodes set to 0.

class impuls.tasks.generate_shapes.MultiErrorObserver(*observers: ErrorObserver)¶

Bases: ErrorObserver

MultiErrorObserver is an ErrorObserver delegating an error callback to multiple other observers.

on_error(ctx: LegContext, error: Any, points: Sequence[tuple[float, float, float]]) → None¶: Callback on an error.

observers: tuple[ErrorObserver, ...]¶

class impuls.tasks.generate_shapes.MultiShapeValidator(*validators: ShapeValidator)¶

Bases: ShapeValidator

MultiShapeValidator is a ShapeValidator delegating the validation to multiple other validators, returning the first encountered error (short circuiting), if any.

validate_leg(ctx: LegContext, points: Sequence[tuple[float, float, float]]) → Any¶

Checks if a shape segment looks correct.

If the shape leg looks correct, returns a false-y value, preferably None.

Otherwise, returns a truthy value representing the reason why the shape looks incorrect. Users are encouraged to keep return values JSON serializable, especially strings.

validators: tuple[ShapeValidator, ...]¶

class impuls.tasks.generate_shapes.RoutxKDTreeStopSnapper(graph: Graph)¶

Bases: StopSnapper

Snap stops to a routx routing graph using a k-d tree.

snap(pt: Waypoint) → int¶

Snaps a waypoint to a corresponding node in the routing graph, and returns that node’s id. If there is no matched node, returns 0.

This method should ignore any existing Waypoint.node and is not required to set that attribute back.

kd_tree: KDTree¶

class impuls.tasks.generate_shapes.RoutxLegRouter(graph: Graph, simplify_epsilon: float = 1e-05)¶

Bases: LegRouter

Generates routes on a routx routing graph; simplifying the results with the Ramer-Douglas-Peucker algorithm.

generate_leg(ctx: LegContext) → Sequence[tuple[float, float, float]]¶

Generates the shape between the two points in the provided context.

If it’s not possible to reach end from start, or the route can’t otherwise be generated (and the issue is to be suppressed), returns an empty sequence ([]).

However, if the start and end were snapped to the same node, a sequence of length one (representing that node) should be returned.

ShapeBuilder ensures this method is not called with any nodes set to 0.

nodes_to_shape_points(nodes: Iterable[int]) → list[tuple[float, float, float]]¶

graph: Graph¶

simplify_epsilon: float¶

Threshold, in decimal degrees, for the RDP simplification algorithm to determine if a node “sticks out” enough to be considered important and preserved.

Defaults to 1e-5, but can be set to a non-finite number (e.g. nan) to disable simplification entirely.

class impuls.tasks.generate_shapes.ShapeBuilder(snapper: StopSnapper, planner: LegPlanner, router: LegRouter, validator: ShapeValidator, fallback: FallbackPolicy, observer: ErrorObserver, stop_locations: Mapping[str, tuple[float, float]])¶

Bases: object

ShapeBuilder orchestrates all necessary single-responsibility objects to actually build and generate shapes.

Users are not really supposed to subclass the builder, if such a necessity arises, please file an issue.

build(stops: Iterable[tuple[str, int]]) → GeneratedShape | None¶: Builds a shape between the provided stops.

build_leg(ctx: LegContext) → Sequence[tuple[float, float, float]]¶

Builds a shape between two waypoints, by calling the StopSnapper and LegRouter; and validating the segment with ShapeValidator.

If the shape fails to generate or is deemed incorrect, calls the ErrorObserver and FallbackPolicy.

get_legs(stops: Iterable[tuple[str, int]]) → list[LegContext]¶: Gets all legs of a shape, by querying the LegPlanner for every stop pair.

on_error(ctx: LegContext, error: Any, points: Sequence[tuple[float, float, float]]) → Sequence[tuple[float, float, float]]¶: Deals with a shape error - invokes the ErrorObserver and gets the fallback shape from the FallbackPolicy.

stop_to_waypoint(stop: tuple[str, int]) → Waypoint¶

fallback: FallbackPolicy¶

observer: ErrorObserver¶

planner: LegPlanner¶

router: LegRouter¶

snapper: StopSnapper¶

stop_locations: Mapping[str, tuple[float, float]]¶

validator: ShapeValidator¶

class impuls.tasks.generate_shapes.ShapeValidator¶

Bases: object

Base class for shape validators - objects which decide if a generated shape leg looks correct.

The default implementation allows all shapes.

validate_leg(ctx: LegContext, points: Sequence[tuple[float, float, float]]) → Any¶

Checks if a shape segment looks correct.

If the shape leg looks correct, returns a false-y value, preferably None.

Otherwise, returns a truthy value representing the reason why the shape looks incorrect. Users are encouraged to keep return values JSON serializable, especially strings.

class impuls.tasks.generate_shapes.StopSnapper¶

Bases: ABC

Abstract base class (interface) for a stop snapper - an object matching stops with nodes in a routing graph.

cached() → CachedStopSnapper¶: Caches all waypoint-to-node matches by wrapping this snapper in a CachedStopSnapper.

distance_limited(get_node_location: Callable[[int], tuple[float, float]], max_distance_m: float = 500.0) → DistanceLimitedStopSnapper¶: Limits the maximum distance of waypoint-to-node matches by wrapping this snapper in a DistanceLimitedStopSnapper.

abstract snap(pt: Waypoint) → int¶

Snaps a waypoint to a corresponding node in the routing graph, and returns that node’s id. If there is no matched node, returns 0.

This method should ignore any existing Waypoint.node and is not required to set that attribute back.

class impuls.tasks.generate_shapes.StraightLineSubstitute¶

Bases: FallbackPolicy

A FallbackPolicy which substitutes failed shape segments by straight lines between waypoints.

distance(start: Waypoint, end: Waypoint) → float¶: Computes the distance between two waypoints, to match with LegRouter units. Defaults to impuls.tools.geo.earth_distance_m() converted to kilometers.

substitute_leg(ctx: LegContext, error: Any) → Sequence[tuple[float, float, float]]¶: Return a substitute shape for the provided leg; or an empty sequence to completely abandon the entire shape.

class impuls.tasks.generate_shapes.Waypoint(id: tuple[str, int] | str, lat: float, lon: float, node: int = 0)¶

Bases: object

Intermediary point used when generating a shape, usually corresponding to a stop.

cache_key() → str¶

id: tuple[str, int] | str¶

ID of the waypoint - either a stop, or some arbitrary point described by a unique string, different to all stop ids.

Both StopKey.id or arbitrary strings here can be used as cache keys in the same lookup table.

lat: float¶: Geographic latitude of the waypoint.

lon: float¶: Geographic longitude of the waypoint.

node: int = 0¶: Identifier of a node in a routing graph corresponding to this waypoint, or 0 when no appropriate node was found.

impuls.tasks.generate_shapes.ShapeKey¶

Unique identifier of a shape - a variable-length tuple of stop keys.

alias of tuple[tuple[str, int], …]

impuls.tasks.generate_shapes.ShapePoint¶

Point in a shape - a latitude, longitude and cumulative distance.

alias of tuple[float, float, float]

impuls.tasks.generate_shapes.StopKey¶

Unique identifier of a stop within a trip - a stop_id and stop_sequence pair.

alias of tuple[str, int]