Projects ¶

class GroupedScopedPropertySetItem(BaseModel):
    """Value for setting property value for a single time of day and category."""

    model_config = ConfigDict(extra="forbid", exclude_none=True)

    category: str | int | None = None
    timespan: TimespanString | None = None
    categories: list[Any] | None = []
    timespans: list[TimespanString] | None = []
    set: Any | None = None
    overwrite_conflicts: bool | None = False
    existing: Any | None = None
    change: int | float | None = None
    _examples = [
        {"category": "hov3", "timespan": ["6:00", "9:00"], "set": 2.0},
        {"category": "hov2", "set": 2.0},
        {"timespan": ["12:00", "2:00"], "change": -1},
    ]

    @model_validator(mode="before")
    @classmethod
    def check_set_or_change(cls, data: dict):
        """Validate that each item has a set or change value."""
        if not isinstance(data, dict):
            return data
        if "set" in data and "change" in data:
            WranglerLogger.warning("Both set and change are set. Ignoring change.")
            data["change"] = None
        return data

    @model_validator(mode="before")
    @classmethod
    def check_categories_or_timespans(cls, data: Any) -> Any:
        """Validate that each item has a category or timespan value."""
        if not isinstance(data, dict):
            return data
        require_any_of = ["category", "timespan", "categories", "timespans"]
        if not any(attr in data for attr in require_any_of):
            msg = f"Require at least one of {require_any_of}"
            raise ValidationError(msg)
        return data

    @field_validator("timespan")
    @classmethod
    def validate_timespan(cls, v):
        """Validate the timespan field."""
        if v is not None:
            return validate_timespan_string(v)
        return v

    @field_validator("timespans", mode="before")
    @classmethod
    def validate_timespans(cls, v):
        """Validate the timespans field."""
        if v is not None:
            return [validate_timespan_string(ts) for ts in v]
        return v

class IndivScopedPropertySetItem(BaseModel):
    """Value for setting property value for a single time of day and category."""

    model_config = ConfigDict(extra="forbid", exclude_none=True)

    category: str | int | None = DEFAULT_CATEGORY
    timespan: TimespanString | None = DEFAULT_TIMESPAN
    set: Any | None = None
    existing: Any | None = None
    overwrite_conflicts: bool | None = False
    change: int | float | None = None
    _examples = [
        {"category": "hov3", "timespan": ["6:00", "9:00"], "set": 2.0},
        {"category": "hov2", "set": 2.0},
        {"timespan": ["12:00", "2:00"], "change": -1},
    ]

    @property
    def timespan_dt(self) -> list[list[datetime]]:
        """Convert timespan to list of datetime objects."""
        return str_to_time_list(self.timespan)

    @model_validator(mode="before")
    @classmethod
    def check_set_or_change(cls, data: dict):
        """Validate that each item has a set or change value."""
        if not isinstance(data, dict):
            return data
        if data.get("set") and data.get("change"):
            WranglerLogger.warning("Both set and change are set. Ignoring change.")
            data["change"] = None

        WranglerLogger.debug(f"Data: {data}")
        if data.get("set") is None and data.get("change") is None:
            msg = f"Require at least one of 'set' or'change' in IndivScopedPropertySetItem"
            WranglerLogger.debug(msg=f"   Found: {data}")
            raise ValueError(msg)
        return data

    @model_validator(mode="before")
    @classmethod
    def check_categories_or_timespans(cls, data: Any) -> Any:
        """Validate that each item has a category or timespan value."""
        if not isinstance(data, dict):
            return data
        require_any_of = ["category", "timespan"]
        if not any(attr in data for attr in require_any_of):
            msg = f"Require at least one of {require_any_of}"
            raise ValidationError(msg)
        return data

    @field_validator("timespan")
    @classmethod
    def validate_timespan(cls, v):
        """Validate the timespan field."""
        if v is not None:
            return validate_timespan_string(v)
        return v

class RoadPropertyChange(RecordModel):
    """Value for setting property value for a time of day and category."""

    model_config = ConfigDict(extra="forbid", exclude_none=True)

    existing: Any | None = None
    change: int | float | None = None
    set: Any | None = None
    scoped: None | ScopedPropertySetList = None
    overwrite_scoped: Literal["conflicting", "all", "error"] | None = None
    existing_value_conflict: Literal["error", "warn", "skip"] | None = None

    require_one_of: ClassVar[OneOf] = [
        ["change", "set"],
    ]

    _examples: ClassVar[list] = [
        {"set": 1},
        {"existing": 2, "change": -1},
        {
            "set": 0,
            "scoped": [
                {"timespan": ["6:00", "9:00"], "value": 2.0},
                {"timespan": ["9:00", "15:00"], "value": 4.0},
            ],
        },
        {
            "set": 0,
            "scoped": [
                {
                    "categories": ["hov3", "hov2"],
                    "timespan": ["6:00", "9:00"],
                    "value": 2.0,
                },
                {"category": "truck", "timespan": ["6:00", "9:00"], "value": 4.0},
            ],
        },
        {
            "set": 0,
            "scoped": [
                {"categories": ["hov3", "hov2"], "value": 2.0},
                {"category": "truck", "value": 4.0},
            ],
        },
    ]

class RoadwayDeletion(RecordModel):
    """Requirements for describing roadway deletion project card (e.g. to delete)."""

    require_any_of: ClassVar[AnyOf] = [["links", "nodes"]]
    model_config = ConfigDict(extra="forbid")

    links: SelectLinksDict | None = None
    nodes: SelectNodesDict | None = None
    clean_shapes: bool = False
    clean_nodes: bool = False

    @field_validator("links")
    @classmethod
    def set_to_all_modes(cls, links: SelectLinksDict | None = None):
        """Set the search mode to 'any' if not specified explicitly."""
        if links is not None and links.modes == DEFAULT_SEARCH_MODES:
            links.modes = DEFAULT_DELETE_MODES
        return links

class ScopedPropertySetList(RootListMixin, RootModel):
    """List of ScopedPropertySetItems used to evaluate and apply changes to roadway properties."""

    root: list[IndivScopedPropertySetItem]

    @model_validator(mode="before")
    @classmethod
    def check_set_or_change(cls, data: list):
        """Validate that each item has a set or change value."""
        data = _grouped_to_indiv_list_of_scopedpropsetitem(data)
        return data

    @model_validator(mode="after")
    def check_conflicting_scopes(self):
        """Check for conflicting scopes in the list of ScopedPropertySetItem."""
        conflicts = []
        for i in self:
            if i.timespan == DEFAULT_TIMESPAN:
                continue
            overlapping_ts_i = self.overlapping_timespans(i.timespan)
            for j in overlapping_ts_i:
                if j == i:
                    continue
                if j.category == i.category:
                    conflicts.append((i, j))
        if conflicts:
            msg = "Conflicting scopes in ScopedPropertySetList"
            WranglerLogger.error(msg + f"\n    Conflicts: {conflicts}")
            raise ScopeConflictError(msg)

        return self

    def overlapping_timespans(self, timespan: TimespanString) -> list[IndivScopedPropertySetItem]:
        """Return a list of items that overlap with the given timespan."""
        timespan_dt = str_to_time_list(timespan)
        return [i for i in self if dt_overlaps(i.timespan_dt, timespan_dt)]

    @property
    def change_items(self) -> list[IndivScopedPropertySetItem]:
        """Filter out items that do not have a change value."""
        WranglerLogger.debug(f"self.root[0]: {self.root[0]}")
        return [i for i in self if i.change is not None]

    @property
    def set_items(self):
        """Filter out items that do not have a set value."""
        return [i for i in self if i.set is not None]

class RoadwaySelectionFormatError(Exception):
    """Raised when there is an issue with the format of a selection."""

class SelectFacility(RecordModel):
    """Roadway Facility Selection."""

    require_one_of: ClassVar[OneOf] = [
        ["links", "nodes", ["links", "from", "to"]],
    ]
    model_config = ConfigDict(extra="forbid")

    links: SelectLinksDict | None = None
    nodes: SelectNodesDict | None = None
    from_: Annotated[SelectNodeDict | None, Field(None, alias="from")]
    to: SelectNodeDict | None = None

    _examples: ClassVar[list[dict]] = [
        {
            "links": {"name": ["Main Street"]},
            "from": {"model_node_id": 1},
            "to": {"model_node_id": 2},
        },
        {"nodes": {"osm_node_id": ["1", "2", "3"]}},
        {"nodes": {"model_node_id": [1, 2, 3]}},
        {"links": {"model_link_id": [1, 2, 3]}},
    ]

class SelectLinksDict(RecordModel):
    """requirements for describing links in the `facility` section of a project card.

    Examples:
    ```python
        {'name': ['Main St'], 'modes': ['drive']}
        {'osm_link_id': ['123456789']}
        {'model_link_id': [123456789], 'modes': ['walk']}
        {'all': 'True', 'modes': ['transit']}
        {'all': 'True', name': ['Main St']}
    ```

    """

    require_conflicts: ClassVar[ConflictsWith] = [
        ["all", "osm_link_id"],
        ["all", "model_link_id"],
        ["all", "name"],
        ["all", "ref"],
        ["osm_link_id", "model_link_id"],
        ["osm_link_id", "name"],
        ["model_link_id", "name"],
    ]
    require_any_of: ClassVar[AnyOf] = [["name", "ref", "osm_link_id", "model_link_id", "all"]]

    model_config = ConfigDict(extra="allow")

    all: bool | None = False
    name: Annotated[list[str] | None, Field(None, min_length=1)]
    ref: Annotated[list[str] | None, Field(None, min_length=1)]
    osm_link_id: Annotated[list[str] | None, Field(None, min_length=1)]
    model_link_id: Annotated[list[int] | None, Field(None, min_length=1)]
    modes: list[str] = DEFAULT_SEARCH_MODES
    ignore_missing: bool | None = True

    _examples: ClassVar[list[dict]] = [
        {"name": ["Main St"], "modes": ["drive"]},
        {"osm_link_id": ["123456789"]},
        {"model_link_id": [123456789], "modes": ["walk"]},
        {"all": "True", "modes": ["transit"]},
    ]

class SelectNodeDict(RecordModel):
    """Selection of a single roadway node in the `facility` section of a project card."""

    require_one_of: ClassVar[OneOf] = [["osm_node_id", "model_node_id"]]
    model_config = ConfigDict(extra="allow")

    osm_node_id: str | None = None
    model_node_id: int | None = None

    _examples: ClassVar[list[dict]] = [{"osm_node_id": "12345"}, {"model_node_id": 67890}]

class SelectNodesDict(RecordModel):
    """Requirements for describing multiple nodes of a project card (e.g. to delete)."""

    require_any_of: ClassVar[AnyOf] = [["osm_node_id", "model_node_id"]]
    model_config = ConfigDict(extra="forbid")

    all: bool | None = False
    osm_node_id: Annotated[list[str] | None, Field(None, min_length=1)]
    model_node_id: Annotated[list[int] | None, Field(min_length=1)]
    ignore_missing: bool | None = True

    _examples: ClassVar[list[dict]] = [
        {"osm_node_id": ["12345", "67890"], "model_node_id": [12345, 67890]},
        {"osm_node_id": ["12345", "67890"]},
        {"model_node_id": [12345, 67890]},
    ]

class SelectRouteProperties(RecordModel):
    """Selection properties for transit routes."""

    route_short_name: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]
    route_long_name: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]
    agency_id: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]
    route_type: Annotated[list[int] | None, Field(None, min_length=1)]

    model_config = ConfigDict(
        extra="allow",
        validate_assignment=True,
        exclude_none=True,
        protected_namespaces=(),
    )

class SelectTransitLinks(RecordModel):
    """Requirements for describing multiple transit links of a project card."""

    require_one_of: ClassVar[OneOf] = [
        ["ab_nodes", "model_link_id"],
    ]

    model_link_id: Annotated[list[int] | None, Field(min_length=1)] = None
    ab_nodes: Annotated[list[TransitABNodesModel] | None, Field(min_length=1)] = None
    require: SelectionRequire | None = "any"

    model_config = ConfigDict(
        extra="forbid",
        validate_assignment=True,
        exclude_none=True,
        protected_namespaces=(),
    )
    _examples: ClassVar[list[dict]] = [
        {
            "ab_nodes": [{"A": "75520", "B": "66380"}, {"A": "66380", "B": "75520"}],
            "type": "any",
        },
        {
            "model_link_id": [123, 321],
            "type": "all",
        },
    ]

class SelectTransitNodes(RecordModel):
    """Requirements for describing multiple transit nodes of a project card (e.g. to delete)."""

    require_any_of: ClassVar[AnyOf] = [
        [
            "model_node_id",
            # "stop_id_GTFS", TODO Not implemented
        ]
    ]

    # stop_id_GTFS: Annotated[Optional[List[ForcedStr]], Field(None, min_length=1)] TODO Not implemented
    model_node_id: Annotated[list[int], Field(min_length=1)]
    require: SelectionRequire | None = "any"

    model_config = ConfigDict(
        extra="forbid",
        validate_assignment=True,
        exclude_none=True,
        protected_namespaces=(),
    )

    _examples: ClassVar[list[dict]] = [
        # {"gtfstop_id": ["stop1", "stop2"], "require": "any"},  TODO Not implemented
        {"model_node_id": [1, 2], "require": "all"},
    ]

class SelectTransitTrips(RecordModel):
    """Selection properties for transit trips."""

    trip_properties: SelectTripProperties | None = None
    route_properties: SelectRouteProperties | None = None
    timespans: Annotated[list[TimespanString] | None, Field(None, min_length=1)]
    nodes: SelectTransitNodes | None = None
    links: SelectTransitLinks | None = None

    model_config = ConfigDict(
        extra="forbid",
        validate_assignment=True,
        exclude_none=True,
        protected_namespaces=(),
    )

    @field_validator("timespans", mode="before")
    @classmethod
    def validate_timespans(cls, v):
        """Validate the timespans field."""
        if v is not None:
            return [validate_timespan_string(ts) for ts in v]
        return v

class SelectTripProperties(RecordModel):
    """Selection properties for transit trips."""

    trip_id: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]
    shape_id: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]
    direction_id: Annotated[int | None, Field(None)]
    service_id: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]
    route_id: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]
    trip_short_name: Annotated[list[ForcedStr] | None, Field(None, min_length=1)]

    model_config = ConfigDict(
        extra="allow",
        validate_assignment=True,
        exclude_none=True,
        protected_namespaces=(),
    )

class TransitABNodesModel(RecordModel):
    """Single transit link model."""

    A: int | None = None  # model_node_id
    B: int | None = None  # model_node_id

    model_config = ConfigDict(
        extra="forbid",
        validate_assignment=True,
        exclude_none=True,
        protected_namespaces=(),
    )