Browse Source

Move flatpy to a separate repo and include it as a submodule.

main
Ben Niemann 3 weeks ago
parent
commit
ef5527be03
  1. 2
      .gitignore
  2. 3
      .gitmodules
  3. 2
      .projectile
  4. 4
      3rdparty/README.md
  5. 1
      3rdparty/flatpy
  6. 2
      appimg/build.sh
  7. 249
      flatpy/flatpy/fbs.jinja2
  8. 183
      flatpy/flatpy/main.py
  9. 145
      flatpy/flatpy/reflection.fbs
  10. 416
      flatpy/flatpy/test_flatpy.py
  11. 0
      flatpy/pytest.ini
  12. 37
      flatpy/setup.py
  13. 2
      noisicaa.nix

2
.gitignore

@ -5,8 +5,6 @@
/.lock-waf*
/build/
/result
/flatpy/flatpy.egg-info/
/flatpy/.env/
/playground/
/website/.venv/
/website/_build/

3
.gitmodules

@ -10,3 +10,6 @@
[submodule "3rdparty/subprocess"]
path = 3rdparty/subprocess
url = https://github.com/benman64/subprocess
[submodule "3rdparty/flatpy"]
path = 3rdparty/flatpy
url = https://git.odahoda.de/pink/flatpy.git

2
.projectile

@ -1,10 +1,10 @@
-/build/
-/.waf*
-/.lock-waf*
-/flatpy/.env/
-/website/.venv/
-/website/_serve/
-/website/_build/
-/3rdparty/qnanopainter/
-/3rdparty/subprocess/
-/3rdparty/dep_sort/
-/3rdparty/flatpy/

4
3rdparty/README.md

@ -11,3 +11,7 @@ Some of the more obscure packages, which are not generally available in distribu
* subprocess: cross platform subprocess library for c++ similar to design of python subprocess
- Source: https://github.com/benman64/subprocess
- License: MIT
* flatpy: An alternative Python code generator for flatbuffers.
- Source: https://git.odahoda.de/pink/flatpy
- License: Apache-2.0

1
3rdparty/flatpy

@ -0,0 +1 @@
Subproject commit bf1a4e391ffcd484f627ba81ae54ffe0c073acb9

2
appimg/build.sh

@ -122,7 +122,7 @@ fi
### flatpy
if [ ! -x /usr/local/bin/flatpy ]; then
python3 -m pip install /src/flatpy/
python3 -m pip install /src/3rdparty/flatpy/
fi
### python build dependencies

249
flatpy/flatpy/fbs.jinja2

@ -1,249 +0,0 @@
# Generated file, do not edit!
import enum
import flatbuffers
from typing import Any, Dict, List, Optional, Union
{% for mod, syms in schema.imports %}
from {{ mod }} import {{ ', '.join(syms) }}
{% endfor %}
NT = flatbuffers.number_types
{% for enum in schema.enums %}
{% if enum.emit %}
class {{ enum.name }}(enum.IntEnum):
{% for val in enum['values'] -%}
{{ val.name }} = {{ val.value }}
{% endfor -%}
{% endif %}
{% endfor %}
{% for obj in schema.objects %}
{% if obj.emit %}
class {{ obj.name }}:
def __init__(self, src: Union[flatbuffers.Table, bytes, bytearray]) -> None:
self._buf = None # type: Optional[bytes]
if isinstance(src, flatbuffers.Table):
self._tab = src
elif isinstance(src, (bytearray, bytes)):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, src, 0)
self._tab = flatbuffers.table.Table(src, n)
self._buf = src
else:
raise TypeError(type(src).__name__)
def Buffer(self) -> bytes:
assert self._buf is not None
return self._buf
{% for field in obj.fields %}
@property
def {{ field.name }}(self) -> {{ field.py_type }}:
{% if field.documentation | length -%}
"""{% for l in field.documentation %}{{ l|trim }}
{% endfor %}"""
{% endif -%}
{% if field.type.base_type == 'String' -%}
o = NT.UOffsetTFlags.py_type(self._tab.Offset({{ field.offset }}))
if o != 0:
return self._tab.String(o + self._tab.Pos).decode('utf-8')
return None
{% elif field.type.base_type == 'Vector' -%}
o = NT.UOffsetTFlags.py_type(self._tab.Offset({{ field.offset }}))
if o != 0:
base = self._tab.Vector(o)
{% if field.type.element == 'String' -%}
v = [] # type: List[str]
for i in range(self._tab.VectorLen(o)):
v.append(self._tab.String(base + i * NT.UOffsetTFlags.bytewidth).decode('utf-8'))
return v
{% elif field.type.element == 'Obj' -%}
v = [] # type: List[flatbuffers.table.Table]
for i in range(self._tab.VectorLen(o)):
o = base + i * NT.UOffsetTFlags.bytewidth
o = self._tab.Indirect(o)
obj = {{ schema.objects[field.type.index].name }}(flatbuffers.table.Table(self._tab.Bytes, o))
v.append(obj)
return v
{% elif field.type.element == 'UByte' -%}
return self._tab.Bytes[base:base+self._tab.VectorLen(o)]
{% else -%}
v = [] # type: List[int]
for i in range(self._tab.VectorLen(o)):
v.append(self._tab.Get(NT.{{ field.type.element | title }}Flags, base + i * NT.{{ field.type.element | title }}Flags.bytewidth))
return v
{% endif %}
return None
{% elif field.type.base_type == 'Obj' -%}
o = NT.UOffsetTFlags.py_type(self._tab.Offset({{ field.offset }}))
if o != 0:
tab = flatbuffers.table.Table(bytearray(), 0)
self._tab.Union(tab, o)
return {{ schema.objects[field.type.index].name }}(tab)
return None
{% elif field.type.base_type == 'Union' -%}
o = NT.UOffsetTFlags.py_type(self._tab.Offset({{ field.offset }}))
if o != 0:
obj = flatbuffers.table.Table(bytearray(), 0)
self._tab.Union(obj, o)
return obj
return None
{% elif field.type.base_type == 'UType' -%}
o = NT.UOffsetTFlags.py_type(self._tab.Offset({{ field.offset }}))
if o != 0:
return self._tab.Get(NT.Uint8Flags, o + self._tab.Pos)
return 0
{% else -%}
o = NT.UOffsetTFlags.py_type(self._tab.Offset({{ field.offset }}))
if o != 0:
return self._tab.Get(NT.{{ field.type.base_type | title }}Flags, o + self._tab.Pos)
return {{ field.default }}
{% endif %}
{% endfor %}
def Create{{obj.name}}(
builder: flatbuffers.Builder,{% if obj.fields %} *,{% endif %}
{%- for field in obj.fields %}
{{ field.name }}: Optional[{{ field.py_type }}]=None,
{%- endfor %}
) -> int:
{%- for field in obj.fields %}
{% if field.type.base_type == 'String' -%}
if {{ field.name }} is not None:
{{ field.name }}_offset = builder.CreateString({{ field.name }})
else:
{{ field.name }}_offset = None
{% elif field.type.base_type == 'Vector' -%}
if {{ field.name }} is not None:
{% if field.type.element == 'String' -%}
elems = [] # type: List[int]
for el in {{ field.name }}:
elems.append(builder.CreateString(el))
builder.StartVector(NT.UOffsetTFlags.bytewidth, len(elems), 0)
for el in reversed(elems):
builder.PrependUOffsetTRelative(el)
{{ field.name }}_offset = builder.EndVector()
{% elif field.type.element == 'Obj' -%}
builder.StartVector(NT.UOffsetTFlags.bytewidth, len({{ field.name }}), 0)
for el in reversed({{ field.name }}):
builder.PrependUOffsetTRelative(el)
{{ field.name }}_offset = builder.EndVector()
{% else -%}
builder.StartVector(NT.{{ field.type.element | title }}Flags.bytewidth, len({{ field.name }}), 0)
for el in reversed({{ field.name }}):
builder.Prepend(NT.{{ field.type.element | title }}Flags, el)
{{ field.name }}_offset = builder.EndVector()
{% endif %}
else:
{{ field.name }}_offset = None
{%- endif %}
{%- endfor %}
builder.StartObject({{ obj.fields | length }})
{%- for field in obj.fields %}
{% if field.type.base_type in ('String', 'Vector') -%}
if {{ field.name }}_offset is not None:
builder.PrependUOffsetTRelativeSlot({{ field.id }}, NT.UOffsetTFlags.py_type({{ field.name }}_offset), 0)
{% elif field.type.base_type in ('Obj', 'Union') -%}
if {{ field.name }} is not None:
builder.PrependUOffsetTRelativeSlot({{ field.id }}, NT.UOffsetTFlags.py_type({{ field.name }}), 0)
{% elif field.type.base_type == 'UType' -%}
if {{ field.name }} is not None:
builder.PrependSlot(NT.Uint8Flags, {{ field.id }}, NT.UOffsetTFlags.py_type({{ field.name }}), 0)
{% else -%}
if {{ field.name }} is not None:
builder.PrependSlot(NT.{{ field.type.base_type | title }}Flags, {{ field.id }}, {{ field.name }}, 0)
{% endif %}
{%- endfor %}
return builder.EndObject()
def Create{{obj.name}}FromJSON(
builder: flatbuffers.Builder, j: Dict[str, Any], strict: bool = True) -> int:
{%- for field in obj.fields %}
{{ field.name }} = j.pop('{{ field.name }}', None)
{% if field.type.base_type == 'String' -%}
if {{ field.name }} is not None:
{{ field.name }}_offset = builder.CreateString({{ field.name }})
else:
{{ field.name }}_offset = None
{% elif field.type.base_type == 'Obj' -%}
if {{ field.name }} is not None:
{{ field.name }}_offset = Create{{ schema.objects[field.type.index].name }}FromJSON(builder, {{ field.name }})
else:
{{ field.name }}_offset = None
{% elif field.type.base_type == 'Vector' -%}
if {{ field.name }} is not None:
{% if field.type.element == 'String' -%}
elems = [] # type: List[int]
for el in {{ field.name }}:
elems.append(builder.CreateString(el))
builder.StartVector(NT.UOffsetTFlags.bytewidth, len(elems), 0)
for el in reversed(elems):
builder.PrependUOffsetTRelative(el)
{{ field.name }}_offset = builder.EndVector()
{% elif field.type.element == 'Obj' -%}
elems = [] # type: List[int]
for el in {{ field.name }}:
elems.append(Create{{ schema.objects[field.type.index].name }}FromJSON(builder, el))
builder.StartVector(NT.UOffsetTFlags.bytewidth, len({{ field.name }}), 0)
for el in reversed(elems):
builder.PrependUOffsetTRelative(el)
{{ field.name }}_offset = builder.EndVector()
{% elif field.type.index >= 0 %}
builder.StartVector(NT.Int32Flags.bytewidth, len({{ field.name }}), 0)
for el in reversed({{ field.name }}):
builder.Prepend(NT.Int32Flags, {{ schema.enums[field.type.index].name }}[el])
{{ field.name }}_offset = builder.EndVector()
{% else -%}
builder.StartVector(NT.{{ field.type.element | title }}Flags.bytewidth, len({{ field.name }}), 0)
for el in reversed({{ field.name }}):
builder.Prepend(NT.{{ field.type.element | title }}Flags, el)
{{ field.name }}_offset = builder.EndVector()
{% endif %}
else:
{{ field.name }}_offset = None
{% elif field.type.index >= 0 %}
if {{ field.name }} is not None:
{{ field.name }} = {{ schema.enums[field.type.index].name }}[{{ field.name }}]
{%- endif %}
{%- endfor %}
if strict and j:
raise ValueError("Unknown field(s) in JSON input: {}".format(", ".join(sorted(j.keys()))))
builder.StartObject({{ obj.fields | length }})
{%- for field in obj.fields %}
{% if field.type.base_type in ('String', 'Vector', 'Obj') -%}
if {{ field.name }}_offset is not None:
builder.PrependUOffsetTRelativeSlot({{ field.id }}, NT.UOffsetTFlags.py_type({{ field.name }}_offset), 0)
{% elif field.type.base_type == 'Union' -%}
if {{ field.name }} is not None:
builder.PrependUOffsetTRelativeSlot({{ field.id }}, NT.UOffsetTFlags.py_type({{ field.name }}), 0)
{% elif field.type.base_type == 'UType' -%}
if {{ field.name }} is not None:
builder.PrependSlot(NT.Uint8Flags, {{ field.id }}, NT.UOffsetTFlags.py_type({{ field.name }}), 0)
{% else -%}
if {{ field.name }} is not None:
builder.PrependSlot(NT.{{ field.type.base_type | title }}Flags, {{ field.id }}, {{ field.name }}, 0)
{% endif %}
{%- endfor %}
return builder.EndObject()
{% endif %}
{% endfor %}

183
flatpy/flatpy/main.py

@ -1,183 +0,0 @@
import argparse
import json
import os.path
import re
import shutil
import subprocess
import stat
import sys
import jinja2
MIN_FLATC_VERSION = (2, 0, 0)
py_types = {
'Bool': 'bool',
'UInt8': 'int',
'UInt16': 'int',
'UInt32': 'int',
'UInt64': 'int',
'Int8': 'int',
'Int16': 'int',
'Int32': 'int',
'Int64': 'int',
'Float32': 'float',
'Float64': 'float',
'String': 'str',
'Vector': 'List',
'Union': 'flatbuffers.Table',
'Obj': 'flatbuffers.Table',
'UType': 'int',
}
defaults = {
'Bool': False,
'UInt8': 0,
'UInt16': 0,
'UInt32': 0,
'UInt64': 0,
'Int8': 0,
'Int16': 0,
'Int32': 0,
'Int64': 0,
'Float32': 0.0,
'Float64': 0.0,
'String': None,
'Vector': None,
'Union': None,
'Obj': None,
'UType': 0,
}
type_aliases = {
'Byte': 'Int8',
'UByte': 'UInt8',
'Short': 'Int16',
'UShort': 'UInt16',
'Int': 'Int32',
'UInt': 'UInt32',
'Long': 'Int64',
'ULong': 'UInt64',
'Float': 'Float32',
'Double': 'Float64',
}
def gen(args, base_name, schema_path):
with open(schema_path, 'r') as fp:
schema = json.load(fp)
for obj in schema['objects']:
for field in obj['fields']:
field['id'] = field.get('id', 0)
field['type']['base_type'] = type_aliases.get(field['type']['base_type'], field['type']['base_type'])
if 'element' in field['type']:
field['type']['element'] = type_aliases.get(field['type']['element'], field['type']['element'])
field['type']['index'] = field['type'].get('index', -1)
field['py_type'] = py_types[field['type']['base_type']]
if 'default_integer' in field:
if field['py_type'] == 'bool':
field['default'] = bool(field['default_integer'])
else:
field['default'] = field['default_integer']
elif 'default_real' in field:
field['default'] = field['default_real']
else:
field['default'] = defaults[field['type']['base_type']]
imports = {}
root_file = '//' + os.path.basename(args.path)
namespace = None
for obj in schema['objects'] + schema['enums']:
obj['emit'] = obj['declaration_file'] == root_file
if '.' in obj['name']:
assert namespace is None or obj['name'].rsplit('.', 1)[0] == namespace, obj['name']
namespace, obj['name'] = obj['name'].rsplit('.', 1)
if obj['declaration_file'] != root_file:
assert obj['declaration_file'].startswith('//')
assert obj['declaration_file'].endswith('.fbs')
assert '/' not in obj['declaration_file'][2:-4]
mod = '.' + obj['declaration_file'][2:-4] + '_generated'
imports.setdefault(mod, []).append(obj['name'])
schema['imports'] = [(m, sorted(s)) for m, s in imports.items()]
with open(os.path.join(os.path.dirname(__file__), 'fbs.jinja2'), 'r') as fp:
template = jinja2.Template(fp.read())
out = template.render(
schema=schema,
)
out_path = os.path.join(args.output, base_name + '_generated.py')
os.makedirs(os.path.dirname(out_path), exist_ok=True)
with open(out_path, 'w') as fp:
fp.write(out)
def main(argv):
argparser = argparse.ArgumentParser()
argparser.add_argument('--flatc', default='flatc')
argparser.add_argument('--output', '-o', default='.')
argparser.add_argument('path')
args = argparser.parse_args(argv[1:])
flatc_path = shutil.which(args.flatc)
if flatc_path is None:
sys.stderr.write("{} is not a valid executable.\n".format(args.flatc))
sys.exit(1)
p = subprocess.run([flatc_path, '--version'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
if p.returncode:
sys.stderr.write("'{} --version' failed.\n".format(flatc_path))
sys.exit(1)
flatc_version_str = p.stdout.decode('ascii', 'ignore').strip()
m = re.match(r'flatc version ([.\d]+)$', flatc_version_str)
if not m:
sys.stderr.write("{} does not look like the flatc compiler.\n".format(flatc_path))
sys.exit(1)
flatc_version = tuple(int(p) for p in m.group(1).split('.'))
if flatc_version < MIN_FLATC_VERSION:
sys.stderr.write("Need flatc V{}, found V{}.\n".format('.'.join(str(p) for p in MIN_FLATC_VERSION), '.'.join(str(p) for p in flatc_version)))
sys.exit(1)
base_name = os.path.splitext(os.path.basename(args.path))[0]
cmd = [
flatc_path,
'--binary',
'--schema',
'--bfbs-builtins',
'--bfbs-comments',
#'--bfbs-filenames', ctx.top_dir,
#'--include-prefix', self.inputs[0].parent.relpath(),
'-o', args.output,
args.path,
]
subprocess.run(cmd, check=True)
bfbs_path = os.path.join(args.output, base_name + '.bfbs')
assert os.path.isfile(bfbs_path), bfbs_path
schema_path = os.path.join(os.path.dirname(__file__), 'reflection.fbs')
assert os.path.isfile(schema_path), schema_path
cmd = [
flatc_path,
'--json',
'--strict-json',
'-o', args.output,
schema_path,
'--',
bfbs_path,
]
subprocess.run(cmd, check=True)
json_path = os.path.join(args.output, base_name + '.json')
assert os.path.isfile(json_path), json_path
gen(args, base_name, json_path)
return 0
def run():
main(sys.argv)

145
flatpy/flatpy/reflection.fbs

@ -1,145 +0,0 @@
// This schema defines objects that represent a parsed schema, like
// the binary version of a .fbs file.
// This could be used to operate on unknown FlatBuffers at runtime.
// It can even ... represent itself (!)
namespace reflection;
// These must correspond to the enum in idl.h.
enum BaseType : byte {
None,
UType,
Bool,
Byte,
UByte,
Short,
UShort,
Int,
UInt,
Long,
ULong,
Float,
Double,
String,
Vector,
Obj, // Used for tables & structs.
Union,
Array,
// Add any new type above this value.
MaxBaseType
}
table Type {
base_type:BaseType;
element:BaseType = None; // Only if base_type == Vector
// or base_type == Array.
index:int = -1; // If base_type == Object, index into "objects" below.
// If base_type == Union, UnionType, or integral derived
// from an enum, index into "enums" below.
fixed_length:uint16 = 0; // Only if base_type == Array.
}
table KeyValue {
key:string (required, key);
value:string;
}
table EnumVal {
name:string (required);
value:long (key);
object:Object (deprecated);
union_type:Type;
documentation:[string];
}
table Enum {
name:string (required, key);
values:[EnumVal] (required); // In order of their values.
is_union:bool = false;
underlying_type:Type (required);
attributes:[KeyValue];
documentation:[string];
/// File that this Enum is declared in.
declaration_file: string;
}
table Field {
name:string (required, key);
type:Type (required);
id:ushort;
offset:ushort; // Offset into the vtable for tables, or into the struct.
default_integer:long = 0;
default_real:double = 0.0;
deprecated:bool = false;
required:bool = false;
key:bool = false;
attributes:[KeyValue];
documentation:[string];
optional:bool = false;
}
table Object { // Used for both tables and structs.
name:string (required, key);
fields:[Field] (required); // Sorted.
is_struct:bool = false;
minalign:int;
bytesize:int; // For structs.
attributes:[KeyValue];
documentation:[string];
/// File that this Object is declared in.
declaration_file: string;
}
table RPCCall {
name:string (required, key);
request:Object (required); // must be a table (not a struct)
response:Object (required); // must be a table (not a struct)
attributes:[KeyValue];
documentation:[string];
}
table Service {
name:string (required, key);
calls:[RPCCall];
attributes:[KeyValue];
documentation:[string];
/// File that this Service is declared in.
declaration_file: string;
}
/// New schema language features that are not supported by old code generators.
enum AdvancedFeatures : ulong (bit_flags) {
AdvancedArrayFeatures,
AdvancedUnionFeatures,
OptionalScalars,
DefaultVectorsAndStrings,
}
/// File specific information.
/// Symbols declared within a file may be recovered by iterating over all
/// symbols and examining the `declaration_file` field.
table SchemaFile {
/// Filename, relative to project root.
filename:string (required, key);
/// Names of included files, relative to project root.
included_filenames:[string];
}
table Schema {
objects:[Object] (required); // Sorted.
enums:[Enum] (required); // Sorted.
file_ident:string;
file_ext:string;
root_table:Object;
services:[Service]; // Sorted.
advanced_features:AdvancedFeatures;
/// All the files used in this compilation. Files are relative to where
/// flatc was invoked.
fbs_files:[SchemaFile]; // Sorted.
}
root_type Schema;
file_identifier "BFBS";
file_extension "bfbs";

416
flatpy/flatpy/test_flatpy.py

@ -1,416 +0,0 @@
import enum
import importlib.util
import os.path
import textwrap
import flatbuffers
import flatpy.main
def compile_fbs(*, outdir, name, contents):
assert name.endswith('.fbs')
with open(os.path.join(outdir, name), 'w') as fp:
fp.write(textwrap.dedent(contents))
try:
rc = flatpy.main.main(['flatpy', '-o', str(outdir), os.path.join(outdir, name)])
finally:
print("---- schema ----------------------------------------------")
with open(os.path.join(outdir, name[:-4] + '.json'), 'r') as fp:
for lineno, line in enumerate(fp, 1):
print('{:4d} {}'.format(lineno, line.rstrip()))
assert rc == 0
mod_name = name[:-4] + '_generated'
mod_path = os.path.join(outdir, mod_name + '.py')
assert os.path.isfile(mod_path), mod_path
print("---- {} ----------------------------------------------".format(mod_name))
with open(mod_path, 'r') as fp:
for lineno, line in enumerate(fp, 1):
print('{:4d} {}'.format(lineno, line.rstrip()))
spec = importlib.util.spec_from_file_location(mod_name, mod_path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def test_simple_fields(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
table Test {
string_field: string;
int8_field: int8;
ubyte_field: ubyte;
uint8_field: uint8;
bool_field: bool;
short_field: short;
int16_field: int16;
ushort_field: ushort;
uint16_field: uint16;
int_field: int;
int32_field: int32;
uint_field: uint;
uint32_field: uint32;
float_field: float;
float32_field: float32;
long_field: long;
int64_field: int64;
ulong_field: ulong;
uint64_field: uint64;
double_field: double;
float64_field: float64;
}
''')
assert isinstance(mod.Test, type)
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
string_field="hello",
int8_field=12,
ubyte_field=13,
uint8_field=14,
bool_field=True,
short_field=15,
int16_field=16,
ushort_field=17,
uint16_field=18,
int_field=19,
int32_field=20,
uint_field=21,
uint32_field=22,
float_field=2.0,
float32_field=3.0,
long_field=23,
int64_field=24,
ulong_field=25,
uint64_field=26,
double_field=4.0,
float64_field=5.0,
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
assert t.string_field == "hello"
assert t.int8_field == 12
assert t.ubyte_field == 13
assert t.uint8_field == 14
assert t.bool_field == True
assert t.short_field == 15
assert t.int16_field == 16
assert t.ushort_field == 17
assert t.uint16_field == 18
assert t.int_field == 19
assert t.int32_field == 20
assert t.uint_field == 21
assert t.uint32_field == 22
assert t.float_field == 2.0
assert t.float32_field == 3.0
assert t.long_field == 23
assert t.int64_field == 24
assert t.ulong_field == 25
assert t.uint64_field == 26
assert t.double_field == 4.0
assert t.float64_field == 5.0
def test_string_vector(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
table Test {
strings: [string];
}
''')
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
strings=["hello", "world"],
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
# TODO: Currently it builds a new list for each attribute access. Replace this with a (cached)
# proxy object, which implements the list interface.
assert t.strings == ["hello", "world"]
def test_enum_vector(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
enum T : int { a, b, c }
table Test {
enums: [T];
}
''')
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
enums=[mod.T.c, mod.T.a],
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
assert t.enums == [mod.T.c, mod.T.a]
def test_float_vector(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
table Test {
values: [float];
}
''')
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
values=[1.0, 2.0, 3.0],
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
assert t.values == [1.0, 2.0, 3.0]
def test_enum(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
enum Some : int {
Val1,
Val2,
Val3
}
table Test {
some : Some;
}
''')
assert issubclass(mod.Some, enum.IntEnum)
assert mod.Some.Val1 == 0
assert mod.Some.Val2 == 1
assert mod.Some.Val3 == 2
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
some=mod.Some.Val2,
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
# TODO: t.some should be an IntEnum instance, not a plain int.
# Problem: the json file does not contain a reference to the enum type for the field, just the numeric base type.
#assert isinstance(t.some, mod.Some)
assert t.some == mod.Some.Val2
def test_table_ref(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
table A {
a1 : string;
}
table Test {
a : A;
}
''')
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
a=mod.CreateA(
bld,
a1="hello"
),
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
assert isinstance(t.a, mod.A)
assert t.a.a1 == "hello"
def test_table_vector(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
table Elem {
a: string;
b: int;
}
table Test {
x: Elem;
v: [Elem];
}
''')
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
v=[mod.CreateElem(bld, a="A", b=1),
mod.CreateElem(bld, a="B", b=2)],
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
# TODO: Currently it builds a new list for each attribute access. Replace this with a (cached)
# proxy object, which implements the list interface.
assert len(t.v) == 2
assert isinstance(t.v[0], mod.Elem)
assert t.v[0].a == "A"
assert t.v[0].b == 1
assert t.v[1].a == "B"
assert t.v[1].b == 2
def test_table_union(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
table A {
a1 : string;
}
table B {
b1 : string;
}
union Payload {
a : A,
b : B,
}
table Test {
payload : Payload;
}
''')
bld = flatbuffers.Builder()
offset = mod.CreateTest(
bld,
payload_type=mod.Payload.a,
payload=mod.CreateA(
bld,
a1="hello"
),
)
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
assert t.payload_type == mod.Payload.a
a = mod.A(t.payload)
assert a.a1 == "hello"
def test_from_json(tmpdir):
mod = compile_fbs(
outdir=tmpdir,
name='test.fbs',
contents='''\
table A {
a1 : string;
a2 : int;
a3 : bool;
}
table B {
b1 : string;
}
enum E : int { X, Y, Z }
table Test {
a : A;
avec : [A];
e : E;
evec : [E];
strvec : [string];
intvec : [int];
}
''')
bld = flatbuffers.Builder()
offset = mod.CreateTestFromJSON(bld, {
'a': {
'a1': 'hello',
'a2': 23
},
'avec': [
{'a1': 'hello1'},
{'a1': 'hello2'},
],
'e': 'X',
'evec': ['Y', 'Z'],
'strvec': ['aa', 'bb'],
'intvec': [3, 4, 5]
})
bld.Finish(offset)
buf = bld.Output()
t = mod.Test(buf)
assert t.a.a1 == "hello"
assert t.a.a2 == 23
assert t.avec[0].a1 == "hello1"
assert t.avec[1].a1 == "hello2"
assert t.e == mod.E.X
assert t.evec == [mod.E.Y, mod.E.Z]
assert t.strvec == ['aa', 'bb']
assert t.intvec == [3, 4, 5]
# def test_imports(tmpdir):
# with open(os.path.join(tmpdir, '__init__.py'), 'w') as fp:
# pass
# mod = compile_fbs(
# outdir=tmpdir,
# name='imported.fbs',
# contents='''\
# table I {
# a : string;
# }
# enum E : int { X, Y, Z }
# ''')
# mod = compile_fbs(
# outdir=tmpdir,
# name='test.fbs',
# contents='''\
# include "imported.fbs";
# table A {
# i : I;
# e : E;
# }
# ''')

0
flatpy/pytest.ini

37
flatpy/setup.py

@ -1,37 +0,0 @@
from setuptools import setup
setup(
name = 'flatpy',
version = '0.1',
author = 'Ben Niemann',
author_email = 'pink@odahoda.de',
# url = 'TODO',
# license = 'TODO',
classifiers = [
'Development Status :: 2 - Pre-Alpha',
#'Environment :: X11 Applications :: Qt',
#'Intended Audience :: End Users/Desktop',
# TODO: 'License :: OSI Approved :: ',
#'Natural Language :: English',
#'Operating System :: POSIX :: Linux',
#'Programming Language :: Cython',
'Programming Language :: Python :: 3 :: Only',
#'Programming Language :: Python :: Implementation :: CPython',
],
packages = ['flatpy'],
package_data={
'flatpy': [
'reflection.fbs',
'fbs.jinja2',
]
},
entry_points = {
'console_scripts': [
'flatpy = flatpy.main:run',
],
},
install_requires=[
'flatbuffers>=2.0.0',
'jinja2>=3.0',
],
)

2
noisicaa.nix

@ -56,7 +56,7 @@ let
_."ruamel.yaml".propagatedBuildInputs.mod = pySelf: oldAttrs: oldVal: oldVal ++ [ pySelf."ruamel_yaml_clib" ];
packagesExtra = [
(mach-nix.buildPythonPackage ./flatpy)
(mach-nix.buildPythonPackage ./3rdparty/flatpy)
];
};

Loading…
Cancel
Save