|
1 | 1 | # Standard library imports |
2 | 2 | import json |
3 | 3 | import os |
4 | | -import uuid |
| 4 | +from typing import Dict, Any, List, TypedDict, cast |
5 | 5 |
|
6 | 6 | # Third party imports |
7 | 7 | import flask |
|
15 | 15 | from opengeodeweb_back.utils_functions import save_all_viewables_and_return_info |
16 | 16 |
|
17 | 17 | routes = flask.Blueprint("create", __name__, url_prefix="/opengeodeweb_back/create") |
18 | | - |
19 | 18 | schemas = os.path.join(os.path.dirname(__file__), "schemas") |
20 | 19 |
|
| 20 | +# --- Type definitions for request validation --- |
| 21 | +class Point(TypedDict): |
| 22 | + x: float |
| 23 | + y: float |
| 24 | + |
| 25 | +class CreatePointRequest(TypedDict): |
| 26 | + name: str |
| 27 | + x: float |
| 28 | + y: float |
| 29 | + z: float |
| 30 | + |
| 31 | +class CreateAOIRequest(TypedDict): |
| 32 | + name: str |
| 33 | + points: List[Point] |
| 34 | + z: float |
| 35 | + |
21 | 36 | # Load schema for point creation |
22 | 37 | with open(os.path.join(schemas, "create_point.json"), "r") as file: |
23 | | - create_point_json = json.load(file) |
| 38 | + create_point_json = cast(Dict[str, Any], json.load(file)) |
24 | 39 |
|
25 | 40 | @routes.route( |
26 | 41 | create_point_json["route"], |
27 | 42 | methods=create_point_json["methods"] |
28 | 43 | ) |
29 | | -def create_point(): |
| 44 | +def create_point() -> flask.Response: |
30 | 45 | """Endpoint to create a single point in 3D space.""" |
31 | 46 | print(f"create_point : {flask.request=}", flush=True) |
32 | | - |
33 | 47 | utils_functions.validate_request(flask.request, create_point_json) |
34 | 48 |
|
35 | | - # Extract data from request |
36 | | - title = flask.request.json["title"] |
37 | | - x = flask.request.json["x"] |
38 | | - y = flask.request.json["y"] |
39 | | - z = flask.request.json["z"] |
40 | | - |
41 | | - # Create the point set |
42 | | - class_ = geode_functions.geode_object_class("PointSet3D") |
43 | | - point_set = class_.create() |
44 | | - builder = geode_functions.create_builder("PointSet3D", point_set) |
45 | | - builder.create_point(opengeode.Point3D([x, y, z])) |
46 | | - builder.set_name(title) |
47 | | - |
48 | | - # Save and get info |
49 | | - result = save_all_viewables_and_return_info( |
50 | | - geode_object="PointSet3D", |
51 | | - data=point_set, |
52 | | - input_file=None, |
53 | | - additional_files=[] |
54 | | - ) |
55 | | - |
56 | | - # Prepare response with the title |
57 | | - response = { |
58 | | - "viewable_file_name": result["viewable_file_name"], |
59 | | - "id": result["id"], |
60 | | - "name": title, |
61 | | - "native_file_name": result["native_file_name"], |
62 | | - "object_type": result["object_type"], |
63 | | - "geode_object": result["geode_object"], |
64 | | - } |
65 | | - |
66 | | - # Add binary_light_viewable if it exists |
67 | | - if "binary_light_viewable" in result: |
68 | | - response["binary_light_viewable"] = result["binary_light_viewable"] |
69 | | - |
70 | | - return flask.make_response(response, 200) |
| 49 | + try: |
| 50 | + # Extract and validate data from request |
| 51 | + request_data = cast(CreatePointRequest, flask.request.json) |
| 52 | + name: str = request_data["name"] |
| 53 | + x: float = request_data["x"] |
| 54 | + y: float = request_data["y"] |
| 55 | + z: float = request_data["z"] |
| 56 | + |
| 57 | + # Create the point set |
| 58 | + class_ = geode_functions.geode_object_class("PointSet3D") |
| 59 | + point_set = class_.create() |
| 60 | + builder = geode_functions.create_builder("PointSet3D", point_set) |
| 61 | + builder.create_point(opengeode.Point3D([x, y, z])) # type: ignore |
| 62 | + builder.set_name(name) |
| 63 | + |
| 64 | + # Save and get info |
| 65 | + result = save_all_viewables_and_return_info( |
| 66 | + geode_object="PointSet3D", |
| 67 | + data=point_set, |
| 68 | + ) |
71 | 69 |
|
| 70 | + # Vérifier que binary_light_viewable est présent |
| 71 | + if "binary_light_viewable" not in result: |
| 72 | + raise ValueError("binary_light_viewable is missing in the result") |
| 73 | + |
| 74 | + # Prepare response |
| 75 | + response: Dict[str, Any] = { |
| 76 | + "viewable_file_name": result["viewable_file_name"], |
| 77 | + "id": result["id"], |
| 78 | + "name": name, |
| 79 | + "native_file_name": result["native_file_name"], |
| 80 | + "object_type": result["object_type"], |
| 81 | + "geode_object": result["geode_object"], |
| 82 | + "binary_light_viewable": result["binary_light_viewable"], |
| 83 | + } |
| 84 | + |
| 85 | + return flask.make_response(response, 200) |
| 86 | + |
| 87 | + except Exception as e: |
| 88 | + return flask.make_response({"error": str(e)}, 500) |
72 | 89 |
|
73 | 90 | # Load schema for AOI creation |
74 | 91 | with open(os.path.join(schemas, "create_aoi.json"), "r") as file: |
75 | | - create_aoi_json = json.load(file) |
| 92 | + create_aoi_json = cast(Dict[str, Any], json.load(file)) |
76 | 93 |
|
77 | 94 | @routes.route( |
78 | 95 | create_aoi_json["route"], |
79 | 96 | methods=create_aoi_json["methods"] |
80 | 97 | ) |
81 | | -def create_aoi(): |
| 98 | +def create_aoi() -> flask.Response: |
82 | 99 | """Endpoint to create an Area of Interest (AOI) as an EdgedCurve3D.""" |
83 | 100 | print(f"create_aoi : {flask.request=}", flush=True) |
84 | | - |
85 | 101 | utils_functions.validate_request(flask.request, create_aoi_json) |
86 | 102 |
|
87 | | - # Extract data from request |
88 | | - name = flask.request.json["name"] |
89 | | - points = flask.request.json["points"] |
90 | | - z = flask.request.json["z"] |
91 | | - |
92 | | - # Create the edged curve |
93 | | - class_ = geode_functions.geode_object_class("EdgedCurve3D") |
94 | | - edged_curve = class_.create() |
95 | | - builder = geode_functions.create_builder("EdgedCurve3D", edged_curve) |
96 | | - builder.set_name(name) |
97 | | - |
98 | | - # Create vertices first |
99 | | - vertex_indices = [] |
100 | | - for point in points: |
101 | | - vertex_id = builder.create_point(opengeode.Point3D([point["x"], point["y"], z])) |
102 | | - vertex_indices.append(vertex_id) |
103 | | - |
104 | | - # Create edges between consecutive vertices and close the loop |
105 | | - num_vertices = len(vertex_indices) |
106 | | - for i in range(num_vertices): |
107 | | - edge_id = builder.create_edge() |
108 | | - next_i = (i + 1) % num_vertices # Wrap around to close the loop |
109 | | - builder.set_edge_vertex( |
110 | | - opengeode.EdgeVertex(edge_id, 0), vertex_indices[i] |
111 | | - ) |
112 | | - builder.set_edge_vertex( |
113 | | - opengeode.EdgeVertex(edge_id, 1), vertex_indices[next_i] |
| 103 | + try: |
| 104 | + # Extract and validate data from request |
| 105 | + request_data = cast(CreateAOIRequest, flask.request.json) |
| 106 | + name: str = request_data["name"] |
| 107 | + points: List[Point] = request_data["points"] |
| 108 | + z: float = request_data["z"] |
| 109 | + |
| 110 | + # Create the edged curve |
| 111 | + class_ = geode_functions.geode_object_class("EdgedCurve3D") |
| 112 | + edged_curve = class_.create() |
| 113 | + builder = geode_functions.create_builder("EdgedCurve3D", edged_curve) |
| 114 | + builder.set_name(name) |
| 115 | + |
| 116 | + # Create vertices first |
| 117 | + vertex_indices: List[int] = [] |
| 118 | + for point in points: |
| 119 | + vertex_id = builder.create_point(opengeode.Point3D([point["x"], point["y"], z])) # type: ignore |
| 120 | + vertex_indices.append(vertex_id) |
| 121 | + |
| 122 | + # Create edges between consecutive vertices and close the loop |
| 123 | + num_vertices = len(vertex_indices) |
| 124 | + for i in range(num_vertices): |
| 125 | + next_i = (i + 1) % num_vertices |
| 126 | + edge_id = builder.create_edge() |
| 127 | + builder.set_edge_vertex(opengeode.EdgeVertex(edge_id, 0), vertex_indices[i]) # type: ignore |
| 128 | + builder.set_edge_vertex(opengeode.EdgeVertex(edge_id, 1), vertex_indices[next_i]) # type: ignore |
| 129 | + |
| 130 | + # Save and get info |
| 131 | + result = save_all_viewables_and_return_info( |
| 132 | + geode_object="EdgedCurve3D", |
| 133 | + data=edged_curve, |
114 | 134 | ) |
115 | 135 |
|
116 | | - # Save and get info |
117 | | - result = save_all_viewables_and_return_info( |
118 | | - geode_object="EdgedCurve3D", |
119 | | - data=edged_curve, |
120 | | - input_file=None, |
121 | | - additional_files=[] |
122 | | - ) |
123 | | - |
124 | | - # Prepare response |
125 | | - response = { |
126 | | - "viewable_file_name": result["viewable_file_name"], |
127 | | - "id": result["id"], |
128 | | - "name": name, |
129 | | - "native_file_name": result["native_file_name"], |
130 | | - "object_type": result["object_type"], |
131 | | - "geode_object": result["geode_object"], |
132 | | - } |
133 | | - |
134 | | - # Add binary_light_viewable if it exists |
135 | | - if "binary_light_viewable" in result: |
136 | | - response["binary_light_viewable"] = result["binary_light_viewable"] |
137 | | - |
138 | | - return flask.make_response(response, 200) |
| 136 | + # Vérifier que binary_light_viewable est présent |
| 137 | + if "binary_light_viewable" not in result: |
| 138 | + raise ValueError("binary_light_viewable is missing in the result") |
| 139 | + |
| 140 | + # Prepare response |
| 141 | + response: Dict[str, Any] = { |
| 142 | + "viewable_file_name": result["viewable_file_name"], |
| 143 | + "id": result["id"], |
| 144 | + "name": name, |
| 145 | + "native_file_name": result["native_file_name"], |
| 146 | + "object_type": result["object_type"], |
| 147 | + "geode_object": result["geode_object"], |
| 148 | + "binary_light_viewable": result["binary_light_viewable"], |
| 149 | + } |
| 150 | + |
| 151 | + return flask.make_response(response, 200) |
| 152 | + |
| 153 | + except Exception as e: |
| 154 | + return flask.make_response({"error": str(e)}, 500) |
0 commit comments