Application

app

Attributes

insert_data module-attribute

insert_data: None = {}

Handle the insert data workflow.

Purpose

Supports the Mappy Streamlit application by executing the insert data workflow.

Parameters:

Name	Type	Description	Default
table_name		str value representing the name of the table.	required
df		pd.DataFrame value used by the insert_data workflow.	required

Classes

ManagedConnection

Bases: Connection

Represent the Managed Connection component for the Mappy Streamlit application.

Purpose

Provides the Managed Connection class used by the Mappy Streamlit application. The class preserves application state, data-access behavior, and UI-facing integration points required by the existing workflow.

Functions:

throw_if

throw_if(name: str, value: object) -> None

Validate that a required value is present.

Purpose

Raises a clear validation error when a required argument is missing, blank, or empty. The helper is used by Streamlit utilities and data workflows before values are consumed by UI, database, or service logic.

Parameters:

Name	Type	Description	Default
name	str	str value used by the throw_if workflow.	required
value	object	object value used by the throw_if workflow.	required

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def throw_if( name: str, value: object ) -> None:
	"""Validate that a required value is present.

	Purpose:
		Raises a clear validation error when a required argument is missing, blank, or
		empty. The helper is used by Streamlit utilities and data workflows before values
		are consumed by UI, database, or service logic.

	Args:
		name: str value used by the `throw_if` workflow.
		value: object value used by the `throw_if` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	if value is None:
		raise ValueError( f'Argument "{name}" cannot be empty!' )

	if isinstance( value, str ) and not value.strip( ):
		raise ValueError( f'Argument "{name}" cannot be empty!' )

	if isinstance( value, (list, tuple, dict, set) ) and len( value ) == 0:
		raise ValueError( f'Argument "{name}" cannot be empty!' )

style_subheaders

style_subheaders() -> None

Handle the style subheaders workflow.

Purpose

Supports the Mappy Streamlit application by executing the style subheaders workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Source code in app.py

def style_subheaders( ) -> None:
	"""Handle the style subheaders workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the style subheaders
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.
	"""
	st.markdown(
		"""
		<style>
		div[data-testid="stMarkdownContainer"] h2,
		div[data-testid="stMarkdownContainer"] h3,
		div[data-testid="stChatMessage"] div[data-testid="stMarkdownContainer"] h2,
		div[data-testid="stChatMessage"] div[data-testid="stMarkdownContainer"] h3 {
			color: rgb(0, 120, 252) !important;
		}
		</style>
		""",
		unsafe_allow_html=True, )

init_state

init_state(key: str, value: Any) -> None

Handle the init state workflow.

Purpose

Supports the Mappy Streamlit application by executing the init state workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
key	str	str value used by the init_state workflow.	required
value	Any	Any value used by the init_state workflow.	required

Source code in app.py

def init_state( key: str, value: Any ) -> None:
	"""Handle the init state workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the init state workflow. The
		function preserves the existing UI behavior, session-state interactions, dataframe
		handling, database access, and service integrations defined by the application
		code.

	Args:
		key: str value used by the `init_state` workflow.
		value: Any value used by the `init_state` workflow.
	"""
	if key not in st.session_state:
		st.session_state[ key ] = value

init_env_state

init_env_state(
    key: str, config_name: str, env_name: str
) -> None

Handle the init env state workflow.

Purpose

Supports the Mappy Streamlit application by executing the init env state workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
key	str	str value used by the init_env_state workflow.	required
config_name	str	str value used by the init_env_state workflow.	required
env_name	str	str value used by the init_env_state workflow.	required

Source code in app.py

def init_env_state( key: str, config_name: str, env_name: str ) -> None:
	"""Handle the init env state workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the init env state workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		key: str value used by the `init_env_state` workflow.
		config_name: str value used by the `init_env_state` workflow.
		env_name: str value used by the `init_env_state` workflow.
	"""
	init_state( key, '' )
	if st.session_state.get( key, '' ) == '':
		default = getattr( cfg, config_name, '' )
		if default:
			st.session_state[ key ] = default
			os.environ[ env_name ] = default

normalize

normalize(obj: str) -> str

Handle the normalize workflow.

Purpose

Supports the Mappy Streamlit application by executing the normalize workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
obj	str	str Runtime value used by the normalize workflow.	required

Returns:

Name	Type	Description
str	str	Result produced by the normalize workflow.

Source code in app.py

def normalize( obj: str ) -> str:
	"""Handle the normalize workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the normalize workflow. The
		function preserves the existing UI behavior, session-state interactions, dataframe
		handling, database access, and service integrations defined by the application
		code.

	Args:
		obj: str Runtime value used by the `normalize` workflow.

	Returns:
		str: Result produced by the `normalize` workflow.
	"""
	if obj is None or isinstance( obj, (str, int, float, bool) ):
		return obj

	if isinstance( obj, dict ):
		return { k: normalize( v ) for k, v in obj.items( ) }

	if isinstance( obj, (list, tuple, set) ):
		return [ normalize( v ) for v in obj ]
	if hasattr( obj, "model_dump" ):
		try:
			return obj.model_dump( )
		except Exception:
			return str( obj )
	return str( obj )

set_blue_divider

set_blue_divider() -> None

Set the blue divider used by the application.

Purpose

Supports the Mappy Streamlit application by executing the set blue divider workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Source code in app.py

def set_blue_divider( ) -> None:
	"""Set the blue divider used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the set blue divider
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.
	"""
	st.markdown( cfg.BLUE_DIVIDER, unsafe_allow_html=True )

log_step

log_step(msg: str) -> None

Handle the log step workflow.

Purpose

Supports the Mappy Streamlit application by executing the log step workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
msg	str	str value used by the log_step workflow.	required

Source code in app.py

def log_step( msg: str ) -> None:
	"""Handle the log step workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the log step workflow. The
		function preserves the existing UI behavior, session-state interactions, dataframe
		handling, database access, and service integrations defined by the application
		code.

	Args:
		msg: str value used by the `log_step` workflow.
	"""
	st.session_state.pipeline_log.append( msg )

render_celestial_map

render_celestial_map(
    asset_root: str = "assets/starmap",
    height: int = 1400,
    latitude: Optional[float] = None,
    longitude: Optional[float] = None,
    tile_url: Optional[str] = None,
    tile_attribution: Optional[str] = None,
    tile_subdomains: Optional[str] = None,
    location: Optional[str] = None,
    zoom: Optional[int] = None,
) -> None

Render the celestial map interface section.

Purpose

Supports the Mappy Streamlit application by executing the render celestial map workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
asset_root	str	str value used by the render_celestial_map workflow. Defaults to 'assets/starmap'.	'assets/starmap'
height	int	int value used by the render_celestial_map workflow. Defaults to 1400.	1400
latitude	Optional[float]	Optional[float] value used by the render_celestial_map workflow. Defaults to None.	None
longitude	Optional[float]	Optional[float] value used by the render_celestial_map workflow. Defaults to None.	None
tile_url	Optional[str]	Optional[str] value used by the render_celestial_map workflow. Defaults to None.	None
tile_attribution	Optional[str]	Optional[str] value used by the render_celestial_map workflow. Defaults to None.	None
tile_subdomains	Optional[str]	Optional[str] value used by the render_celestial_map workflow. Defaults to None.	None
location	Optional[str]	Optional[str] value used by the render_celestial_map workflow. Defaults to None.	None
zoom	Optional[int]	Optional[int] value used by the render_celestial_map workflow. Defaults to None.	None

Source code in app.py

def render_celestial_map( asset_root: str = 'assets/starmap', height: int = 1400,
		latitude: Optional[ float ] = None, longitude: Optional[ float ] = None,
		tile_url: Optional[ str ] = None, tile_attribution: Optional[ str ] = None,
		tile_subdomains: Optional[ str ] = None, location: Optional[ str ] = None,
		zoom: Optional[ int ] = None ) -> None:
	"""Render the celestial map interface section.

	Purpose:
		Supports the Mappy Streamlit application by executing the render celestial map
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		asset_root: str value used by the `render_celestial_map` workflow. Defaults to
			`'assets/starmap'`.
		height: int value used by the `render_celestial_map` workflow. Defaults to `1400`.
		latitude: Optional[float] value used by the `render_celestial_map` workflow. Defaults to
			`None`.
		longitude: Optional[float] value used by the `render_celestial_map` workflow. Defaults to
			`None`.
		tile_url: Optional[str] value used by the `render_celestial_map` workflow. Defaults to
			`None`.
		tile_attribution: Optional[str] value used by the `render_celestial_map` workflow. Defaults to
			`None`.
		tile_subdomains: Optional[str] value used by the `render_celestial_map` workflow. Defaults to
			`None`.
		location: Optional[str] value used by the `render_celestial_map` workflow. Defaults to
			`None`.
		zoom: Optional[int] value used by the `render_celestial_map` workflow. Defaults to
			`None`.
	"""
	try:
		root = Path( asset_root )
		index_path = root / 'index.html'
		style_path = root / 'style.css'
		data_paths = {
				'constellations': root / 'data' / 'constellations.json',
				'lines': root / 'data' / 'constellations.lines.json',
				'stars': root / 'data' / 'stars.6.json',
				'dsos': root / 'data' / 'dsos.bright.json',
				'starnames': root / 'data' / 'starnames.json',
				'planets': root / 'data' / 'planets.json',
				'mw': root / 'data' / 'mw.json',
				'constellationBorders': root / 'data' / 'constellations.borders.json',
				'dsonames': root / 'data' / 'dsonames.json',
		}

		module_paths = [
				root / 'js' / 'modules' / 'CelestialMath.js',
				root / 'js' / 'modules' / 'StarData.js',
				root / 'js' / 'modules' / 'MapRenderer.js',
				root / 'js' / 'modules' / 'LocationPicker.js',
				root / 'js' / 'modules' / 'UIController.js',
				root / 'js' / 'modules' / 'StarDetailsPanel.js',
				root / 'js' / 'modules' / 'ImageExporter.js',
				root / 'js' / 'main.js', ]

		required_paths = [ index_path, style_path, *data_paths.values( ), *module_paths ]
		missing_paths = [ str( path ) for path in required_paths if not path.exists( ) ]
		if missing_paths:
			st.error( 'Celestial Map assets are missing.' )
			st.code( '\n'.join( missing_paths ) )
			return

		html = index_path.read_text( encoding='utf-8' )
		css = style_path.read_text( encoding='utf-8' )
		local_data: Dict[ str, object ] = { }
		for key, path in data_paths.items( ):
			local_data[ key ] = json.loads( path.read_text( encoding='utf-8' ) )

		if has_valid_coordinates( latitude, longitude ):
			default_latitude = float( latitude )
			default_longitude = float( longitude )
		else:
			default_latitude = get_default_latitude( )
			default_longitude = get_default_longitude( )

		if location is not None and str( location ).strip( ):
			default_location = str( location ).strip( )
		else:
			default_location = get_default_location( )

		default_zoom = int( zoom if zoom is not None else st.session_state.get( 'zoom', 8 ) or 8 )
		default_payload = {
				'latitude': default_latitude,
				'longitude': default_longitude,
				'location': default_location,
				'zoom': default_zoom,
				'tileUrl': (
						'https://{s}.basemaps.cartocdn.com/dark_all/{z}/{x}/{y}{r}.png'
				),
				'tileAttribution': (
						'© OpenStreetMap contributors © CARTO'
				),
				'tileSubdomains': 'abcd',
		}

		html = html.replace(
			'<link rel="stylesheet" href="style.css">',
			f'<style>\n{css}\n</style>' )

		for relative_path in [
				'js/modules/CelestialMath.js',
				'js/modules/StarData.js',
				'js/modules/MapRenderer.js',
				'js/modules/LocationPicker.js',
				'js/modules/UIController.js',
				'js/modules/StarDetailsPanel.js',
				'js/modules/ImageExporter.js',
				'js/main.js',
		]:
			html = re.sub( rf'\s*<script\s+src="{re.escape( relative_path )}"></script>',
				'', html, flags=re.IGNORECASE )

		data_script = ('<script>\n'
		               'window.StarMapApp = window.StarMapApp || {};\n'
		               f'window.StarMapApp.LOCAL_DATA = {json.dumps( local_data )};\n'
		               f'window.StarMapApp.DEFAULT_LOCATION = {json.dumps( default_payload )};\n'
		               '</script>')

		inline_scripts = [ data_script ]
		for path in module_paths:
			source = path.read_text( encoding='utf-8' )
			if path.name == 'main.js':
				source = source.replace(
					"this.locationPicker = new LocationPicker('locationPicker', {\n"
					"                onLocationChange: (lat, lon) => this.handleLocationChange(lat, lon)\n"
					"            });",
					"const defaultLocation = window.StarMapApp.DEFAULT_LOCATION || {};\n"
					"            this.locationPicker = new LocationPicker('locationPicker', {\n"
					"                initialLat: defaultLocation.latitude,\n"
					"                initialLon: defaultLocation.longitude,\n"
					"                onLocationChange: (lat, lon) => this.handleLocationChange(lat, lon)\n"
					"            });" )

				source = source.replace(
					"this.selectedCoords = { lat: 30.0444, lon: 31.2357 };",
					"this.selectedCoords = {\n"
					"                lat: Number(defaultLocation.latitude || 30.0444),\n"
					"                lon: Number(defaultLocation.longitude || 31.2357)\n"
					"            };" )

				source = source.replace(
					"this.locationPicker.setView(this.selectedCoords.lat, this.selectedCoords.lon);",
					"this.locationPicker.setView(\n"
					"                    this.selectedCoords.lat,\n"
					"                    this.selectedCoords.lon,\n"
					"                    Number(defaultLocation.zoom || 8)\n"
					"                );\n"
					"                this.ui.updateLocationDisplay(\n"
					"                    this.selectedCoords.lat,\n"
					"                    this.selectedCoords.lon\n"
					"                );" )

			inline_scripts.append( f'<script>\n{source}\n</script>' )

		local_data_override = """
		<script>
		(function () {
		    window.StarMapApp = window.StarMapApp || {};

		    if (!window.StarMapApp.StarData || !window.StarMapApp.LOCAL_DATA) {
		        return;
		    }

		    window.StarMapApp.StarData.prototype.load = async function () {
		        this.data = window.StarMapApp.LOCAL_DATA;
		        return this.data;
		    };
		})();
		</script>
		"""

		inline_scripts.append( local_data_override )
		html = html.replace( '</body>', '\n'.join( inline_scripts ) + '\n</body>' )
		components.html( html, height=int( height ), scrolling=True )
	except Exception as ex:
		st.error( f'Celestial Map failed to render: {ex}' )

has_valid_coordinates

has_valid_coordinates(
    latitude: object, longitude: object
) -> bool

Handle the has valid coordinates workflow.

Purpose

Supports the Mappy Streamlit application by executing the has valid coordinates workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
latitude	object	object value used by the has_valid_coordinates workflow.	required
longitude	object	object value used by the has_valid_coordinates workflow.	required

Returns:

Name	Type	Description
bool	bool	Result produced by the has_valid_coordinates workflow.

Source code in app.py

def has_valid_coordinates( latitude: object, longitude: object ) -> bool:
	"""Handle the has valid coordinates workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the has valid coordinates
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		latitude: object value used by the `has_valid_coordinates` workflow.
		longitude: object value used by the `has_valid_coordinates` workflow.

	Returns:
		bool: Result produced by the `has_valid_coordinates` workflow.
	"""
	try:
		throw_if( 'latitude', latitude )
		throw_if( 'longitude', longitude )
		lat_value = float( latitude )
		lon_value = float( longitude )
	except Exception:
		return False

	if not (-90.0 <= lat_value <= 90.0):
		return False

	if not (-180.0 <= lon_value <= 180.0):
		return False

	if lat_value == 0.0 and lon_value == 0.0:
		return False

	return True

has_valid_global_coordinates

has_valid_global_coordinates() -> bool

Handle the has valid global coordinates workflow.

Purpose

Supports the Mappy Streamlit application by executing the has valid global coordinates workflow. The function preserves the existing UI behavior, session- state interactions, dataframe handling, database access, and service integrations defined by the application code.

Returns:

Name	Type	Description
bool	bool	Result produced by the has_valid_global_coordinates workflow.

Source code in app.py

def has_valid_global_coordinates( ) -> bool:
	"""Handle the has valid global coordinates workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the has valid global
		coordinates workflow. The function preserves the existing UI behavior, session-
		state interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Returns:
		bool: Result produced by the `has_valid_global_coordinates` workflow.
	"""
	return has_valid_coordinates(
		st.session_state.get( 'latitude', None ),
		st.session_state.get( 'longitude', None ) )

set_coordinates

set_coordinates(
    latitude: object, longitude: object
) -> None

Set the coordinates used by the application.

Purpose

Supports the Mappy Streamlit application by executing the set coordinates workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
latitude	object	object value used by the set_coordinates workflow.	required
longitude	object	object value used by the set_coordinates workflow.	required

Source code in app.py

def set_coordinates( latitude: object, longitude: object ) -> None:
	"""Set the coordinates used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the set coordinates
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		latitude: object value used by the `set_coordinates` workflow.
		longitude: object value used by the `set_coordinates` workflow.
	"""
	if not has_valid_coordinates( latitude, longitude ):
		return

	lat_value = float( latitude )
	lon_value = float( longitude )

	st.session_state[ 'latitude' ] = lat_value
	st.session_state[ 'longitude' ] = lon_value
	st.session_state[ 'coordinates' ] = (lat_value, lon_value)

get_location_state

get_location_state() -> Dict[str, object]

Return the location state used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get location state workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Returns:

Type	Description
Dict[str, object]	Dict[str, object]: Result produced by the get_location_state workflow.

Source code in app.py

def get_location_state( ) -> Dict[ str, object ]:
	"""Return the location state used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get location state
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Returns:
		Dict[str, object]: Result produced by the `get_location_state` workflow.
	"""
	return { 'coordinates': st.session_state.get( 'coordinates', ( ) ),
	         'latitude': st.session_state.get( 'latitude', 0.0 ),
	         'longitude': st.session_state.get( 'longitude', 0.0 ),
	         'location': st.session_state.get( 'location', '' ),
	         'country': st.session_state.get( 'country', '' ),
	         'state': st.session_state.get( 'state', '' ),
	         'city': st.session_state.get( 'city', '' ),
	         'zipcode': st.session_state.get( 'zipcode', '' ),
	         'description': st.session_state.get( 'description', '' ),
	         'origin': st.session_state.get( 'origin', '' ),
	         'destination': st.session_state.get( 'destination', '' ),
	         'radius': st.session_state.get( 'radius', 25.0 ),
	         'zoom': st.session_state.get( 'zoom', 8 ),
	         'map_size': st.session_state.get( 'map_size', '600x400' ),
	         'year': st.session_state.get( 'year', dt.datetime.now( ).year ),
	         'month': st.session_state.get( 'month', dt.datetime.now( ).month ),
	         'day': st.session_state.get( 'day', dt.datetime.now( ).day ),
	         'calendar_date': st.session_state.get( 'calendar_date', dt.date.today( ) ),
	         }

set_location_state

set_location_state(
    location: Optional[str] = None,
    city: Optional[str] = None,
    state: Optional[str] = None,
    country: Optional[str] = None,
    zipcode: Optional[str] = None,
    description: Optional[str] = None,
    latitude: Optional[float] = None,
    longitude: Optional[float] = None,
) -> None

Set the location state used by the application.

Purpose

Supports the Mappy Streamlit application by executing the set location state workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
location	Optional[str]	Optional[str] value used by the set_location_state workflow. Defaults to None.	None
city	Optional[str]	Optional[str] value used by the set_location_state workflow. Defaults to None.	None
state	Optional[str]	Optional[str] value used by the set_location_state workflow. Defaults to None.	None
country	Optional[str]	Optional[str] value used by the set_location_state workflow. Defaults to None.	None
zipcode	Optional[str]	Optional[str] value used by the set_location_state workflow. Defaults to None.	None
description	Optional[str]	Optional[str] value used by the set_location_state workflow. Defaults to None.	None
latitude	Optional[float]	Optional[float] value used by the set_location_state workflow. Defaults to None.	None
longitude	Optional[float]	Optional[float] value used by the set_location_state workflow. Defaults to None.	None

Source code in app.py

def set_location_state( location: Optional[ str ] = None, city: Optional[ str ] = None,
		state: Optional[ str ] = None, country: Optional[ str ] = None,
		zipcode: Optional[ str ] = None, description: Optional[ str ] = None,
		latitude: Optional[ float ] = None, longitude: Optional[ float ] = None ) -> None:
	"""Set the location state used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the set location state
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		location: Optional[str] value used by the `set_location_state` workflow. Defaults to
			`None`.
		city: Optional[str] value used by the `set_location_state` workflow. Defaults to
			`None`.
		state: Optional[str] value used by the `set_location_state` workflow. Defaults to
			`None`.
		country: Optional[str] value used by the `set_location_state` workflow. Defaults to
			`None`.
		zipcode: Optional[str] value used by the `set_location_state` workflow. Defaults to
			`None`.
		description: Optional[str] value used by the `set_location_state` workflow. Defaults to
			`None`.
		latitude: Optional[float] value used by the `set_location_state` workflow. Defaults to
			`None`.
		longitude: Optional[float] value used by the `set_location_state` workflow. Defaults to
			`None`.
	"""
	if location is not None:
		st.session_state[ 'location' ] = str( location ).strip( )

	if city is not None:
		st.session_state[ 'city' ] = str( city ).strip( )

	if state is not None:
		st.session_state[ 'state' ] = str( state ).strip( )

	if country is not None:
		st.session_state[ 'country' ] = str( country ).strip( )

	if zipcode is not None:
		st.session_state[ 'zipcode' ] = str( zipcode ).strip( )

	if description is not None:
		st.session_state[ 'description' ] = str( description ).strip( )

	if latitude is not None and longitude is not None:
		set_coordinates( latitude, longitude )

compose_location_from_state

compose_location_from_state() -> str

Handle the compose location from state workflow.

Purpose

Supports the Mappy Streamlit application by executing the compose location from state workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Returns:

Name	Type	Description
str	str	Result produced by the compose_location_from_state workflow.

Source code in app.py

def compose_location_from_state( ) -> str:
	"""Handle the compose location from state workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the compose location from
		state workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Returns:
		str: Result produced by the `compose_location_from_state` workflow.
	"""
	location = str( st.session_state.get( 'location', '' ) ).strip( )
	if location:
		return location

	parts = [ ]

	for key in [ 'city', 'state', 'zipcode', 'country' ]:
		value = st.session_state.get( key, '' )
		text = str( value ).strip( )

		if text and text.lower( ) not in [ 'nan', 'none', 'null' ]:
			parts.append( text )

	return ', '.join( parts )

resolve_table_name

resolve_table_name(
    requested: str, tables: List[str]
) -> Optional[str]

Handle the resolve table name workflow.

Purpose

Supports the Mappy Streamlit application by executing the resolve table name workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
requested	str	str value used by the resolve_table_name workflow.	required
tables	List[str]	List[str] value used by the resolve_table_name workflow.	required

Returns:

Type	Description
Optional[str]	Optional[str]: Result produced by the resolve_table_name workflow.

Source code in app.py

def resolve_table_name( requested: str, tables: List[ str ] ) -> Optional[ str ]:
	"""Handle the resolve table name workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the resolve table name
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		requested: str value used by the `resolve_table_name` workflow.
		tables: List[str] value used by the `resolve_table_name` workflow.

	Returns:
		Optional[str]: Result produced by the `resolve_table_name` workflow.
	"""
	if not requested or not tables:
		return None

	for table in tables:
		if table == requested:
			return table

	requested_lower = requested.lower( )
	for table in tables:
		if str( table ).lower( ) == requested_lower:
			return table

	return None

get_default_location

get_default_location(
    fallback: str = "Washington, DC",
) -> str

Return the default location used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get default location workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
fallback	str	str value used by the get_default_location workflow. Defaults to 'Washington, DC'.	'Washington, DC'

Returns:

Name	Type	Description
str	str	Result produced by the get_default_location workflow.

Source code in app.py

def get_default_location( fallback: str = 'Washington, DC' ) -> str:
	"""Return the default location used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get default location
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		fallback: str value used by the `get_default_location` workflow. Defaults to
			`'Washington, DC'`.

	Returns:
		str: Result produced by the `get_default_location` workflow.
	"""
	location = compose_location_from_state( )
	if location:
		return location

	return fallback

get_default_zipcode

get_default_zipcode(fallback: str = '20001') -> str

Return the default zipcode used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get default zipcode workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
fallback	str	str value used by the get_default_zipcode workflow. Defaults to '20001'.	'20001'

Returns:

Name	Type	Description
str	str	Result produced by the get_default_zipcode workflow.

Source code in app.py

def get_default_zipcode( fallback: str = '20001' ) -> str:
	"""Return the default zipcode used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get default zipcode
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		fallback: str value used by the `get_default_zipcode` workflow. Defaults to `'20001'`.

	Returns:
		str: Result produced by the `get_default_zipcode` workflow.
	"""
	zipcode = str( st.session_state.get( 'zipcode', '' ) ).strip( )
	if zipcode:
		return zipcode

	return fallback

get_default_latitude

get_default_latitude(fallback: float = 38.9072) -> float

Return the default latitude used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get default latitude workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
fallback	float	float value used by the get_default_latitude workflow. Defaults to 38.9072.	38.9072

Returns:

Name	Type	Description
float	float	Result produced by the get_default_latitude workflow.

Source code in app.py

def get_default_latitude( fallback: float = 38.907200 ) -> float:
	"""Return the default latitude used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get default latitude
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		fallback: float value used by the `get_default_latitude` workflow. Defaults to
			`38.9072`.

	Returns:
		float: Result produced by the `get_default_latitude` workflow.
	"""
	if has_valid_global_coordinates( ):
		return float( st.session_state.get( 'latitude', fallback ) )

	return float( fallback )

get_default_longitude

get_default_longitude(fallback: float = -77.0369) -> float

Return the default longitude used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get default longitude workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
fallback	float	float value used by the get_default_longitude workflow. Defaults to -77.0369.	-77.0369

Returns:

Name	Type	Description
float	float	Result produced by the get_default_longitude workflow.

Source code in app.py

def get_default_longitude( fallback: float = -77.036900 ) -> float:
	"""Return the default longitude used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get default longitude
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		fallback: float value used by the `get_default_longitude` workflow. Defaults to
			`-77.0369`.

	Returns:
		float: Result produced by the `get_default_longitude` workflow.
	"""
	if has_valid_global_coordinates( ):
		return float( st.session_state.get( 'longitude', fallback ) )

	return float( fallback )

set_global_coordinates_from_result

set_global_coordinates_from_result(
    latitude: object,
    longitude: object,
    location: Optional[str] = None,
    description: Optional[str] = None,
) -> None

Set the global coordinates from result used by the application.

Purpose

Supports the Mappy Streamlit application by executing the set global coordinates from result workflow. The function preserves the existing UI behavior, session- state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
latitude	object	object value used by the set_global_coordinates_from_result workflow.	required
longitude	object	object value used by the set_global_coordinates_from_result workflow.	required
location	Optional[str]	Optional[str] value used by the set_global_coordinates_from_result workflow. Defaults to None.	None
description	Optional[str]	Optional[str] value used by the set_global_coordinates_from_result workflow. Defaults to None.	None

Source code in app.py

def set_global_coordinates_from_result( latitude: object, longitude: object,
		location: Optional[ str ] = None, description: Optional[ str ] = None ) -> None:
	"""Set the global coordinates from result used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the set global coordinates
		from result workflow. The function preserves the existing UI behavior, session-
		state interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		latitude: object value used by the `set_global_coordinates_from_result` workflow.
		longitude: object value used by the `set_global_coordinates_from_result` workflow.
		location: Optional[str] value used by the `set_global_coordinates_from_result` workflow.
			Defaults to `None`.
		description: Optional[str] value used by the `set_global_coordinates_from_result` workflow.
			Defaults to `None`.
	"""
	if not has_valid_coordinates( latitude, longitude ):
		return

	set_location_state(
		location=location,
		description=description,
		latitude=float( latitude ),
		longitude=float( longitude ) )

create_bounding_box_from_center

create_bounding_box_from_center(
    latitude: object,
    longitude: object,
    delta: float = 0.125,
) -> Dict[str, float]

Create the bounding box from center used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create bounding box from center workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
latitude	object	object value used by the create_bounding_box_from_center workflow.	required
longitude	object	object value used by the create_bounding_box_from_center workflow.	required
delta	float	float value used by the create_bounding_box_from_center workflow. Defaults to 0.125.	0.125

Returns:

Type	Description
Dict[str, float]	Dict[str, float]: Result produced by the create_bounding_box_from_center workflow.

Source code in app.py

def create_bounding_box_from_center( latitude: object, longitude: object,
		delta: float = 0.125 ) -> Dict[ str, float ]:
	"""Create the bounding box from center used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create bounding box from
		center workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		latitude: object value used by the `create_bounding_box_from_center` workflow.
		longitude: object value used by the `create_bounding_box_from_center` workflow.
		delta: float value used by the `create_bounding_box_from_center` workflow. Defaults
			to `0.125`.

	Returns:
		Dict[str, float]: Result produced by the `create_bounding_box_from_center`
			workflow.
	"""
	lat_value = get_default_latitude( )
	lng_value = get_default_longitude( )

	if has_valid_coordinates( latitude, longitude ):
		lat_value = float( latitude )
		lng_value = float( longitude )

	return {
			'west': lng_value - float( delta ),
			'south': lat_value - float( delta ),
			'east': lng_value + float( delta ),
			'north': lat_value + float( delta ),
			'nw_lng': lng_value - float( delta ),
			'nw_lat': lat_value + float( delta ),
			'se_lng': lng_value + float( delta ),
			'se_lat': lat_value - float( delta ),
			'center_lat': lat_value,
			'center_lng': lng_value,
	}

get_geolocation_error_message

get_geolocation_error_message(
    error_code: object, error_message: object
) -> str

Return the geolocation error message used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get geolocation error message workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
error_code	object	object value used by the get_geolocation_error_message workflow.	required
error_message	object	object value used by the get_geolocation_error_message workflow.	required

Returns:

Name	Type	Description
str	str	Result produced by the get_geolocation_error_message workflow.

Source code in app.py

def get_geolocation_error_message( error_code: object, error_message: object ) -> str:
	"""Return the geolocation error message used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get geolocation error
		message workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		error_code: object value used by the `get_geolocation_error_message` workflow.
		error_message: object value used by the `get_geolocation_error_message` workflow.

	Returns:
		str: Result produced by the `get_geolocation_error_message` workflow.
	"""
	try:
		code = int( error_code )
	except Exception:
		code = -1

	message = str( error_message or '' ).strip( )

	if code == 0:
		return f'Browser does not support geolocation. {message}'.strip( )

	if code == 1:
		return f'Browser location permission was denied. {message}'.strip( )

	if code == 2:
		return f'Browser position is unavailable. {message}'.strip( )

	if code == 3:
		return f'Browser location request timed out. {message}'.strip( )

	if message:
		return message

	return 'Browser geolocation failed.'

update_location_from_browser

update_location_from_browser(
    geo: Dict[str, object],
) -> bool

Handle the update location from browser workflow.

Purpose

Supports the Mappy Streamlit application by executing the update location from browser workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
geo	Dict[str, object]	Dict[str, object] value used by the update_location_from_browser workflow.	required

Returns:

Name	Type	Description
bool	bool	Result produced by the update_location_from_browser workflow.

Source code in app.py

def update_location_from_browser( geo: Dict[ str, object ] ) -> bool:
	"""Handle the update location from browser workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the update location from
		browser workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		geo: Dict[str, object] value used by the `update_location_from_browser` workflow.

	Returns:
		bool: Result produced by the `update_location_from_browser` workflow.
	"""
	try:
		if not geo:
			return False

		if 'error' in geo:
			error = geo.get( 'error', { } )
			error_code = None
			error_message = ''

			if isinstance( error, dict ):
				error_code = error.get( 'code', None )
				error_message = error.get( 'message', '' )

			st.session_state[ 'browser_geolocation_error' ] = get_geolocation_error_message(
				error_code,
				error_message )

			if error_code == 1:
				st.session_state[ 'browser_geolocation_permission_denied' ] = True

			st.session_state[ 'browser_geolocation_loaded' ] = False
			return False

		coords = geo.get( 'coords', { } )

		if not coords:
			st.session_state[ 'browser_geolocation_error' ] = (
					'Browser geolocation did not return coordinates.')
			return False

		latitude = coords.get( 'latitude', None )
		longitude = coords.get( 'longitude', None )
		accuracy = coords.get( 'accuracy', None )

		if not has_valid_coordinates( latitude, longitude ):
			st.session_state[ 'browser_geolocation_error' ] = (
					'Browser geolocation returned invalid coordinates.')
			return False

		set_location_state(
			description=f'Browser geolocation. Accuracy: {accuracy} meters.',
			latitude=float( latitude ),
			longitude=float( longitude ) )

		st.session_state[ 'browser_geolocation' ] = geo
		st.session_state[ 'browser_geolocation_loaded' ] = True
		st.session_state[ 'browser_geolocation_permission_denied' ] = False
		st.session_state[ 'browser_geolocation_error' ] = ''

		return True

	except Exception as ex:
		st.session_state[ 'browser_geolocation_error' ] = str( ex )
		return False

bootstrap_browser_geolocation

bootstrap_browser_geolocation(geocoder: Geocoder) -> None

Handle the bootstrap browser geolocation workflow.

Purpose

Supports the Mappy Streamlit application by executing the bootstrap browser geolocation workflow. The function preserves the existing UI behavior, session- state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
geocoder	Geocoder	Geocoder value used by the bootstrap_browser_geolocation workflow.	required

Source code in app.py

def bootstrap_browser_geolocation( geocoder: Geocoder ) -> None:
	"""Handle the bootstrap browser geolocation workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the bootstrap browser
		geolocation workflow. The function preserves the existing UI behavior, session-
		state interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		geocoder: Geocoder value used by the `bootstrap_browser_geolocation` workflow.
	"""
	try:
		if not st.session_state.get( 'browser_geolocation_enabled', True ):
			return

		if st.session_state.get( 'browser_geolocation_loaded', False ):
			return

		if st.session_state.get( 'browser_geolocation_permission_denied', False ):
			return

		geo_payload = get_geolocation( )

		if not geo_payload:
			return

		updated = update_location_from_browser( geo_payload )

		if not updated:
			return

		if not st.session_state.get( 'browser_geolocation_reverse_geocoded', False ):
			latitude = st.session_state.get( 'latitude', None )
			longitude = st.session_state.get( 'longitude', None )

			if geocoder is not None and has_valid_coordinates( latitude, longitude ):
				try:
					location_text = f'{float( latitude ):.6f},{float( longitude ):.6f}'
					geocoder.freeform( location_text )

					if not st.session_state.get( 'location', '' ):
						st.session_state[ 'location' ] = location_text

					st.session_state[ 'browser_geolocation_reverse_geocoded' ] = True

				except Exception:
					st.session_state[ 'location' ] = (
							f'{float( latitude ):.6f},{float( longitude ):.6f}')

		st.rerun( )

	except Exception as ex:
		st.session_state[ 'browser_geolocation_error' ] = str( ex )

create_reports_map

create_reports_map(
    df: DataFrame,
    df_overlay: Optional[DataFrame] = None,
    use_user_location: bool = True,
    key_prefix: str = "reports_map",
) -> None

Create the reports map used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create reports map workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df	DataFrame	pd.DataFrame value used by the create_reports_map workflow.	required
df_overlay	Optional[DataFrame]	Optional[pd.DataFrame] value used by the create_reports_map workflow. Defaults to None.	None
use_user_location	bool	bool value used by the create_reports_map workflow. Defaults to True.	True
key_prefix	str	str value used by the create_reports_map workflow. Defaults to 'reports_map'.	'reports_map'

Source code in app.py

def create_reports_map( df: pd.DataFrame, df_overlay: Optional[ pd.DataFrame ] = None,
		use_user_location: bool = True, key_prefix: str = 'reports_map' ) -> None:
	"""Create the reports map used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create reports map
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		df: pd.DataFrame value used by the `create_reports_map` workflow.
		df_overlay: Optional[pd.DataFrame] value used by the `create_reports_map` workflow.
			Defaults to `None`.
		use_user_location: bool value used by the `create_reports_map` workflow. Defaults to `True`.
		key_prefix: str value used by the `create_reports_map` workflow. Defaults to
			`'reports_map'`.
	"""
	try:
		st.subheader( 'Locations' )
		df_source = pd.DataFrame( ) if df is None else df.copy( )
		df_overlay_source = pd.DataFrame( ) if df_overlay is None else df_overlay.copy( )
		required_cols = [ 'Latitude', 'Longitude' ]

		if not df_source.empty:
			missing_cols = [ col for col in required_cols if col not in df_source.columns ]
			if missing_cols:
				st.warning( f'Map requires missing column(s): {", ".join( missing_cols )}' )
				return

		if not df_overlay_source.empty:
			overlay_missing_cols = [
					col for col in required_cols if col not in df_overlay_source.columns ]

			if overlay_missing_cols:
				df_overlay_source = pd.DataFrame( )

		total_count = len( df_source )
		df_base_map = pd.DataFrame( )

		if not df_source.empty:
			df_source[ 'Latitude' ] = pd.to_numeric( df_source[ 'Latitude' ], errors='coerce' )
			df_source[ 'Longitude' ] = pd.to_numeric( df_source[ 'Longitude' ], errors='coerce' )

			base_mask = (df_source[ 'Latitude' ].notna( )
			             & df_source[ 'Longitude' ].notna( )
			             & df_source[ 'Latitude' ].between( -90.0, 90.0 )
			             & df_source[ 'Longitude' ].between( -180.0, 180.0 )
			             & ~((df_source[ 'Latitude' ] == 0.0)
			                 & (df_source[ 'Longitude' ] == 0.0)))

			df_base_map = df_source.loc[ base_mask ].copy( )

		mapped_count = len( df_base_map )
		missing_count = total_count - mapped_count
		df_overlay_map = pd.DataFrame( )

		if not df_overlay_source.empty:
			df_overlay_source[ 'Latitude' ] = pd.to_numeric( df_overlay_source[ 'Latitude' ],
				errors='coerce' )

			df_overlay_source[ 'Longitude' ] = pd.to_numeric( df_overlay_source[ 'Longitude' ],
				errors='coerce' )

			overlay_mask = (df_overlay_source[ 'Latitude' ].notna( )
			                & df_overlay_source[ 'Longitude' ].notna( )
			                & df_overlay_source[ 'Latitude' ].between( -90.0, 90.0 )
			                & df_overlay_source[ 'Longitude' ].between( -180.0, 180.0 )
			                & ~((df_overlay_source[ 'Latitude' ] == 0.0)
			                    & (df_overlay_source[ 'Longitude' ] == 0.0)))

			df_overlay_map = df_overlay_source.loc[ overlay_mask ].copy( )

		df_user_map = pd.DataFrame( )

		if use_user_location and df_base_map.empty and df_overlay_map.empty:
			user_latitude = st.session_state.get( 'latitude', None )
			user_longitude = st.session_state.get( 'longitude', None )

			if has_valid_coordinates( user_latitude, user_longitude ):
				user_location = compose_location_from_state( )
				user_description = st.session_state.get( 'description', '' )

				if not user_location:
					user_location = (f'{float( user_latitude ):.4f},'
					                 f'{float( user_longitude ):.4f}')

				df_user_map = pd.DataFrame( [ {
						'ID': 'USER-LOCATION',
						'CalendarDate': dt.datetime.now( ).strftime( '%Y-%m-%d %H:%M:%S' ),
						'City': user_location,
						'State': '',
						'Country': '',
						'Latitude': float( user_latitude ),
						'Longitude': float( user_longitude ),
						'Shape': 'User Location',
						'Summary': (user_description
						            or 'Current user location fallback.'), } ] )

		metric_c1, metric_c2, metric_c3, metric_c4 = st.columns( 4, border=True )
		metric_c1.metric( 'Total Records', f'{total_count:,}' )
		metric_c2.metric( 'Mapped Records', f'{mapped_count:,}' )
		metric_c3.metric( 'Missing / Invalid Coordinates', f'{missing_count:,}' )
		metric_c4.metric( 'Overlay / User Records',
			f'{len( df_overlay_map ) + len( df_user_map ):,}' )

		if df_base_map.empty and df_overlay_map.empty and df_user_map.empty:
			st.info( 'No usable base, overlay, or user-location coordinates are available.' )
			return

		if not df_base_map.empty:
			filter_c1, filter_c2, filter_c3, filter_c4 = st.columns(
				[ 0.25, 0.25, 0.25, 0.25 ], border=True )

			with filter_c1:
				if 'Year' in df_base_map.columns:
					year_options = sorted(
						[ str( value ) for value in df_base_map[ 'Year' ].dropna( ).unique( ) ] )
					selected_years = st.multiselect( 'Year', year_options,
						key=f'{key_prefix}_years' )

					if selected_years:
						df_base_map = df_base_map[
							df_base_map[ 'Year' ].astype( str ).isin( selected_years ) ]

			with filter_c2:
				if 'Country' in df_base_map.columns:
					country_options = sorted(
						[ str( value ) for value in df_base_map[ 'Country' ].dropna( ).unique( ) ] )
					selected_countries = st.multiselect( 'Country', country_options,
						key=f'{key_prefix}_countries' )

					if selected_countries:
						df_base_map = df_base_map[
							df_base_map[ 'Country' ].astype( str ).isin( selected_countries ) ]

			with filter_c3:
				if 'State' in df_base_map.columns:
					state_options = sorted(
						[ str( value ) for value in df_base_map[ 'State' ].dropna( ).unique( ) ] )
					selected_states = st.multiselect( 'State', state_options,
						key=f'{key_prefix}_states' )

					if selected_states:
						df_base_map = df_base_map[
							df_base_map[ 'State' ].astype( str ).isin( selected_states ) ]

			with filter_c4:
				if 'Shape' in df_base_map.columns:
					shape_options = sorted(
						[ str( value ) for value in df_base_map[ 'Shape' ].dropna( ).unique( ) ] )
					selected_shapes = st.multiselect( 'Shape', shape_options,
						key=f'{key_prefix}_shapes' )

					if selected_shapes:
						df_base_map = df_base_map[
							df_base_map[ 'Shape' ].astype( str ).isin( selected_shapes ) ]

		if df_base_map.empty and df_overlay_map.empty and df_user_map.empty:
			st.info( 'No mapped records match the selected filters.' )
			return

		map_style_options = {
				'Carto Positron': 'https://basemaps.cartocdn.com/gl/positron-gl-style/style.json',
				'Carto Dark Matter': 'https://basemaps.cartocdn.com/gl/dark-matter-gl-style/style.json',
				'Carto Voyager': 'https://basemaps.cartocdn.com/gl/voyager-gl-style/style.json',
				'Dark': 'dark',
				'Light': 'light',
				'Road': 'road',
				'Satellite': 'satellite',
				'Dark - No Labels': 'dark_no_labels',
				'Light - No Labels': 'light_no_labels',
				'Streamlit Theme': None,
		}

		control_c1, control_c2, control_c3, control_c4 = st.columns(
			[ 0.20, 0.20, 0.30, 0.30 ], border=True )

		with control_c1:
			zoom_level = st.slider( 'Initial Zoom', min_value=0, max_value=50, step=1,
				value=5, key=f'{key_prefix}_zoom' )

		with control_c2:
			point_radius = st.slider( 'Point Radius', min_value=1, max_value=50, value=1,
				step=5, key=f'{key_prefix}_radius' )

		with control_c3:
			selected_map_style = st.selectbox(
				'Map Style',
				list( map_style_options.keys( ) ),
				index=1,
				key=f'{key_prefix}_style' )

			map_style = map_style_options[ selected_map_style ]

		with control_c4:
			if len( df_base_map ) > 100:
				max_records = st.slider( 'Maximum Records', min_value=100,
					max_value=len( df_base_map ), value=min( 2500, len( df_base_map ) ),
					step=500, key=f'{key_prefix}_limit' )

				df_base_map = df_base_map.head( max_records )
			elif not df_base_map.empty:
				st.caption( f'Displaying all {len( df_base_map ):,} mapped base records.' )
			elif not df_overlay_map.empty:
				st.caption( f'Displaying {len( df_overlay_map ):,} overlay record(s).' )
			else:
				st.caption( 'Displaying current user/global location fallback.' )

		for df_target in [ df_base_map, df_overlay_map, df_user_map ]:
			if df_target.empty:
				continue

			for col in [ 'ID', 'CalendarDate', 'City', 'State', 'Country', 'Shape', 'Summary' ]:
				if col not in df_target.columns:
					df_target[ col ] = ''

			df_target[ 'MapSummary' ] = df_target[ 'Summary' ].astype( str ).str.slice( 0, 300 )
			df_target[ 'Position' ] = df_target.apply(
				lambda row: [ float( row[ 'Longitude' ] ), float( row[ 'Latitude' ] ) ],
				axis=1 )

		if not df_base_map.empty:
			df_view = df_base_map
		elif not df_overlay_map.empty:
			df_view = df_overlay_map
		else:
			df_view = df_user_map

		view_state = pdk.ViewState( latitude=float( df_view[ 'Latitude' ].median( ) ),
			longitude=float( df_view[ 'Longitude' ].median( ) ), zoom=zoom_level, pitch=0 )

		layers = [ ]

		if not df_base_map.empty:
			layers.append( pdk.Layer( 'ScatterplotLayer', data=df_base_map,
				get_position='Position', get_radius=point_radius,
				get_fill_color=[ 0, 120, 252, 160 ],
				get_line_color=[ 255, 255, 255, 180 ], line_width_min_pixels=1,
				radius_min_pixels=4, radius_max_pixels=24, filled=True,
				stroked=True, pickable=True ) )

		if not df_overlay_map.empty:
			layers.append( pdk.Layer( 'ScatterplotLayer', data=df_overlay_map,
				get_position='Position', get_radius=max( point_radius * 2, 20000 ),
				get_fill_color=[ 255, 80, 0, 220 ], get_line_color=[ 255, 255, 255, 255 ],
				line_width_min_pixels=2, radius_min_pixels=8, radius_max_pixels=40, filled=True,
				stroked=True, pickable=True ) )

		if not df_user_map.empty:
			layers.append( pdk.Layer( 'ScatterplotLayer', data=df_user_map,
				get_position='Position', get_radius=max( point_radius * 2, 20000 ),
				get_fill_color=[ 0, 180, 80, 230 ],
				get_line_color=[ 255, 255, 255, 255 ], line_width_min_pixels=2,
				radius_min_pixels=10, radius_max_pixels=42, filled=True, stroked=True,
				pickable=True ) )

		tooltip = {
				'html': (
						'<b>ID:</b> {ID}<br/>'
						'<b>Date:</b> {CalendarDate}<br/>'
						'<b>Location:</b> {City}, {State}, {Country}<br/>'
						'<b>Coordinates:</b> {Latitude}, {Longitude}<br/>'
						'<b>Shape:</b> {Shape}<br/>'
						'<b>Summary:</b> {MapSummary}'
				),
				'style': {
						'backgroundColor': 'rgba(0, 0, 0, 0.85)',
						'color': 'white',
						'fontSize': '12px',
				},
		}

		set_blue_divider( )

		deck = pdk.Deck( layers=layers, initial_view_state=view_state,
			map_style=map_style, tooltip=tooltip )

		st.pydeck_chart( deck, use_container_width=True )

		show_table = st.checkbox( 'Show mapped records table', value=False,
			key=f'{key_prefix}_show_table' )

		if show_table:
			if df_base_map.empty:
				if not df_user_map.empty:
					st.info( 'No base mapped records are available.' )
					st.data_editor( df_user_map, key=f'{key_prefix}_user_location_table',
						use_container_width=True, disabled=True )
				else:
					st.info( 'No base mapped records are available to display.' )

			else:
				display_cols = [ 'ID', 'Year', 'Month', 'Day', 'CalendarDate', 'City', 'State',
				                 'Country', 'Latitude', 'Longitude', 'Shape', 'Summary' ]
				display_cols = [ col for col in display_cols if col in df_base_map.columns ]

				st.data_editor( df_base_map[ display_cols ], key=f'{key_prefix}_table',
					use_container_width=True, disabled=True )

	except Exception as e:
		st.error( f'Interactive map failed: {e}' )

compose_location_query

compose_location_query(
    city: object, state: object, country: object
) -> str

Handle the compose location query workflow.

Purpose

Supports the Mappy Streamlit application by executing the compose location query workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
city	object	object value used by the compose_location_query workflow.	required
state	object	object value used by the compose_location_query workflow.	required
country	object	object value used by the compose_location_query workflow.	required

Returns:

Name	Type	Description
str	str	Result produced by the compose_location_query workflow.

Source code in app.py

def compose_location_query( city: object, state: object, country: object ) -> str:
	"""Handle the compose location query workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the compose location query
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		city: object value used by the `compose_location_query` workflow.
		state: object value used by the `compose_location_query` workflow.
		country: object value used by the `compose_location_query` workflow.

	Returns:
		str: Result produced by the `compose_location_query` workflow.
	"""
	parts = [ ]

	for value in [ city, state, country ]:
		if value is None:
			continue

		text = str( value ).strip( )
		if text and text.lower( ) not in [ 'nan', 'none', 'null' ]:
			parts.append( text )

	return ', '.join( parts )

get_value_by_path

get_value_by_path(
    obj: object, path: List[object]
) -> object

Return the value by path used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get value by path workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
obj	object	object value used by the get_value_by_path workflow.	required
path	List[object]	List[object] value used by the get_value_by_path workflow.	required

Returns:

Name	Type	Description
object	object	Result produced by the get_value_by_path workflow.

Source code in app.py

def get_value_by_path( obj: object, path: List[ object ] ) -> object:
	"""Return the value by path used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get value by path
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		obj: object value used by the `get_value_by_path` workflow.
		path: List[object] value used by the `get_value_by_path` workflow.

	Returns:
		object: Result produced by the `get_value_by_path` workflow.
	"""
	current = obj

	for key in path:
		if current is None:
			return None

		if isinstance( current, dict ):
			current = current.get( key )
			continue

		if isinstance( current, (list, tuple) ) and isinstance( key, int ):
			if 0 <= key < len( current ):
				current = current[ key ]
				continue
			return None

		if hasattr( current, str( key ) ):
			current = getattr( current, str( key ) )
			continue

		return None

	return current

extract_coordinates

extract_coordinates(
    result: object,
) -> Tuple[Optional[float], Optional[float]]

Handle the extract coordinates workflow.

Purpose

Supports the Mappy Streamlit application by executing the extract coordinates workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
result	object	object value used by the extract_coordinates workflow.	required

Returns:

Type	Description
Tuple[Optional[float], Optional[float]]	Tuple[Optional[float], Optional[float]]: Result produced by the extract_coordinates workflow.

Source code in app.py

def extract_coordinates( result: object ) -> Tuple[ Optional[ float ], Optional[ float ] ]:
	"""Handle the extract coordinates workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the extract coordinates
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		result: object value used by the `extract_coordinates` workflow.

	Returns:
		Tuple[Optional[float], Optional[float]]: Result produced by the
			`extract_coordinates` workflow.
	"""
	if result is None:
		return None, None

	lat_paths = [
			[ 'lat' ],
			[ 'latitude' ],
			[ 'location', 'lat' ],
			[ 'location', 'latitude' ],
			[ 'geometry', 'location', 'lat' ],
			[ 'geometry', 'location', 'latitude' ],
			[ 'result', 'geometry', 'location', 'lat' ],
			[ 'results', 0, 'geometry', 'location', 'lat' ],
			[ 'candidates', 0, 'geometry', 'location', 'lat' ],
	]

	lon_paths = [
			[ 'lng' ],
			[ 'lon' ],
			[ 'longitude' ],
			[ 'location', 'lng' ],
			[ 'location', 'lon' ],
			[ 'location', 'longitude' ],
			[ 'geometry', 'location', 'lng' ],
			[ 'geometry', 'location', 'lon' ],
			[ 'geometry', 'location', 'longitude' ],
			[ 'result', 'geometry', 'location', 'lng' ],
			[ 'results', 0, 'geometry', 'location', 'lng' ],
			[ 'candidates', 0, 'geometry', 'location', 'lng' ],
	]

	lat_value = None
	lon_value = None

	for path in lat_paths:
		lat_value = get_value_by_path( result, path )
		if lat_value is not None:
			break

	for path in lon_paths:
		lon_value = get_value_by_path( result, path )
		if lon_value is not None:
			break

	try:
		latitude = float( lat_value ) if lat_value is not None else None
		longitude = float( lon_value ) if lon_value is not None else None
	except Exception:
		return None, None

	if not has_valid_coordinates( latitude, longitude ):
		return None, None

	return latitude, longitude

count_missing_report_coordinate_rows

count_missing_report_coordinate_rows(
    table_name: str,
) -> int

Handle the count missing report coordinate rows workflow.

Purpose

Supports the Mappy Streamlit application by executing the count missing report coordinate rows workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table_name	str	str value used by the count_missing_report_coordinate_rows workflow.	required

Returns:

Name	Type	Description
int	int	Result produced by the count_missing_report_coordinate_rows workflow.

Source code in app.py

def count_missing_report_coordinate_rows( table_name: str ) -> int:
	"""Handle the count missing report coordinate rows workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the count missing report
		coordinate rows workflow. The function preserves the existing UI behavior,
		session-state interactions, dataframe handling, database access, and service
		integrations defined by the application code.

	Args:
		table_name: str value used by the `count_missing_report_coordinate_rows` workflow.

	Returns:
		int: Result produced by the `count_missing_report_coordinate_rows` workflow.
	"""
	try:
		throw_if( 'table_name', table_name )
		query = f"""
			SELECT COUNT(*) AS RowCount
			FROM "{table_name}"
			WHERE (
			          Latitude IS NULL
			          OR Longitude IS NULL
			          OR Latitude < -90
			          OR Latitude > 90
			          OR Longitude < -180
			          OR Longitude > 180
			          OR (Latitude = 0 AND Longitude = 0)
			      )
			  AND City IS NOT NULL
			  AND TRIM(City) <> ''
			  AND Country IS NOT NULL
			  AND TRIM(Country) <> '';
		"""
		with create_connection( ) as conn:
			row = conn.execute( query ).fetchone( )
			return int( row[ 0 ] or 0 )

	except Exception as e:
		st.error( f'Unable to count missing coordinate rows: {e}' )
		return 0

read_missing_report_locations

read_missing_report_locations(
    table_name: str, limit: Optional[int] = None
) -> DataFrame

Read the missing report locations for application processing.

Purpose

Supports the Mappy Streamlit application by executing the read missing report locations workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table_name	str	str value used by the read_missing_report_locations workflow.	required
limit	Optional[int]	Optional[int] value used by the read_missing_report_locations workflow. Defaults to None.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: Result produced by the read_missing_report_locations workflow.

Source code in app.py

def read_missing_report_locations( table_name: str, limit: Optional[ int ] = None ) -> pd.DataFrame:
	"""Read the missing report locations for application processing.

	Purpose:
		Supports the Mappy Streamlit application by executing the read missing report
		locations workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		table_name: str value used by the `read_missing_report_locations` workflow.
		limit: Optional[int] value used by the `read_missing_report_locations` workflow.
			Defaults to `None`.

	Returns:
		pd.DataFrame: Result produced by the `read_missing_report_locations` workflow.
	"""
	try:
		throw_if( 'table_name', table_name )
		query = f"""
			SELECT
			    TRIM(City) AS City,
			    TRIM(COALESCE(State, '')) AS State,
			    TRIM(Country) AS Country,
			    COUNT(*) AS RowCount
			FROM "{table_name}"
			WHERE (
			          Latitude IS NULL
			          OR Longitude IS NULL
			          OR Latitude < -90
			          OR Latitude > 90
			          OR Longitude < -180
			          OR Longitude > 180
			          OR (Latitude = 0 AND Longitude = 0)
			      )
			  AND City IS NOT NULL
			  AND TRIM(City) <> ''
			  AND Country IS NOT NULL
			  AND TRIM(Country) <> ''
			GROUP BY TRIM(City), TRIM(COALESCE(State, '')), TRIM(Country)
			ORDER BY RowCount DESC, City, State, Country
		"""
		if limit:
			query += f' LIMIT {int( limit )}'

		with create_connection( ) as conn:
			return pd.read_sql_query( query, conn )

	except Exception as e:
		st.error( f'Unable to read missing report locations: {e}' )
		return pd.DataFrame( )

preview_report_coordinate_updates

preview_report_coordinate_updates(
    table_name: str,
    geocoder: Geocoder,
    places: Place,
    use_places: bool = True,
    limit: Optional[int] = None,
) -> DataFrame

Handle the preview report coordinate updates workflow.

Purpose

Supports the Mappy Streamlit application by executing the preview report coordinate updates workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table_name	str	str value used by the preview_report_coordinate_updates workflow.	required
geocoder	Geocoder	Geocoder value used by the preview_report_coordinate_updates workflow.	required
places	Place	Place value used by the preview_report_coordinate_updates workflow.	required
use_places	bool	bool value used by the preview_report_coordinate_updates workflow. Defaults to True.	True
limit	Optional[int]	Optional[int] value used by the preview_report_coordinate_updates workflow. Defaults to None.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: Result produced by the preview_report_coordinate_updates workflow.

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def preview_report_coordinate_updates( table_name: str, geocoder: Geocoder, places: Place,
		use_places: bool = True, limit: Optional[ int ] = None ) -> pd.DataFrame:
	"""Handle the preview report coordinate updates workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the preview report
		coordinate updates workflow. The function preserves the existing UI behavior,
		session-state interactions, dataframe handling, database access, and service
		integrations defined by the application code.

	Args:
		table_name: str value used by the `preview_report_coordinate_updates` workflow.
		geocoder: Geocoder value used by the `preview_report_coordinate_updates` workflow.
		places: Place value used by the `preview_report_coordinate_updates` workflow.
		use_places: bool value used by the `preview_report_coordinate_updates` workflow. Defaults
			to `True`.
		limit: Optional[int] value used by the `preview_report_coordinate_updates` workflow.
			Defaults to `None`.

	Returns:
		pd.DataFrame: Result produced by the `preview_report_coordinate_updates`
			workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	try:
		throw_if( 'table_name', table_name )
		throw_if( 'geocoder', geocoder )

		df_locations = read_missing_report_locations( table_name, limit )

		if df_locations.empty:
			return pd.DataFrame( )

		records: List[ Dict[ str, object ] ] = [ ]

		for row in df_locations.itertuples( index=False ):
			query = compose_location_query( row.City, row.State, row.Country )

			record = {
					'City': row.City,
					'State': row.State,
					'Country': row.Country,
					'Query': query,
					'RowCount': row.RowCount,
					'Latitude': None,
					'Longitude': None,
					'Source': '',
					'Status': '',
					'Message': '',
			}

			if not query:
				record[ 'Status' ] = 'Skipped'
				record[ 'Message' ] = 'Location query could not be composed.'
				records.append( record )
				continue

			try:
				result = geocoder.freeform( query )
				latitude, longitude = extract_coordinates( result )

				if latitude is None or longitude is None:
					raise NotFound( 'No usable coordinates returned by Geocoder.' )

				record[ 'Latitude' ] = latitude
				record[ 'Longitude' ] = longitude
				record[ 'Source' ] = 'Geocoder'
				record[ 'Status' ] = 'Matched'
				record[ 'Message' ] = 'Resolved by Geocoder.'

			except NotFound as e:
				if use_places and places is not None:
					try:
						result = places.text_to_location( query )
						latitude, longitude = extract_coordinates( result )

						if latitude is None or longitude is None:
							record[ 'Source' ] = 'Places'
							record[ 'Status' ] = 'Failed'
							record[ 'Message' ] = 'No usable coordinates returned by Places.'
						else:
							record[ 'Latitude' ] = latitude
							record[ 'Longitude' ] = longitude
							record[ 'Source' ] = 'Places'
							record[ 'Status' ] = 'Matched'
							record[ 'Message' ] = 'Resolved by Places fallback.'

					except Exception as ex:
						record[ 'Source' ] = 'Places'
						record[ 'Status' ] = 'Failed'
						record[ 'Message' ] = str( ex )
				else:
					record[ 'Source' ] = 'Geocoder'
					record[ 'Status' ] = 'Failed'
					record[ 'Message' ] = str( e )

			except Exception as e:
				record[ 'Source' ] = 'Geocoder'
				record[ 'Status' ] = 'Failed'
				record[ 'Message' ] = str( e )

			records.append( record )

		return pd.DataFrame( records )

	except Exception as e:
		st.error( f'Unable to preview coordinate updates: {e}' )
		return pd.DataFrame( )

apply_report_coordinate_updates

apply_report_coordinate_updates(
    table_name: str, df_updates: DataFrame
) -> int

Apply the report coordinate updates operation.

Purpose

Supports the Mappy Streamlit application by executing the apply report coordinate updates workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table_name	str	str value used by the apply_report_coordinate_updates workflow.	required
df_updates	DataFrame	pd.DataFrame value used by the apply_report_coordinate_updates workflow.	required

Returns:

Name	Type	Description
int	int	Result produced by the apply_report_coordinate_updates workflow.

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def apply_report_coordinate_updates( table_name: str, df_updates: pd.DataFrame ) -> int:
	"""Apply the report coordinate updates operation.

	Purpose:
		Supports the Mappy Streamlit application by executing the apply report coordinate
		updates workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		table_name: str value used by the `apply_report_coordinate_updates` workflow.
		df_updates: pd.DataFrame value used by the `apply_report_coordinate_updates` workflow.

	Returns:
		int: Result produced by the `apply_report_coordinate_updates` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	try:
		throw_if( 'table_name', table_name )

		if df_updates is None or df_updates.empty:
			return 0

		required_cols = [ 'City', 'State', 'Country', 'Latitude', 'Longitude', 'Status' ]
		missing_cols = [ col for col in required_cols if col not in df_updates.columns ]

		if missing_cols:
			raise ValueError( f'Missing update column(s): {", ".join( missing_cols )}' )

		df_apply = df_updates.copy( )
		df_apply = df_apply[ df_apply[ 'Status' ].astype( str ).str.lower( ) == 'matched' ]
		df_apply[ 'Latitude' ] = pd.to_numeric( df_apply[ 'Latitude' ], errors='coerce' )
		df_apply[ 'Longitude' ] = pd.to_numeric( df_apply[ 'Longitude' ], errors='coerce' )

		valid_mask = (
				df_apply[ 'Latitude' ].notna( )
				& df_apply[ 'Longitude' ].notna( )
				& df_apply[ 'Latitude' ].between( -90.0, 90.0 )
				& df_apply[ 'Longitude' ].between( -180.0, 180.0 )
				& ~(
				(df_apply[ 'Latitude' ] == 0.0)
				& (df_apply[ 'Longitude' ] == 0.0)
		)
		)

		df_apply = df_apply.loc[ valid_mask ].copy( )

		if df_apply.empty:
			return 0

		updated_count = 0
		cursor = None

		with create_connection( ) as conn:
			try:
				conn.execute( 'PRAGMA busy_timeout = 30000;' )
				cursor = conn.cursor( )
				cursor.execute( 'BEGIN IMMEDIATE;' )

				for row in df_apply.itertuples( index=False ):
					cursor.execute(
						f"""
						UPDATE "{table_name}"
						SET Latitude = ?,
						    Longitude = ?
						WHERE
							(Latitude IS NULL OR Longitude IS NULL
							 OR Latitude = '' OR Longitude = ''
							 OR Latitude = 0 OR Longitude = 0)
							AND COALESCE(City, '') = COALESCE(?, '')
							AND COALESCE(State, '') = COALESCE(?, '')
							AND COALESCE(Country, '') = COALESCE(?, '');
						""",
						(
								float( row.Latitude ),
								float( row.Longitude ),
								'' if pd.isna( row.City ) else str( row.City ),
								'' if pd.isna( row.State ) else str( row.State ),
								'' if pd.isna( row.Country ) else str( row.Country ),
						) )

					updated_count += int( cursor.rowcount or 0 )

				conn.commit( )

			except Exception:
				conn.rollback( )
				raise

			finally:
				if cursor is not None:
					cursor.close( )

		with create_connection( ) as conn:
			try:
				conn.execute( 'PRAGMA wal_checkpoint(TRUNCATE);' )
			except sqlite3.OperationalError:
				pass

		return int( updated_count )

	except Exception as e:
		st.error( f'Unable to apply coordinate updates: {e}' )
		return 0

append_geocoding_map_result

append_geocoding_map_result(
    query: str, source: str, result: object
) -> None

Handle the append geocoding map result workflow.

Purpose

Supports the Mappy Streamlit application by executing the append geocoding map result workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
query	str	str value used by the append_geocoding_map_result workflow.	required
source	str	str value used by the append_geocoding_map_result workflow.	required
result	object	object value used by the append_geocoding_map_result workflow.	required

Source code in app.py

def append_geocoding_map_result( query: str, source: str, result: object ) -> None:
	"""Handle the append geocoding map result workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the append geocoding map
		result workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		query: str value used by the `append_geocoding_map_result` workflow.
		source: str value used by the `append_geocoding_map_result` workflow.
		result: object value used by the `append_geocoding_map_result` workflow.
	"""
	try:
		throw_if( 'query', query )
		throw_if( 'source', source )

		latitude, longitude = extract_coordinates( result )

		if latitude is None or longitude is None:
			st.warning( 'The geocoding result did not contain usable coordinates for the map.' )
			return

		set_location_state(
			location=query,
			description=f'{source} result for {query}',
			latitude=latitude,
			longitude=longitude )

		df_new = pd.DataFrame(
			[
					{
							'ID': f'GEOCODED-{int( time.time( ) )}',
							'CalendarDate': dt.datetime.now( ).strftime( '%Y-%m-%d %H:%M:%S' ),
							'City': query,
							'State': '',
							'Country': '',
							'Latitude': latitude,
							'Longitude': longitude,
							'Shape': 'Geocoded Result',
							'Summary': f'{source} result for {query}',
							'Source': source,
							'Query': query,
					}
			]
		)

		df_current = st.session_state.get( 'df_geocoding_map_results', pd.DataFrame( ) )

		if df_current is None or df_current.empty:
			st.session_state[ 'df_geocoding_map_results' ] = df_new
		else:
			st.session_state[ 'df_geocoding_map_results' ] = pd.concat(
				[ df_current, df_new ],
				ignore_index=True )

	except Exception as e:
		st.error( f'Unable to append geocoding result to map: {e}' )

initialize_database

initialize_database() -> None

Handle the initialize database workflow.

Purpose

Supports the Mappy Streamlit application by executing the initialize database workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Source code in app.py

def initialize_database( ) -> None:
	"""Handle the initialize database workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the initialize database
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.
	"""
	Path( 'stores/sqlite' ).mkdir( parents=True, exist_ok=True )
	with create_connection( ) as connection:
		conn.execute(
			"""
            CREATE TABLE IF NOT EXISTS chat_history
            (
                id
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                role
                TEXT,
                content
                TEXT
            )
			"""
		)

		conn.execute(
			"""
            CREATE TABLE IF NOT EXISTS embeddings
            (
                id
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                chunk
                TEXT,
                vector
                BLOB
            )
			"""
		)

		conn.execute(
			"""
            CREATE TABLE IF NOT EXISTS Prompts
            (
                PromptsId
                INTEGER
                NOT
                NULL
                PRIMARY
                KEY
                AUTOINCREMENT,
                Caption
                TEXT,
                Name
                TEXT
            (
                80
            ),
                Text TEXT,
                Version TEXT
            (
                80
            ),
                ID TEXT
            (
                80
            )
                )
			"""
		)

		prompt_columns = [ row[ 1 ] for row in
		                   conn.execute( 'PRAGMA table_info("Prompts");' ).fetchall( ) ]

		if 'Caption' not in prompt_columns:
			conn.execute( 'ALTER TABLE "Prompts" ADD COLUMN "Caption" TEXT;' )

		conn.commit( )

create_connection

create_connection() -> Connection

Create the connection used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create connection workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Returns:

Type	Description
Connection	sqlite3.Connection: Result produced by the create_connection workflow.

Source code in app.py

def create_connection( ) -> sqlite3.Connection:
	"""Create the connection used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create connection
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Returns:
		sqlite3.Connection: Result produced by the `create_connection` workflow.
	"""
	conn = sqlite3.connect( cfg.DB_PATH, timeout=60.0, isolation_level=None,
		check_same_thread=False, factory=ManagedConnection )

	conn.execute( 'PRAGMA busy_timeout = 30000;' )
	conn.execute( 'PRAGMA foreign_keys = ON;' )

	try:
		conn.execute( 'PRAGMA journal_mode = WAL;' )
	except sqlite3.OperationalError:
		pass

	return conn

list_tables

list_tables() -> List[str]

Handle the list tables workflow.

Purpose

Supports the Mappy Streamlit application by executing the list tables workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Returns:

Type	Description
List[str]	List[str]: Result produced by the list_tables workflow.

Source code in app.py

def list_tables( ) -> List[ str ]:
	"""Handle the list tables workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the list tables workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Returns:
		List[str]: Result produced by the `list_tables` workflow.
	"""
	with create_connection( ) as conn:
		_query = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name;"
		rows = conn.execute( _query ).fetchall( )
		return [ r[ 0 ] for r in rows ]

create_schema

create_schema(table: str) -> List[Tuple]

Create the schema used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create schema workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the create_schema workflow.	required

Returns:

Type	Description
List[Tuple]	List[Tuple]: Result produced by the create_schema workflow.

Source code in app.py

def create_schema( table: str ) -> List[ Tuple ]:
	"""Create the schema used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create schema workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		table: str value used by the `create_schema` workflow.

	Returns:
		List[Tuple]: Result produced by the `create_schema` workflow.
	"""
	with create_connection( ) as conn:
		return conn.execute( f'PRAGMA table_info("{table}");' ).fetchall( )

read_table

read_table(
    table: str, limit: int = None, offset: int = 0
) -> DataFrame

Read the table for application processing.

Purpose

Supports the Mappy Streamlit application by executing the read table workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the read_table workflow.	required
limit	int	int value used by the read_table workflow. Defaults to None.	None
offset	int	int value used by the read_table workflow. Defaults to 0.	0

Returns:

Type	Description
DataFrame	pd.DataFrame: Result produced by the read_table workflow.

Source code in app.py

def read_table( table: str, limit: int = None, offset: int = 0 ) -> pd.DataFrame:
	"""Read the table for application processing.

	Purpose:
		Supports the Mappy Streamlit application by executing the read table workflow. The
		function preserves the existing UI behavior, session-state interactions, dataframe
		handling, database access, and service integrations defined by the application
		code.

	Args:
		table: str value used by the `read_table` workflow.
		limit: int value used by the `read_table` workflow. Defaults to `None`.
		offset: int value used by the `read_table` workflow. Defaults to `0`.

	Returns:
		pd.DataFrame: Result produced by the `read_table` workflow.
	"""
	if not table:
		return pd.DataFrame( )

	query = f'SELECT * FROM "{table}"'
	if limit:
		query += f' LIMIT {int( limit )} OFFSET {int( offset )}'

	with create_connection( ) as conn:
		cur = conn.cursor( )
		cur.execute( query )

		raw_columns = [ d[ 0 ] for d in (cur.description or [ ]) ]
		rows = cur.fetchall( )

	seen: Dict[ str, int ] = { }
	columns: List[ str ] = [ ]

	for col in raw_columns:
		name = str( col )
		if name not in seen:
			seen[ name ] = 0
			columns.append( name )
		else:
			seen[ name ] += 1
			columns.append( f'{name}_{seen[ name ]}' )

	def _scalarize( value: Any ) -> Any:
		"""Handle the scalarize workflow.

		Purpose:
			Supports the Mappy Streamlit application by executing the scalarize workflow. The
			function preserves the existing UI behavior, session-state interactions, dataframe
			handling, database access, and service integrations defined by the application
			code.

		Args:
			value: Any value used by the `_scalarize` workflow.

		Returns:
			Any: Result produced by the `_scalarize` workflow.
		"""
		if value is None or isinstance( value, (str, int, float, bool) ):
			return value

		if isinstance( value, bytes ):
			try:
				return value.decode( 'utf-8' )
			except Exception:
				return value.hex( )

		if isinstance( value, (list, tuple, set, dict) ):
			try:
				return str( normalize( value ) )
			except Exception:
				return str( value )

		if hasattr( value, 'model_dump' ):
			try:
				return str( value.model_dump( ) )
			except Exception:
				return str( value )

		return str( value )

	normalized_rows: List[ Dict[ str, Any ] ] = [ ]
	for row in rows:
		record: Dict[ str, Any ] = { }
		for idx, col in enumerate( columns ):
			record[ col ] = _scalarize( row[ idx ] )
		normalized_rows.append( record )

	return pd.DataFrame( normalized_rows, columns=columns )

render_table

render_table(df: DataFrame) -> None

Render the table interface section.

Purpose

Supports the Mappy Streamlit application by executing the render table workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df	DataFrame	pd.DataFrame value used by the render_table workflow.	required

Source code in app.py

def render_table( df: pd.DataFrame ) -> None:
	"""Render the table interface section.

	Purpose:
		Supports the Mappy Streamlit application by executing the render table workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		df: pd.DataFrame value used by the `render_table` workflow.
	"""
	if df is None:
		st.info( 'No data available.' )
		return

	try:
		st.data_editor( df, use_container_width=True )
		return
	except Exception:
		pass

	fallback_df = df.copy( )
	fallback_df = fallback_df.where( pd.notnull( fallback_df ), '' )

	for col in fallback_df.columns:
		fallback_df[ col ] = fallback_df[ col ].map(
			lambda x: x if isinstance( x, (str, int, float, bool) ) or x == '' else str( x ) )

	st.markdown( fallback_df.to_html( index=False, escape=True ), unsafe_allow_html=True )

make_display_safe

make_display_safe(df: DataFrame) -> DataFrame

Handle the make display safe workflow.

Purpose

Supports the Mappy Streamlit application by executing the make display safe workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df	DataFrame	pd.DataFrame value used by the make_display_safe workflow.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: Result produced by the make_display_safe workflow.

Source code in app.py

def make_display_safe( df: pd.DataFrame ) -> pd.DataFrame:
	"""Handle the make display safe workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the make display safe
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		df: pd.DataFrame value used by the `make_display_safe` workflow.

	Returns:
		pd.DataFrame: Result produced by the `make_display_safe` workflow.
	"""
	display_df = df.copy( )

	for col in display_df.columns:
		display_df[ col ] = display_df[ col ].map( lambda x: '' if x is None else str( x ) )

	return display_df

drop_table

drop_table(table: str) -> None

Handle the drop table workflow.

Purpose

Supports the Mappy Streamlit application by executing the drop table workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the drop_table workflow.	required

Source code in app.py

def drop_table( table: str ) -> None:
	"""Handle the drop table workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the drop table workflow. The
		function preserves the existing UI behavior, session-state interactions, dataframe
		handling, database access, and service integrations defined by the application
		code.

	Args:
		table: str value used by the `drop_table` workflow.
	"""
	if not table:
		return

	with create_connection( ) as conn:
		conn.execute( f'DROP TABLE IF EXISTS "{table}";' )
		conn.commit( )

create_index

create_index(table: str, column: str) -> None

Create the index used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create index workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the create_index workflow.	required
column	str	str value used by the create_index workflow.	required

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def create_index( table: str, column: str ) -> None:
	"""Create the index used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create index workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		table: str value used by the `create_index` workflow.
		column: str value used by the `create_index` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	if not table or not column:
		return

	# ------------------------------------------------------------------
	# Validate table exists
	# ------------------------------------------------------------------
	tables = list_tables( )
	if table not in tables:
		raise ValueError( 'Invalid table name.' )

	# ------------------------------------------------------------------
	# Validate column exists
	# ------------------------------------------------------------------
	schema = create_schema( table )
	valid_columns = [ col[ 1 ] for col in schema ]

	if column not in valid_columns:
		raise ValueError( 'Invalid column name.' )

	# ------------------------------------------------------------------
	# Sanitize index name (identifier only)
	# ------------------------------------------------------------------
	safe_index_name = re.sub( r"[^0-9a-zA-Z_]+", "_", f"idx_{table}_{column}" )

	# ------------------------------------------------------------------
	# Create index safely (quote identifiers)
	# ------------------------------------------------------------------
	sql = f'CREATE INDEX IF NOT EXISTS "{safe_index_name}" ON "{table}"("{column}");'

	with create_connection( ) as conn:
		conn.execute( sql )
		conn.commit( )

apply_filters

apply_filters(df: DataFrame) -> DataFrame

Apply the filters operation.

Purpose

Supports the Mappy Streamlit application by executing the apply filters workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df	DataFrame	pd.DataFrame value used by the apply_filters workflow.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: Result produced by the apply_filters workflow.

Source code in app.py

def apply_filters( df: pd.DataFrame ) -> pd.DataFrame:
	"""Apply the filters operation.

	Purpose:
		Supports the Mappy Streamlit application by executing the apply filters workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		df: pd.DataFrame value used by the `apply_filters` workflow.

	Returns:
		pd.DataFrame: Result produced by the `apply_filters` workflow.
	"""
	st.subheader( 'Advanced Filters' )
	conditions = [ ]
	col1, col2, col3 = st.columns( 3 )
	column = col1.selectbox( 'Column', df.columns )
	operator = col2.selectbox( 'Operator', [ '=', '!=', '>', '<', '>=', '<=', 'contains' ] )
	value = col3.text_input( 'Value' )
	if value:
		if operator == '=':
			df = df[ df[ column ] == value ]
		elif operator == '!=':
			df = df[ df[ column ] != value ]
		elif operator == '>':
			df = df[ df[ column ].astype( float ) > float( value ) ]
		elif operator == '<':
			df = df[ df[ column ].astype( float ) < float( value ) ]
		elif operator == '>=':
			df = df[ df[ column ].astype( float ) >= float( value ) ]
		elif operator == '<=':
			df = df[ df[ column ].astype( float ) <= float( value ) ]
		elif operator == 'contains':
			df = df[ df[ column ].astype( str ).str.contains( value ) ]

	return df

create_aggregation

create_aggregation(df: DataFrame)

Create the aggregation used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create aggregation workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df	DataFrame	pd.DataFrame value used by the create_aggregation workflow.	required

Source code in app.py

def create_aggregation( df: pd.DataFrame ):
	"""Create the aggregation used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create aggregation
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		df: pd.DataFrame value used by the `create_aggregation` workflow.
	"""
	st.subheader( 'Aggregation Engine' )
	numeric_cols = df.select_dtypes( include=[ 'number' ] ).columns.tolist( )
	if not numeric_cols:
		st.info( 'No numeric columns available.' )
		return

	col = st.selectbox( 'Column', numeric_cols )
	agg = st.selectbox( 'Aggregation', [ 'COUNT', 'SUM', 'AVG', 'MIN', 'MAX', 'MEDIAN' ] )

	if agg == 'COUNT':
		result = df[ col ].count( )
	elif agg == 'SUM':
		result = df[ col ].sum( )
	elif agg == 'AVG':
		result = df[ col ].mean( )
	elif agg == 'MIN':
		result = df[ col ].min( )
	elif agg == 'MAX':
		result = df[ col ].max( )
	elif agg == 'MEDIAN':
		result = df[ col ].median( )

	st.metric( 'Result', result )

create_visualization

create_visualization(df: DataFrame) -> None

Create the visualization used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create visualization workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df	DataFrame	pd.DataFrame value used by the create_visualization workflow.	required

Source code in app.py

def create_visualization( df: pd.DataFrame ) -> None:
	"""Create the visualization used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create visualization
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		df: pd.DataFrame value used by the `create_visualization` workflow.
	"""

	set_blue_divider( )

	st.markdown( '##### Visualization Engine' )

	if df is None or df.empty:
		st.info( 'No data available.' )
		return

	df_plot = df.copy( )

	for col in df_plot.columns:
		if df_plot[ col ].dtype == object:
			df_plot[ col ] = df_plot[ col ].map( lambda x: '' if x is None else str( x ) )

	numeric_cols: List[ str ] = [ ]
	for col in df_plot.columns:
		series_num = pd.to_numeric( df_plot[ col ], errors='coerce' )
		if series_num.notna( ).any( ):
			numeric_cols.append( col )

	categorical_cols: List[ str ] = [ col for col in df_plot.columns if col not in numeric_cols ]

	chart = st.selectbox( 'Chart Type',
		[ 'Histogram', 'Bar', 'Line', 'Scatter', 'Box', 'Pie', 'Correlation' ] )

	if chart == 'Histogram':
		if not numeric_cols:
			st.info( 'No numeric columns available.' )
			return

		col = st.selectbox( 'Column', numeric_cols )
		values = pd.to_numeric( df_plot[ col ], errors='coerce' ).dropna( ).tolist( )

		fig = go.Figure( data=[ go.Histogram( x=values ) ] )
		fig.update_layout( xaxis_title=col, yaxis_title='Count' )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Bar':
		if not numeric_cols:
			st.info( 'No numeric columns available.' )
			return

		x = st.selectbox( 'X', df_plot.columns )
		y = st.selectbox( 'Y', numeric_cols )

		x_values = df_plot[ x ].astype( str ).tolist( )
		y_values = pd.to_numeric( df_plot[ y ], errors='coerce' ).fillna( 0 ).tolist( )

		fig = go.Figure( data=[ go.Bar( x=x_values, y=y_values ) ] )
		fig.update_layout( xaxis_title=x, yaxis_title=y )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Line':
		if not numeric_cols:
			st.info( 'No numeric columns available.' )
			return

		x = st.selectbox( 'X', df_plot.columns )
		y = st.selectbox( 'Y', numeric_cols )

		x_values = df_plot[ x ].astype( str ).tolist( )
		y_values = pd.to_numeric( df_plot[ y ], errors='coerce' ).fillna( 0 ).tolist( )

		fig = go.Figure( data=[ go.Scatter( x=x_values, y=y_values, mode='lines' ) ] )
		fig.update_layout( xaxis_title=x, yaxis_title=y )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Scatter':
		if len( numeric_cols ) < 2:
			st.info( 'At least two numeric columns are required.' )
			return

		x = st.selectbox( 'X', numeric_cols, key='viz_scatter_x' )
		y = st.selectbox( 'Y', numeric_cols, key='viz_scatter_y' )

		x_series = pd.to_numeric( df_plot[ x ], errors='coerce' )
		y_series = pd.to_numeric( df_plot[ y ], errors='coerce' )
		mask = x_series.notna( ) & y_series.notna( )

		x_values = x_series[ mask ].tolist( )
		y_values = y_series[ mask ].tolist( )

		fig = go.Figure( data=[ go.Scatter( x=x_values, y=y_values, mode='markers' ) ] )
		fig.update_layout( xaxis_title=x, yaxis_title=y )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Box':
		if not numeric_cols:
			st.info( 'No numeric columns available.' )
			return

		col = st.selectbox( 'Column', numeric_cols, key='viz_box_col' )
		values = pd.to_numeric( df_plot[ col ], errors='coerce' ).dropna( ).tolist( )

		fig = go.Figure( data=[ go.Box( y=values, name=col ) ] )
		fig.update_layout( yaxis_title=col )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Pie':
		if not categorical_cols:
			st.info( 'No categorical columns available.' )
			return

		col = st.selectbox( 'Category Column', categorical_cols )
		counts = df_plot[ col ].astype( str ).value_counts( )

		fig = go.Figure(
			data=[ go.Pie( labels=counts.index.tolist( ), values=counts.values.tolist( ) ) ] )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Correlation':
		if len( numeric_cols ) < 2:
			st.info( 'At least two numeric columns are required.' )
			return

		corr_df = pd.DataFrame( )
		for col in numeric_cols:
			corr_df[ col ] = pd.to_numeric( df_plot[ col ], errors='coerce' )

		corr = corr_df.corr( )

		fig = go.Figure(
			data=[ go.Heatmap(
				z=corr.values.tolist( ),
				x=corr.columns.tolist( ),
				y=corr.index.tolist( ) ) ] )
		st.plotly_chart( fig, use_container_width=True )

convert_dataframe

convert_dataframe(table_name: str, df: DataFrame) -> None

Handle the convert dataframe workflow.

Purpose

Supports the Mappy Streamlit application by executing the convert dataframe workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table_name	str	str value used by the convert_dataframe workflow.	required
df	DataFrame	pd.DataFrame value used by the convert_dataframe workflow.	required

Source code in app.py

def convert_dataframe( table_name: str, df: pd.DataFrame ) -> None:
	"""Handle the convert dataframe workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the convert dataframe
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		table_name: str value used by the `convert_dataframe` workflow.
		df: pd.DataFrame value used by the `convert_dataframe` workflow.
	"""
	columns = [ ]
	for col in df.columns:
		sql_type = get_sqlite_type( df[ col ].dtype )
		safe_col = col.replace( ' ', '_' )
		columns.append( f'{safe_col} {sql_type}' )

	create_stmt = f'CREATE TABLE IF NOT EXISTS {table_name} ({", ".join( columns )});'
	with create_connection( ) as conn:
		conn.execute( create_stmt )
		conn.commit( )

get_sqlite_type

get_sqlite_type(dtype: str) -> str

Return the sqlite type used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get sqlite type workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
dtype	str	str Runtime value used by the get_sqlite_type workflow.	required

Returns:

Name	Type	Description
str	str	Result produced by the get_sqlite_type workflow.

Source code in app.py

def get_sqlite_type( dtype: str ) -> str:
	"""Return the sqlite type used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get sqlite type
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		dtype: str Runtime value used by the `get_sqlite_type` workflow.

	Returns:
		str: Result produced by the `get_sqlite_type` workflow.
	"""
	dtype_str = str( dtype ).lower( )

	# ------------------------------------------------------------------
	# Integer Types (including nullable Int64)
	# ------------------------------------------------------------------
	if 'int' in dtype_str:
		return 'INTEGER'

	# ------------------------------------------------------------------
	# Float Types
	# ------------------------------------------------------------------
	if 'float' in dtype_str:
		return 'REAL'

	# ------------------------------------------------------------------
	# Boolean
	# ------------------------------------------------------------------
	if 'bool' in dtype_str:
		return 'INTEGER'

	# ------------------------------------------------------------------
	# Datetime
	# ------------------------------------------------------------------
	if 'datetime' in dtype_str:
		return 'TEXT'

	# ------------------------------------------------------------------
	# Categorical
	# ------------------------------------------------------------------
	if 'category' in dtype_str:
		return 'TEXT'

	# ------------------------------------------------------------------
	# Default fallback
	# ------------------------------------------------------------------
	return 'TEXT'

create_custom_table

create_custom_table(table_name: str, columns: list) -> None

Create the custom table used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create custom table workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table_name	str	str value used by the create_custom_table workflow.	required
columns	list	list value used by the create_custom_table workflow.	required

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def create_custom_table( table_name: str, columns: list ) -> None:
	"""Create the custom table used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create custom table
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		table_name: str value used by the `create_custom_table` workflow.
		columns: list value used by the `create_custom_table` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	if not table_name:
		raise ValueError( 'Table name required.' )

	# Validate identifier
	if not re.match( r"^[A-Za-z_][A-Za-z0-9_]*$", table_name ):
		raise ValueError( 'Invalid table name.' )

	col_defs = [ ]

	for col in columns:
		col_name = col[ 'name' ]
		col_type = col[ 'type' ].upper( )

		if not re.match( r"^[A-Za-z_][A-Za-z0-9_]*$", col_name ):
			raise ValueError( f"Invalid column name: {col_name}" )

		definition = f'"{col_name}" {col_type}'

		if col[ 'primary_key' ]:
			definition += ' PRIMARY KEY'
			if col[ 'auto_increment' ] and col_type == 'INTEGER':
				definition += ' AUTOINCREMENT'

		if col[ "not_null" ]:
			definition += " NOT NULL"

		col_defs.append( definition )

	sql = f'CREATE TABLE IF NOT EXISTS "{table_name}" ({", ".join( col_defs )});'

	with create_connection( ) as conn:
		conn.execute( sql )
		conn.commit( )

is_safe_query

is_safe_query(query: str) -> bool

Handle the is safe query workflow.

Purpose

Supports the Mappy Streamlit application by executing the is safe query workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
query	str	str value used by the is_safe_query workflow.	required

Returns:

Name	Type	Description
bool	bool	Result produced by the is_safe_query workflow.

Source code in app.py

def is_safe_query( query: str ) -> bool:
	"""Handle the is safe query workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the is safe query workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		query: str value used by the `is_safe_query` workflow.

	Returns:
		bool: Result produced by the `is_safe_query` workflow.
	"""
	if not query or not isinstance( query, str ):
		return False

	q = query.strip( ).lower( )

	# ------------------------------------------------------------------
	# Block multiple statements
	# ------------------------------------------------------------------
	if ';' in q[ :-1 ]:
		return False

	# ------------------------------------------------------------------
	# Remove SQL comments
	# ------------------------------------------------------------------
	q = re.sub( r"--.*?$", "", q, flags=re.MULTILINE )
	q = re.sub( r"/\*.*?\*/", "", q, flags=re.DOTALL )
	q = q.strip( )

	# ------------------------------------------------------------------
	# Allowed starting keywords
	# ------------------------------------------------------------------
	allowed_starts = ('select', 'with', 'explain', 'pragma')
	if not q.startswith( allowed_starts ):
		return False

	# ------------------------------------------------------------------
	# Block dangerous keywords anywhere
	# ------------------------------------------------------------------
	blocked_keywords = ('insert ', 'update ', 'delete ', 'drop ', 'alter ',
	                    'create ', 'attach ', 'detach ', 'vacuum ', 'replace ', 'trigger ')

	for keyword in blocked_keywords:
		if keyword in q:
			return False

	return True

create_identifier

create_identifier(name: str) -> str

Create the identifier used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create identifier workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
name	str	str value used by the create_identifier workflow.	required

Returns:

Name	Type	Description
str	str	Result produced by the create_identifier workflow.

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def create_identifier( name: str ) -> str:
	"""Create the identifier used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create identifier
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		name: str value used by the `create_identifier` workflow.

	Returns:
		str: Result produced by the `create_identifier` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	if not name or not isinstance( name, str ):
		raise ValueError( 'Invalid Identifier.' )

	safe = re.sub( r'[^0-9a-zA-Z_]', '_', name.strip( ) )
	if not re.match( r'^[A-Za-z_]', safe ):
		safe = f'_{safe}'

	if not safe:
		raise ValueError( 'Invalid identifier after sanitization.' )

	return safe

get_indexes

get_indexes(table: str) -> List[int]

Return the indexes used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get indexes workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the get_indexes workflow.	required

Returns:

Type	Description
List[int]	List[ int ]: Result produced by the get_indexes workflow.

Source code in app.py

def get_indexes( table: str ) -> List[ int ]:
	"""Return the indexes used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get indexes workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		table: str value used by the `get_indexes` workflow.

	Returns:
		List[ int ]: Result produced by the `get_indexes` workflow.
	"""
	with create_connection( ) as conn:
		rows = conn.execute( f'PRAGMA index_list("{table}");' ).fetchall( )
		return rows

add_column

add_column(table: str, column: str, col_type: str)

Handle the add column workflow.

Purpose

Supports the Mappy Streamlit application by executing the add column workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the add_column workflow.	required
column	str	str value used by the add_column workflow.	required
col_type	str	str value used by the add_column workflow.	required

Source code in app.py

def add_column( table: str, column: str, col_type: str ):
	"""Handle the add column workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the add column workflow. The
		function preserves the existing UI behavior, session-state interactions, dataframe
		handling, database access, and service integrations defined by the application
		code.

	Args:
		table: str value used by the `add_column` workflow.
		column: str value used by the `add_column` workflow.
		col_type: str value used by the `add_column` workflow.
	"""
	column = create_identifier( column )
	col_type = col_type.upper( )

	with create_connection( ) as conn:
		conn.execute(
			f'ALTER TABLE "{table}" ADD COLUMN "{column}" {col_type};' )
		conn.commit( )

rename_column

rename_column(
    table_name: str, old_name: str, new_name: str
) -> None

Handle the rename column workflow.

Purpose

Supports the Mappy Streamlit application by executing the rename column workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table_name	str	str value used by the rename_column workflow.	required
old_name	str	str value used by the rename_column workflow.	required
new_name	str	str value used by the rename_column workflow.	required

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def rename_column( table_name: str, old_name: str, new_name: str ) -> None:
	"""Handle the rename column workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the rename column workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		table_name: str value used by the `rename_column` workflow.
		old_name: str value used by the `rename_column` workflow.
		new_name: str value used by the `rename_column` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	if not table_name or not old_name or not new_name:
		return

	with create_connection( ) as conn:
		try:
			conn.execute(
				f'ALTER TABLE "{table_name}" RENAME COLUMN "{old_name}" TO "{new_name}";'
			)
			conn.commit( )
			return
		except Exception:
			pass

		row = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='table' AND name =?
			""",
			(table_name,)
		).fetchone( )

		if not row or not row[ 0 ]:
			raise ValueError( "Table definition not found." )

		create_sql = row[ 0 ]

		indexes = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='index' AND tbl_name=? AND sql IS NOT NULL
			""",
			(table_name,)
		).fetchall( )

		schema = conn.execute( f'PRAGMA table_info("{table_name}");' ).fetchall( )
		cols = [ r[ 1 ] for r in schema ]
		if old_name not in cols:
			raise ValueError( "Column not found." )

		mapped_cols = [ (new_name if c == old_name else c) for c in cols ]

		temp_table = f"{table_name}__rebuild_temp"

		col_defs: List[ str ] = [ ]
		pk_cols = [ r for r in schema if int( r[ 5 ] or 0 ) > 0 ]
		single_pk = len( pk_cols ) == 1

		for row in schema:
			col_name = row[ 1 ]
			col_type = row[ 2 ] or ''
			not_null = int( row[ 3 ] or 0 )
			default_value = row[ 4 ]
			pk = int( row[ 5 ] or 0 )

			out_name = new_name if col_name == old_name else col_name
			col_def = f'"{out_name}" {col_type}'.strip( )

			if not_null:
				col_def += ' NOT NULL'

			if default_value is not None:
				col_def += f' DEFAULT {default_value}'

			if single_pk and pk == 1:
				col_def += ' PRIMARY KEY'

			col_defs.append( col_def )

		new_create_sql = f'CREATE TABLE "{temp_table}" ({", ".join( col_defs )});'

		old_select = ", ".join( [ f'"{c}"' for c in cols ] )
		new_insert = ", ".join( [ f'"{c}"' for c in mapped_cols ] )

		conn.execute( "BEGIN" )
		conn.execute( new_create_sql )
		conn.execute(
			f'INSERT INTO "{temp_table}" ({new_insert}) SELECT {old_select} FROM "{table_name}";'
		)

		conn.execute( f'DROP TABLE "{table_name}";' )
		conn.execute( f'ALTER TABLE "{temp_table}" RENAME TO "{table_name}";' )

		for idx in indexes:
			idx_sql = idx[ 0 ]
			if idx_sql:
				idx_sql = idx_sql.replace( f'"{old_name}"', f'"{new_name}"' )
				conn.execute( idx_sql )

		conn.commit( )

create_profile_table

create_profile_table(table: str) -> object

Create the profile table used by the application.

Purpose

Supports the Mappy Streamlit application by executing the create profile table workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the create_profile_table workflow.	required

Returns:

Name	Type	Description
object	object	Result produced by the create_profile_table workflow.

Source code in app.py

def create_profile_table( table: str ) -> object:
	"""Create the profile table used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the create profile table
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		table: str value used by the `create_profile_table` workflow.

	Returns:
		object: Result produced by the `create_profile_table` workflow.
	"""
	df = read_table( table )
	profile_rows = [ ]
	total_rows = len( df )
	for col in df.columns:
		series = df[ col ]
		null_count = series.isna( ).sum( )
		distinct_count = series.nunique( dropna=True )
		row = \
			{
					'column': col, 'dtype': str( series.dtype ),
					'null_%': round( (null_count / total_rows) * 100, 2 ) if total_rows else 0,
					'distinct_%': round( (
							                     distinct_count / total_rows) * 100,
						2 ) if total_rows else 0,
			}

		if pd.api.types.is_numeric_dtype( series ):
			row[ 'min' ] = series.min( )
			row[ 'max' ] = series.max( )
			row[ 'mean' ] = series.mean( )
		else:
			row[ 'min' ] = None
			row[ 'max' ] = None
			row[ 'mean' ] = None

		profile_rows.append( row )

	return pd.DataFrame( profile_rows )

drop_column

drop_column(table: str, column: str)

Handle the drop column workflow.

Purpose

Supports the Mappy Streamlit application by executing the drop column workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
table	str	str value used by the drop_column workflow.	required
column	str	str value used by the drop_column workflow.	required

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def drop_column( table: str, column: str ):
	"""Handle the drop column workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the drop column workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		table: str value used by the `drop_column` workflow.
		column: str value used by the `drop_column` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	if not table or not column:
		raise ValueError( 'Table and column required.' )

	with create_connection( ) as conn:
		# ------------------------------------------------------------
		# Fetch original CREATE TABLE statement
		# ------------------------------------------------------------
		row = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='table' AND name =?
			""",
			(table,)
		).fetchone( )

		if not row or not row[ 0 ]:
			raise ValueError( 'Table definition not found.' )

		create_sql = row[ 0 ]

		# ------------------------------------------------------------
		# Extract column definitions
		# ------------------------------------------------------------
		open_paren = create_sql.find( "(" )
		close_paren = create_sql.rfind( ")" )

		if open_paren == -1 or close_paren == -1:
			raise ValueError( "Malformed CREATE TABLE statement." )

		inner = create_sql[ open_paren + 1: close_paren ]

		column_defs = [ c.strip( ) for c in inner.split( "," ) ]

		# Remove target column
		new_defs = [ ]
		for col_def in column_defs:
			col_name = col_def.split( )[ 0 ].strip( '"' )
			if col_name != column:
				new_defs.append( col_def )

		if len( new_defs ) == len( column_defs ):
			raise ValueError( "Column not found." )

		# ------------------------------------------------------------
		# Build new CREATE TABLE statement
		# ------------------------------------------------------------
		temp_table = f"{table}_rebuild_temp"

		new_create_sql = (
				f'CREATE TABLE "{temp_table}" ('
				+ ", ".join( new_defs )
				+ ");"
		)

		# ------------------------------------------------------------
		# Begin transaction
		# ------------------------------------------------------------
		conn.execute( "BEGIN" )

		conn.execute( new_create_sql )

		remaining_cols = [
				c.split( )[ 0 ].strip( '"' )
				for c in new_defs
		]

		col_list = ", ".join( [ f'"{c}"' for c in remaining_cols ] )

		conn.execute(
			f'INSERT INTO "{temp_table}" ({col_list}) '
			f'SELECT {col_list} FROM "{table}";'
		)

		# Preserve indexes
		indexes = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='index' AND tbl_name=? AND sql IS NOT NULL
			""",
			(table,)
		).fetchall( )

		conn.execute( f'DROP TABLE "{table}";' )
		conn.execute(
			f'ALTER TABLE "{temp_table}" RENAME TO "{table}";'
		)

		# Recreate indexes
		for idx in indexes:
			idx_sql = idx[ 0 ]
			if column not in idx_sql:
				conn.execute( idx_sql )

		conn.commit( )

rename_table

rename_table(old_name: str, new_name: str) -> None

Handle the rename table workflow.

Purpose

Supports the Mappy Streamlit application by executing the rename table workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
old_name	str	str value used by the rename_table workflow.	required
new_name	str	str value used by the rename_table workflow.	required

Raises:

Type	Description
Exception	Propagates validation, UI, data, database, or provider errors raised by the existing implementation.

Source code in app.py

def rename_table( old_name: str, new_name: str ) -> None:
	"""Handle the rename table workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the rename table workflow.
		The function preserves the existing UI behavior, session-state interactions,
		dataframe handling, database access, and service integrations defined by the
		application code.

	Args:
		old_name: str value used by the `rename_table` workflow.
		new_name: str value used by the `rename_table` workflow.

	Raises:
		Exception: Propagates validation, UI, data, database, or provider errors
			raised by the existing implementation.
	"""
	if not old_name or not new_name:
		return

	with create_connection( ) as conn:
		try:
			conn.execute( f'ALTER TABLE "{old_name}" RENAME TO "{new_name}";' )
			conn.commit( )
			return
		except Exception:
			pass

		row = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='table' AND name =?
			""",
			(old_name,)
		).fetchone( )

		if not row or not row[ 0 ]:
			raise ValueError( "Table definition not found." )

		create_sql = row[ 0 ]

		indexes = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='index' AND tbl_name=? AND sql IS NOT NULL
			""",
			(old_name,)
		).fetchall( )

		open_paren = create_sql.find( "(" )
		if open_paren == -1:
			raise ValueError( "Malformed CREATE TABLE statement." )

		temp_name = f"{new_name}__rebuild_temp"

		conn.execute( "BEGIN" )
		conn.execute( f'CREATE TABLE "{temp_name}" {create_sql[ open_paren: ]}' )

		cols = [ r[ 1 ] for r in conn.execute( f'PRAGMA table_info("{old_name}");' ).fetchall( ) ]
		col_list = ", ".join( [ f'"{c}"' for c in cols ] )

		conn.execute(
			f'INSERT INTO "{temp_name}" ({col_list}) SELECT {col_list} FROM "{old_name}";'
		)

		conn.execute( f'DROP TABLE "{old_name}";' )
		conn.execute( f'ALTER TABLE "{temp_name}" RENAME TO "{new_name}";' )

		for idx in indexes:
			idx_sql = idx[ 0 ]
			if idx_sql:
				idx_sql = idx_sql.replace( f'ON "{old_name}"', f'ON "{new_name}"' )
				conn.execute( idx_sql )

		conn.commit( )

has_loaded_dataset

has_loaded_dataset(df_frame: object) -> bool

Handle the has loaded dataset workflow.

Purpose

Supports the Mappy Streamlit application by executing the has loaded dataset workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df_frame	object	object value used by the has_loaded_dataset workflow.	required

Returns:

Name	Type	Description
bool	bool	Result produced by the has_loaded_dataset workflow.

Source code in app.py

def has_loaded_dataset( df_frame: object ) -> bool:
	"""Handle the has loaded dataset workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the has loaded dataset
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		df_frame: object value used by the `has_loaded_dataset` workflow.

	Returns:
		bool: Result produced by the `has_loaded_dataset` workflow.
	"""
	return (
			isinstance( df_frame, pd.DataFrame )
			and not df_frame.empty
			and len( df_frame.columns ) > 0
	)

get_loaded_dataset

get_loaded_dataset() -> DataFrame | None

Return the loaded dataset used by the application.

Purpose

Supports the Mappy Streamlit application by executing the get loaded dataset workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: Result produced by the get_loaded_dataset workflow.

Source code in app.py

def get_loaded_dataset( ) -> pd.DataFrame | None:
	"""Return the loaded dataset used by the application.

	Purpose:
		Supports the Mappy Streamlit application by executing the get loaded dataset
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Returns:
		pd.DataFrame | None: Result produced by the `get_loaded_dataset` workflow.
	"""
	df_frame = st.session_state.get( 'df_dataset', None )
	if not has_loaded_dataset( df_frame ):
		return None

	return df_frame.copy( )

store_loaded_dataset

store_loaded_dataset(
    df_dataset: DataFrame,
    df_original: DataFrame | None = None,
) -> None

Handle the store loaded dataset workflow.

Purpose

Supports the Mappy Streamlit application by executing the store loaded dataset workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Parameters:

Name	Type	Description	Default
df_dataset	DataFrame	pd.DataFrame value used by the store_loaded_dataset workflow.	required
df_original	DataFrame | None	pd.DataFrame | None value used by the store_loaded_dataset workflow. Defaults to None.	None

Source code in app.py

def store_loaded_dataset( df_dataset: pd.DataFrame,
		df_original: pd.DataFrame | None = None ) -> None:
	"""Handle the store loaded dataset workflow.

	Purpose:
		Supports the Mappy Streamlit application by executing the store loaded dataset
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.

	Args:
		df_dataset: pd.DataFrame value used by the `store_loaded_dataset` workflow.
		df_original: pd.DataFrame | None value used by the `store_loaded_dataset` workflow.
			Defaults to `None`.
	"""
	if not has_loaded_dataset( df_dataset ):
		return

	df_source = df_dataset.copy( )
	df_base = df_original.copy( ) if isinstance( df_original, pd.DataFrame ) else df_source.copy( )

	st.session_state[ 'df_raw' ] = df_source.copy( )
	st.session_state[ 'df_original' ] = df_base.copy( )
	st.session_state[ 'df_dataset' ] = df_source.copy( )

clear_webscrape_state

clear_webscrape_state() -> None

Clear the webscrape state state.

Purpose

Supports the Mappy Streamlit application by executing the clear webscrape state workflow. The function preserves the existing UI behavior, session-state interactions, dataframe handling, database access, and service integrations defined by the application code.

Source code in app.py

def clear_webscrape_state( ) -> None:
	"""Clear the webscrape state state.

	Purpose:
		Supports the Mappy Streamlit application by executing the clear webscrape state
		workflow. The function preserves the existing UI behavior, session-state
		interactions, dataframe handling, database access, and service integrations
		defined by the application code.
	"""
	st.session_state[ 'webscrape_clear_request' ] = True