mark/surplus
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

s+: touch up conversion to shareabletext + tests

Browse source
This commit is contained in:
Mark Joshwel 2023-09-02 09:35:33 +00:00
parent f62c31d685
commit cc90b5e196
3 changed files with 344 additions and 159 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

191
surplus.future.ipynb
View file

@ -113,6 +113,26 @@
"## Query Types" "## Query Types"
] ]
}, },
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"Result(value=Latlong(latitude=1.3336875, longitude=103.7746875), error=None)"
]
},
"execution_count": 3,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"PlusCodeQuery(code=\"6PH58QMF+FV\").to_lat_long_coord(geocoder=default_geocoder)"
]
},
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 4, "execution_count": 4,
@ -130,7 +150,11 @@
} }
], ],
"source": [ "source": [
"PlusCodeQuery(code=\"6PH58QMF+FV\").to_lat_long_coord(geocoder=default_geocoder)" "plus_code = LocalCodeQuery(code=\"8QMF+FV\", locality=\"Singapore\").to_full_plus_code(\n",
" geocoder=default_geocoder\n",
")\n",
"\n",
"PlusCodeQuery(code=plus_code.get()).to_lat_long_coord(geocoder=default_geocoder)"
] ]
}, },
{ {
@ -150,11 +174,9 @@
} }
], ],
"source": [ "source": [
"plus_code = LocalCodeQuery(code=\"8QMF+FV\", locality=\"Singapore\").to_full_plus_code(\n", "LocalCodeQuery(code=\"8QMF+FV\", locality=\"Singapore\").to_lat_long_coord(\n",
" geocoder=default_geocoder\n", " geocoder=default_geocoder\n",
")\n", ")"
"\n",
"PlusCodeQuery(code=plus_code.get()).to_lat_long_coord(geocoder=default_geocoder)"
] ]
}, },
{ {
@ -165,7 +187,7 @@
{ {
"data": { "data": {
"text/plain": [ "text/plain": [
"Result(value=Latlong(latitude=1.3336875, longitude=103.7746875), error=None)" "Result(value=Latlong(latitude=1.33318835, longitude=103.77461234638255), error=None)"
] ]
}, },
"execution_count": 6, "execution_count": 6,
@ -174,9 +196,9 @@
} }
], ],
"source": [ "source": [
"LocalCodeQuery(code=\"8QMF+FV\", locality=\"Singapore\").to_lat_long_coord(\n", "LatlongQuery(\n",
" geocoder=default_geocoder\n", " latlong=Latlong(latitude=1.33318835, longitude=103.77461234638255)\n",
")" ").to_lat_long_coord(geocoder=default_geocoder)"
] ]
}, },
{ {
@ -195,28 +217,6 @@
"output_type": "execute_result" "output_type": "execute_result"
} }
], ],
"source": [
"LatlongQuery(\n",
" latlong=Latlong(latitude=1.33318835, longitude=103.77461234638255)\n",
").to_lat_long_coord(geocoder=default_geocoder)"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"Result(value=Latlong(latitude=1.33318835, longitude=103.77461234638255), error=None)"
]
},
"execution_count": 2,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [ "source": [
"StringQuery(query=\"Ngee Ann Polytechnic\").to_lat_long_coord(geocoder=default_geocoder)" "StringQuery(query=\"Ngee Ann Polytechnic\").to_lat_long_coord(geocoder=default_geocoder)"
] ]
@ -225,7 +225,13 @@
"cell_type": "markdown", "cell_type": "markdown",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## Reverser Dictionary Output" "## return dictionary of `reverser` function\n",
"\n",
"all the necessary keys (see `SHAREABLE_TEXT_LINE_*` constants) should be at the top-level of the dictionary.\n",
"these keys will be casted into strings for safety guarantee when shareable text lines are being generated.\n",
"\n",
"while not necessary, consider keeping the original response dict under the \"raw\" key.\n",
"helps with debugging using `-d/--debug`!"
] ]
}, },
{ {
@ -234,21 +240,116 @@
"metadata": {}, "metadata": {},
"outputs": [], "outputs": [],
"source": [ "source": [
"{\n", "reverser_return = {\n",
" \"amenity\": \"\",\n", " \"amenity\": \"Ngee Ann Polytechnic\",\n",
" \"house_number\": \"\",\n", " \"house_number\": \"535\",\n",
" \"road\": \"\",\n", " \"road\": \"Clementi Road\",\n",
" \"suburb\": \"\",\n", " \"suburb\": \"Bukit Timah\",\n",
" \"city\": \"\",\n", " \"city\": \"Singapore\",\n",
" \"county\": \"\",\n", " \"county\": \"Northwest\",\n",
" \"ISO3166-2-lvl6\": \"\",\n", " \"ISO3166-2-lvl6\": \"SG-03\",\n",
" \"postcode\": \"\",\n", " \"postcode\": \"599489\",\n",
" \"country\": \"\",\n", " \"country\": \"Singapore\",\n",
" \"country_code\": \"\",\n", " \"country_code\": \"sg\",\n",
" \"original response\": \"{'place_id': 297946059, 'licence': 'Data © OpenStreetMap contributors, ODbL 1.0. http://osm.org/copyright', 'osm_type': 'relation', 'osm_id': 2535118, 'lat': '1.33318835', 'lon': '103.77461234638255', 'class': 'amenity', 'type': 'university', 'place_rank': 30, 'importance': 0.34662169301918117, 'addresstype': 'amenity', 'name': 'Ngee Ann Polytechnic', 'display_name': 'Ngee Ann Polytechnic, 535, Clementi Road, Bukit Timah, Singapore, Northwest, 599489, Singapore', 'address': {'amenity': 'Ngee Ann Polytechnic', 'house_number': '535', 'road': 'Clementi Road', 'suburb': 'Bukit Timah', 'city': 'Singapore', 'county': 'Northwest', 'ISO3166-2-lvl6': 'SG-03', 'postcode': '599489', 'country': 'Singapore', 'country_code': 'sg'}, 'boundingbox': ['1.3289692', '1.3372184', '103.7701481', '103.7783945']}\",\n", " \"raw\": {\n",
" \"latitude\": \"1.33318835\",\n", " \"place_id\": 297946059,\n",
" \"licence\": \"Data © OpenStreetMap contributors, ODbL 1.0. http://osm.org/copyright\",\n",
" \"osm_type\": \"relation\",\n",
" \"osm_id\": 2535118,\n",
" \"lat\": \"1.33318835\",\n",
" \"lon\": \"103.77461234638255\",\n",
" \"class\": \"amenity\",\n",
" \"type\": \"university\",\n",
" \"place_rank\": 30,\n",
" \"importance\": 0.34662169301918117,\n",
" \"addresstype\": \"amenity\",\n",
" \"name\": \"Ngee Ann Polytechnic\",\n",
" \"display_name\": \"Ngee Ann Polytechnic, 535, Clementi Road, Bukit Timah, Singapore, Northwest, 599489, Singapore\",\n",
" \"address\": {\n",
" \"amenity\": \"Ngee Ann Polytechnic\",\n",
" \"house_number\": \"535\",\n",
" \"road\": \"Clementi Road\",\n",
" \"suburb\": \"Bukit Timah\",\n",
" \"city\": \"Singapore\",\n",
" \"county\": \"Northwest\",\n",
" \"ISO3166-2-lvl6\": \"SG-03\",\n",
" \"postcode\": \"599489\",\n",
" \"country\": \"Singapore\",\n",
" \"country_code\": \"sg\",\n",
" },\n",
" \"boundingbox\": [\"1.3289692\", \"1.3372184\", \"103.7701481\", \"103.7783945\"],\n",
" },\n",
" \"latitude\": 1.33318835,\n",
" \"longitude\": 103.77461234638255,\n",
"}" "}"
] ]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'ISO3166-2-lvl6': 'SG-03',\n",
" 'amenity': 'Ngee Ann Polytechnic',\n",
" 'city': 'Singapore',\n",
" 'country': 'Singapore',\n",
" 'country_code': 'sg',\n",
" 'county': 'Northwest',\n",
" 'house_number': '535',\n",
" 'latitude': 1.33318835,\n",
" 'longitude': 103.77461234638255,\n",
" 'postcode': '599489',\n",
" 'raw': {'address': {'ISO3166-2-lvl6': 'SG-03',\n",
" 'amenity': 'Ngee Ann Polytechnic',\n",
" 'city': 'Singapore',\n",
" 'country': 'Singapore',\n",
" 'country_code': 'sg',\n",
" 'county': 'Northwest',\n",
" 'house_number': '535',\n",
" 'postcode': '599489',\n",
" 'road': 'Clementi Road',\n",
" 'suburb': 'Bukit Timah'},\n",
" 'addresstype': 'amenity',\n",
" 'boundingbox': ['1.3289692',\n",
" '1.3372184',\n",
" '103.7701481',\n",
" '103.7783945'],\n",
" 'class': 'amenity',\n",
" 'display_name': 'Ngee Ann Polytechnic, 535, Clementi Road, Bukit '\n",
" 'Timah, Singapore, Northwest, 599489, Singapore',\n",
" 'importance': 0.34662169301918117,\n",
" 'lat': '1.33318835',\n",
" 'licence': 'Data © OpenStreetMap contributors, ODbL 1.0. '\n",
" 'http://osm.org/copyright',\n",
" 'lon': '103.77461234638255',\n",
" 'name': 'Ngee Ann Polytechnic',\n",
" 'osm_id': 2535118,\n",
" 'osm_type': 'relation',\n",
" 'place_id': 297946059,\n",
" 'place_rank': 30,\n",
" 'type': 'university'},\n",
" 'road': 'Clementi Road',\n",
" 'suburb': 'Bukit Timah'}\n"
]
}
],
"source": [
"import pprint\n",
"\n",
"latlong = LocalCodeQuery(code=\"8QMF+FV\", locality=\"Singapore\").to_lat_long_coord(\n",
" default_geocoder\n",
")\n",
"if not latlong:\n",
" latlong.cry()\n",
"\n",
"else:\n",
" location = default_reverser(latlong.get())\n",
" pprint.pprint(location)"
]
} }
], ],
"metadata": { "metadata": {

234
surplus.py
View file

@ -58,7 +58,7 @@ from pluscodes.openlocationcode import ( # type: ignore # isort: skip
VERSION: Final[tuple[int, int, int]] = (2, 0, 0) VERSION: Final[tuple[int, int, int]] = (2, 0, 0)
USER_AGENT: Final[str] = "surplus" USER_AGENT: Final[str] = "surplus"
OUTPUT_LINE_0_KEYS: Final[tuple[str, ...]] = ( SHAREABLE_TEXT_LINE_0_KEYS: Final[tuple[str, ...]] = (
"emergency", "emergency",
"historic", "historic",
"military", "military",
@ -82,15 +82,14 @@ OUTPUT_LINE_0_KEYS: Final[tuple[str, ...]] = (
"tunnel", "tunnel",
"waterway", "waterway",
) )
OUTPUT_LINE_1_KEYS: Final[tuple[str, ...]] = ("building",) SHAREABLE_TEXT_LINE_1_KEYS: Final[tuple[str, ...]] = ("building",)
OUTPUT_LINE_2_KEYS: Final[tuple[str, ...]] = ("highway",) SHAREABLE_TEXT_LINE_2_KEYS: Final[tuple[str, ...]] = ("highway",)
OUTPUT_LINE_3_NAME: Final[tuple[str, ...]] = ("house_name", "road") SHAREABLE_TEXT_LINE_3_KEYS: Final[tuple[str, ...]] = (
OUTPUT_LINE_3_KEYS: Final[tuple[str, ...]] = (
"house_number", "house_number",
"house_name", "house_name",
"road", "road",
) )
OUTPUT_LINE_4_KEYS: Final[tuple[str, ...]] = ( SHAREABLE_TEXT_LINE_4_KEYS: Final[tuple[str, ...]] = (
"residential", "residential",
"neighbourhood", "neighbourhood",
"allotments", "allotments",
@ -105,8 +104,8 @@ OUTPUT_LINE_4_KEYS: Final[tuple[str, ...]] = (
"town", "town",
"village", "village",
) )
OUTPUT_LINE_5_KEYS: Final[tuple[str, ...]] = ("postcode",) SHAREABLE_TEXT_LINE_5_KEYS: Final[tuple[str, ...]] = ("postcode",)
OUTPUT_LINE_6_KEYS: Final[tuple[str, ...]] = ( SHAREABLE_TEXT_LINE_6_KEYS: Final[tuple[str, ...]] = (
"region", "region",
"county", "county",
"state", "state",
@ -114,6 +113,12 @@ OUTPUT_LINE_6_KEYS: Final[tuple[str, ...]] = (
"country", "country",
"continent", "continent",
) )
SHAREABLE_TEXT_NAMES: Final[tuple[str, ...]] = (
SHAREABLE_TEXT_LINE_0_KEYS
+ SHAREABLE_TEXT_LINE_1_KEYS
+ SHAREABLE_TEXT_LINE_2_KEYS
+ ("house_name", "road")
)
# exceptions # exceptions
@ -276,6 +281,7 @@ class PlusCodeQuery(NamedTuple):
return Result[Latlong]( return Result[Latlong](
EMPTY_LATLONG, EMPTY_LATLONG,
error=IncompletePlusCodeError( error=IncompletePlusCodeError(
"PlusCodeQuery.to_lat_long_coord: "
"Plus Code is not full-length (e.g., 6PH58QMF+FX)" "Plus Code is not full-length (e.g., 6PH58QMF+FX)"
), ),
) )
@ -435,7 +441,7 @@ def default_geocoder(place: str) -> Latlong:
) )
def default_reverser(latlong: Latlong) -> dict[str, str]: def default_reverser(latlong: Latlong) -> dict[str, Any]:
"""default geocoder for surplus, uses OpenStreetMap Nominatim""" """default geocoder for surplus, uses OpenStreetMap Nominatim"""
location: _geopy_Location | None = _geopy_Nominatim(user_agent=USER_AGENT).reverse( location: _geopy_Location | None = _geopy_Nominatim(user_agent=USER_AGENT).reverse(
str(latlong) str(latlong)
@ -444,21 +450,14 @@ def default_reverser(latlong: Latlong) -> dict[str, str]:
if location is None: if location is None:
raise NoSuitableLocationError(f"could not reverse '{str(latlong)}'") raise NoSuitableLocationError(f"could not reverse '{str(latlong)}'")
location_dict: dict[str, str] = {} location_dict: dict[str, Any] = {}
for key in (address := location.raw.get("address", {})): for key in (address := location.raw.get("address", {})):
location_dict[key] = str(address.get(key, "")) location_dict[key] = address.get(key, "")
_bounding_box_default: list[float] = [0.0] * 4 location_dict["raw"] = location.raw
location_dict["latitude"] = location.latitude
location_dict["original response"] = str(location.raw) location_dict["longitude"] = location.longitude
location_dict["latitude"] = str(location.longitude)
location_dict["latitude"] = str(location.latitude)
# location_dict["boundingbox1"] = str(location.raw.get("boundingbox", _bounding_box_default)[0])
# location_dict["boundingbox2"] = str(location.raw.get("boundingbox", _bounding_box_default)[1])
# location_dict["boundingbox3"] = str(location.raw.get("boundingbox", _bounding_box_default)[2])
# location_dict["boundingbox4"] = str(location.raw.get("boundingbox", _bounding_box_default)[3])
return location_dict return location_dict
@ -468,12 +467,13 @@ class Behaviour(NamedTuple):
typing.NamedTuple representing expected behaviour of surplus typing.NamedTuple representing expected behaviour of surplus
arguments arguments
query: list[str] query: str | list[str]
original user-passed query string split by spaces str: original user-passed query string
list[str]: original user-passed query string split by spaces
geocoder: Callable[[str], Latlong] geocoder: Callable[[str], Latlong]
name string to location function, must take in a string and return a Latlong. name string to location function, must take in a string and return a Latlong.
exceptions are handled by the caller. exceptions are handled by the caller.
reverser: Callable[[str], dict[str, str]] reverser: Callable[[str], dict[str, Any]]
Latlong object to dictionary function, must take in a string and return a Latlong object to dictionary function, must take in a string and return a
dict. exceptions are handled by the caller. dict. exceptions are handled by the caller.
stderr: TextIO = stderr stderr: TextIO = stderr
@ -488,9 +488,9 @@ class Behaviour(NamedTuple):
what type to convert query to what type to convert query to
""" """
query: list[str] query: str | list[str]
geocoder: Callable[[str], Latlong] = default_geocoder geocoder: Callable[[str], Latlong] = default_geocoder
reverser: Callable[[Latlong], dict[str, str]] = default_reverser reverser: Callable[[Latlong], dict[str, Any]] = default_reverser
stderr: TextIO = stderr stderr: TextIO = stderr
stdout: TextIO = stdout stdout: TextIO = stdout
debug: bool = False debug: bool = False
@ -527,9 +527,16 @@ def parse_query(
validator = _PlusCode_Validator() validator = _PlusCode_Validator()
portion_plus_code: str = "" portion_plus_code: str = ""
portion_locality: str = "" portion_locality: str = ""
original_query: str = ""
split_query: list[str] = []
original_query = " ".join(behaviour.query) if isinstance(behaviour.query, str):
split_query = behaviour.query original_query = behaviour.query
split_query = behaviour.query.split(" ")
else:
original_query = " ".join(behaviour.query)
split_query = behaviour.query
for word in split_query: for word in split_query:
if validator.is_valid(word): if validator.is_valid(word):
@ -547,19 +554,19 @@ def parse_query(
error="unable to find a Plus Code", error="unable to find a Plus Code",
) )
# did find plus code, but not full-length. :(
if not validator.is_full(portion_plus_code):
return Result[Query](
LatlongQuery(EMPTY_LATLONG),
error=IncompletePlusCodeError(
"Plus Code is not full-length (e.g., 6PH58QMF+FX)"
),
)
# found a plus code! # found a plus code!
portion_locality = original_query.replace(portion_plus_code, "") portion_locality = original_query.replace(portion_plus_code, "")
portion_locality = portion_locality.strip().strip(",").strip() portion_locality = portion_locality.strip().strip(",").strip()
# did find plus code, but not full-length. :(
if (portion_locality == "") and (not validator.is_full(portion_plus_code)):
return Result[Query](
LatlongQuery(EMPTY_LATLONG),
error=IncompletePlusCodeError(
"_match_plus_code: Plus Code is not full-length (e.g., 6PH58QMF+FX)"
),
)
if behaviour.debug: if behaviour.debug:
behaviour.stderr.write(f"debug: {portion_plus_code=}, {portion_locality=}\n") behaviour.stderr.write(f"debug: {portion_plus_code=}, {portion_locality=}\n")
@ -589,7 +596,7 @@ def parse_query(
behaviour.stderr.write(f"debug: {behaviour.query=}\n") behaviour.stderr.write(f"debug: {behaviour.query=}\n")
# check if empty # check if empty
if behaviour.query == []: if (behaviour.query == []) or (behaviour.query == ""):
return Result[Query]( return Result[Query](
LatlongQuery(EMPTY_LATLONG), LatlongQuery(EMPTY_LATLONG),
error="query is empty", error="query is empty",
@ -604,28 +611,43 @@ def parse_query(
if isinstance(mpc_result.error, IncompletePlusCodeError): if isinstance(mpc_result.error, IncompletePlusCodeError):
return mpc_result # propagate back up to caller return mpc_result # propagate back up to caller
# handle query
original_query: str = ""
split_query: list[str] = []
if isinstance(behaviour.query, str):
original_query = behaviour.query
split_query = behaviour.query.split(" ")
else:
original_query = " ".join(behaviour.query)
split_query = behaviour.query
if behaviour.debug:
behaviour.stderr.write(f"debug: {split_query=}\ndebug: {original_query=}\n")
# not a plus/local code, try to match for latlong or string query # not a plus/local code, try to match for latlong or string query
match behaviour.query: match split_query:
case [single]: case [single]:
# possibly a: # possibly a:
# comma-seperated single-word-long latlong coord # comma-seperated single-word-long latlong coord
# (fallback) single word string query # (fallback) single word string query
if "," not in single: # no comma, not a latlong coord if "," not in single: # no comma, not a latlong coord
return Result[Query](StringQuery(" ".join(behaviour.query))) return Result[Query](StringQuery(original_query))
else: # has comma, possibly a latlong coord else: # has comma, possibly a latlong coord
split_query: list[str] = single.split(",") comma_split_single: list[str] = single.split(",")
if len(split_query) > 2: if len(comma_split_single) > 2:
return Result[Query]( return Result[Query](
LatlongQuery(EMPTY_LATLONG), LatlongQuery(EMPTY_LATLONG),
error="unable to parse latlong coord", error="unable to parse latlong coord",
) )
try: # try to type cast query try: # try to type cast query
latitude = float(split_query[0].strip(",")) latitude = float(comma_split_single[0].strip(","))
longitude = float(split_query[-1].strip(",")) longitude = float(comma_split_single[-1].strip(","))
except ValueError: # not a latlong coord, fallback except ValueError: # not a latlong coord, fallback
return Result[Query](StringQuery(single)) return Result[Query](StringQuery(single))
@ -650,7 +672,7 @@ def parse_query(
longitude = float(right_single.strip(",")) longitude = float(right_single.strip(","))
except ValueError: # not a latlong coord, fallback except ValueError: # not a latlong coord, fallback
return Result[Query](StringQuery(" ".join(behaviour.query))) return Result[Query](StringQuery(original_query))
else: # are floats, so is a latlong coord else: # are floats, so is a latlong coord
return Result[Query]( return Result[Query](
@ -661,7 +683,7 @@ def parse_query(
# possibly a: # possibly a:
# (fallback) space-seperated string query # (fallback) space-seperated string query
return Result[Query](StringQuery(" ".join(behaviour.query))) return Result[Query](StringQuery(original_query))
def handle_args() -> Behaviour: def handle_args() -> Behaviour:
@ -736,15 +758,16 @@ def _unique(l: Sequence[str]) -> list[str]:
return list(unique.keys()) return list(unique.keys())
def _generate_text(location: dict[str, str], debug: bool = False) -> str: def _generate_text(
location: dict[str, Any], behaviour: Behaviour, debug: bool = False
) -> str:
"""(internal function) generate shareable text from location dict""" """(internal function) generate shareable text from location dict"""
def _generate_text_line( def _generate_text_line(
line_number: int, line_number: int,
line_keys: Sequence[str], line_keys: Sequence[str],
seperator: str = ", ", seperator: str = ", ",
element_check: Callable[[str], bool] = lambda e: True, filter: Callable[[str], list[bool]] = lambda e: [True],
debug: bool = False,
) -> str: ) -> str:
""" """
(internal function) generate a line of shareable text from a list of keys (internal function) generate a line of shareable text from a list of keys
@ -754,11 +777,10 @@ def _generate_text(location: dict[str, str], debug: bool = False) -> str:
line number to prefix with line number to prefix with
line_keys: Sequence[str] line_keys: Sequence[str]
list of keys to .get() from location dict list of keys to .get() from location dict
element_check: Callable[[str], bool] = lambda e: True filter: Callable[[str], list[bool]] = lambda e: True
function that takes in a string and returns a bool, used to filter function that takes in a string and returns a list of bools, used to
elements from line_keys filter elements from line_keys. list will be passed to all(). if all
debug: bool = False returns True, then the element is kept.
whether to prefix line with line_number
returns str returns str
""" """
@ -766,60 +788,93 @@ def _generate_text(location: dict[str, str], debug: bool = False) -> str:
line_prefix: str = f"{line_number}\t" if debug else "" line_prefix: str = f"{line_number}\t" if debug else ""
basket: list[str] = [] basket: list[str] = []
for detail in _unique([location.get(detail, "") for detail in line_keys]): for detail in _unique([str(location.get(detail, "")) for detail in line_keys]):
if detail == "": if detail == "":
continue continue
if element_check(detail): # filtering: if all(filter(detail)) returns True,
# then the element is kept/added to 'basket'
if filter_status := all(detail_check := filter(detail)) is True:
if debug:
behaviour.stderr.write(
"debug: _generate_line_text: "
f"{str(detail_check):<20} -> {str(filter_status):<5} "
f"-------- '{detail}'\n"
)
basket.append(detail) basket.append(detail)
line = line_prefix + seperator.join(basket) else: # filter function returned False, so element is filtered/skipped
if debug:
behaviour.stderr.write(
"debug: _generate_line_text: "
f"{str(detail_check):<20} -> {str(filter_status):<5}"
f" filtered '{detail}'\n"
)
continue
line = line_prefix + seperator.join(basket)
return (line + "\n") if (line != "") else "" return (line + "\n") if (line != "") else ""
# iso3166-2 handling: this allows surplus to have special key arrangements for a
# specific iso3166-2 code for edge cases
# (https://en.wikipedia.org/wiki/ISO_3166-2)
# get iso3166-2 before doing anything
iso3166_2: str = ""
for key in location:
if key.startswith("iso3166"):
iso3166_2 = location.get(key, "")
# skeleton code to allow for changing keys based on iso3166-2 code
st_line0_keys = SHAREABLE_TEXT_LINE_0_KEYS
st_line1_keys = SHAREABLE_TEXT_LINE_1_KEYS
st_line2_keys = SHAREABLE_TEXT_LINE_2_KEYS
st_line3_keys = SHAREABLE_TEXT_LINE_3_KEYS
st_line4_keys = SHAREABLE_TEXT_LINE_4_KEYS
st_line5_keys = SHAREABLE_TEXT_LINE_5_KEYS
st_line6_keys = SHAREABLE_TEXT_LINE_6_KEYS
st_names = SHAREABLE_TEXT_NAMES
match iso3166_2.split("-"):
case _:
pass
# start generating text
text: list[str] = [] text: list[str] = []
seen_names: list[str] = [ seen_names: list[str] = [
detail detail
for detail in _unique( for detail in _unique(
[ [str(location.get(location_key, "")) for location_key in st_names]
location.get(location_key, "")
for location_key in (
OUTPUT_LINE_0_KEYS
+ OUTPUT_LINE_1_KEYS
+ OUTPUT_LINE_2_KEYS
+ OUTPUT_LINE_3_NAME
)
]
) )
if detail != "" if detail != ""
] ]
general_global_info: list[str] = [ general_global_info: list[str] = [
location.get(detail, "") for detail in OUTPUT_LINE_6_KEYS str(location.get(detail, "")) for detail in st_line6_keys
] ]
text.append(_generate_text_line(0, OUTPUT_LINE_0_KEYS, debug=debug)) text.append(_generate_text_line(0, st_line0_keys))
text.append(_generate_text_line(1, OUTPUT_LINE_1_KEYS, debug=debug)) text.append(_generate_text_line(1, st_line1_keys))
text.append(_generate_text_line(2, OUTPUT_LINE_2_KEYS, debug=debug)) text.append(_generate_text_line(2, st_line2_keys))
text.append(_generate_text_line(3, OUTPUT_LINE_3_KEYS, seperator=" ", debug=debug)) text.append(_generate_text_line(3, st_line3_keys, seperator=" "))
text.append( text.append(
_generate_text_line( _generate_text_line(
4, 4,
OUTPUT_LINE_4_KEYS, st_line4_keys,
element_check=lambda ak: all( filter=lambda ak: [
[ # everything here should be True if the element is to be kept
ak not in general_global_info, ak not in general_global_info,
any(True if (ak not in sn) else False for sn in seen_names), not any(True if (ak in sn) else False for sn in seen_names),
] ],
),
debug=debug,
) )
) )
text.append(_generate_text_line(5, OUTPUT_LINE_5_KEYS, debug=debug)) text.append(_generate_text_line(5, st_line5_keys))
text.append(_generate_text_line(6, OUTPUT_LINE_6_KEYS, debug=debug)) text.append(_generate_text_line(6, st_line6_keys))
return "".join(text) return "".join(_unique(text)).rstrip()
def surplus( def surplus(
@ -852,8 +907,9 @@ def surplus(
if behaviour.debug: if behaviour.debug:
behaviour.stderr.write(f"debug: {latlong.get()=}\n") behaviour.stderr.write(f"debug: {latlong.get()=}\n")
# reverse location and handle result
try: try:
location: dict[str, str] = behaviour.reverser(latlong.get()) location: dict[str, Any] = behaviour.reverser(latlong.get())
except Exception as exc: except Exception as exc:
return Result[str]("", error=exc) return Result[str]("", error=exc)
@ -861,17 +917,21 @@ def surplus(
if behaviour.debug: if behaviour.debug:
behaviour.stderr.write(f"debug: {location=}\n") behaviour.stderr.write(f"debug: {location=}\n")
# generate text
if behaviour.debug: if behaviour.debug:
behaviour.stderr.write( behaviour.stderr.write(
_generate_text( _generate_text(
location=location, location=location,
behaviour=behaviour,
debug=behaviour.debug, debug=behaviour.debug,
) )
+ "\n"
) )
text = _generate_text( text = _generate_text(
location=location, location=location,
).rstrip() behaviour=behaviour,
)
return Result[str](text) return Result[str](text)
@ -918,13 +978,13 @@ def cli() -> int:
# parse query and handle result # parse query and handle result
query = parse_query(behaviour=behaviour) query = parse_query(behaviour=behaviour)
if behaviour.debug:
behaviour.stderr.write(f"debug: {query=}\n")
if not query: if not query:
behaviour.stderr.write(f"error: {query.cry(string=not behaviour.debug)}\n") behaviour.stderr.write(f"error: {query.cry(string=not behaviour.debug)}\n")
return -1 return -1
if behaviour.debug:
behaviour.stderr.write(f"debug: {query.get()=}\n")
# run surplus # run surplus
text = surplus( text = surplus(
query=query.get(), query=query.get(),

78
test.py
View file

@ -39,6 +39,7 @@ from typing import Final, NamedTuple
import surplus import surplus
INDENT: Final[int] = 3 INDENT: Final[int] = 3
MINIMUM_PASS_RATE: Final[float] = 0.7 # because results can be flaky
class ContinuityTest(NamedTuple): class ContinuityTest(NamedTuple):
@ -54,10 +55,8 @@ class TestFailure(NamedTuple):
tests: list[ContinuityTest] = [ tests: list[ContinuityTest] = [
ContinuityTest( ContinuityTest(
query="8RPQ+JW Singapore", query="8R3M+F8 Singapore",
expected=( expected=("Wisma Atria\n" "435 Orchard Road\n" "238877\n" "Central, Singapore"),
"Caldecott Stn Exit 4\n" "Toa Payoh Link\n" "298106\n" "Central, Singapore"
),
), ),
ContinuityTest( ContinuityTest(
query="9R3J+R9 Singapore", query="9R3J+R9 Singapore",
@ -73,18 +72,32 @@ tests: list[ContinuityTest] = [
query="3RQ3+HW3 Pemping, Batam City, Riau Islands, Indonesia", query="3RQ3+HW3 Pemping, Batam City, Riau Islands, Indonesia",
expected=("Batam\n" "Kepulauan Riau, Indonesia"), expected=("Batam\n" "Kepulauan Riau, Indonesia"),
), ),
# ContinuityTest(
# query="CQPP+QM9 Singapore",
# expected=(
# "Woodlands Integrated Transport Hub\n" "738343\n" "Northwest, Singapore"
# ),
# ),
ContinuityTest( ContinuityTest(
query="8RRX+75Q Singapore", query="St Lucia, Queensland, Australia G227+XF",
expected=( expected=(
"Braddell Station/Blk 106\n" "The University of Queensland\n"
"Lorong 1 Toa Payoh\n" "Macquarie Street\n"
"319758\n" "St Lucia, Greater Brisbane\n"
"4072\n"
"Queensland, Australia"
),
),
ContinuityTest(
query="Ngee Ann Polytechnic, Singapore",
expected=(
"Ngee Ann Polytechnic\n"
"535 Clementi Road\n"
"Bukit Timah\n"
"599489\n"
"Northwest, Singapore"
),
),
ContinuityTest(
query="1.3521, 103.8198",
expected=(
"MacRitchie Nature Trail"
"Central Water Catchment\n"
"574325\n"
"Central, Singapore" "Central, Singapore"
), ),
), ),
@ -108,29 +121,29 @@ def main() -> int:
for idx, test in enumerate(tests, start=1): for idx, test in enumerate(tests, start=1):
print(f"[{idx}/{len(tests)}] {test.query}") print(f"[{idx}/{len(tests)}] {test.query}")
output: str = "" output: str = ""
behaviour = surplus.Behaviour(test.query)
try: try:
query = surplus.parse_query(query=test.query) query = surplus.parse_query(behaviour)
if query[0] is False: if not query:
raise QueryParseFailure(f"test query parse result returned False") raise QueryParseFailure(query.cry())
result = surplus.surplus(query=query[1]) result = surplus.surplus(query.get(), behaviour)
if result[0] is False: if not result:
raise SurplusFailure(result[1]) raise SurplusFailure(result.cry())
output = result[1] output = result.get()
print(indent(text=output, prefix=INDENT * " "))
if output != test.expected: if output != test.expected:
raise ContinuityFailure(f"test did not match output") raise ContinuityFailure("did not match output")
except Exception as exc: except Exception as exc:
failures.append(TestFailure(test=test, exception=exc, output=output)) failures.append(TestFailure(test=test, exception=exc, output=output))
stderr.write(indent(text="(fail)", prefix=INDENT * " ") + "\n") stderr.write(indent(text="(fail)", prefix=INDENT * " ") + "\n\n")
else: else:
stderr.write(indent(text="(pass)", prefix=INDENT * " ") + "\n\n") stderr.write(indent(text="(pass)", prefix=INDENT * " ") + "\n\n")
@ -153,9 +166,20 @@ def main() -> int:
+ (indent(text=fail.output, prefix=(2 * INDENT) * " ")) + (indent(text=fail.output, prefix=(2 * INDENT) * " "))
) )
print(f"\ncomplete: {len(tests) - len(failures)} passed, {len(failures)} failed") passes = len(tests) - len(failures)
pass_rate = round(passes / len(tests), 2)
return len(failures) print(
f"complete: {passes} passed, {len(failures)} failed "
f"({pass_rate * 100:.0f}%/{pass_rate * 100:.0f}%)"
)
if passes < MINIMUM_PASS_RATE:
print("continuity pass rate is under minimum, test suite failed ;<")
return 1
print("continuity tests passed :>")
return 0
if __name__ == "__main__": if __name__ == "__main__":