|
| 1 | +# ruff: noqa: S603 |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +import argparse |
| 5 | +import dataclasses |
| 6 | +import importlib.util |
| 7 | +import inspect |
| 8 | +from pathlib import Path |
| 9 | +import subprocess |
| 10 | +import sys |
| 11 | +import types |
| 12 | +import typing |
| 13 | +from typing import Any, get_args, get_origin |
| 14 | + |
| 15 | + |
| 16 | +def _module_from_path(module_path: Path) -> types.ModuleType: |
| 17 | + spec = importlib.util.spec_from_file_location(module_path.stem, str(module_path)) |
| 18 | + if spec is None or spec.loader is None: |
| 19 | + msg = f"Unable to load module from {module_path}" |
| 20 | + raise RuntimeError(msg) |
| 21 | + module = importlib.util.module_from_spec(spec) |
| 22 | + sys.modules[module_path.stem] = module |
| 23 | + spec.loader.exec_module(module) |
| 24 | + return module |
| 25 | + |
| 26 | + |
| 27 | +def _compute_dotted_import(module_path: Path) -> str: |
| 28 | + # Try to compute dotted path relative to repo's src directory |
| 29 | + repo_root = Path(__file__).resolve().parents[1] |
| 30 | + src_dir = repo_root / "src" |
| 31 | + try: |
| 32 | + rel = module_path.resolve().relative_to(src_dir.resolve()) |
| 33 | + dotted = ".".join(rel.with_suffix("").parts) |
| 34 | + return dotted |
| 35 | + except Exception: |
| 36 | + # Fallback: use module name only (may fail if not on sys.path) |
| 37 | + return module_path.stem |
| 38 | + |
| 39 | + |
| 40 | +def _is_dataclass_type(tp: Any) -> bool: |
| 41 | + try: |
| 42 | + return inspect.isclass(tp) and dataclasses.is_dataclass(tp) |
| 43 | + except Exception: |
| 44 | + return False |
| 45 | + |
| 46 | + |
| 47 | +def _unwrap_optional(tp: Any) -> Any: |
| 48 | + origin = get_origin(tp) |
| 49 | + if origin is None: |
| 50 | + return tp |
| 51 | + if origin in (list, dict, tuple): |
| 52 | + return tp |
| 53 | + if ( |
| 54 | + origin is typing.Union |
| 55 | + or str(origin) == "typing.Union" |
| 56 | + or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) |
| 57 | + ): |
| 58 | + args = [a for a in get_args(tp) if a is not type(None)] |
| 59 | + return args[0] if args else tp |
| 60 | + return tp |
| 61 | + |
| 62 | + |
| 63 | +def _choose_union_member(tp: Any) -> Any: |
| 64 | + origin = get_origin(tp) |
| 65 | + if origin is None: |
| 66 | + return tp |
| 67 | + if ( |
| 68 | + origin is typing.Union |
| 69 | + or str(origin) == "typing.Union" |
| 70 | + or (getattr(types, "UnionType", None) is not None and origin is types.UnionType) |
| 71 | + ): |
| 72 | + args = list(get_args(tp)) |
| 73 | + for preferred in (str, int, float, bool): |
| 74 | + if preferred in args: |
| 75 | + return preferred |
| 76 | + for arg in args: |
| 77 | + if arg is not type(None): |
| 78 | + return arg |
| 79 | + return tp |
| 80 | + |
| 81 | + |
| 82 | +def _collect_dataclass_types( |
| 83 | + cls: type[Any], seen: set[type[Any]] | None = None |
| 84 | +) -> set[type[Any]]: |
| 85 | + if seen is None: |
| 86 | + seen = set() |
| 87 | + if cls in seen: |
| 88 | + return seen |
| 89 | + seen.add(cls) |
| 90 | + try: |
| 91 | + type_hints = typing.get_type_hints(cls, include_extras=True) |
| 92 | + except Exception: |
| 93 | + type_hints = {} |
| 94 | + for f in dataclasses.fields(cls): |
| 95 | + tp = type_hints.get(f.name, f.type) |
| 96 | + tp = _choose_union_member(tp) |
| 97 | + tp = _unwrap_optional(tp) |
| 98 | + origin = get_origin(tp) |
| 99 | + args = get_args(tp) |
| 100 | + if origin is list and args: |
| 101 | + inner = _unwrap_optional(_choose_union_member(args[0])) |
| 102 | + if _is_dataclass_type(inner): |
| 103 | + _collect_dataclass_types(inner, seen) |
| 104 | + elif _is_dataclass_type(tp): |
| 105 | + _collect_dataclass_types(tp, seen) |
| 106 | + return seen |
| 107 | + |
| 108 | + |
| 109 | +def _py_literal_for_scalar(tp: Any, field_name: str, list_index: int | None) -> str: |
| 110 | + lname = field_name.lower() |
| 111 | + if "identifier" in lname: |
| 112 | + suffix = (list_index + 1) if isinstance(list_index, int) else 1 |
| 113 | + base = field_name.replace("_", "-").lower() |
| 114 | + return f'"{base}-{suffix:03d}"' |
| 115 | + if tp is str: |
| 116 | + return f'"example {field_name.replace("_", " ")}"' |
| 117 | + if tp is int: |
| 118 | + return "1" |
| 119 | + if tp is float: |
| 120 | + return "1.0" |
| 121 | + if tp is bool: |
| 122 | + return "True" |
| 123 | + return f'"example {field_name.replace("_", " ")}"' |
| 124 | + |
| 125 | + |
| 126 | +def _render_value_expr( |
| 127 | + tp: Any, field_name: str, indent: int, list_index: int | None |
| 128 | +) -> str: |
| 129 | + tp = _choose_union_member(tp) |
| 130 | + tp = _unwrap_optional(tp) |
| 131 | + origin = get_origin(tp) |
| 132 | + args = get_args(tp) |
| 133 | + sp = " " * indent |
| 134 | + # Dicts: generate one representative key/value pair |
| 135 | + if origin is dict and args: |
| 136 | + _, val_tp = args[0], args[1] |
| 137 | + key_expr = _py_literal_for_scalar(str, f"{field_name}_key", list_index) |
| 138 | + val_expr = _render_value_expr( |
| 139 | + val_tp, f"{field_name}_value", indent + 4, list_index |
| 140 | + ) |
| 141 | + return "{\n" + f"{sp} {key_expr}: {val_expr}\n" + f"{sp}" + "}" |
| 142 | + if origin is list and args: |
| 143 | + inner = args[0] |
| 144 | + items = [] |
| 145 | + for i in range(2): |
| 146 | + items.append(_render_value_expr(inner, field_name, indent + 4, i)) |
| 147 | + # Determine if items are multiline (start with a class constructor or open bracket) |
| 148 | + multiline = any( |
| 149 | + x.strip().startswith(tuple("ABCDEFGHIJKLMNOPQRSTUVWXYZ")) |
| 150 | + or x.strip().startswith("[") |
| 151 | + or "\n" in x |
| 152 | + for x in items |
| 153 | + ) |
| 154 | + if multiline: |
| 155 | + inner_joined = ",\n".join(items) |
| 156 | + return f"[\n{inner_joined}\n{sp}]" |
| 157 | + return f"[{', '.join(items)}]" |
| 158 | + if _is_dataclass_type(tp): |
| 159 | + return _render_dataclass_expr(tp, indent, list_index) |
| 160 | + # scalar |
| 161 | + chosen = _choose_union_member(tp) |
| 162 | + chosen = _unwrap_optional(chosen) |
| 163 | + return _py_literal_for_scalar(chosen, field_name, list_index) |
| 164 | + |
| 165 | + |
| 166 | +def _render_dataclass_expr(cls: type[Any], indent: int, list_index: int | None) -> str: |
| 167 | + sp = " " * indent |
| 168 | + lines = [f"{sp}{cls.__name__}("] |
| 169 | + try: |
| 170 | + module_globals = sys.modules.get(cls.__module__).__dict__ |
| 171 | + except Exception: |
| 172 | + module_globals = None |
| 173 | + try: |
| 174 | + type_hints = typing.get_type_hints( |
| 175 | + cls, globalns=module_globals, localns=module_globals, include_extras=True |
| 176 | + ) |
| 177 | + except Exception: |
| 178 | + type_hints = {} |
| 179 | + for f in dataclasses.fields(cls): |
| 180 | + key = f.name |
| 181 | + tp = type_hints.get(f.name, f.type) |
| 182 | + val_expr = _render_value_expr(tp, key, indent + 4, list_index) |
| 183 | + lines.append(f"{sp} {key}={val_expr},") |
| 184 | + lines.append(f"{sp})") |
| 185 | + return "\n".join(lines) |
| 186 | + |
| 187 | + |
| 188 | +def generate_dummy_data_py( |
| 189 | + mapper_file: Path, top_class_name: str, output: Path |
| 190 | +) -> None: |
| 191 | + # Ensure repo 'src' is on sys.path for module imports |
| 192 | + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) |
| 193 | + module = _module_from_path(mapper_file) |
| 194 | + |
| 195 | + top_cls = getattr(module, top_class_name, None) |
| 196 | + if top_cls is None or not _is_dataclass_type(top_cls): |
| 197 | + msg = f"Dataclass '{top_class_name}' not found in {mapper_file}" |
| 198 | + raise RuntimeError(msg) |
| 199 | + |
| 200 | + # Collect all dataclasses used so we can import them in the generated file |
| 201 | + dataclass_types = _collect_dataclass_types(top_cls) |
| 202 | + # Ensure top class first in import list, followed by others sorted |
| 203 | + class_names = [ |
| 204 | + top_cls.__name__, |
| 205 | + *sorted([c.__name__ for c in dataclass_types if c is not top_cls]), |
| 206 | + ] |
| 207 | + |
| 208 | + dotted_import = _compute_dotted_import(mapper_file) |
| 209 | + header = f"""from __future__ import annotations |
| 210 | +
|
| 211 | +from dataclasses import asdict |
| 212 | +import json |
| 213 | +
|
| 214 | +from {dotted_import} import ( |
| 215 | + {", ".join(class_names)} |
| 216 | +) |
| 217 | +
|
| 218 | +
|
| 219 | +def build_dummy_data() -> {top_cls.__name__}: |
| 220 | +""" |
| 221 | + body_expr = _render_dataclass_expr(top_cls, indent=4, list_index=None) |
| 222 | + |
| 223 | + content = header + " return " + body_expr |
| 224 | + |
| 225 | + output.parent.mkdir(parents=True, exist_ok=True) |
| 226 | + output.write_text(content, encoding="utf-8") |
| 227 | + |
| 228 | + |
| 229 | +def main() -> None: |
| 230 | + parser = argparse.ArgumentParser( |
| 231 | + description="Generate a Python file that builds dummy Data-like object for any mapper." |
| 232 | + ) |
| 233 | + parser.add_argument( |
| 234 | + "--module-file", |
| 235 | + required=True, |
| 236 | + type=Path, |
| 237 | + help="Path to the mapper module file (e.g., schema_mappers/.../mass_spectrometry.py).", |
| 238 | + ) |
| 239 | + parser.add_argument( |
| 240 | + "--output", |
| 241 | + required=True, |
| 242 | + type=Path, |
| 243 | + help="Path to write the generated Python file.", |
| 244 | + ) |
| 245 | + parser.add_argument( |
| 246 | + "--top-class", |
| 247 | + default="Data", |
| 248 | + help="Top-level dataclass name to start from. Defaults to 'Data'.", |
| 249 | + ) |
| 250 | + args = parser.parse_args() |
| 251 | + generate_dummy_data_py(args.module_file, args.top_class, args.output) |
| 252 | + print(f"Wrote dummy data generator to: {args.output}") |
| 253 | + output_path = str(args.output) |
| 254 | + subprocess.run(["ruff", "check", "--fix", output_path], check=True) # noqa: S607 |
| 255 | + |
| 256 | + |
| 257 | +if __name__ == "__main__": |
| 258 | + main() |
0 commit comments