Road Conditions Translator

Translators/RoadConditions/Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.10-slim
+
+ENV PYTHONUNBUFFERED TRUE
+
+ENV WORKDIR /app
+
+WORKDIR $WORKDIR
+
+RUN apt-get update && apt-get install -y libgeos-dev
+
+COPY ./requirements.txt .
+
+RUN pip install -r requirements.txt
+
+COPY . .
+
+ENV FLASK_APP app/main.py
+
+CMD ["python3", "-m", "flask", "run", "--host", "0.0.0.0", "--port", "8082"]

Translators/RoadConditions/README.md

@@ -0,0 +1,53 @@
+# Road Conditions Translator
+
+## Table of Contents
+
+- [About](#about)
+- [Getting Started](#getting_started)
+- [Usage](#usage)
+- [Contributing](../CONTRIBUTING.md)
+
+## About <a name = "about"></a>
+
+Python Road Conditions to TIM message translator that is designed to pull messages from the CDOT Road Conditions feed and translate to TIM messages.
+
+## Getting Started <a name = "getting_started"></a>
+
+These instructions will get you a copy of the project up and running on your local machine for development and testing purposes. See [deployment](#deployment) for notes on how to deploy the project on a live system.
+
+### Prerequisites
+
+This project supports Python >= 3.5. Packages required to run the translator can be installed via [pip](https://pip.pypa.io/en/stable/)
+
+```bash
+pip install -r requirements.txt
+```
+
+Alternatively, if you are running VSCode there is a task available to run this on your behalf. This can be accessed under Task Explorer -> vscode -> pipinstall.
+
+The scripts also require access to the CDOT Postgres database. This can be accessed by setting the environment variables in the .env file. The .env file is not included in the repository for security reasons, however a sample.env file has been provided to show structure required. 
+
+In addition to the environment variables for accessing the CDOT Postgres database, the scripts also require the following environment variables to scrape the Road Conditions endpoint and deposit the resulting TIMs:
+<ol>
+    <li>CDOT_FEED_ENDPOINT - the CDOT Data Feed URL</li>
+    <li>CDOT_FEED_API_KEY - the API key for accessing the CDOT Road Conditions data</li>
+    <li>TIM_MANAGER_ENDPOINT - the TIM Manager URL where translated TIMs will be submitted </li>
+</ol>
+
+### Testing
+Unit tests are ran with the python pytest module. To run the tests, run the following command from the root of the project:
+
+```bash
+python -m pytest ./tests
+```
+
+Again, if you are running VSCode there is a task available to run this on your behalf. This can be accessed under Task Explorer -> vscode -> python test and coverage. This task includes a unit test coverage report to indicate how much of the code is covered by unit tests.
+
+## Usage <a name = "usage"></a>
+
+### Running the Translator Locally
+Using VSCode, a simple launch.json file has been provided to allow debugging the application. This can be accessed under the Run & Debug tab. The default configuration will run the translator using the functions framework. This runs the translator as a REST service accessed on http://localhost:8082. The translator can be tested by sending a POST request to this endpoint.
+
+
+### Running the Translator via Docker
+The Road Conditions to TIM translator can also be run locally using Docker. The translator Dockerfile can be found under Translators/RoadConditions/. Additionally, there is a docker-compose file which builds and runs the translator. Setting the environment variable RUN_LOCAL to true will run the translator REST service as a flask application that can be accessed on http://localhost:8082. Alternatively, leaving RUN_LOCAL blank will run the translator one time immediately after the build has finished. 

Translators/RoadConditions/itis_codes.py

@@ -0,0 +1,74 @@
+from enum import Enum
+
+
+class ItisCodes(Enum):
+    SPEED_LIMIT = '368'
+    ACCIDENT = '513'
+    INCIDENT = '531'
+    HAZARDOUS_MATERIAL_SPILL = '550'
+    CLOSED = '770'
+    CLOSED_FOR_SEASON = '774'
+    REDUCED_ONE_LANE = '777'
+    AVALANCHE_CONTROL_ACTIVITIES = '1042'
+    ROAD_CONSTRUCTION = '1025'
+    HERD_OF_ANIMALS_ON_ROADWAY = '1292'
+    ROCKFALL = '1309'
+    LANDSLIDE = '1310'
+    DELAYS = '1537'
+    WIDE_LOAD = '2050'
+    NO_TRAILERS = '2568'
+    WIDTH_LIMIT = '2573'
+    HEIGHT_LIMIT = '2574'
+    WILD_FIRE = '3084'
+    WEATHER_EMERGENCY = '3201'
+    MAJOR_EVENT = '3841'
+    NO_PARKING_SPACES_AVAILABLE = '4103'
+    FEW_PARKING_SPACES_AVAILABLE = '4104'
+    SPACES_AVAILABLE = '4105'
+    NO_PARKING_INFO_AVAILABLE = '4223'
+    SEVERE_WEATHER = '4865'
+    SNOW = '4868'
+    WINTER_STORM = '4871'
+    RAIN = '4885'
+    STRONG_WINDS = '5127'
+    FOG = '5378'
+    VISIBILITY_REDUCED = '5383'
+    BLOWING_SNOW = '5985'
+    BLACK_ICE = '5908'
+    WET_PAVEMENT = '5895'
+    ICE = '5906'
+    ICY_PATCHES = '5907'
+    SNOW_DRIFTS = '5927'
+    GRAVEL_ROAD_SURFACE = '5933'
+    DRY_PAVEMENT = '6011'
+    DIRT_ROAD_SURFACE = '6016'
+    MILLED_ROAD_SURFACE = '6017'
+    SNOW_TIRES_OR_CHAINS_REQUIRED = '6156'
+    LOOK_OUT_FOR_WORKERS = '6952'
+    KEEP_TO_RIGHT = '7425'
+    KEEP_TO_LEFT = '7426'
+    REDUCE_YOUR_SPEED = '7443'
+    DRIVE_CAREFUL = '7169'
+    DRIVE_WITH_EXTREME_CAUTION = '7170'
+    INCREASE_FOLLOWING_DISTANCE = '7173'
+    PREPARE_TO_STOP = '7186'
+    STOP_AT_NEXT_SAFE_PLACE = '7188'
+    ONLY_TRAVEL_IF_NECESSARY = '7189'
+    FALLING_ROCKS = '1203'
+
+ItisCodeExtraKeywords = {
+    "herd of animals on roadway": "herd of animals on the roadway",
+    "rockfall": "rock fall",
+    "wildfire": "wild fire",
+    "keep to right": "keep right",
+    "keep to left": "keep left",
+    "reduce your speed": "reduce speed",
+    "drive careful": "drive carefully",
+    "stop at next safe place": "stop at the next safe place",
+    "only travel if necessary": "only necessary travel",
+    "falling rocks": "falling rock",
+    "icy patches": "icy spots",
+    "snow": "snow packed spots",
+    "closed for season": "seasonal closure",
+    "ice": "icy",
+}

Translators/RoadConditions/main.py

@@ -0,0 +1,65 @@
+import json
+import requests
+import copy
+import logging
+import os
+from flask import request, Flask
+from tim_translator import translate
+
+app = Flask(__name__)
+
+log_level = os.environ.get('LOGGING_LEVEL', 'INFO')
+logging.basicConfig(format='%(levelname)s:%(message)s', level=log_level)
+
+@app.route('/', methods=['POST'])
+def entry():
+    if request.method == 'OPTIONS':
+        headers = {
+            'Access-Control-Allow-Origin': '*',
+            'Access-Control-Allow-Methods': 'POST',
+            'Access-Control-Allow-Headers': 'Content-Type',
+            'Access-Control-Max-Age': '3600'
+        }
+
+        return ('', 204, headers)
+
+    headers = {
+        'Access-Control-Allow-Origin': '*'
+    }
+
+    result = RC_tim_translator()
+    logging.info(result)
+
+    return (result, 200, headers)
+
+def RC_tim_translator():
+    logging.info('Road Conditions TIM Translator Timer Called...')
+
+    # Scrape the CDOT endpoint to get current list of road condition features
+    geoJSON =  json.loads(requests.get(f'https://{os.getenv("CDOT_FEED_ENDPOINT")}/api/v1/roadConditions?apiKey={os.getenv("CDOT_FEED_API_KEY")}').content.decode('utf-8'))
+
+    tim_list = translate(geoJSON)
+
+    logging.info('Pushing TIMs to TIM Manager...')
+
+    tim_list_copy = copy.deepcopy(tim_list)
+    tim_all_clear_list = {"timRcList": [tim for tim in tim_list_copy["timRcList"] if len(tim["itisCodes"]) == 0]}
+    tim_list["timRcList"] = [tim for tim in tim_list["timRcList"] if len(tim["itisCodes"]) > 0]
+
+    return_value = requests.put(f'{os.getenv("TIM_MANAGER_ENDPOINT")}/submit-rc-ac', json=tim_all_clear_list)
+    if (return_value.status_code == 200):
+        logging.info(f'Successfully submitted {len(tim_all_clear_list["timRcList"])} All Clear TIMs to TIM Manager')
+
+    return_value = requests.post(f'{os.getenv("TIM_MANAGER_ENDPOINT")}/create-update-rc-tim', json=tim_list)
+    if (return_value.status_code == 200):
+        return f'Successfully pushed {len(tim_list["timRcList"])} TIMs to TIM Manager'
+
+    return f'Error pushing TIMs to TIM Manager: {return_value.content}'
+
+# Run via flask app if running locally else just run translator directly
+if (os.getenv("RUN_LOCAL") == "true"):
+    if __name__ == '__main__':
+        app.run()
+else:
+    res_str = RC_tim_translator()
+    logging.info(res_str)

Translators/RoadConditions/pgquery.py

@@ -0,0 +1,86 @@
+import os
+import sqlalchemy
+import logging
+
+db_config = {
+    # Pool size is the maximum number of permanent connections to keep.
+    "pool_size": 5,
+    # Temporarily exceeds the set pool_size if no connections are available.
+    "max_overflow": 2,
+    # Maximum number of seconds to wait when retrieving a
+    # new connection from the pool. After the specified amount of time, an
+    # exception will be thrown.
+    "pool_timeout": 30,  # 30 seconds
+    # 'pool_recycle' is the maximum number of seconds a connection can persist.
+    # Connections that live longer than the specified amount of time will be
+    # reestablished
+    "pool_recycle": 60  # 1 minutes
+}
+
+db = None
+
+def init_tcp_connection_engine(db_user, db_pass, db_name, db_hostname, db_port):
+    logging.info(f"Creating DB pool")
+    pool = sqlalchemy.create_engine(
+        # Equivalent URL:
+        # postgresql+pg8000://<db_user>:<db_pass>@<db_host>:<db_port>/<db_name>
+        sqlalchemy.engine.url.URL.create(
+            drivername="postgresql+pg8000",
+            username=db_user,  # e.g. "my-database-user"
+            password=db_pass,  # e.g. "my-database-password"
+            host=db_hostname,  # e.g. "127.0.0.1"
+            port=db_port,  # e.g. 5432
+            database=db_name  # e.g. "my-database-name"
+        ),
+        **db_config
+    )
+    #pool.dialect.description_encoding = None
+    logging.info("DB pool created!")
+    return pool
+
+def init_socket_connection_engine(db_user, db_pass, db_name, unix_query):
+  logging.info(f"Creating DB pool")
+  pool = sqlalchemy.create_engine(
+    # Equivalent URL:
+    # postgresql+pg8000://<db_user>:<db_pass>@/<db_name>?unix_sock=/cloudsql/<cloud_sql_instance_name>
+    sqlalchemy.engine.url.URL.create(
+      drivername="postgresql+pg8000",
+      username=db_user,  # e.g. "my-database-user"
+      password=db_pass,  # e.g. "my-database-password"
+      database=db_name,  # e.g. "my-database-name"
+      query=unix_query
+    ),
+    **db_config
+  )
+  logging.info("DB pool created!")
+  return pool
+
+
+def init_connection_engine():
+    db_user = os.environ["DB_USER"]
+    db_pass = os.environ["DB_PASS"]
+    db_name = os.environ["DB_NAME"]
+    if("INSTANCE_CONNECTION_NAME" in os.environ):
+        instance_connection_name = os.environ["INSTANCE_CONNECTION_NAME"]
+        unix_query = {
+            "unix_sock": f"/cloudsql/{instance_connection_name}/.s.PGSQL.5432"
+        }
+        return init_socket_connection_engine(db_user, db_pass, db_name, unix_query)
+    else:
+        db_host = os.environ["DB_HOST"]
+        # Extract host and port from db_host
+        host_args = db_host.split(":")
+        db_hostname, db_port = host_args[0], int(host_args[1])
+        return init_tcp_connection_engine(db_user, db_pass, db_name, db_hostname, db_port)
+
+
+def query_db(query):
+    global db
+    if db is None:
+        db = init_connection_engine()
+
+    logging.info("DB connection starting...")
+    with db.connect() as conn:
+        logging.debug("Executing query...")
+        data = conn.execute(query).fetchall()
+        return data

Translators/RoadConditions/requirements.txt

@@ -0,0 +1,11 @@
+functions-framework==3.4.0
+flask==2.0.2
+google-cloud-error-reporting==1.4.1
+pyproj==3.6.1
+requests
+shapely==2.0.4
+sqlalchemy==1.4.22
+pg8000==1.29.6
+marshmallow
+redis==4.6.0
+werkzeug==2.2.2

Translators/RoadConditions/sample.env

@@ -0,0 +1,10 @@
+dual_carriageway_endpoint=https://dtdapps.coloradodot.info/arcgis/rest/services/LRS/Routes_withDEC/MapServer/exts/CdotLrsAccessRounded
+DB_HOST=
+DB_NAME=
+DB_PASS=
+DB_USER=
+LOGGING_LEVEL=INFO
+CDOT_FEED_ENDPOINT=
+CDOT_FEED_API_KEY=
+TIM_MANAGER_ENDPOINT=
+RUN_LOCAL=

Translators/RoadConditions/tim_generator.py

@@ -0,0 +1,31 @@
+from itis_codes import ItisCodes, ItisCodeExtraKeywords
+from string import digits
+
+def get_itis_codes(feature):
+    itisCodes = []
+
+    # need to iterate over entries & split to check for keywords
+    for entry in feature.get_current_conditions():
+        itis_codes = entry["conditionDescription"].split(",")
+        for code in itis_codes:
+            code = code.translate({ord(k): None for k in digits}).replace("-", "").strip()
+            if code == "forecast text included":
+                continue
+            for key in ItisCodes:
+                searchKey = key.name.replace("_", " ").lower()
+                if searchKey in code.lower() and key.value not in itisCodes:
+                    itisCodes.append(key.value)
+                elif searchKey in ItisCodeExtraKeywords:
+                    if (
+                        ItisCodeExtraKeywords[searchKey] in code.lower()
+                        and key.value not in itisCodes
+                    ):
+                        itisCodes.append(key.value)
+
+    return itisCodes
+
+def get_geometry(geometry):
+    annotated_geometry = []
+    for coord in geometry:
+        annotated_geometry.append({"latitude": coord[1], "longitude": coord[0]})
+    return annotated_geometry