API Documentation

Core Classes

The fundamental classes that form the backbone of Network Wrangler:

Scenario objects manage how a collection of projects is applied to the networks.

Scenarios are built from a base scenario and a list of project cards.

A project card is a YAML file (or similar) that describes a change to the network. The project card can contain multiple changes, each of which is applied to the network in sequence.

Create a Scenario

Instantiate a scenario by seeding it with a base scenario and optionally some project cards.

from network_wrangler import create_scenario

my_scenario = create_scenario(
    base_scenario=my_base_year_scenario,
    card_search_dir=project_card_directory,
    filter_tags=["baseline2050"],
)

A base_year_scenario is a dictionary representation of key components of a scenario:

• road_net: RoadwayNetwork instance
• transit_net: TransitNetwork instance
• applied_projects: list of projects that have been applied to the base scenario so that the scenario knows if there will be conflicts with future projects or if a future project’s pre-requisite is satisfied.
• conflicts: dictionary of conflicts for project that have been applied to the base scenario so that the scenario knows if there will be conflicts with future projects.

my_base_year_scenario = {
    "road_net": load_from_roadway_dir(STPAUL_DIR),
    "transit_net": load_transit(STPAUL_DIR),
    "applied_projects": [],
    "conflicts": {},
}

Add Projects to a Scenario

In addition to adding projects when you create the scenario, project cards can be added to a scenario using the add_project_cards method.

from projectcard import read_cards

project_card_dict = read_cards(card_location, filter_tags=["Baseline2030"], recursive=True)
my_scenario.add_project_cards(project_card_dict.values())

Where card_location can be a single path, list of paths, a directory, or a glob pattern.

Apply Projects to a Scenario

Projects can be applied to a scenario using the apply_all_projects method. Before applying projects, the scenario will check that all pre-requisites are satisfied, that there are no conflicts, and that the projects are in the planned projects list.

If you want to check the order of projects before applying them, you can use the queued_projects prooperty.

my_scenario.queued_projects
my_scenario.apply_all_projects()

You can review the resulting scenario, roadway network, and transit networks.

my_scenario.applied_projects
my_scenario.road_net.links_gdf.explore()
my_scenario.transit_net.feed.shapes_gdf.explore()

Write a Scenario to Disk

Scenarios (and their networks) can be written to disk using the write method which in addition to writing out roadway and transit networks, will serialize the scenario to a yaml-like file and can also write out the project cards that have been applied.

my_scenario.write(
    "output_dir",
    "scenario_name_to_use",
    overwrite=True,
    projects_write=True,
    file_format="parquet",
)

Example Serialized Scenario File

applied_projects: &id001
- project a
- project b
base_scenario:
applied_projects: *id001
roadway:
    dir: /Users/elizabeth/Documents/urbanlabs/MetCouncil/NetworkWrangler/working/network_wrangler/examples/small
    file_format: geojson
transit:
    dir: /Users/elizabeth/Documents/urbanlabs/MetCouncil/NetworkWrangler/working/network_wrangler/examples/small
config:
CPU:
    EST_PD_READ_SPEED:
    csv: 0.03
    geojson: 0.03
    json: 0.15
    parquet: 0.005
    txt: 0.04
IDS:
    ML_LINK_ID_METHOD: range
    ML_LINK_ID_RANGE: &id002 !!python/tuple
    - 950000
    - 999999
    ML_LINK_ID_SCALAR: 15000
    ML_NODE_ID_METHOD: range
    ML_NODE_ID_RANGE: *id002
    ML_NODE_ID_SCALAR: 15000
    ROAD_SHAPE_ID_METHOD: scalar
    ROAD_SHAPE_ID_SCALAR: 1000
    TRANSIT_SHAPE_ID_METHOD: scalar
    TRANSIT_SHAPE_ID_SCALAR: 1000000
MODEL_ROADWAY:
    ADDITIONAL_COPY_FROM_GP_TO_ML: []
    ADDITIONAL_COPY_TO_ACCESS_EGRESS: []
    ML_OFFSET_METERS: -10
conflicts: {}
corequisites: {}
name: first_scenario
prerequisites: {}
roadway:
dir: /Users/elizabeth/Documents/urbanlabs/MetCouncil/NetworkWrangler/working/network_wrangler/tests/out/first_scenario/roadway
file_format: parquet
transit:
dir: /Users/elizabeth/Documents/urbanlabs/MetCouncil/NetworkWrangler/working/network_wrangler/tests/out/first_scenario/transit
file_format: txt

Load a scenario from disk

And if you want to reload scenario that you “wrote”, you can use the load_scenario function.

from network_wrangler import load_scenario

my_scenario = load_scenario("output_dir/scenario_name_to_use_scenario.yml")

network_wrangler.scenario.BASE_SCENARIO_SUGGESTED_PROPS module-attribute

BASE_SCENARIO_SUGGESTED_PROPS = ['road_net', 'transit_net', 'applied_projects', 'conflicts']

List of card types that that will be applied to the transit network.

network_wrangler.scenario.ROADWAY_CARD_TYPES module-attribute

ROADWAY_CARD_TYPES = ['roadway_property_change', 'roadway_deletion', 'roadway_addition', 'pycode']

List of card types that that will be applied to the transit network AFTER being applied to the roadway network.

network_wrangler.scenario.TRANSIT_CARD_TYPES module-attribute

TRANSIT_CARD_TYPES = ['transit_property_change', 'transit_routing_change', 'transit_route_addition', 'transit_service_deletion']

List of card types that that will be applied to the roadway network.

network_wrangler.scenario.Scenario

Holds information about a scenario.

Typical usage example:

my_base_year_scenario = {
    "road_net": load_roadway(
        links_file=STPAUL_LINK_FILE,
        nodes_file=STPAUL_NODE_FILE,
        shapes_file=STPAUL_SHAPE_FILE,
    ),
    "transit_net": load_transit(STPAUL_DIR),
}

# create a future baseline scenario from base by searching for all cards in dir w/ baseline tag
project_card_directory = Path(STPAUL_DIR) / "project_cards"
my_scenario = create_scenario(
    base_scenario=my_base_year_scenario,
    card_search_dir=project_card_directory,
    filter_tags=["baseline2050"],
)

# check project card queue and then apply the projects
my_scenario.queued_projects
my_scenario.apply_all_projects()

# check applied projects, write it out, and create a summary report.
my_scenario.applied_projects
my_scenario.write("baseline")
my_scenario.summary

# Add some projects to create a build scenario based on a list of files.
build_card_filenames = [
    "3_multiple_roadway_attribute_change.yml",
    "road.prop_changes.segment.yml",
    "4_simple_managed_lane.yml",
]
my_scenario.add_projects_from_files(build_card_filenames)
my_scenario.write("build2050")
my_scenario.summary

Attributes:

• base_scenario (dict) –

  dictionary representation of a scenario

• road_net (Optional[RoadwayNetwork]) –

  instance of RoadwayNetwork for the scenario

• transit_net (Optional[TransitNetwork]) –

  instance of TransitNetwork for the scenario

• project_cards (dict[str, ProjectCard]) –

  Mapping[ProjectCard.name,ProjectCard] Storage of all project cards by name.

• queued_projects –

  Projects which are “shovel ready” - have had pre-requisits checked and done any required re-ordering. Similar to a git staging, project cards aren’t recognized in this collecton once they are moved to applied.

• applied_projects (list[str]) –

  list of project names that have been applied

• projects –

  list of all projects either planned, queued, or applied

• prerequisites (dict[str, list[str]]) –

  dictionary storing prerequiste info as projectA: [prereqs-for-projectA]

• corequisites (dict[str, list[str]]) –

  dictionary storing corequisite info asprojectA: [coreqs-for-projectA]

• conflicts (dict[str, list[str]]) –

  dictionary storing conflict info as projectA: [conflicts-for-projectA]

• config –

  WranglerConfig instance.

Source code in network_wrangler/scenario.py

class Scenario:
    """Holds information about a scenario.

    Typical usage example:

    ```python
    my_base_year_scenario = {
        "road_net": load_roadway(
            links_file=STPAUL_LINK_FILE,
            nodes_file=STPAUL_NODE_FILE,
            shapes_file=STPAUL_SHAPE_FILE,
        ),
        "transit_net": load_transit(STPAUL_DIR),
    }

    # create a future baseline scenario from base by searching for all cards in dir w/ baseline tag
    project_card_directory = Path(STPAUL_DIR) / "project_cards"
    my_scenario = create_scenario(
        base_scenario=my_base_year_scenario,
        card_search_dir=project_card_directory,
        filter_tags=["baseline2050"],
    )

    # check project card queue and then apply the projects
    my_scenario.queued_projects
    my_scenario.apply_all_projects()

    # check applied projects, write it out, and create a summary report.
    my_scenario.applied_projects
    my_scenario.write("baseline")
    my_scenario.summary

    # Add some projects to create a build scenario based on a list of files.
    build_card_filenames = [
        "3_multiple_roadway_attribute_change.yml",
        "road.prop_changes.segment.yml",
        "4_simple_managed_lane.yml",
    ]
    my_scenario.add_projects_from_files(build_card_filenames)
    my_scenario.write("build2050")
    my_scenario.summary
    ```

    Attributes:
        base_scenario: dictionary representation of a scenario
        road_net: instance of RoadwayNetwork for the scenario
        transit_net: instance of TransitNetwork for the scenario
        project_cards: Mapping[ProjectCard.name,ProjectCard] Storage of all project cards by name.
        queued_projects: Projects which are "shovel ready" - have had pre-requisits checked and
            done any required re-ordering. Similar to a git staging, project cards aren't
            recognized in this collecton once they are moved to applied.
        applied_projects: list of project names that have been applied
        projects: list of all projects either planned, queued, or applied
        prerequisites:  dictionary storing prerequiste info as `projectA: [prereqs-for-projectA]`
        corequisites:  dictionary storing corequisite info as`projectA: [coreqs-for-projectA]`
        conflicts: dictionary storing conflict info as `projectA: [conflicts-for-projectA]`
        config: WranglerConfig instance.
    """

    def __init__(
        self,
        base_scenario: Union[Scenario, dict],
        project_card_list: Optional[list[ProjectCard]] = None,
        config: Optional[Union[WranglerConfig, dict, Path, list[Path]]] = None,
        name: str = "",
    ):
        """Constructor.

        Args:
            base_scenario: A base scenario object to base this isntance off of, or a dict which
                describes the scenario attributes including applied projects and respective
                conflicts. `{"applied_projects": [],"conflicts":{...}}`
            project_card_list: Optional list of ProjectCard instances to add to planned projects.
                Defaults to None.
            config: WranglerConfig instance or a dictionary of configuration settings or a path to
                one or more configuration files. Configurations that are not explicity set will
                default to the values in the default configuration in
                `/configs/wrangler/default.yml`.
            name: Optional name for the scenario.
        """
        WranglerLogger.info("Creating Scenario")
        self.config = load_wrangler_config(config)

        if project_card_list is None:
            project_card_list = []

        if isinstance(base_scenario, Scenario):
            base_scenario = base_scenario.__dict__

        self.base_scenario: dict = extract_base_scenario_metadata(base_scenario)

        if not set(BASE_SCENARIO_SUGGESTED_PROPS) <= set(base_scenario.keys()):
            WranglerLogger.warning(
                f"Base_scenario doesn't contain {BASE_SCENARIO_SUGGESTED_PROPS}"
            )
        self.name: str = name
        # if the base scenario had roadway or transit networks, use them as the basis.
        self.road_net: Optional[RoadwayNetwork] = copy.deepcopy(
            base_scenario.pop("road_net", None)
        )

        self.transit_net: Optional[TransitNetwork] = copy.deepcopy(
            base_scenario.pop("transit_net", None)
        )
        if self.road_net and self.transit_net:
            self.transit_net.road_net = self.road_net

        # Set configs for networks to be the same as scenario.
        if isinstance(self.road_net, RoadwayNetwork):
            self.road_net.config = self.config
        if isinstance(self.transit_net, TransitNetwork):
            self.transit_net.config = self.config

        self.project_cards: dict[str, ProjectCard] = {}
        self._planned_projects: list[str] = []
        self._queued_projects = None
        self.applied_projects: list[str] = base_scenario.pop("applied_projects", [])

        self.prerequisites: dict[str, list[str]] = base_scenario.pop("prerequisites", {})
        self.corequisites: dict[str, list[str]] = base_scenario.pop("corequisites", {})
        self.conflicts: dict[str, list[str]] = base_scenario.pop("conflicts", {})

        for p in project_card_list:
            self._add_project(p)

    @property
    def projects(self):
        """Returns a list of all projects in the scenario: applied and planned."""
        return self.applied_projects + self._planned_projects

    @property
    def queued_projects(self):
        """Returns a list version of _queued_projects queue.

        Queued projects are thos that have been planned, have all pre-requisites satisfied, and
        have been ordered based on pre-requisites.

        If no queued projects, will dynamically generate from planned projects based on
        pre-requisites and return the queue.
        """
        if not self._queued_projects:
            self._check_projects_requirements_satisfied(self._planned_projects)
            self._queued_projects = self.order_projects(self._planned_projects)
        return list(self._queued_projects)

    def __str__(self):
        """String representation of the Scenario object."""
        s = [f"{key}: {value}" for key, value in self.__dict__.items()]
        return "\n".join(s)

    def _add_dependencies(self, project_name, dependencies: dict) -> None:
        """Add dependencies from a project card to relevant scenario variables.

        Updates existing "prerequisites", "corequisites" and "conflicts".
        Lowercases everything to enable string matching.

        Args:
            project_name: name of project you are adding dependencies for.
            dependencies: Dictionary of depndencies by dependency type and list of associated
                projects.
        """
        project_name = project_name.lower()

        for d, v in dependencies.items():
            _dep = list(map(str.lower, v))
            WranglerLogger.debug(f"Adding {_dep} to {project_name} dependency table.")
            self.__dict__[d].update({project_name: _dep})

    def _add_project(
        self,
        project_card: ProjectCard,
        validate: bool = True,
        filter_tags: Optional[list[str]] = None,
    ) -> None:
        """Adds a single ProjectCard instances to the Scenario.

        Checks that a project of same name is not already in scenario.
        If selected, will validate ProjectCard before adding.
        If provided, will only add ProjectCard if it matches at least one filter_tags.

        Resets scenario queued_projects.

        Args:
            project_card (ProjectCard): ProjectCard instance to add to scenario.
            validate (bool, optional): If True, will validate the projectcard before
                being adding it to the scenario. Defaults to True.
            filter_tags: If used, will only add the project card if
                its tags match one or more of these filter_tags. Defaults to []
                which means no tag-filtering will occur.

        """
        filter_tags = filter_tags or []
        project_name = project_card.project.lower()
        filter_tags = list(map(str.lower, filter_tags))

        if project_name in self.projects:
            msg = f"Names not unique from existing scenario projects: {project_card.project}"
            raise ProjectCardError(msg)

        if filter_tags and set(project_card.tags).isdisjoint(set(filter_tags)):
            WranglerLogger.debug(
                f"Skipping {project_name} - no overlapping tags with {filter_tags}."
            )
            return

        if validate:
            project_card.validate()

        WranglerLogger.info(f"Adding {project_name} to scenario.")
        self.project_cards[project_name] = project_card
        self._planned_projects.append(project_name)
        self._queued_projects = None
        self._add_dependencies(project_name, project_card.dependencies)

    def add_project_cards(
        self,
        project_card_list: list[ProjectCard],
        validate: bool = True,
        filter_tags: Optional[list[str]] = None,
    ) -> None:
        """Adds a list of ProjectCard instances to the Scenario.

        Checks that a project of same name is not already in scenario.
        If selected, will validate ProjectCard before adding.
        If provided, will only add ProjectCard if it matches at least one filter_tags.

        Args:
            project_card_list: List of ProjectCard instances to add to
                scenario.
            validate (bool, optional): If True, will require each ProjectCard is validated before
                being added to scenario. Defaults to True.
            filter_tags: If used, will filter ProjectCard instances
                and only add those whose tags match one or more of these filter_tags.
                Defaults to [] - which means no tag-filtering will occur.
        """
        filter_tags = filter_tags or []
        for p in project_card_list:
            self._add_project(p, validate=validate, filter_tags=filter_tags)

    def _check_projects_requirements_satisfied(self, project_list: list[str]):
        """Checks all requirements are satisified to apply this specific set of projects.

        Including:
        1. has an associaed project card
        2. is in scenario's planned projects
        3. pre-requisites satisfied
        4. co-requisies satisfied by applied or co-applied projects
        5. no conflicing applied or co-applied projects

        Args:
            project_list: list of projects to check requirements for.
        """
        self._check_projects_planned(project_list)
        self._check_projects_have_project_cards(project_list)
        self._check_projects_prerequisites(project_list)
        self._check_projects_corequisites(project_list)
        self._check_projects_conflicts(project_list)

    def _check_projects_planned(self, project_names: list[str]) -> None:
        """Checks that a list of projects are in the scenario's planned projects."""
        _missing_ps = [p for p in project_names if p not in self._planned_projects]
        if _missing_ps:
            msg = f"Projects are not in planned projects: \n {_missing_ps}. \
                Add them by using add_project_cards()."
            WranglerLogger.debug(msg)
            raise ValueError(msg)

    def _check_projects_have_project_cards(self, project_list: list[str]) -> bool:
        """Checks that a list of projects has an associated project card in the scenario."""
        _missing = [p for p in project_list if p not in self.project_cards]
        if _missing:
            WranglerLogger.error(
                f"Projects referenced which are missing project cards: {_missing}"
            )
            return False
        return True

    def _check_projects_prerequisites(self, project_names: list[str]) -> None:
        """Check a list of projects' pre-requisites have been or will be applied to scenario."""
        if set(project_names).isdisjoint(set(self.prerequisites.keys())):
            return
        _prereqs = []
        for p in project_names:
            _prereqs += self.prerequisites.get(p, [])
        _projects_applied = self.applied_projects + project_names
        _missing = list(set(_prereqs) - set(_projects_applied))
        if _missing:
            WranglerLogger.debug(
                f"project_names: {project_names}\nprojects_have_or_will_be_applied: \
                    {_projects_applied}\nmissing: {_missing}"
            )
            msg = f"Missing {len(_missing)} pre-requisites."
            raise ScenarioPrerequisiteError(msg)

    def _check_projects_corequisites(self, project_names: list[str]) -> None:
        """Check a list of projects' co-requisites have been or will be applied to scenario."""
        if set(project_names).isdisjoint(set(self.corequisites.keys())):
            return
        _coreqs = []
        for p in project_names:
            _coreqs += self.corequisites.get(p, [])
        _projects_applied = self.applied_projects + project_names
        _missing = list(set(_coreqs) - set(_projects_applied))
        if _missing:
            WranglerLogger.debug(
                f"project_names: {project_names}\nprojects_have_or_will_be_applied: \
                    {_projects_applied}\nmissing: {_missing}"
            )
            msg = f"Missing {len(_missing)} corequisites."
            raise ScenarioCorequisiteError(msg)

    def _check_projects_conflicts(self, project_names: list[str]) -> None:
        """Checks that list of projects' conflicts have not been or will be applied to scenario."""
        # WranglerLogger.debug("Checking Conflicts...")
        projects_to_check = project_names + self.applied_projects
        # WranglerLogger.debug(f"\nprojects_to_check:{projects_to_check}\nprojects_with_conflicts:{set(self.conflicts.keys())}")
        if set(projects_to_check).isdisjoint(set(self.conflicts.keys())):
            # WranglerLogger.debug("Projects have no conflicts to check")
            return
        _conflicts = []
        for p in project_names:
            _conflicts += self.conflicts.get(p, [])
        _conflict_problems = [p for p in _conflicts if p in projects_to_check]
        if _conflict_problems:
            WranglerLogger.warning(f"Conflict Problems: \n{_conflict_problems}")
            _conf_dict = {
                k: v
                for k, v in self.conflicts.items()
                if k in projects_to_check and not set(v).isdisjoint(set(_conflict_problems))
            }
            WranglerLogger.debug(f"Problematic Conflicts: \n{_conf_dict}")
            msg = f"Found {len(_conflict_problems)} conflicts: {_conflict_problems}"
            raise ScenarioConflictError(msg)

    def order_projects(self, project_list: list[str]) -> deque:
        """Orders a list of projects based on moving up pre-requisites into a deque.

        Args:
            project_list: list of projects to order

        Returns: deque for applying projects.
        """
        project_list = [p.lower() for p in project_list]
        assert self._check_projects_have_project_cards(project_list)

        # build prereq (adjacency) list for topological sort
        adjacency_list: dict[str, list] = defaultdict(list)
        visited_list: dict[str, bool] = defaultdict(bool)

        for project in project_list:
            visited_list[project] = False
            if not self.prerequisites.get(project):
                continue
            for prereq in self.prerequisites[project]:
                # this will always be true, else would have been flagged in missing \
                # prerequsite check, but just in case
                if prereq.lower() in project_list:
                    if adjacency_list.get(prereq.lower()):
                        adjacency_list[prereq.lower()].append(project)
                    else:
                        adjacency_list[prereq.lower()] = [project]

        # sorted_project_names is topological sorted project card names (based on prerequsiite)
        _ordered_projects = topological_sort(
            adjacency_list=adjacency_list, visited_list=visited_list
        )

        if set(_ordered_projects) != set(project_list):
            _missing = list(set(project_list) - set(_ordered_projects))
            msg = f"Project sort resulted in missing projects: {_missing}"
            raise ValueError(msg)

        project_deque = deque(_ordered_projects)

        WranglerLogger.debug(f"Ordered Projects: \n{project_deque}")

        return project_deque

    def apply_all_projects(self):
        """Applies all planned projects in the queue."""
        # Call this to make sure projects are appropriately queued in hidden variable.
        self.queued_projects  # noqa: B018

        # Use hidden variable.
        while self._queued_projects:
            self._apply_project(self._queued_projects.popleft())

        # set this so it will trigger re-queuing any more projects.
        self._queued_projects = None

    def _apply_change(self, change: Union[ProjectCard, SubProject]) -> None:
        """Applies a specific change specified in a project card.

        Change type must be in at least one of:
        - ROADWAY_CARD_TYPES
        - TRANSIT_CARD_TYPES

        Args:
            change: a project card or subproject card
        """
        if change.change_type in ROADWAY_CARD_TYPES:
            if not self.road_net:
                msg = "Missing Roadway Network"
                raise ValueError(msg)
            if change.change_type in SECONDARY_TRANSIT_CARD_TYPES and self.transit_net:
                self.road_net.apply(change, transit_net=self.transit_net)
            else:
                self.road_net.apply(change)
        if change.change_type in TRANSIT_CARD_TYPES:
            if not self.transit_net:
                msg = "Missing Transit Network"
                raise ValueError(msg)
            self.transit_net.apply(change)

        if change.change_type not in ROADWAY_CARD_TYPES + TRANSIT_CARD_TYPES:
            msg = f"Project {change.project}: Don't understand project cat: {change.change_type}"
            raise ProjectCardError(msg)

    def _apply_project(self, project_name: str) -> None:
        """Applies project card to scenario.

        If a list of changes is specified in referenced project card, iterates through each change.

        Args:
            project_name (str): name of project to be applied.
        """
        project_name = project_name.lower()

        WranglerLogger.info(
            f"Applying {project_name} from file:\
                            {self.project_cards[project_name].file}"
        )

        p = self.project_cards[project_name]
        WranglerLogger.debug(f"types: {p.change_types}")
        WranglerLogger.debug(f"type: {p.change_type}")
        if p._sub_projects:
            for sp in p._sub_projects:
                WranglerLogger.debug(f"- applying subproject: {sp.change_type}")
                self._apply_change(sp)

        else:
            self._apply_change(p)

        self._planned_projects.remove(project_name)
        self.applied_projects.append(project_name)

    def apply_projects(self, project_list: list[str]):
        """Applies a specific list of projects from the planned project queue.

        Will order the list of projects based on pre-requisites.

        NOTE: does not check co-requisites b/c that isn't possible when applying a single project.

        Args:
            project_list: List of projects to be applied. All need to be in the planned project
                queue.
        """
        project_list = [p.lower() for p in project_list]

        self._check_projects_requirements_satisfied(project_list)
        ordered_project_queue = self.order_projects(project_list)

        while ordered_project_queue:
            self._apply_project(ordered_project_queue.popleft())

        # Set so that when called again it will retrigger queueing from planned projects.
        self._ordered_projects = None

    def write(
        self,
        path: Path,
        name: str,
        overwrite: bool = True,
        roadway_write: bool = True,
        transit_write: bool = True,
        projects_write: bool = True,
        roadway_convert_complex_link_properties_to_single_field: bool = False,
        roadway_out_dir: Optional[Path] = None,
        roadway_prefix: Optional[str] = None,
        roadway_file_format: RoadwayFileTypes = "parquet",
        roadway_true_shape: bool = False,
        transit_out_dir: Optional[Path] = None,
        transit_prefix: Optional[str] = None,
        transit_file_format: TransitFileTypes = "txt",
        projects_out_dir: Optional[Path] = None,
    ) -> Path:
        """Writes scenario networks and summary to disk and returns path to scenario file.

        Args:
            path: Path to write scenario networks and scenario summary to.
            name: Name to use.
            overwrite: If True, will overwrite the files if they already exist.
            roadway_write: If True, will write out the roadway network.
            transit_write: If True, will write out the transit network.
            projects_write: If True, will write out the project cards.
            roadway_convert_complex_link_properties_to_single_field: If True, will convert complex
                link properties to a single field.
            roadway_out_dir: Path to write the roadway network files to.
            roadway_prefix: Prefix to add to the file name.
            roadway_file_format: File format to write the roadway network to
            roadway_true_shape: If True, will write the true shape of the roadway network
            transit_out_dir: Path to write the transit network files to.
            transit_prefix: Prefix to add to the file name.
            transit_file_format: File format to write the transit network to
            projects_out_dir: Path to write the project cards to.
        """
        path = Path(path)
        path.mkdir(parents=True, exist_ok=True)

        if self.road_net and roadway_write:
            if roadway_out_dir is None:
                roadway_out_dir = path / "roadway"
            roadway_out_dir.mkdir(parents=True, exist_ok=True)

            write_roadway(
                net=self.road_net,
                out_dir=roadway_out_dir,
                prefix=roadway_prefix or name,
                convert_complex_link_properties_to_single_field=roadway_convert_complex_link_properties_to_single_field,
                file_format=roadway_file_format,
                true_shape=roadway_true_shape,
                overwrite=overwrite,
            )
        if self.transit_net and transit_write:
            if transit_out_dir is None:
                transit_out_dir = path / "transit"
            transit_out_dir.mkdir(parents=True, exist_ok=True)
            write_transit(
                self.transit_net,
                out_dir=transit_out_dir,
                prefix=transit_prefix or name,
                file_format=transit_file_format,
                overwrite=overwrite,
            )
        if projects_write:
            if projects_out_dir is None:
                projects_out_dir = path / "projects"
            write_applied_projects(
                self,
                out_dir=projects_out_dir,
                overwrite=overwrite,
            )

        scenario_data = self.summary
        if transit_write:
            scenario_data["transit"] = {
                "dir": str(transit_out_dir),
                "file_format": transit_file_format,
            }
        if roadway_write:
            scenario_data["roadway"] = {
                "dir": str(roadway_out_dir),
                "file_format": roadway_file_format,
            }
        if projects_write:
            scenario_data["project_cards"] = {"dir": str(projects_out_dir)}
        scenario_file_path = Path(path) / f"{name}_scenario.yml"
        with scenario_file_path.open("w") as f:
            yaml.dump(scenario_data, f, default_flow_style=False, allow_unicode=True)
        return scenario_file_path

    @property
    def summary(self) -> dict:
        """A high level summary of the created scenario and public attributes."""
        skip = ["road_net", "base_scenario", "transit_net", "project_cards", "config"]
        summary_dict = {
            k: v for k, v in self.__dict__.items() if not k.startswith("_") and k not in skip
        }
        summary_dict["config"] = self.config.to_dict()

        """
        # Handle nested dictionary for "base_scenario"
        skip_base = ["project_cards"]
        if "base_scenario" in self.__dict__:
            base_summary_dict = {
                k: v
                for k, v in self.base_scenario.items()
                if not k.startswith("_") and k not in skip_base
            }
            summary_dict["base_scenario"] = base_summary_dict
        """

        return summary_dict

network_wrangler.scenario.Scenario.projects property

projects

Returns a list of all projects in the scenario: applied and planned.

network_wrangler.scenario.Scenario.queued_projects property

queued_projects

Returns a list version of _queued_projects queue.

Queued projects are thos that have been planned, have all pre-requisites satisfied, and have been ordered based on pre-requisites.

If no queued projects, will dynamically generate from planned projects based on pre-requisites and return the queue.

network_wrangler.scenario.Scenario.summary property

summary

A high level summary of the created scenario and public attributes.

network_wrangler.scenario.Scenario.__init__

__init__(base_scenario, project_card_list=None, config=None, name='')

Constructor.

Parameters:

• base_scenario (Union[Scenario, dict]) –

  A base scenario object to base this isntance off of, or a dict which describes the scenario attributes including applied projects and respective conflicts. {"applied_projects": [],"conflicts":{...}}

• project_card_list (Optional[list[ProjectCard]], default: None ) –

  Optional list of ProjectCard instances to add to planned projects. Defaults to None.

• config (Optional[Union[WranglerConfig, dict, Path, list[Path]]], default: None ) –

  WranglerConfig instance or a dictionary of configuration settings or a path to one or more configuration files. Configurations that are not explicity set will default to the values in the default configuration in /configs/wrangler/default.yml.

• name (str, default: '' ) –

  Optional name for the scenario.

Source code in network_wrangler/scenario.py

def __init__(
    self,
    base_scenario: Union[Scenario, dict],
    project_card_list: Optional[list[ProjectCard]] = None,
    config: Optional[Union[WranglerConfig, dict, Path, list[Path]]] = None,
    name: str = "",
):
    """Constructor.

    Args:
        base_scenario: A base scenario object to base this isntance off of, or a dict which
            describes the scenario attributes including applied projects and respective
            conflicts. `{"applied_projects": [],"conflicts":{...}}`
        project_card_list: Optional list of ProjectCard instances to add to planned projects.
            Defaults to None.
        config: WranglerConfig instance or a dictionary of configuration settings or a path to
            one or more configuration files. Configurations that are not explicity set will
            default to the values in the default configuration in
            `/configs/wrangler/default.yml`.
        name: Optional name for the scenario.
    """
    WranglerLogger.info("Creating Scenario")
    self.config = load_wrangler_config(config)

    if project_card_list is None:
        project_card_list = []

    if isinstance(base_scenario, Scenario):
        base_scenario = base_scenario.__dict__

    self.base_scenario: dict = extract_base_scenario_metadata(base_scenario)

    if not set(BASE_SCENARIO_SUGGESTED_PROPS) <= set(base_scenario.keys()):
        WranglerLogger.warning(
            f"Base_scenario doesn't contain {BASE_SCENARIO_SUGGESTED_PROPS}"
        )
    self.name: str = name
    # if the base scenario had roadway or transit networks, use them as the basis.
    self.road_net: Optional[RoadwayNetwork] = copy.deepcopy(
        base_scenario.pop("road_net", None)
    )

    self.transit_net: Optional[TransitNetwork] = copy.deepcopy(
        base_scenario.pop("transit_net", None)
    )
    if self.road_net and self.transit_net:
        self.transit_net.road_net = self.road_net

    # Set configs for networks to be the same as scenario.
    if isinstance(self.road_net, RoadwayNetwork):
        self.road_net.config = self.config
    if isinstance(self.transit_net, TransitNetwork):
        self.transit_net.config = self.config

    self.project_cards: dict[str, ProjectCard] = {}
    self._planned_projects: list[str] = []
    self._queued_projects = None
    self.applied_projects: list[str] = base_scenario.pop("applied_projects", [])

    self.prerequisites: dict[str, list[str]] = base_scenario.pop("prerequisites", {})
    self.corequisites: dict[str, list[str]] = base_scenario.pop("corequisites", {})
    self.conflicts: dict[str, list[str]] = base_scenario.pop("conflicts", {})

    for p in project_card_list:
        self._add_project(p)

network_wrangler.scenario.Scenario.__str__

__str__()

String representation of the Scenario object.

Source code in network_wrangler/scenario.py

def __str__(self):
    """String representation of the Scenario object."""
    s = [f"{key}: {value}" for key, value in self.__dict__.items()]
    return "\n".join(s)

network_wrangler.scenario.Scenario.add_project_cards

add_project_cards(project_card_list, validate=True, filter_tags=None)

Adds a list of ProjectCard instances to the Scenario.

Checks that a project of same name is not already in scenario. If selected, will validate ProjectCard before adding. If provided, will only add ProjectCard if it matches at least one filter_tags.

Parameters:

• project_card_list (list[ProjectCard]) –

  List of ProjectCard instances to add to scenario.

• validate (bool, default: True ) –

  If True, will require each ProjectCard is validated before being added to scenario. Defaults to True.

• filter_tags (Optional[list[str]], default: None ) –

  If used, will filter ProjectCard instances and only add those whose tags match one or more of these filter_tags. Defaults to [] - which means no tag-filtering will occur.

Source code in network_wrangler/scenario.py

def add_project_cards(
    self,
    project_card_list: list[ProjectCard],
    validate: bool = True,
    filter_tags: Optional[list[str]] = None,
) -> None:
    """Adds a list of ProjectCard instances to the Scenario.

    Checks that a project of same name is not already in scenario.
    If selected, will validate ProjectCard before adding.
    If provided, will only add ProjectCard if it matches at least one filter_tags.

    Args:
        project_card_list: List of ProjectCard instances to add to
            scenario.
        validate (bool, optional): If True, will require each ProjectCard is validated before
            being added to scenario. Defaults to True.
        filter_tags: If used, will filter ProjectCard instances
            and only add those whose tags match one or more of these filter_tags.
            Defaults to [] - which means no tag-filtering will occur.
    """
    filter_tags = filter_tags or []
    for p in project_card_list:
        self._add_project(p, validate=validate, filter_tags=filter_tags)

network_wrangler.scenario.Scenario.apply_all_projects

apply_all_projects()

Applies all planned projects in the queue.

Source code in network_wrangler/scenario.py

def apply_all_projects(self):
    """Applies all planned projects in the queue."""
    # Call this to make sure projects are appropriately queued in hidden variable.
    self.queued_projects  # noqa: B018

    # Use hidden variable.
    while self._queued_projects:
        self._apply_project(self._queued_projects.popleft())

    # set this so it will trigger re-queuing any more projects.
    self._queued_projects = None

network_wrangler.scenario.Scenario.apply_projects

apply_projects(project_list)

Applies a specific list of projects from the planned project queue.

Will order the list of projects based on pre-requisites.

NOTE: does not check co-requisites b/c that isn’t possible when applying a single project.

Parameters:

• project_list (list[str]) –

  List of projects to be applied. All need to be in the planned project queue.

Source code in network_wrangler/scenario.py

def apply_projects(self, project_list: list[str]):
    """Applies a specific list of projects from the planned project queue.

    Will order the list of projects based on pre-requisites.

    NOTE: does not check co-requisites b/c that isn't possible when applying a single project.

    Args:
        project_list: List of projects to be applied. All need to be in the planned project
            queue.
    """
    project_list = [p.lower() for p in project_list]

    self._check_projects_requirements_satisfied(project_list)
    ordered_project_queue = self.order_projects(project_list)

    while ordered_project_queue:
        self._apply_project(ordered_project_queue.popleft())

    # Set so that when called again it will retrigger queueing from planned projects.
    self._ordered_projects = None

network_wrangler.scenario.Scenario.order_projects

order_projects(project_list)

Orders a list of projects based on moving up pre-requisites into a deque.

Parameters:

• project_list (list[str]) –

  list of projects to order

Source code in network_wrangler/scenario.py

def order_projects(self, project_list: list[str]) -> deque:
    """Orders a list of projects based on moving up pre-requisites into a deque.

    Args:
        project_list: list of projects to order

    Returns: deque for applying projects.
    """
    project_list = [p.lower() for p in project_list]
    assert self._check_projects_have_project_cards(project_list)

    # build prereq (adjacency) list for topological sort
    adjacency_list: dict[str, list] = defaultdict(list)
    visited_list: dict[str, bool] = defaultdict(bool)

    for project in project_list:
        visited_list[project] = False
        if not self.prerequisites.get(project):
            continue
        for prereq in self.prerequisites[project]:
            # this will always be true, else would have been flagged in missing \
            # prerequsite check, but just in case
            if prereq.lower() in project_list:
                if adjacency_list.get(prereq.lower()):
                    adjacency_list[prereq.lower()].append(project)
                else:
                    adjacency_list[prereq.lower()] = [project]

    # sorted_project_names is topological sorted project card names (based on prerequsiite)
    _ordered_projects = topological_sort(
        adjacency_list=adjacency_list, visited_list=visited_list
    )

    if set(_ordered_projects) != set(project_list):
        _missing = list(set(project_list) - set(_ordered_projects))
        msg = f"Project sort resulted in missing projects: {_missing}"
        raise ValueError(msg)

    project_deque = deque(_ordered_projects)

    WranglerLogger.debug(f"Ordered Projects: \n{project_deque}")

    return project_deque

network_wrangler.scenario.Scenario.write

write(path, name, overwrite=True, roadway_write=True, transit_write=True, projects_write=True, roadway_convert_complex_link_properties_to_single_field=False, roadway_out_dir=None, roadway_prefix=None, roadway_file_format='parquet', roadway_true_shape=False, transit_out_dir=None, transit_prefix=None, transit_file_format='txt', projects_out_dir=None)

Writes scenario networks and summary to disk and returns path to scenario file.

Parameters:

• path (Path) –

  Path to write scenario networks and scenario summary to.

• name (str) –

  Name to use.

• overwrite (bool, default: True ) –

  If True, will overwrite the files if they already exist.

• roadway_write (bool, default: True ) –

  If True, will write out the roadway network.

• transit_write (bool, default: True ) –

  If True, will write out the transit network.

• projects_write (bool, default: True ) –

  If True, will write out the project cards.

• roadway_convert_complex_link_properties_to_single_field (bool, default: False ) –

  If True, will convert complex link properties to a single field.

• roadway_out_dir (Optional[Path], default: None ) –

  Path to write the roadway network files to.

• roadway_prefix (Optional[str], default: None ) –

  Prefix to add to the file name.

• roadway_file_format (RoadwayFileTypes, default: 'parquet' ) –

  File format to write the roadway network to

• roadway_true_shape (bool, default: False ) –

  If True, will write the true shape of the roadway network

• transit_out_dir (Optional[Path], default: None ) –

  Path to write the transit network files to.

• transit_prefix (Optional[str], default: None ) –

  Prefix to add to the file name.

• transit_file_format (TransitFileTypes, default: 'txt' ) –

  File format to write the transit network to

• projects_out_dir (Optional[Path], default: None ) –

  Path to write the project cards to.

Source code in network_wrangler/scenario.py

def write(
    self,
    path: Path,
    name: str,
    overwrite: bool = True,
    roadway_write: bool = True,
    transit_write: bool = True,
    projects_write: bool = True,
    roadway_convert_complex_link_properties_to_single_field: bool = False,
    roadway_out_dir: Optional[Path] = None,
    roadway_prefix: Optional[str] = None,
    roadway_file_format: RoadwayFileTypes = "parquet",
    roadway_true_shape: bool = False,
    transit_out_dir: Optional[Path] = None,
    transit_prefix: Optional[str] = None,
    transit_file_format: TransitFileTypes = "txt",
    projects_out_dir: Optional[Path] = None,
) -> Path:
    """Writes scenario networks and summary to disk and returns path to scenario file.

    Args:
        path: Path to write scenario networks and scenario summary to.
        name: Name to use.
        overwrite: If True, will overwrite the files if they already exist.
        roadway_write: If True, will write out the roadway network.
        transit_write: If True, will write out the transit network.
        projects_write: If True, will write out the project cards.
        roadway_convert_complex_link_properties_to_single_field: If True, will convert complex
            link properties to a single field.
        roadway_out_dir: Path to write the roadway network files to.
        roadway_prefix: Prefix to add to the file name.
        roadway_file_format: File format to write the roadway network to
        roadway_true_shape: If True, will write the true shape of the roadway network
        transit_out_dir: Path to write the transit network files to.
        transit_prefix: Prefix to add to the file name.
        transit_file_format: File format to write the transit network to
        projects_out_dir: Path to write the project cards to.
    """
    path = Path(path)
    path.mkdir(parents=True, exist_ok=True)

    if self.road_net and roadway_write:
        if roadway_out_dir is None:
            roadway_out_dir = path / "roadway"
        roadway_out_dir.mkdir(parents=True, exist_ok=True)

        write_roadway(
            net=self.road_net,
            out_dir=roadway_out_dir,
            prefix=roadway_prefix or name,
            convert_complex_link_properties_to_single_field=roadway_convert_complex_link_properties_to_single_field,
            file_format=roadway_file_format,
            true_shape=roadway_true_shape,
            overwrite=overwrite,
        )
    if self.transit_net and transit_write:
        if transit_out_dir is None:
            transit_out_dir = path / "transit"
        transit_out_dir.mkdir(parents=True, exist_ok=True)
        write_transit(
            self.transit_net,
            out_dir=transit_out_dir,
            prefix=transit_prefix or name,
            file_format=transit_file_format,
            overwrite=overwrite,
        )
    if projects_write:
        if projects_out_dir is None:
            projects_out_dir = path / "projects"
        write_applied_projects(
            self,
            out_dir=projects_out_dir,
            overwrite=overwrite,
        )

    scenario_data = self.summary
    if transit_write:
        scenario_data["transit"] = {
            "dir": str(transit_out_dir),
            "file_format": transit_file_format,
        }
    if roadway_write:
        scenario_data["roadway"] = {
            "dir": str(roadway_out_dir),
            "file_format": roadway_file_format,
        }
    if projects_write:
        scenario_data["project_cards"] = {"dir": str(projects_out_dir)}
    scenario_file_path = Path(path) / f"{name}_scenario.yml"
    with scenario_file_path.open("w") as f:
        yaml.dump(scenario_data, f, default_flow_style=False, allow_unicode=True)
    return scenario_file_path

network_wrangler.scenario.build_scenario_from_config

build_scenario_from_config(scenario_config)

Builds a scenario from a dictionary configuration.

Parameters:

• scenario_config (Union[Path, list[Path], ScenarioConfig, dict]) –

  Path to a configuration file, list of paths, or a dictionary of configuration.

Source code in network_wrangler/scenario.py

def build_scenario_from_config(
    scenario_config: Union[Path, list[Path], ScenarioConfig, dict],
) -> Scenario:
    """Builds a scenario from a dictionary configuration.

    Args:
        scenario_config: Path to a configuration file, list of paths, or a dictionary of
            configuration.
    """
    WranglerLogger.info(f"Building Scenario from Configuration: {scenario_config}")
    scenario_config = load_scenario_config(scenario_config)
    WranglerLogger.debug(f"{pprint.pformat(scenario_config)}")

    base_scenario = create_base_scenario(
        **scenario_config.base_scenario.to_dict(), config=scenario_config.wrangler_config
    )

    my_scenario = create_scenario(
        base_scenario=base_scenario,
        config=scenario_config.wrangler_config,
        **scenario_config.projects.to_dict(),
    )

    my_scenario.apply_all_projects()

    write_args = _scenario_output_config_to_scenario_write(scenario_config.output_scenario)
    my_scenario.write(**write_args, name=scenario_config.name)
    return my_scenario

network_wrangler.scenario.create_base_scenario

create_base_scenario(roadway=None, transit=None, applied_projects=None, conflicts=None, config=DefaultConfig)

Creates a base scenario dictionary from roadway and transit network files.

Parameters:

• roadway (Optional[dict], default: None ) –

  kwargs for load_roadway_from_dir

• transit (Optional[dict], default: None ) –

  kwargs for load_transit from dir

• applied_projects (Optional[list], default: None ) –

  list of projects that have been applied to the base scenario.

• conflicts (Optional[dict], default: None ) –

  dictionary of conflicts that have been identified in the base scenario. Takes the format of {"projectA": ["projectB", "projectC"]} showing that projectA, which has been applied, conflicts with projectB and projectC and so they shouldn’t be applied in the future.

• config (WranglerConfig, default: DefaultConfig ) –

  WranglerConfig instance.

Source code in network_wrangler/scenario.py

def create_base_scenario(
    roadway: Optional[dict] = None,
    transit: Optional[dict] = None,
    applied_projects: Optional[list] = None,
    conflicts: Optional[dict] = None,
    config: WranglerConfig = DefaultConfig,
) -> dict:
    """Creates a base scenario dictionary from roadway and transit network files.

    Args:
        roadway: kwargs for load_roadway_from_dir
        transit: kwargs for load_transit from dir
        applied_projects: list of projects that have been applied to the base scenario.
        conflicts: dictionary of conflicts that have been identified in the base scenario.
            Takes the format of `{"projectA": ["projectB", "projectC"]}` showing that projectA,
            which has been applied, conflicts with projectB and projectC and so they shouldn't be
            applied in the future.
        config: WranglerConfig instance.
    """
    applied_projects = applied_projects or []
    conflicts = conflicts or {}
    if roadway:
        road_net = load_roadway_from_dir(**roadway, config=config)
    else:
        road_net = None
        WranglerLogger.info(
            "No roadway directory specified, base scenario will have empty roadway network."
        )

    if transit:
        transit_net = load_transit(**transit, config=config)
        if roadway:
            transit_net.road_net = road_net
    else:
        transit_net = None
        WranglerLogger.info(
            "No transit directory specified, base scenario will have empty transit network."
        )

    base_scenario = {
        "road_net": road_net,
        "transit_net": transit_net,
        "applied_projects": applied_projects,
        "conflicts": conflicts,
    }

    return base_scenario

network_wrangler.scenario.create_scenario

create_scenario(base_scenario=None, name=datetime.now().strftime('%Y%m%d%H%M%S'), project_card_list=None, project_card_filepath=None, filter_tags=None, config=None)

Creates scenario from a base scenario and adds project cards.

Project cards can be added using any/all of the following methods: 1. List of ProjectCard instances 2. List of ProjectCard files 3. Directory and optional glob search to find project card files in

Checks that a project of same name is not already in scenario. If selected, will validate ProjectCard before adding. If provided, will only add ProjectCard if it matches at least one filter_tags.

Parameters:

• base_scenario (Optional[Union[Scenario, dict]], default: None ) –

  base Scenario scenario instances of dictionary of attributes.

• name (str, default: strftime('%Y%m%d%H%M%S') ) –

  Optional name for the scenario. Defaults to current datetime.

• project_card_list –

  List of ProjectCard instances to create Scenario from. Defaults to [].

• project_card_filepath (Optional[Union[list[Path], Path]], default: None ) –

  where the project card is. A single path, list of paths,

• filter_tags (Optional[list[str]], default: None ) –

  If used, will only add the project card if its tags match one or more of these filter_tags. Defaults to [] which means no tag-filtering will occur.

• config (Optional[Union[dict, Path, list[Path], WranglerConfig]], default: None ) –

  Optional wrangler configuration file or dictionary or instance. Defaults to default config.

Source code in network_wrangler/scenario.py

def create_scenario(
    base_scenario: Optional[Union[Scenario, dict]] = None,
    name: str = datetime.now().strftime("%Y%m%d%H%M%S"),
    project_card_list=None,
    project_card_filepath: Optional[Union[list[Path], Path]] = None,
    filter_tags: Optional[list[str]] = None,
    config: Optional[Union[dict, Path, list[Path], WranglerConfig]] = None,
) -> Scenario:
    """Creates scenario from a base scenario and adds project cards.

    Project cards can be added using any/all of the following methods:
    1. List of ProjectCard instances
    2. List of ProjectCard files
    3. Directory and optional glob search to find project card files in

    Checks that a project of same name is not already in scenario.
    If selected, will validate ProjectCard before adding.
    If provided, will only add ProjectCard if it matches at least one filter_tags.

    Args:
        base_scenario: base Scenario scenario instances of dictionary of attributes.
        name: Optional name for the scenario. Defaults to current datetime.
        project_card_list: List of ProjectCard instances to create Scenario from. Defaults
            to [].
        project_card_filepath: where the project card is.  A single path, list of paths,
        a directory, or a glob pattern. Defaults to None.
        filter_tags: If used, will only add the project card if
            its tags match one or more of these filter_tags. Defaults to []
            which means no tag-filtering will occur.
        config: Optional wrangler configuration file or dictionary or instance. Defaults to
            default config.
    """
    base_scenario = base_scenario or {}
    project_card_list = project_card_list or []
    filter_tags = filter_tags or []

    scenario = Scenario(base_scenario, config=config, name=name)

    if project_card_filepath:
        project_card_list += list(
            read_cards(project_card_filepath, filter_tags=filter_tags).values()
        )

    if project_card_list:
        scenario.add_project_cards(project_card_list, filter_tags=filter_tags)

    return scenario

network_wrangler.scenario.extract_base_scenario_metadata

extract_base_scenario_metadata(base_scenario)

Extract metadata from base scenario rather than keeping all of big files.

Useful for summarizing a scenario.

Source code in network_wrangler/scenario.py

def extract_base_scenario_metadata(base_scenario: dict) -> dict:
    """Extract metadata from base scenario rather than keeping all of big files.

    Useful for summarizing a scenario.
    """
    _skip_copy = ["road_net", "transit_net", "config"]
    out_dict = {k: v for k, v in base_scenario.items() if k not in _skip_copy}
    if isinstance(base_scenario.get("road_net"), RoadwayNetwork):
        nodes_file_path = base_scenario["road_net"].nodes_df.attrs.get("source_file", None)
        if nodes_file_path is not None:
            out_dict["roadway"] = {
                "dir": str(Path(nodes_file_path).parent),
                "file_format": str(nodes_file_path.suffix).lstrip("."),
            }
    if isinstance(base_scenario.get("transit_net"), TransitNetwork):
        feed_path = base_scenario["transit_net"].feed.feed_path
        if feed_path is not None:
            out_dict["transit"] = {"dir": str(feed_path)}
    return out_dict

network_wrangler.scenario.load_scenario

load_scenario(scenario_data, name=datetime.now().strftime('%Y%m%d%H%M%S'))

Loads a scenario from a file written by Scenario.write() as the base scenario.

Parameters:

• scenario_data (Union[dict, Path]) –

  Scenario data as a dict or path to scenario data file

• name (str, default: strftime('%Y%m%d%H%M%S') ) –

  Optional name for the scenario. Defaults to current datetime.

Source code in network_wrangler/scenario.py

def load_scenario(
    scenario_data: Union[dict, Path],
    name: str = datetime.now().strftime("%Y%m%d%H%M%S"),
) -> Scenario:
    """Loads a scenario from a file written by Scenario.write() as the base scenario.

    Args:
        scenario_data: Scenario data as a dict or path to scenario data file
        name: Optional name for the scenario. Defaults to current datetime.
    """
    if not isinstance(scenario_data, dict):
        WranglerLogger.debug(f"Loading Scenario from file: {scenario_data}")
        scenario_data = load_dict(scenario_data)
    else:
        WranglerLogger.debug("Loading Scenario from dict.")

    base_scenario_data = {
        "roadway": scenario_data.get("roadway"),
        "transit": scenario_data.get("transit"),
        "applied_projects": scenario_data.get("applied_projects", []),
        "conflicts": scenario_data.get("conflicts", {}),
    }
    base_scenario = _load_base_scenario_from_config(
        base_scenario_data, config=scenario_data["config"]
    )
    my_scenario = create_scenario(
        base_scenario=base_scenario, name=name, config=scenario_data["config"]
    )
    return my_scenario

network_wrangler.scenario.write_applied_projects

write_applied_projects(scenario, out_dir, overwrite=True)

Summarizes all projects in a scenario to folder.

Parameters:

• scenario (Scenario) –

  Scenario instance to summarize.

• out_dir (Path) –

  Path to write the project cards.

• overwrite (bool, default: True ) –

  If True, will overwrite the files if they already exist.

Source code in network_wrangler/scenario.py

def write_applied_projects(scenario: Scenario, out_dir: Path, overwrite: bool = True) -> None:
    """Summarizes all projects in a scenario to folder.

    Args:
        scenario: Scenario instance to summarize.
        out_dir: Path to write the project cards.
        overwrite: If True, will overwrite the files if they already exist.
    """
    outdir = Path(out_dir)
    prep_dir(out_dir, overwrite=overwrite)

    for p in scenario.applied_projects:
        if p in scenario.project_cards:
            card = scenario.project_cards[p]
        elif p in scenario.base_scenario["project_cards"]:
            card = scenario.base_scenario["project_cards"][p]
        else:
            continue
        filename = Path(card.__dict__.get("file", f"{p}.yml")).name
        outpath = outdir / filename
        write_card(card, outpath)

Roadway Network class and functions for Network Wrangler.

Used to represent a roadway network and perform operations on it.

Usage:

from network_wrangler import load_roadway_from_dir, write_roadway

net = load_roadway_from_dir("my_dir")
net.get_selection({"links": [{"name": ["I 35E"]}]})
net.apply("my_project_card.yml")

write_roadway(net, "my_out_prefix", "my_dir", file_format="parquet")

network_wrangler.roadway.network.RoadwayNetwork

Bases: BaseModel

Representation of a Roadway Network.

Typical usage example:

net = load_roadway(
    links_file=MY_LINK_FILE,
    nodes_file=MY_NODE_FILE,
    shapes_file=MY_SHAPE_FILE,
)
my_selection = {
    "link": [{"name": ["I 35E"]}],
    "A": {"osm_node_id": "961117623"},  # start searching for segments at A
    "B": {"osm_node_id": "2564047368"},
}
net.get_selection(my_selection)

my_change = [
    {
        'property': 'lanes',
        'existing': 1,
        'set': 2,
    },
    {
        'property': 'drive_access',
        'set': 0,
    },
]

my_net.apply_roadway_feature_change(
    my_net.get_selection(my_selection),
    my_change
)

    net.model_net
    net.is_network_connected(mode="drive", nodes=self.m_nodes_df, links=self.m_links_df)
    _, disconnected_nodes = net.assess_connectivity(
        mode="walk",
        ignore_end_nodes=True,
        nodes=self.m_nodes_df,
        links=self.m_links_df
    )
    write_roadway(net,filename=my_out_prefix, path=my_dir, for_model = True)

Attributes:

• nodes_df (RoadNodesTable) –

  dataframe of of node records.

• links_df (RoadLinksTable) –

  dataframe of link records and associated properties.

• shapes_df (RoadShapesTable) –

  dataframe of detailed shape records This is lazily created iff it is called because shapes files can be expensive to read.

• _selections (dict) –

  dictionary of stored roadway selection objects, mapped by RoadwayLinkSelection.sel_key or RoadwayNodeSelection.sel_key in case they are made repeatedly.

• network_hash (str) –

  dynamic property of the hashed value of links_df and nodes_df. Used for quickly identifying if a network has changed since various expensive operations have taken place (i.e. generating a ModelRoadwayNetwork or a network graph)

• model_net (ModelRoadwayNetwork) –

  referenced ModelRoadwayNetwork object which will be lazily created if None or if the network_hash has changed.

• config (WranglerConfig) –

  wrangler configuration object

Source code in network_wrangler/roadway/network.py

class RoadwayNetwork(BaseModel):
    """Representation of a Roadway Network.

    Typical usage example:

    ```py
    net = load_roadway(
        links_file=MY_LINK_FILE,
        nodes_file=MY_NODE_FILE,
        shapes_file=MY_SHAPE_FILE,
    )
    my_selection = {
        "link": [{"name": ["I 35E"]}],
        "A": {"osm_node_id": "961117623"},  # start searching for segments at A
        "B": {"osm_node_id": "2564047368"},
    }
    net.get_selection(my_selection)

    my_change = [
        {
            'property': 'lanes',
            'existing': 1,
            'set': 2,
        },
        {
            'property': 'drive_access',
            'set': 0,
        },
    ]

    my_net.apply_roadway_feature_change(
        my_net.get_selection(my_selection),
        my_change
    )

        net.model_net
        net.is_network_connected(mode="drive", nodes=self.m_nodes_df, links=self.m_links_df)
        _, disconnected_nodes = net.assess_connectivity(
            mode="walk",
            ignore_end_nodes=True,
            nodes=self.m_nodes_df,
            links=self.m_links_df
        )
        write_roadway(net,filename=my_out_prefix, path=my_dir, for_model = True)
    ```

    Attributes:
        nodes_df (RoadNodesTable): dataframe of of node records.
        links_df (RoadLinksTable): dataframe of link records and associated properties.
        shapes_df (RoadShapesTable): dataframe of detailed shape records  This is lazily
            created iff it is called because shapes files can be expensive to read.
        _selections (dict): dictionary of stored roadway selection objects, mapped by
            `RoadwayLinkSelection.sel_key` or `RoadwayNodeSelection.sel_key` in case they are
                made repeatedly.
        network_hash: dynamic property of the hashed value of links_df and nodes_df. Used for
            quickly identifying if a network has changed since various expensive operations have
            taken place (i.e. generating a ModelRoadwayNetwork or a network graph)
        model_net (ModelRoadwayNetwork): referenced `ModelRoadwayNetwork` object which will be
            lazily created if None or if the `network_hash` has changed.
        config (WranglerConfig): wrangler configuration object
    """

    model_config = {"arbitrary_types_allowed": True}

    nodes_df: pd.DataFrame
    links_df: pd.DataFrame
    _shapes_df: Optional[pd.DataFrame] = None

    _links_file: Optional[Path] = None
    _nodes_file: Optional[Path] = None
    _shapes_file: Optional[Path] = None

    config: WranglerConfig = DefaultConfig

    _model_net: Optional[ModelRoadwayNetwork] = None
    _selections: dict[str, Selections] = {}
    _modal_graphs: dict[str, dict] = defaultdict(lambda: {"graph": None, "hash": None})

    @field_validator("config")
    def validate_config(cls, v):
        """Validate config."""
        return load_wrangler_config(v)

    @field_validator("nodes_df", "links_df")
    def coerce_crs(cls, v):
        """Coerce crs of nodes_df and links_df to LAT_LON_CRS."""
        if v.crs != LAT_LON_CRS:
            WranglerLogger.warning(
                f"CRS of links_df ({v.crs}) doesn't match network crs {LAT_LON_CRS}. \
                    Changing to network crs."
            )
            v.to_crs(LAT_LON_CRS)
        return v

    @property
    def shapes_df(self) -> pd.DataFrame:
        """Load and return RoadShapesTable.

        If not already loaded, will read from shapes_file and return. If shapes_file is None,
        will return an empty dataframe with the right schema. If shapes_df is already set, will
        return that.
        """
        if (self._shapes_df is None or self._shapes_df.empty) and self._shapes_file is not None:
            self._shapes_df = read_shapes(
                self._shapes_file,
                filter_to_shape_ids=self.links_df.shape_id.to_list(),
                config=self.config,
            )
        # if there is NONE, then at least create an empty dataframe with right schema
        elif self._shapes_df is None:
            self._shapes_df = empty_df_from_datamodel(RoadShapesTable)
            self._shapes_df.set_index("shape_id_idx", inplace=True)

        return self._shapes_df

    @shapes_df.setter
    def shapes_df(self, value):
        self._shapes_df = df_to_shapes_df(value, config=self.config)

    @property
    def network_hash(self) -> str:
        """Hash of the links and nodes dataframes."""
        _value = str.encode(self.links_df.df_hash() + "-" + self.nodes_df.df_hash())

        _hash = hashlib.sha256(_value).hexdigest()
        return _hash

    @property
    def model_net(self) -> ModelRoadwayNetwork:
        """Return a ModelRoadwayNetwork object for this network."""
        if self._model_net is None or self._model_net._net_hash != self.network_hash:
            self._model_net = ModelRoadwayNetwork(self)
        return self._model_net

    @property
    def summary(self) -> dict:
        """Quick summary dictionary of number of links, nodes."""
        d = {
            "links": len(self.links_df),
            "nodes": len(self.nodes_df),
        }
        return d

    @property
    def link_shapes_df(self) -> gpd.GeoDataFrame:
        """Add shape geometry to links if available.

        returns: shapes merged to links dataframe
        """
        _links_df = copy.deepcopy(self.links_df)
        link_shapes_df = _links_df.merge(
            self.shapes_df,
            left_on="shape_id",
            right_on="shape_id",
            how="left",
        )
        link_shapes_df["geometry"] = link_shapes_df["geometry_y"].combine_first(
            link_shapes_df["geometry_x"]
        )
        link_shapes_df = link_shapes_df.drop(columns=["geometry_x", "geometry_y"])
        link_shapes_df = link_shapes_df.set_geometry("geometry")
        return link_shapes_df

    def get_property_by_timespan_and_group(
        self,
        link_property: str,
        category: Optional[Union[str, int]] = DEFAULT_CATEGORY,
        timespan: Optional[TimespanString] = DEFAULT_TIMESPAN,
        strict_timespan_match: bool = False,
        min_overlap_minutes: int = 60,
    ) -> Any:
        """Returns a new dataframe with model_link_id and link property by category and timespan.

        Convenience method for backward compatability.

        Args:
            link_property: link property to query
            category: category to query or a list of categories. Defaults to DEFAULT_CATEGORY.
            timespan: timespan to query in the form of ["HH:MM","HH:MM"].
                Defaults to DEFAULT_TIMESPAN.
            strict_timespan_match: If True, will only return links that match the timespan exactly.
                Defaults to False.
            min_overlap_minutes: If strict_timespan_match is False, will return links that overlap
                with the timespan by at least this many minutes. Defaults to 60.
        """
        from .links.scopes import prop_for_scope  # noqa: PLC0415

        return prop_for_scope(
            self.links_df,
            link_property,
            timespan=timespan,
            category=category,
            strict_timespan_match=strict_timespan_match,
            min_overlap_minutes=min_overlap_minutes,
        )

    def get_selection(
        self,
        selection_dict: Union[dict, SelectFacility],
        overwrite: bool = False,
    ) -> Union[RoadwayNodeSelection, RoadwayLinkSelection]:
        """Return selection if it already exists, otherwise performs selection.

        Args:
            selection_dict (dict): SelectFacility dictionary.
            overwrite: if True, will overwrite any previously cached searches. Defaults to False.
        """
        key = _create_selection_key(selection_dict)
        if (key in self._selections) and not overwrite:
            WranglerLogger.debug(f"Using cached selection from key: {key}")
            return self._selections[key]

        if isinstance(selection_dict, SelectFacility):
            selection_data = selection_dict
        elif isinstance(selection_dict, SelectLinksDict):
            selection_data = SelectFacility(links=selection_dict)
        elif isinstance(selection_dict, SelectNodesDict):
            selection_data = SelectFacility(nodes=selection_dict)
        elif isinstance(selection_dict, dict):
            selection_data = SelectFacility(**selection_dict)
        else:
            msg = "selection_dict arg must be a dictionary or SelectFacility model."
            WranglerLogger.error(
                msg + f" Received: {selection_dict} of type {type(selection_dict)}"
            )
            raise SelectionError(msg)

        WranglerLogger.debug(f"Getting selection from key: {key}")
        if "links" in selection_data.fields:
            return RoadwayLinkSelection(self, selection_dict)
        if "nodes" in selection_data.fields:
            return RoadwayNodeSelection(self, selection_dict)
        msg = "Selection data should have either 'links' or 'nodes'."
        WranglerLogger.error(msg + f" Received: {selection_dict}")
        raise SelectionError(msg)

    def modal_graph_hash(self, mode) -> str:
        """Hash of the links in order to detect a network change from when graph created."""
        _value = str.encode(self.links_df.df_hash() + "-" + mode)
        _hash = hashlib.sha256(_value).hexdigest()

        return _hash

    def get_modal_graph(self, mode) -> MultiDiGraph:
        """Return a networkx graph of the network for a specific mode.

        Args:
            mode: mode of the network, one of `drive`,`transit`,`walk`, `bike`
        """
        from .graph import net_to_graph  # noqa: PLC0415

        if self._modal_graphs[mode]["hash"] != self.modal_graph_hash(mode):
            self._modal_graphs[mode]["graph"] = net_to_graph(self, mode)

        return self._modal_graphs[mode]["graph"]

    def apply(
        self,
        project_card: Union[ProjectCard, dict],
        transit_net: Optional[TransitNetwork] = None,
        **kwargs,
    ) -> RoadwayNetwork:
        """Wrapper method to apply a roadway project, returning a new RoadwayNetwork instance.

        Args:
            project_card: either a dictionary of the project card object or ProjectCard instance
            transit_net: optional transit network which will be used to if project requires as
                noted in `SECONDARY_TRANSIT_CARD_TYPES`.  If no transit network is provided, will
                skip anything related to transit network.
            **kwargs: keyword arguments to pass to project application
        """
        if not (isinstance(project_card, (ProjectCard, SubProject))):
            project_card = ProjectCard(project_card)

        # project_card.validate()
        if not project_card.valid:
            msg = f"Project card {project_card.project} not valid."
            WranglerLogger.error(msg)
            raise ProjectCardError(msg)

        if project_card._sub_projects:
            for sp in project_card._sub_projects:
                WranglerLogger.debug(f"- applying subproject: {sp.change_type}")
                self._apply_change(sp, transit_net=transit_net, **kwargs)
            return self
        return self._apply_change(project_card, transit_net=transit_net, **kwargs)

    def _apply_change(
        self,
        change: Union[ProjectCard, SubProject],
        transit_net: Optional[TransitNetwork] = None,
    ) -> RoadwayNetwork:
        """Apply a single change: a single-project project or a sub-project."""
        if not isinstance(change, SubProject):
            WranglerLogger.info(f"Applying Project to Roadway Network: {change.project}")

        if change.change_type == "roadway_property_change":
            return apply_roadway_property_change(
                self,
                self.get_selection(change.roadway_property_change["facility"]),
                change.roadway_property_change["property_changes"],
                project_name=change.project,
            )

        if change.change_type == "roadway_addition":
            return apply_new_roadway(
                self,
                change.roadway_addition,
                project_name=change.project,
            )

        if change.change_type == "roadway_deletion":
            return apply_roadway_deletion(
                self,
                change.roadway_deletion,
                transit_net=transit_net,
            )

        if change.change_type == "pycode":
            return apply_calculated_roadway(self, change.pycode)
        WranglerLogger.error(f"Couldn't find project in: \n{change.__dict__}")
        msg = f"Invalid Project Card Category: {change.change_type}"
        raise ProjectCardError(msg)

    def links_with_link_ids(self, link_ids: list[int]) -> pd.DataFrame:
        """Return subset of links_df based on link_ids list."""
        return filter_links_to_ids(self.links_df, link_ids)

    def links_with_nodes(self, node_ids: list[int]) -> pd.DataFrame:
        """Return subset of links_df based on node_ids list."""
        return filter_links_to_node_ids(self.links_df, node_ids)

    def nodes_in_links(self) -> pd.DataFrame:
        """Returns subset of self.nodes_df that are in self.links_df."""
        return filter_nodes_to_links(self.links_df, self.nodes_df)

    def node_coords(self, model_node_id: int) -> tuple:
        """Return coordinates (x, y) of a node based on model_node_id."""
        try:
            node = self.nodes_df[self.nodes_df.model_node_id == model_node_id]
        except ValueError as err:
            msg = f"Node with model_node_id {model_node_id} not found."
            WranglerLogger.error(msg)
            raise NodeNotFoundError(msg) from err
        return node.geometry.x.values[0], node.geometry.y.values[0]

    def add_links(
        self,
        add_links_df: pd.DataFrame,
        in_crs: int = LAT_LON_CRS,
    ):
        """Validate combined links_df with LinksSchema before adding to self.links_df.

        Args:
            add_links_df: Dataframe of additional links to add.
            in_crs: crs of input data. Defaults to LAT_LON_CRS.
        """
        dupe_recs = self.links_df.model_link_id.isin(add_links_df.model_link_id)

        if dupe_recs.any():
            dupe_ids = self.links_df.loc[dupe_recs, "model_link_id"]
            WranglerLogger.error(
                f"Cannot add links with model_link_id already in network: {dupe_ids}"
            )
            msg = "Cannot add links with model_link_id already in network."
            raise LinkAddError(msg)

        if add_links_df.attrs.get("name") != "road_links":
            add_links_df = data_to_links_df(add_links_df, nodes_df=self.nodes_df, in_crs=in_crs)
        self.links_df = validate_df_to_model(
            concat_with_attr([self.links_df, add_links_df], axis=0), RoadLinksTable
        )

    def add_nodes(
        self,
        add_nodes_df: pd.DataFrame,
        in_crs: int = LAT_LON_CRS,
    ):
        """Validate combined nodes_df with NodesSchema before adding to self.nodes_df.

        Args:
            add_nodes_df: Dataframe of additional nodes to add.
            in_crs: crs of input data. Defaults to LAT_LON_CRS.
        """
        dupe_ids = self.nodes_df.model_node_id.isin(add_nodes_df.model_node_id)
        if dupe_ids.any():
            WranglerLogger.error(
                f"Cannot add nodes with model_node_id already in network: {dupe_ids}"
            )
            msg = "Cannot add nodes with model_node_id already in network."
            raise NodeAddError(msg)

        if add_nodes_df.attrs.get("name") != "road_nodes":
            add_nodes_df = data_to_nodes_df(add_nodes_df, in_crs=in_crs, config=self.config)
        self.nodes_df = validate_df_to_model(
            concat_with_attr([self.nodes_df, add_nodes_df], axis=0), RoadNodesTable
        )
        if self.nodes_df.attrs.get("name") != "road_nodes":
            msg = f"Expected nodes_df to have name 'road_nodes', got {self.nodes_df.attrs.get('name')}"
            raise NotNodesError(msg)

    def add_shapes(
        self,
        add_shapes_df: pd.DataFrame,
        in_crs: int = LAT_LON_CRS,
    ):
        """Validate combined shapes_df with RoadShapesTable efore adding to self.shapes_df.

        Args:
            add_shapes_df: Dataframe of additional shapes to add.
            in_crs: crs of input data. Defaults to LAT_LON_CRS.
        """
        dupe_ids = self.shapes_df.shape_id.isin(add_shapes_df.shape_id)
        if dupe_ids.any():
            msg = "Cannot add shapes with shape_id already in network."
            WranglerLogger.error(msg + f"\nDuplicates: {dupe_ids}")
            raise ShapeAddError(msg)

        if add_shapes_df.attrs.get("name") != "road_shapes":
            add_shapes_df = df_to_shapes_df(add_shapes_df, in_crs=in_crs, config=self.config)

        WranglerLogger.debug(f"add_shapes_df: \n{add_shapes_df}")
        WranglerLogger.debug(f"self.shapes_df: \n{self.shapes_df}")

        self.shapes_df = validate_df_to_model(
            concat_with_attr([self.shapes_df, add_shapes_df], axis=0), RoadShapesTable
        )

    def delete_links(
        self,
        selection_dict: Union[dict, SelectLinksDict],
        clean_nodes: bool = False,
        clean_shapes: bool = False,
        transit_net: Optional[TransitNetwork] = None,
    ):
        """Deletes links based on selection dictionary and optionally associated nodes and shapes.

        Args:
            selection_dict (SelectLinks): Dictionary describing link selections as follows:
                `all`: Optional[bool] = False. If true, will select all.
                `name`: Optional[list[str]]
                `ref`: Optional[list[str]]
                `osm_link_id`:Optional[list[str]]
                `model_link_id`: Optional[list[int]]
                `modes`: Optional[list[str]]. Defaults to "any"
                `ignore_missing`: if true, will not error when defaults to True.
                ...plus any other link property to select on top of these.
            clean_nodes (bool, optional): If True, will clean nodes uniquely associated with
                deleted links. Defaults to False.
            clean_shapes (bool, optional): If True, will clean nodes uniquely associated with
                deleted links. Defaults to False.
            transit_net (TransitNetwork, optional): If provided, will check TransitNetwork and
                warn if deletion breaks transit shapes. Defaults to None.
        """
        if not isinstance(selection_dict, SelectLinksDict):
            selection_dict = SelectLinksDict(**selection_dict)
        selection_dict = selection_dict.model_dump(exclude_none=True, by_alias=True)
        selection = self.get_selection({"links": selection_dict})
        if isinstance(selection, RoadwayNodeSelection):
            msg = "Selection should be for links, but got nodes."
            raise SelectionError(msg)
        if clean_nodes:
            node_ids_to_delete = node_ids_unique_to_link_ids(
                selection.selected_links, selection.selected_links_df, self.nodes_df
            )
            WranglerLogger.debug(
                f"Dropping nodes associated with dropped links: \n{node_ids_to_delete}"
            )
            self.nodes_df = delete_nodes_by_ids(self.nodes_df, del_node_ids=node_ids_to_delete)

        if clean_shapes:
            shape_ids_to_delete = shape_ids_unique_to_link_ids(
                selection.selected_links, selection.selected_links_df, self.shapes_df
            )
            WranglerLogger.debug(
                f"Dropping shapes associated with dropped links: \n{shape_ids_to_delete}"
            )
            self.shapes_df = delete_shapes_by_ids(
                self.shapes_df, del_shape_ids=shape_ids_to_delete
            )

        self.links_df = delete_links_by_ids(
            self.links_df,
            selection.selected_links,
            ignore_missing=selection.ignore_missing,
            transit_net=transit_net,
        )

    def delete_nodes(
        self,
        selection_dict: Union[dict, SelectNodesDict],
        remove_links: bool = False,
    ) -> None:
        """Deletes nodes from roadway network. Wont delete nodes used by links in network.

        Args:
            selection_dict: dictionary of node selection criteria in the form of a SelectNodesDict.
            remove_links: if True, will remove any links that are associated with the nodes.
                If False, will only remove nodes if they are not associated with any links.
                Defaults to False.

        Raises:
            NodeDeletionError: If not ignore_missing and selected nodes to delete aren't in network
        """
        if not isinstance(selection_dict, SelectNodesDict):
            selection_dict = SelectNodesDict(**selection_dict)
        selection_dict = selection_dict.model_dump(exclude_none=True, by_alias=True)
        _selection = self.get_selection({"nodes": selection_dict})
        assert isinstance(_selection, RoadwayNodeSelection)  # for mypy
        selection: RoadwayNodeSelection = _selection
        if remove_links:
            del_node_ids = selection.selected_nodes
            link_ids = self.links_with_nodes(selection.selected_nodes).model_link_id.to_list()
            WranglerLogger.info(f"Removing {len(link_ids)} links associated with nodes.")
            self.delete_links({"model_link_id": link_ids})
        else:
            unused_node_ids = node_ids_without_links(self.nodes_df, self.links_df)
            del_node_ids = list(set(selection.selected_nodes).intersection(unused_node_ids))

        self.nodes_df = delete_nodes_by_ids(
            self.nodes_df, del_node_ids, ignore_missing=selection.ignore_missing
        )

    def clean_unused_shapes(self):
        """Removes any unused shapes from network that aren't referenced by links_df."""
        from .shapes.shapes import shape_ids_without_links  # noqa: PLC0415

        del_shape_ids = shape_ids_without_links(self.shapes_df, self.links_df)
        self.shapes_df = self.shapes_df.drop(del_shape_ids)

    def clean_unused_nodes(self):
        """Removes any unused nodes from network that aren't referenced by links_df.

        NOTE: does not check if these nodes are used by transit, so use with caution.
        """
        from .nodes.nodes import node_ids_without_links  # noqa: PLC0415

        node_ids = node_ids_without_links(self.nodes_df, self.links_df)
        self.nodes_df = self.nodes_df.drop(node_ids)

    def move_nodes(
        self,
        node_geometry_change_table: pd.DataFrame,
    ):
        """Moves nodes based on updated geometry along with associated links and shape geometry.

        Args:
            node_geometry_change_table: a table with model_node_id, X, Y, and CRS.
        """
        node_geometry_change_table = NodeGeometryChangeTable(node_geometry_change_table)
        node_ids = node_geometry_change_table.model_node_id.to_list()
        WranglerLogger.debug(f"Moving nodes: {node_ids}")
        self.nodes_df = edit_node_geometry(self.nodes_df, node_geometry_change_table)
        self.links_df = edit_link_geometry_from_nodes(self.links_df, self.nodes_df, node_ids)
        self.shapes_df = edit_shape_geometry_from_nodes(
            self.shapes_df, self.links_df, self.nodes_df, node_ids
        )

    def has_node(self, model_node_id: int) -> bool:
        """Queries if network has node based on model_node_id.

        Args:
            model_node_id: model_node_id to check for.
        """
        has_node = self.nodes_df[self.nodes_df.model_node_id].isin([model_node_id]).any()

        return has_node

    def has_link(self, ab: tuple) -> bool:
        """Returns true if network has links with AB values.

        Args:
            ab: Tuple of values corresponding with A and B.
        """
        sel_a, sel_b = ab
        has_link = (
            self.links_df[self.links_df[["A", "B"]]].isin_dict({"A": sel_a, "B": sel_b}).any()
        )
        return has_link

    def is_connected(self, mode: str) -> bool:
        """Determines if the network graph is "strongly" connected.

        A graph is strongly connected if each vertex is reachable from every other vertex.

        Args:
            mode:  mode of the network, one of `drive`,`transit`,`walk`, `bike`
        """
        is_connected = nx.is_strongly_connected(self.get_modal_graph(mode))

        return is_connected

network_wrangler.roadway.network.RoadwayNetwork.link_shapes_df property

link_shapes_df

Add shape geometry to links if available.

returns: shapes merged to links dataframe

network_wrangler.roadway.network.RoadwayNetwork.model_net property

model_net

Return a ModelRoadwayNetwork object for this network.

network_wrangler.roadway.network.RoadwayNetwork.network_hash property

network_hash

Hash of the links and nodes dataframes.

network_wrangler.roadway.network.RoadwayNetwork.shapes_df property writable

shapes_df

Load and return RoadShapesTable.

If not already loaded, will read from shapes_file and return. If shapes_file is None, will return an empty dataframe with the right schema. If shapes_df is already set, will return that.

network_wrangler.roadway.network.RoadwayNetwork.summary property

summary

Quick summary dictionary of number of links, nodes.

network_wrangler.roadway.network.RoadwayNetwork.add_links

add_links(add_links_df, in_crs=LAT_LON_CRS)

Validate combined links_df with LinksSchema before adding to self.links_df.

Parameters:

• add_links_df (DataFrame) –

  Dataframe of additional links to add.

• in_crs (int, default: LAT_LON_CRS ) –

  crs of input data. Defaults to LAT_LON_CRS.

Source code in network_wrangler/roadway/network.py

def add_links(
    self,
    add_links_df: pd.DataFrame,
    in_crs: int = LAT_LON_CRS,
):
    """Validate combined links_df with LinksSchema before adding to self.links_df.

    Args:
        add_links_df: Dataframe of additional links to add.
        in_crs: crs of input data. Defaults to LAT_LON_CRS.
    """
    dupe_recs = self.links_df.model_link_id.isin(add_links_df.model_link_id)

    if dupe_recs.any():
        dupe_ids = self.links_df.loc[dupe_recs, "model_link_id"]
        WranglerLogger.error(
            f"Cannot add links with model_link_id already in network: {dupe_ids}"
        )
        msg = "Cannot add links with model_link_id already in network."
        raise LinkAddError(msg)

    if add_links_df.attrs.get("name") != "road_links":
        add_links_df = data_to_links_df(add_links_df, nodes_df=self.nodes_df, in_crs=in_crs)
    self.links_df = validate_df_to_model(
        concat_with_attr([self.links_df, add_links_df], axis=0), RoadLinksTable
    )

network_wrangler.roadway.network.RoadwayNetwork.add_nodes

add_nodes(add_nodes_df, in_crs=LAT_LON_CRS)

Validate combined nodes_df with NodesSchema before adding to self.nodes_df.

Parameters:

• add_nodes_df (DataFrame) –

  Dataframe of additional nodes to add.

• in_crs (int, default: LAT_LON_CRS ) –

  crs of input data. Defaults to LAT_LON_CRS.

Source code in network_wrangler/roadway/network.py

def add_nodes(
    self,
    add_nodes_df: pd.DataFrame,
    in_crs: int = LAT_LON_CRS,
):
    """Validate combined nodes_df with NodesSchema before adding to self.nodes_df.

    Args:
        add_nodes_df: Dataframe of additional nodes to add.
        in_crs: crs of input data. Defaults to LAT_LON_CRS.
    """
    dupe_ids = self.nodes_df.model_node_id.isin(add_nodes_df.model_node_id)
    if dupe_ids.any():
        WranglerLogger.error(
            f"Cannot add nodes with model_node_id already in network: {dupe_ids}"
        )
        msg = "Cannot add nodes with model_node_id already in network."
        raise NodeAddError(msg)

    if add_nodes_df.attrs.get("name") != "road_nodes":
        add_nodes_df = data_to_nodes_df(add_nodes_df, in_crs=in_crs, config=self.config)
    self.nodes_df = validate_df_to_model(
        concat_with_attr([self.nodes_df, add_nodes_df], axis=0), RoadNodesTable
    )
    if self.nodes_df.attrs.get("name") != "road_nodes":
        msg = f"Expected nodes_df to have name 'road_nodes', got {self.nodes_df.attrs.get('name')}"
        raise NotNodesError(msg)

network_wrangler.roadway.network.RoadwayNetwork.add_shapes

add_shapes(add_shapes_df, in_crs=LAT_LON_CRS)

Validate combined shapes_df with RoadShapesTable efore adding to self.shapes_df.

Parameters:

• add_shapes_df (DataFrame) –

  Dataframe of additional shapes to add.

• in_crs (int, default: LAT_LON_CRS ) –

  crs of input data. Defaults to LAT_LON_CRS.

Source code in network_wrangler/roadway/network.py

def add_shapes(
    self,
    add_shapes_df: pd.DataFrame,
    in_crs: int = LAT_LON_CRS,
):
    """Validate combined shapes_df with RoadShapesTable efore adding to self.shapes_df.

    Args:
        add_shapes_df: Dataframe of additional shapes to add.
        in_crs: crs of input data. Defaults to LAT_LON_CRS.
    """
    dupe_ids = self.shapes_df.shape_id.isin(add_shapes_df.shape_id)
    if dupe_ids.any():
        msg = "Cannot add shapes with shape_id already in network."
        WranglerLogger.error(msg + f"\nDuplicates: {dupe_ids}")
        raise ShapeAddError(msg)

    if add_shapes_df.attrs.get("name") != "road_shapes":
        add_shapes_df = df_to_shapes_df(add_shapes_df, in_crs=in_crs, config=self.config)

    WranglerLogger.debug(f"add_shapes_df: \n{add_shapes_df}")
    WranglerLogger.debug(f"self.shapes_df: \n{self.shapes_df}")

    self.shapes_df = validate_df_to_model(
        concat_with_attr([self.shapes_df, add_shapes_df], axis=0), RoadShapesTable
    )

network_wrangler.roadway.network.RoadwayNetwork.apply

apply(project_card, transit_net=None, **kwargs)

Wrapper method to apply a roadway project, returning a new RoadwayNetwork instance.

Parameters:

• project_card (Union[ProjectCard, dict]) –

  either a dictionary of the project card object or ProjectCard instance

• transit_net (Optional[TransitNetwork], default: None ) –

  optional transit network which will be used to if project requires as noted in SECONDARY_TRANSIT_CARD_TYPES. If no transit network is provided, will skip anything related to transit network.

• **kwargs –

  keyword arguments to pass to project application

Source code in network_wrangler/roadway/network.py

def apply(
    self,
    project_card: Union[ProjectCard, dict],
    transit_net: Optional[TransitNetwork] = None,
    **kwargs,
) -> RoadwayNetwork:
    """Wrapper method to apply a roadway project, returning a new RoadwayNetwork instance.

    Args:
        project_card: either a dictionary of the project card object or ProjectCard instance
        transit_net: optional transit network which will be used to if project requires as
            noted in `SECONDARY_TRANSIT_CARD_TYPES`.  If no transit network is provided, will
            skip anything related to transit network.
        **kwargs: keyword arguments to pass to project application
    """
    if not (isinstance(project_card, (ProjectCard, SubProject))):
        project_card = ProjectCard(project_card)

    # project_card.validate()
    if not project_card.valid:
        msg = f"Project card {project_card.project} not valid."
        WranglerLogger.error(msg)
        raise ProjectCardError(msg)

    if project_card._sub_projects:
        for sp in project_card._sub_projects:
            WranglerLogger.debug(f"- applying subproject: {sp.change_type}")
            self._apply_change(sp, transit_net=transit_net, **kwargs)
        return self
    return self._apply_change(project_card, transit_net=transit_net, **kwargs)

network_wrangler.roadway.network.RoadwayNetwork.clean_unused_nodes

clean_unused_nodes()

Removes any unused nodes from network that aren’t referenced by links_df.

NOTE: does not check if these nodes are used by transit, so use with caution.

Source code in network_wrangler/roadway/network.py

def clean_unused_nodes(self):
    """Removes any unused nodes from network that aren't referenced by links_df.

    NOTE: does not check if these nodes are used by transit, so use with caution.
    """
    from .nodes.nodes import node_ids_without_links  # noqa: PLC0415

    node_ids = node_ids_without_links(self.nodes_df, self.links_df)
    self.nodes_df = self.nodes_df.drop(node_ids)

network_wrangler.roadway.network.RoadwayNetwork.clean_unused_shapes

clean_unused_shapes()

Removes any unused shapes from network that aren’t referenced by links_df.

Source code in network_wrangler/roadway/network.py

def clean_unused_shapes(self):
    """Removes any unused shapes from network that aren't referenced by links_df."""
    from .shapes.shapes import shape_ids_without_links  # noqa: PLC0415

    del_shape_ids = shape_ids_without_links(self.shapes_df, self.links_df)
    self.shapes_df = self.shapes_df.drop(del_shape_ids)

network_wrangler.roadway.network.RoadwayNetwork.coerce_crs

coerce_crs(v)

Coerce crs of nodes_df and links_df to LAT_LON_CRS.

Source code in network_wrangler/roadway/network.py

@field_validator("nodes_df", "links_df")
def coerce_crs(cls, v):
    """Coerce crs of nodes_df and links_df to LAT_LON_CRS."""
    if v.crs != LAT_LON_CRS:
        WranglerLogger.warning(
            f"CRS of links_df ({v.crs}) doesn't match network crs {LAT_LON_CRS}. \
                Changing to network crs."
        )
        v.to_crs(LAT_LON_CRS)
    return v

network_wrangler.roadway.network.RoadwayNetwork.delete_links

delete_links(selection_dict, clean_nodes=False, clean_shapes=False, transit_net=None)

Deletes links based on selection dictionary and optionally associated nodes and shapes.

Parameters:

• selection_dict (SelectLinks) –

  Dictionary describing link selections as follows: all: Optional[bool] = False. If true, will select all. name: Optional[list[str]] ref: Optional[list[str]] osm_link_id:Optional[list[str]] model_link_id: Optional[list[int]] modes: Optional[list[str]]. Defaults to “any” ignore_missing: if true, will not error when defaults to True. …plus any other link property to select on top of these.

• clean_nodes (bool, default: False ) –

  If True, will clean nodes uniquely associated with deleted links. Defaults to False.

• clean_shapes (bool, default: False ) –

  If True, will clean nodes uniquely associated with deleted links. Defaults to False.

• transit_net (TransitNetwork, default: None ) –

  If provided, will check TransitNetwork and warn if deletion breaks transit shapes. Defaults to None.

Source code in network_wrangler/roadway/network.py

def delete_links(
    self,
    selection_dict: Union[dict, SelectLinksDict],
    clean_nodes: bool = False,
    clean_shapes: bool = False,
    transit_net: Optional[TransitNetwork] = None,
):
    """Deletes links based on selection dictionary and optionally associated nodes and shapes.

    Args:
        selection_dict (SelectLinks): Dictionary describing link selections as follows:
            `all`: Optional[bool] = False. If true, will select all.
            `name`: Optional[list[str]]
            `ref`: Optional[list[str]]
            `osm_link_id`:Optional[list[str]]
            `model_link_id`: Optional[list[int]]
            `modes`: Optional[list[str]]. Defaults to "any"
            `ignore_missing`: if true, will not error when defaults to True.
            ...plus any other link property to select on top of these.
        clean_nodes (bool, optional): If True, will clean nodes uniquely associated with
            deleted links. Defaults to False.
        clean_shapes (bool, optional): If True, will clean nodes uniquely associated with
            deleted links. Defaults to False.
        transit_net (TransitNetwork, optional): If provided, will check TransitNetwork and
            warn if deletion breaks transit shapes. Defaults to None.
    """
    if not isinstance(selection_dict, SelectLinksDict):
        selection_dict = SelectLinksDict(**selection_dict)
    selection_dict = selection_dict.model_dump(exclude_none=True, by_alias=True)
    selection = self.get_selection({"links": selection_dict})
    if isinstance(selection, RoadwayNodeSelection):
        msg = "Selection should be for links, but got nodes."
        raise SelectionError(msg)
    if clean_nodes:
        node_ids_to_delete = node_ids_unique_to_link_ids(
            selection.selected_links, selection.selected_links_df, self.nodes_df
        )
        WranglerLogger.debug(
            f"Dropping nodes associated with dropped links: \n{node_ids_to_delete}"
        )
        self.nodes_df = delete_nodes_by_ids(self.nodes_df, del_node_ids=node_ids_to_delete)

    if clean_shapes:
        shape_ids_to_delete = shape_ids_unique_to_link_ids(
            selection.selected_links, selection.selected_links_df, self.shapes_df
        )
        WranglerLogger.debug(
            f"Dropping shapes associated with dropped links: \n{shape_ids_to_delete}"
        )
        self.shapes_df = delete_shapes_by_ids(
            self.shapes_df, del_shape_ids=shape_ids_to_delete
        )

    self.links_df = delete_links_by_ids(
        self.links_df,
        selection.selected_links,
        ignore_missing=selection.ignore_missing,
        transit_net=transit_net,
    )

network_wrangler.roadway.network.RoadwayNetwork.delete_nodes

delete_nodes(selection_dict, remove_links=False)

Deletes nodes from roadway network. Wont delete nodes used by links in network.

Parameters:

• selection_dict (Union[dict, SelectNodesDict]) –

  dictionary of node selection criteria in the form of a SelectNodesDict.

• remove_links (bool, default: False ) –

  if True, will remove any links that are associated with the nodes. If False, will only remove nodes if they are not associated with any links. Defaults to False.

Raises:

• NodeDeletionError –

  If not ignore_missing and selected nodes to delete aren’t in network

Source code in network_wrangler/roadway/network.py

def delete_nodes(
    self,
    selection_dict: Union[dict, SelectNodesDict],
    remove_links: bool = False,
) -> None:
    """Deletes nodes from roadway network. Wont delete nodes used by links in network.

    Args:
        selection_dict: dictionary of node selection criteria in the form of a SelectNodesDict.
        remove_links: if True, will remove any links that are associated with the nodes.
            If False, will only remove nodes if they are not associated with any links.
            Defaults to False.

    Raises:
        NodeDeletionError: If not ignore_missing and selected nodes to delete aren't in network
    """
    if not isinstance(selection_dict, SelectNodesDict):
        selection_dict = SelectNodesDict(**selection_dict)
    selection_dict = selection_dict.model_dump(exclude_none=True, by_alias=True)
    _selection = self.get_selection({"nodes": selection_dict})
    assert isinstance(_selection, RoadwayNodeSelection)  # for mypy
    selection: RoadwayNodeSelection = _selection
    if remove_links:
        del_node_ids = selection.selected_nodes
        link_ids = self.links_with_nodes(selection.selected_nodes).model_link_id.to_list()
        WranglerLogger.info(f"Removing {len(link_ids)} links associated with nodes.")
        self.delete_links({"model_link_id": link_ids})
    else:
        unused_node_ids = node_ids_without_links(self.nodes_df, self.links_df)
        del_node_ids = list(set(selection.selected_nodes).intersection(unused_node_ids))

    self.nodes_df = delete_nodes_by_ids(
        self.nodes_df, del_node_ids, ignore_missing=selection.ignore_missing
    )

network_wrangler.roadway.network.RoadwayNetwork.get_modal_graph

get_modal_graph(mode)

Return a networkx graph of the network for a specific mode.

Parameters:

• mode –

  mode of the network, one of drive,transit,walk, bike

Source code in network_wrangler/roadway/network.py

def get_modal_graph(self, mode) -> MultiDiGraph:
    """Return a networkx graph of the network for a specific mode.

    Args:
        mode: mode of the network, one of `drive`,`transit`,`walk`, `bike`
    """
    from .graph import net_to_graph  # noqa: PLC0415

    if self._modal_graphs[mode]["hash"] != self.modal_graph_hash(mode):
        self._modal_graphs[mode]["graph"] = net_to_graph(self, mode)

    return self._modal_graphs[mode]["graph"]

network_wrangler.roadway.network.RoadwayNetwork.get_property_by_timespan_and_group

get_property_by_timespan_and_group(link_property, category=DEFAULT_CATEGORY, timespan=DEFAULT_TIMESPAN, strict_timespan_match=False, min_overlap_minutes=60)

Returns a new dataframe with model_link_id and link property by category and timespan.

Convenience method for backward compatability.

Parameters:

• link_property (str) –

  link property to query

• category (Optional[Union[str, int]], default: DEFAULT_CATEGORY ) –

  category to query or a list of categories. Defaults to DEFAULT_CATEGORY.

• timespan (Optional[TimespanString], default: DEFAULT_TIMESPAN ) –

  timespan to query in the form of [“HH:MM”,”HH:MM”]. Defaults to DEFAULT_TIMESPAN.

• strict_timespan_match (bool, default: False ) –

  If True, will only return links that match the timespan exactly. Defaults to False.

• min_overlap_minutes (int, default: 60 ) –

  If strict_timespan_match is False, will return links that overlap with the timespan by at least this many minutes. Defaults to 60.

Source code in network_wrangler/roadway/network.py

def get_property_by_timespan_and_group(
    self,
    link_property: str,
    category: Optional[Union[str, int]] = DEFAULT_CATEGORY,
    timespan: Optional[TimespanString] = DEFAULT_TIMESPAN,
    strict_timespan_match: bool = False,
    min_overlap_minutes: int = 60,
) -> Any:
    """Returns a new dataframe with model_link_id and link property by category and timespan.

    Convenience method for backward compatability.

    Args:
        link_property: link property to query
        category: category to query or a list of categories. Defaults to DEFAULT_CATEGORY.
        timespan: timespan to query in the form of ["HH:MM","HH:MM"].
            Defaults to DEFAULT_TIMESPAN.
        strict_timespan_match: If True, will only return links that match the timespan exactly.
            Defaults to False.
        min_overlap_minutes: If strict_timespan_match is False, will return links that overlap
            with the timespan by at least this many minutes. Defaults to 60.
    """
    from .links.scopes import prop_for_scope  # noqa: PLC0415

    return prop_for_scope(
        self.links_df,
        link_property,
        timespan=timespan,
        category=category,
        strict_timespan_match=strict_timespan_match,
        min_overlap_minutes=min_overlap_minutes,
    )

network_wrangler.roadway.network.RoadwayNetwork.get_selection

get_selection(selection_dict, overwrite=False)

Return selection if it already exists, otherwise performs selection.

Parameters:

• selection_dict (dict) –

  SelectFacility dictionary.

• overwrite (bool, default: False ) –

  if True, will overwrite any previously cached searches. Defaults to False.

Source code in network_wrangler/roadway/network.py

def get_selection(
    self,
    selection_dict: Union[dict, SelectFacility],
    overwrite: bool = False,
) -> Union[RoadwayNodeSelection, RoadwayLinkSelection]:
    """Return selection if it already exists, otherwise performs selection.

    Args:
        selection_dict (dict): SelectFacility dictionary.
        overwrite: if True, will overwrite any previously cached searches. Defaults to False.
    """
    key = _create_selection_key(selection_dict)
    if (key in self._selections) and not overwrite:
        WranglerLogger.debug(f"Using cached selection from key: {key}")
        return self._selections[key]

    if isinstance(selection_dict, SelectFacility):
        selection_data = selection_dict
    elif isinstance(selection_dict, SelectLinksDict):
        selection_data = SelectFacility(links=selection_dict)
    elif isinstance(selection_dict, SelectNodesDict):
        selection_data = SelectFacility(nodes=selection_dict)
    elif isinstance(selection_dict, dict):
        selection_data = SelectFacility(**selection_dict)
    else:
        msg = "selection_dict arg must be a dictionary or SelectFacility model."
        WranglerLogger.error(
            msg + f" Received: {selection_dict} of type {type(selection_dict)}"
        )
        raise SelectionError(msg)

    WranglerLogger.debug(f"Getting selection from key: {key}")
    if "links" in selection_data.fields:
        return RoadwayLinkSelection(self, selection_dict)
    if "nodes" in selection_data.fields:
        return RoadwayNodeSelection(self, selection_dict)
    msg = "Selection data should have either 'links' or 'nodes'."
    WranglerLogger.error(msg + f" Received: {selection_dict}")
    raise SelectionError(msg)

network_wrangler.roadway.network.RoadwayNetwork.has_link

has_link(ab)

Returns true if network has links with AB values.

Parameters:

• ab (tuple) –

  Tuple of values corresponding with A and B.

Source code in network_wrangler/roadway/network.py

def has_link(self, ab: tuple) -> bool:
    """Returns true if network has links with AB values.

    Args:
        ab: Tuple of values corresponding with A and B.
    """
    sel_a, sel_b = ab
    has_link = (
        self.links_df[self.links_df[["A", "B"]]].isin_dict({"A": sel_a, "B": sel_b}).any()
    )
    return has_link

network_wrangler.roadway.network.RoadwayNetwork.has_node

has_node(model_node_id)

Queries if network has node based on model_node_id.

Parameters:

• model_node_id (int) –

  model_node_id to check for.

Source code in network_wrangler/roadway/network.py

def has_node(self, model_node_id: int) -> bool:
    """Queries if network has node based on model_node_id.

    Args:
        model_node_id: model_node_id to check for.
    """
    has_node = self.nodes_df[self.nodes_df.model_node_id].isin([model_node_id]).any()

    return has_node

network_wrangler.roadway.network.RoadwayNetwork.is_connected

is_connected(mode)

Determines if the network graph is “strongly” connected.

A graph is strongly connected if each vertex is reachable from every other vertex.

Parameters:

• mode (str) –

  mode of the network, one of drive,transit,walk, bike

Source code in network_wrangler/roadway/network.py

def is_connected(self, mode: str) -> bool:
    """Determines if the network graph is "strongly" connected.

    A graph is strongly connected if each vertex is reachable from every other vertex.

    Args:
        mode:  mode of the network, one of `drive`,`transit`,`walk`, `bike`
    """
    is_connected = nx.is_strongly_connected(self.get_modal_graph(mode))

    return is_connected

network_wrangler.roadway.network.RoadwayNetwork.links_with_link_ids

links_with_link_ids(link_ids)

Return subset of links_df based on link_ids list.

Source code in network_wrangler/roadway/network.py

def links_with_link_ids(self, link_ids: list[int]) -> pd.DataFrame:
    """Return subset of links_df based on link_ids list."""
    return filter_links_to_ids(self.links_df, link_ids)

network_wrangler.roadway.network.RoadwayNetwork.links_with_nodes

links_with_nodes(node_ids)

Return subset of links_df based on node_ids list.

Source code in network_wrangler/roadway/network.py

def links_with_nodes(self, node_ids: list[int]) -> pd.DataFrame:
    """Return subset of links_df based on node_ids list."""
    return filter_links_to_node_ids(self.links_df, node_ids)

network_wrangler.roadway.network.RoadwayNetwork.modal_graph_hash

modal_graph_hash(mode)

Hash of the links in order to detect a network change from when graph created.

Source code in network_wrangler/roadway/network.py

def modal_graph_hash(self, mode) -> str:
    """Hash of the links in order to detect a network change from when graph created."""
    _value = str.encode(self.links_df.df_hash() + "-" + mode)
    _hash = hashlib.sha256(_value).hexdigest()

    return _hash

network_wrangler.roadway.network.RoadwayNetwork.move_nodes

move_nodes(node_geometry_change_table)

Moves nodes based on updated geometry along with associated links and shape geometry.

Parameters:

• node_geometry_change_table (DataFrame) –

  a table with model_node_id, X, Y, and CRS.

Source code in network_wrangler/roadway/network.py

def move_nodes(
    self,
    node_geometry_change_table: pd.DataFrame,
):
    """Moves nodes based on updated geometry along with associated links and shape geometry.

    Args:
        node_geometry_change_table: a table with model_node_id, X, Y, and CRS.
    """
    node_geometry_change_table = NodeGeometryChangeTable(node_geometry_change_table)
    node_ids = node_geometry_change_table.model_node_id.to_list()
    WranglerLogger.debug(f"Moving nodes: {node_ids}")
    self.nodes_df = edit_node_geometry(self.nodes_df, node_geometry_change_table)
    self.links_df = edit_link_geometry_from_nodes(self.links_df, self.nodes_df, node_ids)
    self.shapes_df = edit_shape_geometry_from_nodes(
        self.shapes_df, self.links_df, self.nodes_df, node_ids
    )

network_wrangler.roadway.network.RoadwayNetwork.node_coords

node_coords(model_node_id)

Return coordinates (x, y) of a node based on model_node_id.

Source code in network_wrangler/roadway/network.py

def node_coords(self, model_node_id: int) -> tuple:
    """Return coordinates (x, y) of a node based on model_node_id."""
    try:
        node = self.nodes_df[self.nodes_df.model_node_id == model_node_id]
    except ValueError as err:
        msg = f"Node with model_node_id {model_node_id} not found."
        WranglerLogger.error(msg)
        raise NodeNotFoundError(msg) from err
    return node.geometry.x.values[0], node.geometry.y.values[0]

network_wrangler.roadway.network.RoadwayNetwork.nodes_in_links

nodes_in_links()

Returns subset of self.nodes_df that are in self.links_df.

Source code in network_wrangler/roadway/network.py

def nodes_in_links(self) -> pd.DataFrame:
    """Returns subset of self.nodes_df that are in self.links_df."""
    return filter_nodes_to_links(self.links_df, self.nodes_df)

network_wrangler.roadway.network.RoadwayNetwork.validate_config

validate_config(v)

Validate config.

Source code in network_wrangler/roadway/network.py

@field_validator("config")
def validate_config(cls, v):
    """Validate config."""
    return load_wrangler_config(v)

network_wrangler.roadway.network.add_incident_link_data_to_nodes

add_incident_link_data_to_nodes(links_df, nodes_df, link_variables=None)

Add data from links going to/from nodes to node.

Parameters:

• links_df (DataFrame) –

  Will assess connectivity of this links list

• nodes_df (DataFrame) –

  Will assess connectivity of this nodes list

• link_variables (Optional[list], default: None ) –

  list of columns in links dataframe to add to incident nodes

Returns:

• DataFrame –

  nodes DataFrame with link data where length is N*number of links going in/out

Source code in network_wrangler/roadway/network.py

def add_incident_link_data_to_nodes(
    links_df: pd.DataFrame,
    nodes_df: pd.DataFrame,
    link_variables: Optional[list] = None,
) -> pd.DataFrame:
    """Add data from links going to/from nodes to node.

    Args:
        links_df: Will assess connectivity of this links list
        nodes_df: Will assess connectivity of this nodes list
        link_variables: list of columns in links dataframe to add to incident nodes

    Returns:
        nodes DataFrame with link data where length is N*number of links going in/out
    """
    WranglerLogger.debug("Adding following link data to nodes: ".format())
    link_variables = link_variables or []

    _link_vals_to_nodes = [x for x in link_variables if x in links_df.columns]
    if link_variables not in _link_vals_to_nodes:
        WranglerLogger.warning(
            f"Following columns not in links_df and wont be added to nodes: {list(set(link_variables) - set(_link_vals_to_nodes))} "
        )

    _nodes_from_links_A = nodes_df.merge(
        links_df[[links_df.A, *_link_vals_to_nodes]],
        how="outer",
        left_on=nodes_df.model_node_id,
        right_on=links_df.A,
    )
    _nodes_from_links_B = nodes_df.merge(
        links_df[[links_df.B, *_link_vals_to_nodes]],
        how="outer",
        left_on=nodes_df.model_node_id,
        right_on=links_df.B,
    )
    _nodes_from_links_ab = concat_with_attr([_nodes_from_links_A, _nodes_from_links_B])

    return _nodes_from_links_ab

TransitNetwork class for representing a transit network.

Transit Networks are represented as a Wrangler-flavored GTFS Feed and optionally mapped to a RoadwayNetwork object. The TransitNetwork object is the primary object for managing transit networks in Wrangler.

Usage:

```python
import network_wrangler as wr

t = wr.load_transit(stpaul_gtfs)
t.road_net = wr.load_roadway(stpaul_roadway)
t = t.apply(project_card)
write_transit(t, "output_dir")
```

network_wrangler.transit.network.TransitNetwork

Representation of a Transit Network.

Typical usage example:

import network_wrangler as wr

tc = wr.load_transit(stpaul_gtfs)

Attributes:

• feed –

  gtfs feed object with interlinked tables.

• road_net (RoadwayNetwork) –

  Associated roadway network object.

• graph (MultiDiGraph) –

  Graph for associated roadway network object.

• config (WranglerConfig) –

  Configuration object for the transit network.

• feed_path (str) –

  Where the feed was read in from.

• validated_frequencies (bool) –

  The frequencies have been validated.

• validated_road_network_consistency –

  The network has been validated against the road network.

Source code in network_wrangler/transit/network.py

class TransitNetwork:
    """Representation of a Transit Network.

    Typical usage example:
    ``` py
    import network_wrangler as wr

    tc = wr.load_transit(stpaul_gtfs)
    ```

    Attributes:
        feed: gtfs feed object with interlinked tables.
        road_net (RoadwayNetwork): Associated roadway network object.
        graph (nx.MultiDiGraph): Graph for associated roadway network object.
        config (WranglerConfig): Configuration object for the transit network.
        feed_path (str): Where the feed was read in from.
        validated_frequencies (bool): The frequencies have been validated.
        validated_road_network_consistency (): The network has been validated against
            the road network.
    """

    TIME_COLS: ClassVar = ["arrival_time", "departure_time", "start_time", "end_time"]

    def __init__(self, feed: Feed, config: WranglerConfig = DefaultConfig) -> None:
        """Constructor for TransitNetwork.

        Args:
            feed: Feed object representing the transit network gtfs tables
            config: WranglerConfig object. Defaults to DefaultConfig.
        """
        WranglerLogger.debug("Creating new TransitNetwork.")

        self._road_net: Optional[RoadwayNetwork] = None
        self.feed: Feed = feed
        self.graph: nx.MultiDiGraph = None
        self.config: WranglerConfig = config
        # initialize
        self._consistent_with_road_net = False

        # cached selections
        self._selections: dict[str, TransitSelection] = {}

    @property
    def feed_path(self):
        """Pass through property from Feed."""
        return self.feed.feed_path

    @property
    def applied_projects(self) -> list[str]:
        """List of projects applied to the network.

        Note: This may or may not return a full accurate account of all the applied projects.
        For better project accounting, please leverage the scenario object.
        """
        return _get_applied_projects_from_tables(self.feed)

    @property
    def feed(self):
        """Feed associated with the transit network."""
        return self._feed

    @feed.setter
    def feed(self, feed: Feed):
        if not isinstance(feed, Feed):
            msg = f"TransitNetwork's feed value must be a valid Feed instance. \
                             This is a {type(feed)}."
            WranglerLogger.error(msg)
            raise TransitValidationError(msg)
        if self._road_net is None or transit_road_net_consistency(feed, self._road_net):
            self._feed = feed
            self._stored_feed_hash = copy.deepcopy(feed.hash)
        else:
            msg = "Can't assign Feed inconsistent with set Roadway Network."
            WranglerLogger.error(msg)
            raise TransitRoadwayConsistencyError(msg)

    @property
    def road_net(self) -> Union[None, RoadwayNetwork]:
        """Roadway network associated with the transit network."""
        return self._road_net

    @road_net.setter
    def road_net(self, road_net_in: RoadwayNetwork):
        if road_net_in is None or road_net_in.__class__.__name__ != "RoadwayNetwork":
            msg = f"TransitNetwork's road_net: value must be a valid RoadwayNetwork instance. \
                             This is a {type(road_net_in)}."
            WranglerLogger.error(msg)
            raise TransitValidationError(msg)
        if transit_road_net_consistency(self.feed, road_net_in):
            self._road_net = road_net_in
            self._stored_road_net_hash = copy.deepcopy(road_net_in.network_hash)
            self._consistent_with_road_net = True
        else:
            msg = "Can't assign inconsistent RoadwayNetwork - Roadway Network not \
                   set, but can be referenced separately."
            WranglerLogger.error(msg)
            raise TransitRoadwayConsistencyError(msg)

    @property
    def feed_hash(self):
        """Return the hash of the feed."""
        return self.feed.hash

    @property
    def consistent_with_road_net(self) -> bool:
        """Indicate if road_net is consistent with transit network.

        Will return True if road_net is None, but provide a warning.

        Checks the network hash of when consistency was last evaluated. If transit network or
        roadway network has changed, will re-evaluate consistency and return the updated value and
        update self._stored_road_net_hash.

        Returns:
            Boolean indicating if road_net is consistent with transit network.
        """
        if self.road_net is None:
            WranglerLogger.warning("Roadway Network not set, cannot accurately check consistency.")
            return True
        updated_road = self.road_net.network_hash != self._stored_road_net_hash
        updated_feed = self.feed_hash != self._stored_feed_hash

        if updated_road or updated_feed:
            self._consistent_with_road_net = transit_road_net_consistency(self.feed, self.road_net)
            self._stored_road_net_hash = copy.deepcopy(self.road_net.network_hash)
            self._stored_feed_hash = copy.deepcopy(self.feed_hash)
        return self._consistent_with_road_net

    def __deepcopy__(self, memo):
        """Returns copied TransitNetwork instance with deep copy of Feed but not roadway net."""
        COPY_REF_NOT_VALUE = ["_road_net"]
        # Create a new, empty instance
        copied_net = self.__class__.__new__(self.__class__)
        # Return the new TransitNetwork instance
        attribute_dict = vars(self)

        # Copy the attributes to the new instance
        for attr_name, attr_value in attribute_dict.items():
            # WranglerLogger.debug(f"Copying {attr_name}")
            if attr_name in COPY_REF_NOT_VALUE:
                # If the attribute is in the COPY_REF_NOT_VALUE list, assign the reference
                setattr(copied_net, attr_name, attr_value)
            else:
                # WranglerLogger.debug(f"making deep copy: {attr_name}")
                # For other attributes, perform a deep copy
                setattr(copied_net, attr_name, copy.deepcopy(attr_value, memo))

        return copied_net

    def deepcopy(self):
        """Returns copied TransitNetwork instance with deep copy of Feed but not roadway net."""
        return copy.deepcopy(self)

    @property
    def stops_gdf(self) -> gpd.GeoDataFrame:
        """Return stops as a GeoDataFrame using set roadway geometry."""
        ref_nodes = self.road_net.nodes_df if self.road_net is not None else None
        return to_points_gdf(self.feed.stops, ref_nodes_df=ref_nodes)

    @property
    def shapes_gdf(self) -> gpd.GeoDataFrame:
        """Return aggregated shapes as a GeoDataFrame using set roadway geometry."""
        ref_nodes = self.road_net.nodes_df if self.road_net is not None else None
        return shapes_to_trip_shapes_gdf(self.feed.shapes, ref_nodes_df=ref_nodes)

    @property
    def shape_links_gdf(self) -> gpd.GeoDataFrame:
        """Return shape-links as a GeoDataFrame using set roadway geometry."""
        ref_nodes = self.road_net.nodes_df if self.road_net is not None else None
        return shapes_to_shape_links_gdf(self.feed.shapes, ref_nodes_df=ref_nodes)

    @property
    def stop_time_links_gdf(self) -> gpd.GeoDataFrame:
        """Return stop-time-links as a GeoDataFrame using set roadway geometry."""
        ref_nodes = self.road_net.nodes_df if self.road_net is not None else None
        return stop_times_to_stop_time_links_gdf(
            self.feed.stop_times, self.feed.stops, ref_nodes_df=ref_nodes
        )

    @property
    def stop_times_points_gdf(self) -> gpd.GeoDataFrame:
        """Return stop-time-points as a GeoDataFrame using set roadway geometry."""
        ref_nodes = self.road_net.nodes_df if self.road_net is not None else None

        return stop_times_to_stop_time_points_gdf(
            self.feed.stop_times, self.feed.stops, ref_nodes_df=ref_nodes
        )

    def get_selection(
        self,
        selection_dict: dict,
        overwrite: bool = False,
    ) -> TransitSelection:
        """Return selection if it already exists, otherwise performs selection.

        Will raise an error if no trips found.

        Args:
            selection_dict (dict): _description_
            overwrite: if True, will overwrite any previously cached searches. Defaults to False.

        Returns:
            Selection: Selection object
        """
        key = dict_to_hexkey(selection_dict)

        if (key not in self._selections) or overwrite:
            WranglerLogger.debug(f"Performing selection from key: {key}")
            self._selections[key] = TransitSelection(self, selection_dict)
        else:
            WranglerLogger.debug(f"Using cached selection from key: {key}")

        if not self._selections[key]:
            msg = f"No links or nodes found for selection dict: \n {selection_dict}"
            WranglerLogger.error(msg)
            raise TransitSelectionEmptyError(msg)
        return self._selections[key]

    def apply(self, project_card: Union[ProjectCard, dict], **kwargs) -> TransitNetwork:
        """Wrapper method to apply a roadway project, returning a new TransitNetwork instance.

        Args:
            project_card: either a dictionary of the project card object or ProjectCard instance
            **kwargs: keyword arguments to pass to project application
        """
        if not (isinstance(project_card, (ProjectCard, SubProject))):
            project_card = ProjectCard(project_card)

        if not project_card.valid:
            msg = f"Project card {project_card.project} not valid."
            WranglerLogger.error(msg)
            raise ProjectCardError(msg)

        if project_card._sub_projects:
            for sp in project_card._sub_projects:
                WranglerLogger.debug(f"- applying subproject: {sp.change_type}")
                self._apply_change(sp, **kwargs)
            return self
        return self._apply_change(project_card, **kwargs)

    def _apply_change(
        self,
        change: Union[ProjectCard, SubProject],
        reference_road_net: Optional[RoadwayNetwork] = None,
    ) -> TransitNetwork:
        """Apply a single change: a single-project project or a sub-project."""
        if not isinstance(change, SubProject):
            WranglerLogger.info(f"Applying Project to Transit Network: {change.project}")

        if change.change_type == "transit_property_change":
            return apply_transit_property_change(
                self,
                self.get_selection(change.transit_property_change["service"]),
                change.transit_property_change["property_changes"],
                project_name=change.project,
            )

        if change.change_type == "transit_routing_change":
            return apply_transit_routing_change(
                self,
                self.get_selection(change.transit_routing_change["service"]),
                change.transit_routing_change["routing"],
                reference_road_net=reference_road_net,
                project_name=change.project,
            )

        if change.change_type == "pycode":
            return apply_calculated_transit(self, change.pycode)

        if change.change_type == "transit_route_addition":
            return apply_transit_route_addition(
                self,
                change.transit_route_addition,
                reference_road_net=reference_road_net,
            )
        if change.change_type == "transit_service_deletion":
            return apply_transit_service_deletion(
                self,
                self.get_selection(change.transit_service_deletion["service"]),
                clean_shapes=change.transit_service_deletion.get("clean_shapes"),
                clean_routes=change.transit_service_deletion.get("clean_routes"),
            )
        msg = f"Not a currently valid transit project: {change}."
        WranglerLogger.error(msg)
        raise NotImplementedError(msg)

network_wrangler.transit.network.TransitNetwork.applied_projects property

applied_projects

List of projects applied to the network.

Note: This may or may not return a full accurate account of all the applied projects. For better project accounting, please leverage the scenario object.

network_wrangler.transit.network.TransitNetwork.consistent_with_road_net property

consistent_with_road_net

Indicate if road_net is consistent with transit network.

Will return True if road_net is None, but provide a warning.

Checks the network hash of when consistency was last evaluated. If transit network or roadway network has changed, will re-evaluate consistency and return the updated value and update self._stored_road_net_hash.

Returns:

• bool –

  Boolean indicating if road_net is consistent with transit network.

network_wrangler.transit.network.TransitNetwork.feed property writable

feed

Feed associated with the transit network.

network_wrangler.transit.network.TransitNetwork.feed_hash property

feed_hash

Return the hash of the feed.

network_wrangler.transit.network.TransitNetwork.feed_path property

feed_path

Pass through property from Feed.

network_wrangler.transit.network.TransitNetwork.road_net property writable

road_net

Roadway network associated with the transit network.

network_wrangler.transit.network.TransitNetwork.shape_links_gdf property

shape_links_gdf

Return shape-links as a GeoDataFrame using set roadway geometry.

network_wrangler.transit.network.TransitNetwork.shapes_gdf property

shapes_gdf

Return aggregated shapes as a GeoDataFrame using set roadway geometry.

network_wrangler.transit.network.TransitNetwork.stop_time_links_gdf property

stop_time_links_gdf

Return stop-time-links as a GeoDataFrame using set roadway geometry.

network_wrangler.transit.network.TransitNetwork.stop_times_points_gdf property

stop_times_points_gdf

Return stop-time-points as a GeoDataFrame using set roadway geometry.

network_wrangler.transit.network.TransitNetwork.stops_gdf property

stops_gdf

Return stops as a GeoDataFrame using set roadway geometry.

network_wrangler.transit.network.TransitNetwork.__deepcopy__

__deepcopy__(memo)

Returns copied TransitNetwork instance with deep copy of Feed but not roadway net.

Source code in network_wrangler/transit/network.py

def __deepcopy__(self, memo):
    """Returns copied TransitNetwork instance with deep copy of Feed but not roadway net."""
    COPY_REF_NOT_VALUE = ["_road_net"]
    # Create a new, empty instance
    copied_net = self.__class__.__new__(self.__class__)
    # Return the new TransitNetwork instance
    attribute_dict = vars(self)

    # Copy the attributes to the new instance
    for attr_name, attr_value in attribute_dict.items():
        # WranglerLogger.debug(f"Copying {attr_name}")
        if attr_name in COPY_REF_NOT_VALUE:
            # If the attribute is in the COPY_REF_NOT_VALUE list, assign the reference
            setattr(copied_net, attr_name, attr_value)
        else:
            # WranglerLogger.debug(f"making deep copy: {attr_name}")
            # For other attributes, perform a deep copy
            setattr(copied_net, attr_name, copy.deepcopy(attr_value, memo))

    return copied_net

network_wrangler.transit.network.TransitNetwork.__init__

__init__(feed, config=DefaultConfig)

Constructor for TransitNetwork.

Parameters:

• feed (Feed) –

  Feed object representing the transit network gtfs tables

• config (WranglerConfig, default: DefaultConfig ) –

  WranglerConfig object. Defaults to DefaultConfig.

Source code in network_wrangler/transit/network.py

def __init__(self, feed: Feed, config: WranglerConfig = DefaultConfig) -> None:
    """Constructor for TransitNetwork.

    Args:
        feed: Feed object representing the transit network gtfs tables
        config: WranglerConfig object. Defaults to DefaultConfig.
    """
    WranglerLogger.debug("Creating new TransitNetwork.")

    self._road_net: Optional[RoadwayNetwork] = None
    self.feed: Feed = feed
    self.graph: nx.MultiDiGraph = None
    self.config: WranglerConfig = config
    # initialize
    self._consistent_with_road_net = False

    # cached selections
    self._selections: dict[str, TransitSelection] = {}

network_wrangler.transit.network.TransitNetwork.apply

apply(project_card, **kwargs)

Wrapper method to apply a roadway project, returning a new TransitNetwork instance.

Parameters:

• project_card (Union[ProjectCard, dict]) –

  either a dictionary of the project card object or ProjectCard instance

• **kwargs –

  keyword arguments to pass to project application

Source code in network_wrangler/transit/network.py

def apply(self, project_card: Union[ProjectCard, dict], **kwargs) -> TransitNetwork:
    """Wrapper method to apply a roadway project, returning a new TransitNetwork instance.

    Args:
        project_card: either a dictionary of the project card object or ProjectCard instance
        **kwargs: keyword arguments to pass to project application
    """
    if not (isinstance(project_card, (ProjectCard, SubProject))):
        project_card = ProjectCard(project_card)

    if not project_card.valid:
        msg = f"Project card {project_card.project} not valid."
        WranglerLogger.error(msg)
        raise ProjectCardError(msg)

    if project_card._sub_projects:
        for sp in project_card._sub_projects:
            WranglerLogger.debug(f"- applying subproject: {sp.change_type}")
            self._apply_change(sp, **kwargs)
        return self
    return self._apply_change(project_card, **kwargs)

network_wrangler.transit.network.TransitNetwork.deepcopy

deepcopy()

Returns copied TransitNetwork instance with deep copy of Feed but not roadway net.

Source code in network_wrangler/transit/network.py

def deepcopy(self):
    """Returns copied TransitNetwork instance with deep copy of Feed but not roadway net."""
    return copy.deepcopy(self)

network_wrangler.transit.network.TransitNetwork.get_selection

get_selection(selection_dict, overwrite=False)

Return selection if it already exists, otherwise performs selection.

Will raise an error if no trips found.

Parameters:

• selection_dict (dict) –

  description

• overwrite (bool, default: False ) –

  if True, will overwrite any previously cached searches. Defaults to False.

Returns:

• Selection ( TransitSelection ) –

  Selection object

Source code in network_wrangler/transit/network.py

def get_selection(
    self,
    selection_dict: dict,
    overwrite: bool = False,
) -> TransitSelection:
    """Return selection if it already exists, otherwise performs selection.

    Will raise an error if no trips found.

    Args:
        selection_dict (dict): _description_
        overwrite: if True, will overwrite any previously cached searches. Defaults to False.

    Returns:
        Selection: Selection object
    """
    key = dict_to_hexkey(selection_dict)

    if (key not in self._selections) or overwrite:
        WranglerLogger.debug(f"Performing selection from key: {key}")
        self._selections[key] = TransitSelection(self, selection_dict)
    else:
        WranglerLogger.debug(f"Using cached selection from key: {key}")

    if not self._selections[key]:
        msg = f"No links or nodes found for selection dict: \n {selection_dict}"
        WranglerLogger.error(msg)
        raise TransitSelectionEmptyError(msg)
    return self._selections[key]

network_wrangler.models._base.db.DbForeignKeys module-attribute

DbForeignKeys = dict[str, TableForeignKeys]

Mapping of tables that have fields that other tables use as fks.

{ <table>:{<field>:[(<table using FK>,<field using fk>)]} }

Example

{“stops”: {“stop_id”: [ (“stops”, “parent_station”), (“stop_times”, “stop_id”) ]} }

network_wrangler.models._base.db.TableForeignKeys module-attribute

TableForeignKeys = dict[str, tuple[str, str]]

Dict of each table’s foreign keys.

{ <table>:{<field>:[<fk_table>,<fk_field>]} }

Example

{“stops”: {“parent_station”: (“stops”, “stop_id”)} “stop_times”: {“stop_id”: (“stops”, “stop_id”)} {“trip_id”: (“trips”, “trip_id”)} }

network_wrangler.models._base.db.TablePrimaryKeys module-attribute

TablePrimaryKeys = list[str]

TableForeignKeys is a dictionary of foreign keys for a single table.

Uses the form

{:[,]}

Example

{“parent_station”: (“stops”, “stop_id”)}

network_wrangler.models._base.db.DBModelMixin

An mixin class for interrelated pandera DataFrameModel tables.

Contains a bunch of convenience methods and overrides the dunder methods deepcopy and eq.

Methods:

• hash –

  hash of tables

• deepcopy –

  deepcopy of tables which references a custom deepcopy

• get_table –

  retrieve table by name

• table_names_with_field –

  returns tables in table_names with field name

Attr

Where metadata variable _fk = {:[,]}

e.g.: _fk = {"parent_station": ["stops", "stop_id"]}

Source code in network_wrangler/models/_base/db.py

class DBModelMixin:
    """An mixin class for interrelated pandera DataFrameModel tables.

    Contains a bunch of convenience methods and overrides the dunder methods
        __deepcopy__ and __eq__.

    Methods:
        hash: hash of tables
        deepcopy: deepcopy of tables which references a custom __deepcopy__
        get_table: retrieve table by name
        table_names_with_field: returns tables in `table_names` with field name

    Attr:
        table_names: list of dataframe table names that are required as part of this "db"
            schema.
        optional_table_names: list of optional table names that will be added to `table_names` iff
            they are found.
        hash: creates a hash of tables found in `table_names` to track if they change.
        tables: dataframes corresponding to each table_name in `table_names`
        tables_dict: mapping of `<table_name>:<table>` dataframe
        _table_models: mapping of `<table_name>:<DataFrameModel>` to use for validation when
            `__setattr__` is called.
        _converters: mapping of `<table_name>:<converter_method>` where converter method should
            have a function signature of `(<table>, self.**kwargs)` .  Called on `__setattr__` if
            initial validation fails.

    Where metadata variable _fk = {<table_field>:[<fk table>,<fk field>]}

    e.g.: `_fk = {"parent_station": ["stops", "stop_id"]}`

    """

    # list of optional tables which are added to table_names if they are found.
    optional_table_names: ClassVar[list[str]] = []

    # list of interrelated tables.
    table_names: ClassVar[list[str]] = []

    # mapping of which Pandera DataFrameModel to validate the table to.
    _table_models: ClassVar[dict[str, DataFrameModel]] = {}

    # mapping of <table_name>:<conversion method> to use iff df validation fails.
    _converters: ClassVar[dict[str, Callable]] = {}

    def __setattr__(self, key, value):
        """Override the default setattr behavior to handle DataFrame validation.

        Note: this is NOT called when a dataframe is mutated in place!

        Args:
            key (str): The attribute name.
            value: The value to be assigned to the attribute.

        Raises:
            SchemaErrors: If the DataFrame does not conform to the schema.
            ForeignKeyError: If doesn't validate to foreign key.
        """
        if isinstance(value, pd.DataFrame):
            WranglerLogger.debug(f"Validating + coercing value to {key}")
            df = self.validate_coerce_table(key, value)
            super().__setattr__(key, df)
        else:
            super().__setattr__(key, value)

    def validate_coerce_table(self, table_name: str, table: pd.DataFrame) -> pd.DataFrame:
        if table_name not in self._table_models:
            return table
        table_model = self._table_models[table_name]
        converter = self._converters.get(table_name)
        try:
            validated_df = validate_df_to_model(table, table_model)
        except SchemaErrors as e:
            if not converter:
                raise e
            WranglerLogger.debug(
                f"Initial validation failed as {table_name}. \
                                Attempting to convert using: {converter}"
            )
            # Note that some converters may have dependency on other attributes being set first
            converted_df = converter(table, **self.__dict__)
            validated_df = validate_df_to_model(converted_df, table_model)

        # Do this in both directions so that ordering of tables being added doesn't matter.
        self.check_table_fks(table_name, table=validated_df)
        self.check_referenced_fks(table_name, table=validated_df)
        return validated_df

    def initialize_tables(self, **kwargs):
        """Initializes the tables for the database.

        Args:
            **kwargs: Keyword arguments representing the tables to be initialized.

        Raises:
            RequiredTableError: If any required tables are missing in the initialization.
        """
        # Flag missing required tables
        _missing_tables = [t for t in self.table_names if t not in kwargs]
        if _missing_tables:
            msg = f"Missing required tables: {_missing_tables}"
            raise RequiredTableError(msg)

        # Add provided optional tables
        _opt_tables = [k for k in kwargs if k in self.optional_table_names]
        self.table_names += _opt_tables

        # Set tables in order
        for table in self.table_names:
            WranglerLogger.info(f"Initializing {table}")
            self.__setattr__(table, kwargs[table])

    @classmethod
    def fks(cls) -> DbForeignKeys:
        """Return the fk field constraints as `{ <table>:{<field>:[<fk_table>,<fk_field>]} }`."""
        fk_fields = {}
        for table_name, table_model in cls._table_models.items():
            config = table_model.Config
            if not hasattr(config, "_fk"):
                continue
            fk_fields[table_name] = config._fk
        return fk_fields

    @classmethod
    def fields_as_fks(cls) -> DbForeignKeyUsage:
        """Returns mapping of tables that have fields that other tables use as fks.

        `{ <table>:{<field>:[(<table using FK>,<field using fk>)]} }`

        Useful for knowing if you should check FK validation when changing a field value.
        """
        pks_as_fks: defaultdict = defaultdict(lambda: defaultdict(list))
        for t, field_fk in cls.fks().items():
            for f, fk in field_fk.items():
                fk_table, fk_field = fk
                pks_as_fks[fk_table][fk_field].append((t, f))
        return {k: dict(v) for k, v in pks_as_fks.items()}

    def check_referenced_fk(
        self, pk_table_name: str, pk_field: str, pk_table: Optional[pd.DataFrame] = None
    ) -> bool:
        """True if table.field has the values referenced in any table referencing fields as fk.

        For example. If routes.route_id is referenced in trips table, we need to check that
        if a route_id is deleted, it isn't referenced in trips.route_id.
        """
        msg = f"Checking tables which referenced {pk_table_name}.{pk_field} as an FK"
        # WranglerLogger.debug(msg)
        if pk_table is None:
            pk_table = self.get_table(pk_table_name)

        if pk_field not in pk_table:
            WranglerLogger.warning(
                f"Foreign key value {pk_field} not in {pk_table_name} - \
                 skipping fk validation"
            )
            return True

        fields_as_fks: DbForeignKeyUsage = self.fields_as_fks()

        if pk_table_name not in fields_as_fks:
            return True
        if pk_field not in fields_as_fks[pk_table_name]:
            return True

        all_valid = True

        for ref_table_name, ref_field in fields_as_fks[pk_table_name][pk_field]:
            if ref_table_name not in self.table_names:
                WranglerLogger.debug(
                    f"Referencing table {ref_table_name} not in self.table_names - \
                    skipping fk validation."
                )
                continue

            try:
                ref_table = self.get_table(ref_table_name)
            except RequiredTableError:
                WranglerLogger.debug(
                    f"Referencing table {ref_table_name} not yet set in \
                     {type(self)} - skipping fk validation."
                )
                continue

            if ref_field not in ref_table:
                WranglerLogger.debug(
                    f"Referencing field {ref_field} not in {ref_table_name} - \
                    skipping fk validation."
                )
                continue

            valid, _missing = fk_in_pk(pk_table[pk_field], ref_table[ref_field])
            all_valid = all_valid and valid
            if _missing:
                WranglerLogger.error(
                    f"Following values missing from {pk_table_name}.{pk_field} that \
                      are referenced by {ref_table}: \n{_missing}"
                )
        return all_valid

    def check_referenced_fks(self, table_name: str, table: Optional[pd.DataFrame] = None) -> bool:
        """True if this table has the values referenced in any table referencing fields as fk.

        For example. If routes.route_id is referenced in trips table, we need to check that
        if a route_id is deleted, it isn't referenced in trips.route_id.
        """
        # WranglerLogger.debug(f"Checking referenced foreign keys for {table_name}")
        all_valid = True
        if table is None:
            table = self.get_table(table_name)
        all_valid = True
        for field in self.fields_as_fks().get(table_name, {}):
            valid = self.check_referenced_fk(table_name, field, pk_table=table)
            all_valid = valid and all_valid
        return all_valid

    def check_table_fks(
        self, table_name: str, table: Optional[pd.DataFrame] = None, raise_error: bool = True
    ) -> bool:
        """Return True if the foreign key fields in table have valid references.

        Note: will return true and give a warning if the specified foreign key table doesn't exist.
        """
        # WranglerLogger.debug(f"Checking foreign keys for {table_name}")
        fks = self.fks()
        if table_name not in fks:
            return True
        if table is None:
            table = self.get_table(table_name)
        all_valid = True
        for field, fk in fks[table_name].items():
            # WranglerLogger.debug(f"Checking {table_name}.{field} foreign key")
            pkref_table_name, pkref_field = fk
            # WranglerLogger.debug(f"Looking for PK in {pkref_table_name}.{pkref_field}.")
            if field not in table:
                WranglerLogger.warning(
                    f"Foreign key value {field} not in {table_name} -\
                    skipping validation"
                )
                continue

            if pkref_table_name not in self.table_names:
                WranglerLogger.debug(
                    f"PK table {pkref_table_name} for specified FK \
                    {table_name}.{field} not in table list - skipping validation."
                )
                continue
            try:
                pkref_table = self.get_table(pkref_table_name)
            except RequiredTableError:
                WranglerLogger.debug(
                    f"PK table {pkref_table_name} for specified FK \
                    {table_name}.{field} not in {type(self)}-  \
                    skipping validation."
                )
                continue
            if pkref_field not in pkref_table:
                WranglerLogger.error(
                    f"!!! {pkref_table_name} missing {pkref_field} field used as FK\
                                    ref in {table_name}.{field}."
                )
                all_valid = False
                continue
            if len(pkref_table) < SMALL_RECS:
                pass
                # WranglerLogger.debug(f"PK values:\n{pkref_table[pkref_field]}.")
            # WranglerLogger.debug(f"Checking {table_name}.{field} foreign key")
            valid, missing = fk_in_pk(pkref_table[pkref_field], table[field])
            if missing:
                WranglerLogger.error(
                    f"!!! {pkref_table_name}.{pkref_field} missing values used as FK\
                      in {table_name}.{field}: \n_missing"
                )
            all_valid = valid and all_valid

        if not all_valid:
            if raise_error:
                msg = f"FK fields/ values referenced in {table_name} missing."
                raise ForeignKeyValueError(msg)
            return False
        return True

    def check_fks(self) -> bool:
        """Check all FKs in set of tables."""
        all_valid = True
        for table_name in self.table_names:
            valid = self.check_table_fks(
                table_name, self.tables_dict[table_name], raise_error=False
            )
            all_valid = valid and all_valid
        return all_valid

    @property
    def tables(self) -> list[DataFrameModel]:
        return [self.__dict__[t] for t in self.table_names]

    @property
    def tables_dict(self) -> dict[str, DataFrameModel]:
        num_records = [len(self.get_table(t)) for t in self.table_names]
        return pd.DataFrame({"Table": self.table_names, "Records": num_records})

    @property
    def describe_df(self) -> pd.DataFrame:
        return pd.DataFrame({t: len(self.get_table(t)) for t in self.table_names})

    def get_table(self, table_name: str) -> pd.DataFrame:
        """Get table by name."""
        if table_name not in self.table_names:
            msg = f"{table_name} table not in db."
            raise ValueError(msg)
        if table_name not in self.__dict__:
            msg = f"Required table not set yet: {table_name}"
            raise RequiredTableError(msg)
        return self.__dict__[table_name]

    def table_names_with_field(self, field: str) -> list[str]:
        """Returns tables in the class instance which contain the field."""
        return [t for t in self.table_names if field in self.get_table(t).columns]

    @property
    def hash(self) -> str:
        """A hash representing the contents of the tables in self.table_names."""
        _table_hashes = [self.get_table(t).df_hash() for t in self.table_names]
        _value = str.encode("-".join(_table_hashes))

        _hash = hashlib.sha256(_value).hexdigest()
        return _hash

    def __eq__(self, other):
        """Override the default Equals behavior."""
        if isinstance(other, self.__class__):
            return self.hash == other.hash
        return False

    def __deepcopy__(self, memo):
        """Custom implementation of __deepcopy__ method.

        This method is called by copy.deepcopy() to create a deep copy of the object.

        Args:
            memo (dict): Dictionary to track objects already copied during deepcopy.

        Returns:
            Feed: A deep copy of the db object.
        """
        # Create a new, empty instance of the Feed class
        new_instance = self.__class__.__new__(self.__class__)

        # Copy all attributes to the new instance
        for attr_name, attr_value in self.__dict__.items():
            # Handle pandera DataFrameModel objects specially
            if (
                hasattr(attr_value, "__class__")
                and hasattr(attr_value.__class__, "__name__")
                and "DataFrameModel" in attr_value.__class__.__name__
            ):
                # For pandera DataFrameModel objects, copy the underlying DataFrame and recreate the model
                # This avoids the timestamp corruption issue with copy.deepcopy()
                try:
                    # Get the underlying DataFrame
                    if hasattr(attr_value, "_obj"):
                        df_copy = attr_value._obj.copy(deep=True)
                    elif hasattr(attr_value, "data"):
                        df_copy = attr_value.data.copy(deep=True)
                    else:
                        # For newer pandera versions, try direct access
                        df_copy = attr_value.copy(deep=True)

                    # Recreate the DataFrameModel object with the copied DataFrame
                    new_table = attr_value.__class__(df_copy)

                    setattr(new_instance, attr_name, new_table)
                except Exception as e:
                    # Fallback to regular deep copy if the above fails
                    setattr(new_instance, attr_name, copy.deepcopy(attr_value, memo))
            elif isinstance(attr_value, pd.DataFrame):
                # For plain pandas DataFrames, use deep copy
                setattr(new_instance, attr_name, attr_value.copy(deep=True))
            else:
                # For all other objects, use regular deep copy
                setattr(new_instance, attr_name, copy.deepcopy(attr_value, memo))

        return new_instance

    def deepcopy(self):
        """Convenience method to exceute deep copy of instance."""
        return copy.deepcopy(self)

    def __hash__(self):
        """Hash based on the hashes of the tables in table_names."""
        return hash(tuple((name, self.get_table(name).to_csv()) for name in self.table_names))

network_wrangler.models._base.db.DBModelMixin.hash property

hash

A hash representing the contents of the tables in self.table_names.

network_wrangler.models._base.db.DBModelMixin.__deepcopy__

__deepcopy__(memo)

Custom implementation of deepcopy method.

This method is called by copy.deepcopy() to create a deep copy of the object.

Parameters:

• memo (dict) –

  Dictionary to track objects already copied during deepcopy.

Returns:

• Feed –

  A deep copy of the db object.

Source code in network_wrangler/models/_base/db.py

def __deepcopy__(self, memo):
    """Custom implementation of __deepcopy__ method.

    This method is called by copy.deepcopy() to create a deep copy of the object.

    Args:
        memo (dict): Dictionary to track objects already copied during deepcopy.

    Returns:
        Feed: A deep copy of the db object.
    """
    # Create a new, empty instance of the Feed class
    new_instance = self.__class__.__new__(self.__class__)

    # Copy all attributes to the new instance
    for attr_name, attr_value in self.__dict__.items():
        # Handle pandera DataFrameModel objects specially
        if (
            hasattr(attr_value, "__class__")
            and hasattr(attr_value.__class__, "__name__")
            and "DataFrameModel" in attr_value.__class__.__name__
        ):
            # For pandera DataFrameModel objects, copy the underlying DataFrame and recreate the model
            # This avoids the timestamp corruption issue with copy.deepcopy()
            try:
                # Get the underlying DataFrame
                if hasattr(attr_value, "_obj"):
                    df_copy = attr_value._obj.copy(deep=True)
                elif hasattr(attr_value, "data"):
                    df_copy = attr_value.data.copy(deep=True)
                else:
                    # For newer pandera versions, try direct access
                    df_copy = attr_value.copy(deep=True)

                # Recreate the DataFrameModel object with the copied DataFrame
                new_table = attr_value.__class__(df_copy)

                setattr(new_instance, attr_name, new_table)
            except Exception as e:
                # Fallback to regular deep copy if the above fails
                setattr(new_instance, attr_name, copy.deepcopy(attr_value, memo))
        elif isinstance(attr_value, pd.DataFrame):
            # For plain pandas DataFrames, use deep copy
            setattr(new_instance, attr_name, attr_value.copy(deep=True))
        else:
            # For all other objects, use regular deep copy
            setattr(new_instance, attr_name, copy.deepcopy(attr_value, memo))

    return new_instance

network_wrangler.models._base.db.DBModelMixin.__eq__

__eq__(other)

Override the default Equals behavior.

Source code in network_wrangler/models/_base/db.py

def __eq__(self, other):
    """Override the default Equals behavior."""
    if isinstance(other, self.__class__):
        return self.hash == other.hash
    return False

network_wrangler.models._base.db.DBModelMixin.__hash__

__hash__()

Hash based on the hashes of the tables in table_names.

Source code in network_wrangler/models/_base/db.py

def __hash__(self):
    """Hash based on the hashes of the tables in table_names."""
    return hash(tuple((name, self.get_table(name).to_csv()) for name in self.table_names))

network_wrangler.models._base.db.DBModelMixin.__setattr__

__setattr__(key, value)

Override the default setattr behavior to handle DataFrame validation.

Note: this is NOT called when a dataframe is mutated in place!

Parameters:

• key (str) –

  The attribute name.

• value –

  The value to be assigned to the attribute.

Raises:

• SchemaErrors –

  If the DataFrame does not conform to the schema.

• ForeignKeyError –

  If doesn’t validate to foreign key.

Source code in network_wrangler/models/_base/db.py

def __setattr__(self, key, value):
    """Override the default setattr behavior to handle DataFrame validation.

    Note: this is NOT called when a dataframe is mutated in place!

    Args:
        key (str): The attribute name.
        value: The value to be assigned to the attribute.

    Raises:
        SchemaErrors: If the DataFrame does not conform to the schema.
        ForeignKeyError: If doesn't validate to foreign key.
    """
    if isinstance(value, pd.DataFrame):
        WranglerLogger.debug(f"Validating + coercing value to {key}")
        df = self.validate_coerce_table(key, value)
        super().__setattr__(key, df)
    else:
        super().__setattr__(key, value)

network_wrangler.models._base.db.DBModelMixin.check_fks

check_fks()

Check all FKs in set of tables.

Source code in network_wrangler/models/_base/db.py

def check_fks(self) -> bool:
    """Check all FKs in set of tables."""
    all_valid = True
    for table_name in self.table_names:
        valid = self.check_table_fks(
            table_name, self.tables_dict[table_name], raise_error=False
        )
        all_valid = valid and all_valid
    return all_valid

network_wrangler.models._base.db.DBModelMixin.check_referenced_fk

check_referenced_fk(pk_table_name, pk_field, pk_table=None)

True if table.field has the values referenced in any table referencing fields as fk.

For example. If routes.route_id is referenced in trips table, we need to check that if a route_id is deleted, it isn’t referenced in trips.route_id.

Source code in network_wrangler/models/_base/db.py

def check_referenced_fk(
    self, pk_table_name: str, pk_field: str, pk_table: Optional[pd.DataFrame] = None
) -> bool:
    """True if table.field has the values referenced in any table referencing fields as fk.

    For example. If routes.route_id is referenced in trips table, we need to check that
    if a route_id is deleted, it isn't referenced in trips.route_id.
    """
    msg = f"Checking tables which referenced {pk_table_name}.{pk_field} as an FK"
    # WranglerLogger.debug(msg)
    if pk_table is None:
        pk_table = self.get_table(pk_table_name)

    if pk_field not in pk_table:
        WranglerLogger.warning(
            f"Foreign key value {pk_field} not in {pk_table_name} - \
             skipping fk validation"
        )
        return True

    fields_as_fks: DbForeignKeyUsage = self.fields_as_fks()

    if pk_table_name not in fields_as_fks:
        return True
    if pk_field not in fields_as_fks[pk_table_name]:
        return True

    all_valid = True

    for ref_table_name, ref_field in fields_as_fks[pk_table_name][pk_field]:
        if ref_table_name not in self.table_names:
            WranglerLogger.debug(
                f"Referencing table {ref_table_name} not in self.table_names - \
                skipping fk validation."
            )
            continue

        try:
            ref_table = self.get_table(ref_table_name)
        except RequiredTableError:
            WranglerLogger.debug(
                f"Referencing table {ref_table_name} not yet set in \
                 {type(self)} - skipping fk validation."
            )
            continue

        if ref_field not in ref_table:
            WranglerLogger.debug(
                f"Referencing field {ref_field} not in {ref_table_name} - \
                skipping fk validation."
            )
            continue

        valid, _missing = fk_in_pk(pk_table[pk_field], ref_table[ref_field])
        all_valid = all_valid and valid
        if _missing:
            WranglerLogger.error(
                f"Following values missing from {pk_table_name}.{pk_field} that \
                  are referenced by {ref_table}: \n{_missing}"
            )
    return all_valid

network_wrangler.models._base.db.DBModelMixin.check_referenced_fks

check_referenced_fks(table_name, table=None)

True if this table has the values referenced in any table referencing fields as fk.

For example. If routes.route_id is referenced in trips table, we need to check that if a route_id is deleted, it isn’t referenced in trips.route_id.

Source code in network_wrangler/models/_base/db.py

def check_referenced_fks(self, table_name: str, table: Optional[pd.DataFrame] = None) -> bool:
    """True if this table has the values referenced in any table referencing fields as fk.

    For example. If routes.route_id is referenced in trips table, we need to check that
    if a route_id is deleted, it isn't referenced in trips.route_id.
    """
    # WranglerLogger.debug(f"Checking referenced foreign keys for {table_name}")
    all_valid = True
    if table is None:
        table = self.get_table(table_name)
    all_valid = True
    for field in self.fields_as_fks().get(table_name, {}):
        valid = self.check_referenced_fk(table_name, field, pk_table=table)
        all_valid = valid and all_valid
    return all_valid

network_wrangler.models._base.db.DBModelMixin.check_table_fks

check_table_fks(table_name, table=None, raise_error=True)

Return True if the foreign key fields in table have valid references.

Note: will return true and give a warning if the specified foreign key table doesn’t exist.

Source code in network_wrangler/models/_base/db.py

def check_table_fks(
    self, table_name: str, table: Optional[pd.DataFrame] = None, raise_error: bool = True
) -> bool:
    """Return True if the foreign key fields in table have valid references.

    Note: will return true and give a warning if the specified foreign key table doesn't exist.
    """
    # WranglerLogger.debug(f"Checking foreign keys for {table_name}")
    fks = self.fks()
    if table_name not in fks:
        return True
    if table is None:
        table = self.get_table(table_name)
    all_valid = True
    for field, fk in fks[table_name].items():
        # WranglerLogger.debug(f"Checking {table_name}.{field} foreign key")
        pkref_table_name, pkref_field = fk
        # WranglerLogger.debug(f"Looking for PK in {pkref_table_name}.{pkref_field}.")
        if field not in table:
            WranglerLogger.warning(
                f"Foreign key value {field} not in {table_name} -\
                skipping validation"
            )
            continue

        if pkref_table_name not in self.table_names:
            WranglerLogger.debug(
                f"PK table {pkref_table_name} for specified FK \
                {table_name}.{field} not in table list - skipping validation."
            )
            continue
        try:
            pkref_table = self.get_table(pkref_table_name)
        except RequiredTableError:
            WranglerLogger.debug(
                f"PK table {pkref_table_name} for specified FK \
                {table_name}.{field} not in {type(self)}-  \
                skipping validation."
            )
            continue
        if pkref_field not in pkref_table:
            WranglerLogger.error(
                f"!!! {pkref_table_name} missing {pkref_field} field used as FK\
                                ref in {table_name}.{field}."
            )
            all_valid = False
            continue
        if len(pkref_table) < SMALL_RECS:
            pass
            # WranglerLogger.debug(f"PK values:\n{pkref_table[pkref_field]}.")
        # WranglerLogger.debug(f"Checking {table_name}.{field} foreign key")
        valid, missing = fk_in_pk(pkref_table[pkref_field], table[field])
        if missing:
            WranglerLogger.error(
                f"!!! {pkref_table_name}.{pkref_field} missing values used as FK\
                  in {table_name}.{field}: \n_missing"
            )
        all_valid = valid and all_valid

    if not all_valid:
        if raise_error:
            msg = f"FK fields/ values referenced in {table_name} missing."
            raise ForeignKeyValueError(msg)
        return False
    return True

network_wrangler.models._base.db.DBModelMixin.deepcopy

deepcopy()

Convenience method to exceute deep copy of instance.

Source code in network_wrangler/models/_base/db.py

def deepcopy(self):
    """Convenience method to exceute deep copy of instance."""
    return copy.deepcopy(self)

network_wrangler.models._base.db.DBModelMixin.fields_as_fks classmethod

fields_as_fks()

Returns mapping of tables that have fields that other tables use as fks.

{ <table>:{<field>:[(<table using FK>,<field using fk>)]} }

Useful for knowing if you should check FK validation when changing a field value.

Source code in network_wrangler/models/_base/db.py

@classmethod
def fields_as_fks(cls) -> DbForeignKeyUsage:
    """Returns mapping of tables that have fields that other tables use as fks.

    `{ <table>:{<field>:[(<table using FK>,<field using fk>)]} }`

    Useful for knowing if you should check FK validation when changing a field value.
    """
    pks_as_fks: defaultdict = defaultdict(lambda: defaultdict(list))
    for t, field_fk in cls.fks().items():
        for f, fk in field_fk.items():
            fk_table, fk_field = fk
            pks_as_fks[fk_table][fk_field].append((t, f))
    return {k: dict(v) for k, v in pks_as_fks.items()}

network_wrangler.models._base.db.DBModelMixin.fks classmethod

fks()

Return the fk field constraints as { <table>:{<field>:[<fk_table>,<fk_field>]} }.

Source code in network_wrangler/models/_base/db.py

@classmethod
def fks(cls) -> DbForeignKeys:
    """Return the fk field constraints as `{ <table>:{<field>:[<fk_table>,<fk_field>]} }`."""
    fk_fields = {}
    for table_name, table_model in cls._table_models.items():
        config = table_model.Config
        if not hasattr(config, "_fk"):
            continue
        fk_fields[table_name] = config._fk
    return fk_fields

network_wrangler.models._base.db.DBModelMixin.get_table

get_table(table_name)

Get table by name.

Source code in network_wrangler/models/_base/db.py

def get_table(self, table_name: str) -> pd.DataFrame:
    """Get table by name."""
    if table_name not in self.table_names:
        msg = f"{table_name} table not in db."
        raise ValueError(msg)
    if table_name not in self.__dict__:
        msg = f"Required table not set yet: {table_name}"
        raise RequiredTableError(msg)
    return self.__dict__[table_name]

network_wrangler.models._base.db.DBModelMixin.initialize_tables

initialize_tables(**kwargs)

Initializes the tables for the database.

Parameters:

• **kwargs –

  Keyword arguments representing the tables to be initialized.

Raises:

• RequiredTableError –

  If any required tables are missing in the initialization.

Source code in network_wrangler/models/_base/db.py

def initialize_tables(self, **kwargs):
    """Initializes the tables for the database.

    Args:
        **kwargs: Keyword arguments representing the tables to be initialized.

    Raises:
        RequiredTableError: If any required tables are missing in the initialization.
    """
    # Flag missing required tables
    _missing_tables = [t for t in self.table_names if t not in kwargs]
    if _missing_tables:
        msg = f"Missing required tables: {_missing_tables}"
        raise RequiredTableError(msg)

    # Add provided optional tables
    _opt_tables = [k for k in kwargs if k in self.optional_table_names]
    self.table_names += _opt_tables

    # Set tables in order
    for table in self.table_names:
        WranglerLogger.info(f"Initializing {table}")
        self.__setattr__(table, kwargs[table])

network_wrangler.models._base.db.DBModelMixin.table_names_with_field

table_names_with_field(field)

Returns tables in the class instance which contain the field.

Source code in network_wrangler/models/_base/db.py

def table_names_with_field(self, field: str) -> list[str]:
    """Returns tables in the class instance which contain the field."""
    return [t for t in self.table_names if field in self.get_table(t).columns]

Configuration

Classes and utilities for configuring Network Wrangler behavior:

Configuration for parameters for Network Wrangler.

Users can change a handful of parameters which control the way Wrangler runs. These parameters can be saved as a wrangler config file which can be read in repeatedly to make sure the same parameters are used each time.

Usage

At runtime, you can specify configurable parameters at the scenario level which will then also be assigned and accessible to the roadway and transit networks.

create_scenario(...config = myconfig)

Or if you are not using Scenario functionality, you can specify the config when you read in a RoadwayNetwork.

load_roadway_from_dir(**roadway, config=myconfig)
load_transit(**transit, config=myconfig)

my_config can be a:

• Path to a config file in yaml/toml/json (recommended),
• List of paths to config files (in case you want to split up various sub-configurations)
• Dictionary which is in the same structure of a config file, or
• A WranglerConfig() instance.

If not provided, Wrangler will use reasonable defaults.

Default Wrangler Configuration Values

If not explicitly provided, the following default values are used:

IDS:
    TRANSIT_SHAPE_ID_METHOD: scalar
    TRANSIT_SHAPE_ID_SCALAR: 1000000
    ROAD_SHAPE_ID_METHOD: scalar
    ROAD_SHAPE_ID_SCALAR: 1000
    ML_LINK_ID_METHOD: range
    ML_LINK_ID_RANGE: (950000, 999999)
    ML_LINK_ID_SCALAR: 15000
    ML_NODE_ID_METHOD: range
    ML_NODE_ID_RANGE: (950000, 999999)
    ML_NODE_ID_SCALAR: 15000
EDITS:
    EXISTING_VALUE_CONFLIC: warn
    OVERWRITE_SCOPED: conflicting
MODEL_ROADWAY:
    ML_OFFSET_METERS: int = -10
    ADDITIONAL_COPY_FROM_GP_TO_ML: []
    ADDITIONAL_COPY_TO_ACCESS_EGRESS: []
CPU:
    EST_PD_READ_SPEED:
        csv: 0.03
        parquet: 0.005
        geojson: 0.03
        json: 0.15
        txt: 0.04

Extended usage

Load the default configuration:

from network_wrangler.configs import DefaultConfig

Access the configuration:

from network_wrangler.configs import DefaultConfig
DefaultConfig.MODEL_ROADWAY.ML_OFFSET_METERS
>> -10

Modify the default configuration in-line:

from network_wrangler.configs import DefaultConfig

DefaultConfig.MODEL_ROADWAY.ML_OFFSET_METERS = 20

Load a configuration from a file:

from network_wrangler.configs import load_wrangler_config

config = load_wrangler_config("path/to/config.yaml")

Set a configuration value:

config.MODEL_ROADWAY.ML_OFFSET_METERS = 10

network_wrangler.configs.wrangler.CpuConfig

Bases: ConfigItem

CPU Configuration - Will not change any outcomes.

Attributes:

• EST_PD_READ_SPEED (dict[str, float]) –

  Read sec / MB - WILL DEPEND ON SPECIFIC COMPUTER

Source code in network_wrangler/configs/wrangler.py

@dataclass
class CpuConfig(ConfigItem):
    """CPU Configuration -  Will not change any outcomes.

    Attributes:
        EST_PD_READ_SPEED: Read sec / MB - WILL DEPEND ON SPECIFIC COMPUTER
    """

    EST_PD_READ_SPEED: dict[str, float] = Field(
        default_factory=lambda: {
            "csv": 0.03,
            "parquet": 0.005,
            "geojson": 0.03,
            "json": 0.15,
            "txt": 0.04,
        }
    )

network_wrangler.configs.wrangler.EditsConfig

Bases: ConfigItem

Configuration for Edits.

Attributes:

• EXISTING_VALUE_CONFLICT (Literal['warn', 'error', 'skip']) –

  Only used if ‘existing’ provided in project card and existing doesn’t match the existing network value. One of error, warn, or skip. error will raise an error, warn will warn the user, and skip will skip the change for that specific property (note it will still apply any remaining property changes). Defaults to warn. Can be overridden by setting existing_value_conflict in a roadway_property_change project card.

• OVERWRITE_SCOPED (Literal['conflicting', 'all', 'error']) –

  How to handle conflicts with existing values. Should be one of “conflicting”, “all”, or False. “conflicting” will only overwrite values where the scope only partially overlaps with the existing value. “all” will overwrite all the scoped values. “error” will error if there is any overlap. Default is “conflicting”. Can be changed at the project-level by setting overwrite_scoped in a roadway_property_change project card.

Source code in network_wrangler/configs/wrangler.py

@dataclass
class EditsConfig(ConfigItem):
    """Configuration for Edits.

    Attributes:
        EXISTING_VALUE_CONFLICT: Only used if 'existing' provided in project card and
            `existing` doesn't match the existing network value. One of `error`, `warn`, or `skip`.
            `error` will raise an error, `warn` will warn the user, and `skip` will skip the change
            for that specific property (note it will still apply any remaining property changes).
            Defaults to `warn`. Can be overridden by setting `existing_value_conflict` in
            a `roadway_property_change` project card.

        OVERWRITE_SCOPED: How to handle conflicts with existing values.
            Should be one of "conflicting", "all", or False.
            "conflicting" will only overwrite values where the scope only partially overlaps with
            the existing value. "all" will overwrite all the scoped values. "error" will error if
            there is any overlap. Default is "conflicting". Can be changed at the project-level
            by setting `overwrite_scoped` in a `roadway_property_change` project card.
    """

    EXISTING_VALUE_CONFLICT: Literal["warn", "error", "skip"] = "warn"
    OVERWRITE_SCOPED: Literal["conflicting", "all", "error"] = "conflicting"

network_wrangler.configs.wrangler.IdGenerationConfig

Bases: ConfigItem

Model Roadway Configuration.

Attributes:

• TRANSIT_SHAPE_ID_METHOD (Literal['scalar']) –

  method for creating a shape_id for a transit shape. Should be “scalar”.

• TRANSIT_SHAPE_ID_SCALAR (int) –

  scalar value to add to general purpose lane to create a shape_id for a transit shape.

• ROAD_SHAPE_ID_METHOD (Literal['scalar']) –

  method for creating a shape_id for a roadway shape. Should be “scalar”.

• ROAD_SHAPE_ID_SCALAR (int) –

  scalar value to add to general purpose lane to create a shape_id for a roadway shape.

• ML_LINK_ID_METHOD (Literal['range', 'scalar']) –

  method for creating a model_link_id for an associated link for a parallel managed lane.

• ML_LINK_ID_RANGE (tuple[int, int]) –

  range of model_link_ids to use when creating an associated link for a parallel managed lane.

• ML_LINK_ID_SCALAR (int) –

  scalar value to add to general purpose lane to create a model_link_id when creating an associated link for a parallel managed lane.

• ML_NODE_ID_METHOD (Literal['range', 'scalar']) –

  method for creating a model_node_id for an associated node for a parallel managed lane.

• ML_NODE_ID_RANGE (tuple[int, int]) –

  range of model_node_ids to use when creating an associated node for a parallel managed lane.

• ML_NODE_ID_SCALAR (int) –

  scalar value to add to general purpose lane node ides create a model_node_id when creating an associated nodes for parallel managed lane.

Source code in network_wrangler/configs/wrangler.py

@dataclass
class IdGenerationConfig(ConfigItem):
    """Model Roadway Configuration.

    Attributes:
        TRANSIT_SHAPE_ID_METHOD: method for creating a shape_id for a transit shape.
            Should be "scalar".
        TRANSIT_SHAPE_ID_SCALAR: scalar value to add to general purpose lane to create a
            shape_id for a transit shape.
        ROAD_SHAPE_ID_METHOD: method for creating a shape_id for a roadway shape.
            Should be "scalar".
        ROAD_SHAPE_ID_SCALAR: scalar value to add to general purpose lane to create a
            shape_id for a roadway shape.
        ML_LINK_ID_METHOD: method for creating a model_link_id for an associated
            link for a parallel managed lane.
        ML_LINK_ID_RANGE: range of model_link_ids to use when creating an associated
            link for a parallel managed lane.
        ML_LINK_ID_SCALAR: scalar value to add to general purpose lane to create a
            model_link_id when creating an associated link for a parallel managed lane.
        ML_NODE_ID_METHOD: method for creating a model_node_id for an associated node
            for a parallel managed lane.
        ML_NODE_ID_RANGE: range of model_node_ids to use when creating an associated
            node for a parallel managed lane.
        ML_NODE_ID_SCALAR: scalar value to add to general purpose lane node ides create
            a model_node_id when creating an associated nodes for parallel managed lane.
    """

    TRANSIT_SHAPE_ID_METHOD: Literal["scalar"] = "scalar"
    TRANSIT_SHAPE_ID_SCALAR: int = 1000000
    ROAD_SHAPE_ID_METHOD: Literal["scalar"] = "scalar"
    ROAD_SHAPE_ID_SCALAR: int = 1000
    ML_LINK_ID_METHOD: Literal["range", "scalar"] = "scalar"
    ML_LINK_ID_RANGE: tuple[int, int] = (950000, 999999)
    ML_LINK_ID_SCALAR: int = 3000000
    ML_NODE_ID_METHOD: Literal["range", "scalar"] = "range"
    ML_NODE_ID_RANGE: tuple[int, int] = (950000, 999999)
    ML_NODE_ID_SCALAR: int = 15000

network_wrangler.configs.wrangler.ModelRoadwayConfig

Bases: ConfigItem

Model Roadway Configuration.

Attributes:

• ML_OFFSET_METERS (int) –

  Offset in meters for managed lanes.

• ADDITIONAL_COPY_FROM_GP_TO_ML (list[str]) –

  Additional fields to copy from general purpose to managed lanes.

• ADDITIONAL_COPY_TO_ACCESS_EGRESS (list[str]) –

  Additional fields to copy to access and egress links.

Source code in network_wrangler/configs/wrangler.py

@dataclass
class ModelRoadwayConfig(ConfigItem):
    """Model Roadway Configuration.

    Attributes:
        ML_OFFSET_METERS: Offset in meters for managed lanes.
        ADDITIONAL_COPY_FROM_GP_TO_ML: Additional fields to copy from general purpose to managed
            lanes.
        ADDITIONAL_COPY_TO_ACCESS_EGRESS: Additional fields to copy to access and egress links.
    """

    ML_OFFSET_METERS: int = -10
    ADDITIONAL_COPY_FROM_GP_TO_ML: list[str] = Field(default_factory=list)
    ADDITIONAL_COPY_TO_ACCESS_EGRESS: list[str] = Field(default_factory=list)

network_wrangler.configs.wrangler.WranglerConfig

Bases: ConfigItem

Configuration for Network Wrangler.

Attributes:

• IDS (IdGenerationConfig) –

  Parameteters governing how new ids are generated.

• MODEL_ROADWAY (ModelRoadwayConfig) –

  Parameters governing how the model roadway is created.

• CPU (CpuConfig) –

  Parameters for accessing CPU information. Will not change any outcomes.

• EDITS (EditsConfig) –

  Parameters governing how edits are handled.

Source code in network_wrangler/configs/wrangler.py

@dataclass
class WranglerConfig(ConfigItem):
    """Configuration for Network Wrangler.

    Attributes:
        IDS: Parameteters governing how new ids are generated.
        MODEL_ROADWAY: Parameters governing how the model roadway is created.
        CPU: Parameters for accessing CPU information. Will not change any outcomes.
        EDITS: Parameters governing how edits are handled.
    """

    IDS: IdGenerationConfig = IdGenerationConfig()
    MODEL_ROADWAY: ModelRoadwayConfig = ModelRoadwayConfig()
    CPU: CpuConfig = CpuConfig()
    EDITS: EditsConfig = EditsConfig()

Scenario configuration for Network Wrangler.

You can build a scenario and write out the output from a scenario configuration file using the code below. This is very useful when you are running a specific scenario with minor variations over again because you can enter your config file into version control. In addition to the completed roadway and transit files, the output will provide a record of how the scenario was run.

Usage

    from scenario import build_scenario_from_config
    my_scenario = build_scenario_from_config(my_scenario_config)

Where my_scenario_config can be a:

• Path to a scenario config file in yaml/toml/json (recommended),
• Dictionary which is in the same structure of a scenario config file, or
• A ScenarioConfig() instance.

Notes on relative paths in scenario configs

• Relative paths are recognized by a preceeding “.”.
• Relative paths within output_scenario for roadway, transit, and project_cards are interpreted to be relative to output_scenario.path.
• All other relative paths are interpreted to be relative to directory of the scenario config file. (Or if scenario config is provided as a dictionary, relative paths will be interpreted as relative to the current working directory.)

Example Scenario Config

name: "my_scenario"
base_scenario:
    roadway:
        dir: "path/to/roadway_network"
        file_format: "geojson"
        read_in_shapes: True
    transit:
        dir: "path/to/transit_network"
        file_format: "txt"
    applied_projects:
        - "project1"
        - "project2"
    conflicts:
        "project3": ["project1", "project2"]
        "project4": ["project1"]
projects:
    project_card_filepath:
        - "path/to/projectA.yaml"
        - "path/to/projectB.yaml"
    filter_tags:
        - "tag1"
output_scenario:
    overwrite: True
    roadway:
        out_dir: "path/to/output/roadway"
        prefix: "my_scenario"
        file_format: "geojson"
        true_shape: False
    transit:
        out_dir: "path/to/output/transit"
        prefix: "my_scenario"
        file_format: "txt"
    project_cards:
        out_dir: "path/to/output/project_cards"

wrangler_config: "path/to/wrangler_config.yaml"

Extended Usage

Load a configuration from a file:

from network_wrangler.configs import load_scenario_config

my_scenario_config = load_scenario_config("path/to/config.yaml")

Access the configuration:

my_scenario_config.base_transit_network.path
>> path/to/transit_network

network_wrangler.configs.scenario.ProjectCardOutputConfig

Bases: ConfigItem

Configuration for outputing project cards in a scenario.

Attributes:

• out_dir –

  Path to write the project card files to if you don’t want to use the default.

• write –

  If True, will write the project cards. Defaults to True.

Source code in network_wrangler/configs/scenario.py

class ProjectCardOutputConfig(ConfigItem):
    """Configuration for outputing project cards in a scenario.

    Attributes:
        out_dir: Path to write the project card files to if you don't want to use the default.
        write: If True, will write the project cards. Defaults to True.
    """

    def __init__(
        self,
        base_path: Path = DEFAULT_BASE_DIR,
        out_dir: Path = DEFAULT_PROJECT_OUT_DIR,
        write: bool = DEFAULT_PROJECT_WRITE,
    ):
        """Constructor for ProjectCardOutputConfig."""
        if out_dir is not None and not Path(out_dir).is_absolute():
            self.out_dir = (base_path / Path(out_dir)).resolve()
        else:
            self.out_dir = Path(out_dir)
        self.write = write

network_wrangler.configs.scenario.ProjectsConfig

Bases: ConfigItem

Configuration for projects in a scenario.

Attributes:

• project_card_filepath –

  where the project card is. A single path, list of paths, a directory, or a glob pattern. Defaults to None.

• filter_tags –

  List of tags to filter the project cards by.

Source code in network_wrangler/configs/scenario.py

class ProjectsConfig(ConfigItem):
    """Configuration for projects in a scenario.

    Attributes:
        project_card_filepath: where the project card is.  A single path, list of paths,
            a directory, or a glob pattern. Defaults to None.
        filter_tags: List of tags to filter the project cards by.
    """

    def __init__(
        self,
        base_path: Path = DEFAULT_BASE_DIR,
        project_card_filepath: ProjectCardFilepaths = DEFAULT_PROJECT_IN_PATHS,
        filter_tags: list[str] = DEFAULT_PROJECT_TAGS,
    ):
        """Constructor for ProjectsConfig."""
        self.project_card_filepath = _resolve_rel_paths(project_card_filepath, base_path=base_path)
        self.filter_tags = filter_tags

network_wrangler.configs.scenario.RoadwayNetworkInputConfig

Bases: ConfigItem

Configuration for the road network in a scenario.

Attributes:

• dir –

  Path to directory with roadway network files.

• file_format –

  File format for the roadway network files. Should be one of RoadwayFileTypes. Defaults to “geojson”.

• read_in_shapes –

  If True, will read in the shapes of the roadway network. Defaults to False.

• boundary_geocode –

  Geocode of the boundary. Will use this to filter the roadway network.

• boundary_file –

  Path to the boundary file. If provided and both boundary_gdf and boundary_geocode are not provided, will use this to filter the roadway network.

Source code in network_wrangler/configs/scenario.py

class RoadwayNetworkInputConfig(ConfigItem):
    """Configuration for the road network in a scenario.

    Attributes:
        dir: Path to directory with roadway network files.
        file_format: File format for the roadway network files. Should be one of RoadwayFileTypes.
            Defaults to "geojson".
        read_in_shapes: If True, will read in the shapes of the roadway network. Defaults to False.
        boundary_geocode: Geocode of the boundary. Will use this to filter the roadway network.
        boundary_file: Path to the boundary file. If provided and both boundary_gdf and
            boundary_geocode are not provided, will use this to filter the roadway network.
    """

    def __init__(
        self,
        base_path: Path = DEFAULT_BASE_DIR,
        dir: Path = DEFAULT_ROADWAY_IN_DIR,
        file_format: RoadwayFileTypes = DEFAULT_ROADWAY_IN_FORMAT,
        read_in_shapes: bool = DEFAULT_ROADWAY_SHAPE_READ,
        boundary_geocode: Optional[str] = None,
        boundary_file: Optional[Path] = None,
    ):
        """Constructor for RoadwayNetworkInputConfig."""
        if dir is not None and not Path(dir).is_absolute():
            self.dir = (base_path / Path(dir)).resolve()
        else:
            self.dir = Path(dir)
        self.file_format = file_format
        self.read_in_shapes = read_in_shapes
        self.boundary_geocode = boundary_geocode
        self.boundary_file = boundary_file

network_wrangler.configs.scenario.RoadwayNetworkOutputConfig

Bases: ConfigItem

Configuration for writing out the resulting roadway network for a scenario.

Attributes:

• out_dir –

  Path to write the roadway network files to if you don’t want to use the default.

• prefix –

  Prefix to add to the file name. If not provided will use the scenario name.

• file_format –

  File format to write the roadway network to. Should be one of RoadwayFileTypes. Defaults to “geojson”.

• true_shape –

  If True, will write the true shape of the roadway network. Defaults to False.

• write –

  If True, will write the roadway network. Defaults to True.

Source code in network_wrangler/configs/scenario.py

class RoadwayNetworkOutputConfig(ConfigItem):
    """Configuration for writing out the resulting roadway network for a scenario.

    Attributes:
        out_dir: Path to write the roadway network files to if you don't want to use the default.
        prefix: Prefix to add to the file name. If not provided will use the scenario name.
        file_format: File format to write the roadway network to. Should be one of
            RoadwayFileTypes. Defaults to "geojson".
        true_shape: If True, will write the true shape of the roadway network. Defaults to False.
        write: If True, will write the roadway network. Defaults to True.
    """

    def __init__(
        self,
        out_dir: Path = DEFAULT_ROADWAY_OUT_DIR,
        base_path: Path = DEFAULT_BASE_DIR,
        convert_complex_link_properties_to_single_field: bool = False,
        prefix: Optional[str] = None,
        file_format: RoadwayFileTypes = DEFAULT_ROADWAY_OUT_FORMAT,
        true_shape: bool = False,
        write: bool = DEFAULT_ROADWAY_WRITE,
    ):
        """Constructor for RoadwayNetworkOutputConfig."""
        if out_dir is not None and not Path(out_dir).is_absolute():
            self.out_dir = (base_path / Path(out_dir)).resolve()
        else:
            self.out_dir = Path(out_dir)

        self.convert_complex_link_properties_to_single_field = (
            convert_complex_link_properties_to_single_field
        )
        self.prefix = prefix
        self.file_format = file_format
        self.true_shape = true_shape
        self.write = write

network_wrangler.configs.scenario.ScenarioConfig

Bases: ConfigItem

Scenario configuration for Network Wrangler.

Attributes:

• base_path –

  base path of the scenario. Defaults to cwd.

• name –

  Name of the scenario.

• base_scenario –

  information about the base scenario

• projects –

  information about the projects to apply on top of the base scenario

• output_scenario –

  information about how to output the scenario

• wrangler_config –

  wrangler configuration to use

Source code in network_wrangler/configs/scenario.py

class ScenarioConfig(ConfigItem):
    """Scenario configuration for Network Wrangler.

    Attributes:
        base_path: base path of the scenario. Defaults to cwd.
        name: Name of the scenario.
        base_scenario: information about the base scenario
        projects: information about the projects to apply on top of the base scenario
        output_scenario: information about how to output the scenario
        wrangler_config: wrangler configuration to use
    """

    def __init__(
        self,
        base_scenario: dict,
        projects: dict,
        output_scenario: dict,
        base_path: Path = DEFAULT_BASE_DIR,
        name: str = DEFAULT_SCENARIO_NAME,
        wrangler_config=DefaultConfig,
    ):
        """Constructor for ScenarioConfig."""
        self.base_path = Path(base_path) if base_path is not None else Path.cwd()
        self.name = name
        self.base_scenario = ScenarioInputConfig(**base_scenario, base_path=base_path)
        self.projects = ProjectsConfig(**projects, base_path=base_path)
        self.output_scenario = ScenarioOutputConfig(**output_scenario, base_path=base_path)
        self.wrangler_config = wrangler_config

network_wrangler.configs.scenario.ScenarioInputConfig

Bases: ConfigItem

Configuration for the writing the output of a scenario.

Attributes:

• roadway (Optional[RoadwayNetworkInputConfig]) –

  Configuration for writing out the roadway network.

• transit (Optional[TransitNetworkInputConfig]) –

  Configuration for writing out the transit network.

• applied_projects –

  List of projects to apply to the base scenario.

• conflicts –

  Dict of projects that conflict with the applied_projects.

Source code in network_wrangler/configs/scenario.py

class ScenarioInputConfig(ConfigItem):
    """Configuration for the writing the output of a scenario.

    Attributes:
        roadway: Configuration for writing out the roadway network.
        transit: Configuration for writing out the transit network.
        applied_projects: List of projects to apply to the base scenario.
        conflicts: Dict of projects that conflict with the applied_projects.
    """

    def __init__(
        self,
        base_path: Path = DEFAULT_BASE_DIR,
        roadway: Optional[dict] = None,
        transit: Optional[dict] = None,
        applied_projects: Optional[list[str]] = None,
        conflicts: Optional[dict] = None,
    ):
        """Constructor for ScenarioInputConfig."""
        if roadway is not None:
            self.roadway: Optional[RoadwayNetworkInputConfig] = RoadwayNetworkInputConfig(
                **roadway, base_path=base_path
            )
        else:
            self.roadway = None

        if transit is not None:
            self.transit: Optional[TransitNetworkInputConfig] = TransitNetworkInputConfig(
                **transit, base_path=base_path
            )
        else:
            self.transit = None

        self.applied_projects = applied_projects if applied_projects is not None else []
        self.conflicts = conflicts if conflicts is not None else {}

network_wrangler.configs.scenario.ScenarioOutputConfig

Bases: ConfigItem

Configuration for the writing the output of a scenario.

Attributes:

• roadway –

  Configuration for writing out the roadway network.

• transit –

  Configuration for writing out the transit network.

• project_cards (Optional[ProjectCardOutputConfig]) –

  Configuration for writing out the project cards.

• overwrite –

  If True, will overwrite the files if they already exist. Defaults to True

Source code in network_wrangler/configs/scenario.py

class ScenarioOutputConfig(ConfigItem):
    """Configuration for the writing the output of a scenario.

    Attributes:
        roadway: Configuration for writing out the roadway network.
        transit: Configuration for writing out the transit network.
        project_cards: Configuration for writing out the project cards.
        overwrite: If True, will overwrite the files if they already exist. Defaults to True
    """

    def __init__(
        self,
        path: Path = DEFAULT_OUTPUT_DIR,
        base_path: Path = DEFAULT_BASE_DIR,
        roadway: Optional[dict] = None,
        transit: Optional[dict] = None,
        project_cards: Optional[dict] = None,
        overwrite: bool = True,
    ):
        """Constructor for ScenarioOutputConfig."""
        if not Path(path).is_absolute():
            self.path = (base_path / Path(path)).resolve()
        else:
            self.path = Path(path)

        roadway = roadway if roadway else RoadwayNetworkOutputConfig().to_dict()
        transit = transit if transit else TransitNetworkOutputConfig().to_dict()
        self.roadway = RoadwayNetworkOutputConfig(**roadway, base_path=self.path)
        self.transit = TransitNetworkOutputConfig(**transit, base_path=self.path)

        if project_cards is not None:
            self.project_cards: Optional[ProjectCardOutputConfig] = ProjectCardOutputConfig(
                **project_cards, base_path=self.path
            )
        else:
            self.project_cards = None

        self.overwrite = overwrite

network_wrangler.configs.scenario.TransitNetworkInputConfig

Bases: ConfigItem

Configuration for the transit network in a scenario.

Attributes:

• dir –

  Path to the transit network files. Defaults to “.”.

• file_format –

  File format for the transit network files. Should be one of TransitFileTypes. Defaults to “txt”.

Source code in network_wrangler/configs/scenario.py

class TransitNetworkInputConfig(ConfigItem):
    """Configuration for the transit network in a scenario.

    Attributes:
        dir: Path to the transit network files. Defaults to ".".
        file_format: File format for the transit network files. Should be one of TransitFileTypes.
            Defaults to "txt".
    """

    def __init__(
        self,
        base_path: Path = DEFAULT_BASE_DIR,
        dir: Path = DEFAULT_TRANSIT_IN_DIR,
        file_format: TransitFileTypes = DEFAULT_TRANSIT_IN_FORMAT,
    ):
        """Constructor for TransitNetworkInputConfig."""
        if dir is not None and not Path(dir).is_absolute():
            self.feed = (base_path / Path(dir)).resolve()
        else:
            self.feed = Path(dir)
        self.file_format = file_format

network_wrangler.configs.scenario.TransitNetworkOutputConfig

Bases: ConfigItem

Configuration for the transit network in a scenario.

Attributes:

• out_dir –

  Path to write the transit network files to if you don’t want to use the default.

• prefix –

  Prefix to add to the file name. If not provided will use the scenario name.

• file_format –

  File format to write the transit network to. Should be one of TransitFileTypes. Defaults to “txt”.

• write –

  If True, will write the transit network. Defaults to True.

Source code in network_wrangler/configs/scenario.py

class TransitNetworkOutputConfig(ConfigItem):
    """Configuration for the transit network in a scenario.

    Attributes:
        out_dir: Path to write the transit network files to if you don't want to use the default.
        prefix: Prefix to add to the file name. If not provided will use the scenario name.
        file_format: File format to write the transit network to. Should be one of
            TransitFileTypes. Defaults to "txt".
        write: If True, will write the transit network. Defaults to True.
    """

    def __init__(
        self,
        base_path: Path = DEFAULT_BASE_DIR,
        out_dir: Path = DEFAULT_TRANSIT_OUT_DIR,
        prefix: Optional[str] = None,
        file_format: TransitFileTypes = DEFAULT_TRANSIT_OUT_FORMAT,
        write: bool = DEFAULT_TRANSIT_WRITE,
    ):
        """Constructor for TransitNetworkOutputCOnfig."""
        if out_dir is not None and not Path(out_dir).is_absolute():
            self.out_dir = (base_path / Path(out_dir)).resolve()
        else:
            self.out_dir = Path(out_dir)
        self.write = write
        self.prefix = prefix
        self.file_format = file_format

Configuration utilities.

network_wrangler.configs.utils.ConfigItem

Base class to add partial dict-like interface to configuration.

Allow use of .items() [“X”] and .get(“X”) .to_dict() from configuration.

Not to be constructed directly. To be used a mixin for dataclasses representing config schema. Do not use “get” “to_dict”, or “items” for key names.

Source code in network_wrangler/configs/utils.py

class ConfigItem:
    """Base class to add partial dict-like interface to  configuration.

    Allow use of .items() ["X"] and .get("X") .to_dict() from configuration.

    Not to be constructed directly. To be used a mixin for dataclasses
    representing config schema.
    Do not use "get" "to_dict", or "items" for key names.
    """

    base_path: Optional[Path] = None

    def __getitem__(self, key):
        """Return the value for key if key is in the dictionary, else default."""
        return getattr(self, key)

    def items(self):
        """A set-like object providing a view on D's items."""
        return self.__dict__.items()

    def to_dict(self):
        """Convert the configuration to a dictionary."""
        result = {}
        for key, value in self.__dict__.items():
            if isinstance(value, ConfigItem):
                result[key] = value.to_dict()
            else:
                result[key] = value
        return result

    def get(self, key, default=None):
        """Return the value for key if key is in the dictionary, else default."""
        return self.__dict__.get(key, default)

    def update(self, data: Union[Path, list[Path], dict]):
        """Update the configuration with a dictionary of new values."""
        if not isinstance(data, dict):
            WranglerLogger.info(f"Updating configuration with {data}.")
            data = load_merge_dict(data)

        self.__dict__.update(data)
        return self

    def resolve_paths(self, base_path):
        """Resolve relative paths in the configuration."""
        base_path = Path(base_path)
        for key, value in self.__dict__.items():
            if isinstance(value, ConfigItem):
                value.resolve_paths(base_path)
            elif isinstance(value, str) and value.startswith("."):
                resolved_path = (base_path / value).resolve()
                setattr(self, key, str(resolved_path))

network_wrangler.configs.utils.ConfigItem.__getitem__

__getitem__(key)

Return the value for key if key is in the dictionary, else default.

Source code in network_wrangler/configs/utils.py

def __getitem__(self, key):
    """Return the value for key if key is in the dictionary, else default."""
    return getattr(self, key)

network_wrangler.configs.utils.ConfigItem.get

get(key, default=None)

Return the value for key if key is in the dictionary, else default.

Source code in network_wrangler/configs/utils.py

def get(self, key, default=None):
    """Return the value for key if key is in the dictionary, else default."""
    return self.__dict__.get(key, default)

network_wrangler.configs.utils.ConfigItem.items

items()

A set-like object providing a view on D’s items.

Source code in network_wrangler/configs/utils.py

def items(self):
    """A set-like object providing a view on D's items."""
    return self.__dict__.items()

network_wrangler.configs.utils.ConfigItem.resolve_paths

resolve_paths(base_path)

Resolve relative paths in the configuration.

Source code in network_wrangler/configs/utils.py

def resolve_paths(self, base_path):
    """Resolve relative paths in the configuration."""
    base_path = Path(base_path)
    for key, value in self.__dict__.items():
        if isinstance(value, ConfigItem):
            value.resolve_paths(base_path)
        elif isinstance(value, str) and value.startswith("."):
            resolved_path = (base_path / value).resolve()
            setattr(self, key, str(resolved_path))

network_wrangler.configs.utils.ConfigItem.to_dict

to_dict()

Convert the configuration to a dictionary.

Source code in network_wrangler/configs/utils.py

def to_dict(self):
    """Convert the configuration to a dictionary."""
    result = {}
    for key, value in self.__dict__.items():
        if isinstance(value, ConfigItem):
            result[key] = value.to_dict()
        else:
            result[key] = value
    return result

network_wrangler.configs.utils.ConfigItem.update

update(data)

Update the configuration with a dictionary of new values.

Source code in network_wrangler/configs/utils.py

def update(self, data: Union[Path, list[Path], dict]):
    """Update the configuration with a dictionary of new values."""
    if not isinstance(data, dict):
        WranglerLogger.info(f"Updating configuration with {data}.")
        data = load_merge_dict(data)

    self.__dict__.update(data)
    return self

network_wrangler.configs.utils.find_configs_in_dir

find_configs_in_dir(dir, config_type)

Find configuration files in the directory that match *config<ext>.

Source code in network_wrangler/configs/utils.py

def find_configs_in_dir(dir: Union[Path, list[Path]], config_type) -> list[Path]:
    """Find configuration files in the directory that match `*config<ext>`."""
    config_files: list[Path] = []
    if isinstance(dir, list):
        for d in dir:
            config_files.extend(find_configs_in_dir(d, config_type))
    elif dir.is_dir():
        dir = Path(dir)
        for ext in SUPPORTED_CONFIG_EXTENSIONS:
            config_like_files = list(dir.glob(f"*config{ext}"))
            config_files.extend(find_configs_in_dir(config_like_files, config_type))
    elif dir.is_file():
        try:
            config_type(load_dict(dir))
        except ValidationError:
            return config_files
        config_files.append(dir)

    if config_files:
        return [Path(config_file) for config_file in config_files]
    return []