will
25592fa6c2
ci / test (push) Successful in 6m14s
History-free snapshot of the private development repository at 77a82eb. Code, docs and the public catalogue only: no plant or job knowledge layers, no client, job or producer names (tools/scrub.py clean against the widened term list; a second word-level grep clean). The intro deck is left out because its slide images cannot be checked for job drawings.
1372 lines
56 KiB
Python
1372 lines
56 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""The reinforcing index's Revit-free half: eligibility, features, staleness.
|
|
|
|
No Revit, no model. Everything here is the part of the index that has to be
|
|
right before a single piece of geometry is read -- who may be a donor, what a
|
|
feature record says, what gets rebuilt when something moves, and the
|
|
distinction between a piece that has no openings and a piece nobody could
|
|
measure.
|
|
|
|
The numbers in the feature tests are the measured WP157/WP168 rectangles from
|
|
the worked example: the two panels are different lengths and their two big
|
|
openings sit at the identical ABSOLUTE feet-from-start, which is exactly the
|
|
match a proportional record would have thrown away.
|
|
"""
|
|
# pylint: disable=import-error,invalid-name,broad-except
|
|
import json
|
|
import os
|
|
|
|
from vibe import reinforcecore as core
|
|
|
|
|
|
# --------------------------------------------------------------- eligibility
|
|
|
|
def test_reinforced_and_not_on_hold_is_the_whole_rule():
|
|
assert core.is_eligible(1, 0) is True
|
|
assert core.is_eligible(1, None) is True
|
|
|
|
|
|
def test_on_hold_disqualifies_even_a_reinforced_piece():
|
|
"""The measured case: the worked example's second donor is on hold."""
|
|
assert core.is_eligible(1, 1) is False
|
|
|
|
|
|
def test_unreinforced_is_never_a_donor():
|
|
assert core.is_eligible(0, 0) is False
|
|
assert core.is_eligible(None, None) is False
|
|
assert core.is_eligible(None, 0) is False
|
|
|
|
|
|
# ------------------------------------------------------------------ features
|
|
|
|
def test_feature_records_absolute_position_not_a_fraction():
|
|
"""Same rectangle, two panels of different lengths -> identical fields."""
|
|
a = core.feature('WP157', 0, 6.5, 10.583, 0.354, 4.479, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
b = core.feature('WP168', 0, 6.5, 10.583, 0.354, 4.479, True, 0.0,
|
|
24.917, 11.5, 8.0)
|
|
for key in ('x0', 'x1', 'y0', 'y1', 'w', 'h', 'cx', 'cy'):
|
|
assert a[key] == b[key], key
|
|
assert a['piece_length_ft'] != b['piece_length_ft']
|
|
|
|
|
|
def test_feature_is_flat_and_two_dimensional():
|
|
f = core.feature('WP157', 0, 24.25, 27.625, 6.25, 11.5, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
assert f['w'] == 3.375 and f['h'] == 5.25
|
|
assert f['cx'] == 20.938 or round(f['cx'], 3) == round((24.25 + 27.625) / 2, 3)
|
|
assert f['cy'] == 8.875
|
|
for value in f.values():
|
|
assert not isinstance(value, (dict, list)), 'features must stay flat'
|
|
|
|
|
|
def test_feature_carries_no_category():
|
|
"""A blockout and a notch produce the same shape of record, on purpose."""
|
|
blockout = core.feature('WP157', 0, 24.25, 27.625, 6.25, 11.5, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
notch = core.feature('WP157', 1, 0.0, 0.875, 1.125, 3.0, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
assert blockout['kind'] == notch['kind'] == core.KIND_OPENING
|
|
|
|
|
|
def test_unknown_penetration_is_none_not_false():
|
|
f = core.feature('WP157', 0, 0.0, 0.667, 11.292, 11.5, None, None,
|
|
27.625, 11.5, 8.0)
|
|
assert f['through'] is None
|
|
assert f['concrete_in'] is None
|
|
|
|
|
|
def test_blind_recess_keeps_its_concrete_depth():
|
|
f = core.feature('WP157', 0, 0.0, 0.667, 11.292, 11.5, False, 3.5,
|
|
27.625, 11.5, 8.0)
|
|
assert f['through'] is False
|
|
assert f['concrete_in'] == 3.5
|
|
|
|
|
|
def test_features_are_reindexed_in_a_stable_order():
|
|
late = core.feature('WP157', 0, 24.25, 27.625, 6.25, 11.5, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
early = core.feature('WP157', 1, 0.0, 0.875, 1.125, 3.0, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
rec = core.piece_record('WP157', 1, 'sig', 1, 0, features=[late, early])
|
|
assert [f['x0'] for f in rec['features']] == [0.0, 24.25]
|
|
assert [f['index'] for f in rec['features']] == [0, 1]
|
|
|
|
|
|
# ------------------------------------------------- could-not-judge vs nothing
|
|
|
|
def test_no_openings_and_could_not_measure_are_different_answers():
|
|
measured = core.piece_record('WP393', 1, 'sig', 1, 0, features=[])
|
|
unread = core.piece_record('WP999', 2, 'sig', 1, 0,
|
|
error='no concrete solid')
|
|
assert measured['features'] == []
|
|
assert measured['error'] is None
|
|
assert unread['features'] is None
|
|
assert unread['error']
|
|
|
|
|
|
def test_unjudged_pieces_are_held_back_from_the_donor_list_and_reported():
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[]),
|
|
'B': core.piece_record('B', 2, 's', 1, 0, error='no panel frame'),
|
|
}
|
|
names = [r['name'] for r in core.donors(cache)]
|
|
assert names == ['A']
|
|
assert [r['name'] for r in core.donors(cache, include_unjudged=True)] \
|
|
== ['A', 'B']
|
|
plan = {'build': [], 'fresh': ['A', 'B']}
|
|
assert core.coverage(cache, plan)['unjudged'] == ['B']
|
|
|
|
|
|
# ----------------------------------------------------------------- staleness
|
|
|
|
def test_signature_moves_when_a_member_is_added_or_removed():
|
|
base = core.signature(1, 0, [10, 11, 12])
|
|
assert core.signature(1, 0, [10, 11, 12, 13]) != base
|
|
assert core.signature(1, 0, [10, 11]) != base
|
|
|
|
|
|
def test_signature_ignores_member_order():
|
|
assert core.signature(1, 0, [12, 10, 11]) == core.signature(1, 0, [10, 11, 12])
|
|
|
|
|
|
def test_signature_moves_when_either_flag_moves():
|
|
base = core.signature(1, 0, [10, 11])
|
|
assert core.signature(0, 0, [10, 11]) != base
|
|
assert core.signature(1, 1, [10, 11]) != base
|
|
|
|
|
|
def test_signature_cannot_see_a_geometry_edit_in_place():
|
|
"""The honest limit, asserted so nobody assumes otherwise.
|
|
|
|
Moving or resizing an opening changes no member id and no flag, so the
|
|
signature is unchanged and the cached record stays. `rebuild_all` is the
|
|
only answer to this, and it is cheap.
|
|
"""
|
|
before = core.signature(1, 0, [10, 11, 12])
|
|
after = core.signature(1, 0, [10, 11, 12])
|
|
assert before == after
|
|
|
|
|
|
# ---------------------------------------------------------------- the plan
|
|
|
|
def _live(**kw):
|
|
return kw
|
|
|
|
|
|
def test_plan_builds_new_and_stale_reuses_fresh():
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 'sigA', 1, 0, features=[]),
|
|
'B': core.piece_record('B', 2, 'sigB', 1, 0, features=[]),
|
|
}
|
|
live = _live(A=('sigA', 1, 0), # unchanged
|
|
B=('sigB2', 1, 0), # moved
|
|
C=('sigC', 1, 0)) # new
|
|
plan = core.plan(cache, live)
|
|
assert plan['fresh'] == ['A']
|
|
assert plan['build'] == ['B', 'C']
|
|
assert plan['drop'] == []
|
|
|
|
|
|
def test_plan_drops_a_piece_that_has_gone_on_hold():
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {'A': core.piece_record('A', 1, 'sigA', 1, 0, features=[])}
|
|
plan = core.plan(cache, _live(A=('sigA2', 1, 1)))
|
|
assert plan['build'] == [] and plan['fresh'] == []
|
|
assert plan['drop'] == ['A']
|
|
|
|
|
|
def test_plan_picks_up_a_piece_taken_off_hold():
|
|
cache = core.new_cache('m')
|
|
plan = core.plan(cache, _live(A=('sigA', 1, 0)))
|
|
assert plan['build'] == ['A']
|
|
|
|
|
|
def test_plan_drops_a_piece_that_left_the_model():
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {'A': core.piece_record('A', 1, 'sigA', 1, 0, features=[])}
|
|
assert core.plan(cache, {})['drop'] == ['A']
|
|
|
|
|
|
def test_a_schema_bump_rebuilds_everything():
|
|
cache = core.new_cache('m')
|
|
cache['schema'] = core.SCHEMA - 1
|
|
cache['pieces'] = {'A': core.piece_record('A', 1, 'sigA', 1, 0, features=[])}
|
|
plan = core.plan(cache, _live(A=('sigA', 1, 0)))
|
|
assert plan['build'] == ['A']
|
|
|
|
|
|
# --------------------------------------------------- partial builds are ok
|
|
|
|
def test_a_partial_build_is_kept_and_says_what_is_missing():
|
|
cache = core.new_cache('m')
|
|
live = _live(A=('sA', 1, 0), B=('sB', 1, 0), C=('sC', 1, 0))
|
|
plan = core.plan(cache, live)
|
|
# only A got built before the budget ran out
|
|
cache = core.apply_plan(cache, plan,
|
|
[core.piece_record('A', 1, 'sA', 1, 0, features=[])],
|
|
built_at='now')
|
|
cov = core.coverage(cache, plan)
|
|
assert cov['eligible'] == 3
|
|
assert cov['indexed'] == 1
|
|
assert cov['missing'] == ['B', 'C']
|
|
assert cov['complete'] is False
|
|
|
|
|
|
def test_a_finished_build_reports_complete():
|
|
cache = core.new_cache('m')
|
|
live = _live(A=('sA', 1, 0))
|
|
plan = core.plan(cache, live)
|
|
cache = core.apply_plan(cache, plan,
|
|
[core.piece_record('A', 1, 'sA', 1, 0, features=[])])
|
|
assert core.coverage(cache, plan)['complete'] is True
|
|
|
|
|
|
# ------------------------------------------------------------------- io
|
|
|
|
def test_round_trip_through_disk(tmp_path):
|
|
path = str(tmp_path / 'reinforce' / 'index.json')
|
|
cache = core.new_cache('model')
|
|
f = core.feature('A', 0, 6.5, 10.583, 0.354, 4.479, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
cache['pieces'] = {'A': core.piece_record('A', 1, 'sA', 1, 0, features=[f])}
|
|
core.save(path, cache)
|
|
assert os.path.isfile(path)
|
|
back = core.load(path)
|
|
assert back['pieces']['A']['features'][0]['x0'] == 6.5
|
|
assert back['doc_key'] == 'model'
|
|
|
|
|
|
def test_a_damaged_cache_costs_a_rebuild_and_says_so(tmp_path):
|
|
path = str(tmp_path / 'index.json')
|
|
with open(path, 'wb') as handle:
|
|
handle.write(b'{not json')
|
|
cache = core.load(path)
|
|
assert cache['pieces'] == {}
|
|
assert cache['_load_error']
|
|
|
|
|
|
def test_an_old_schema_is_not_read_as_data(tmp_path):
|
|
path = str(tmp_path / 'index.json')
|
|
with open(path, 'wb') as handle:
|
|
handle.write(json.dumps({'schema': 0, 'pieces': {'A': {}}}).encode('utf-8'))
|
|
cache = core.load(path)
|
|
assert cache['pieces'] == {}
|
|
assert cache['_load_error']
|
|
|
|
|
|
def test_a_missing_cache_is_an_empty_one_not_an_error():
|
|
cache = core.load(None)
|
|
assert cache['schema'] == core.SCHEMA
|
|
assert cache['pieces'] == {}
|
|
assert '_load_error' not in cache
|
|
|
|
|
|
def test_all_features_is_flat_across_donors():
|
|
cache = core.new_cache('m')
|
|
fa = core.feature('A', 0, 1.0, 2.0, 1.0, 2.0, True, 0.0, 20.0, 10.0, 8.0)
|
|
fb = core.feature('B', 0, 3.0, 4.0, 1.0, 2.0, True, 0.0, 20.0, 10.0, 8.0)
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[fa]),
|
|
'B': core.piece_record('B', 2, 's', 1, 0, features=[fb]),
|
|
'C': core.piece_record('C', 3, 's', 1, 0, error='unreadable'),
|
|
}
|
|
feats = core.all_features(cache)
|
|
assert len(feats) == 2
|
|
assert sorted(f['piece'] for f in feats) == ['A', 'B']
|
|
|
|
|
|
def test_disk_form_is_packed_but_the_caller_sees_the_full_feature(tmp_path):
|
|
"""Derived fields are recomputed on load, not stored 690 times over."""
|
|
path = str(tmp_path / 'index.json')
|
|
cache = core.new_cache('m')
|
|
f = core.feature('A', 0, 6.5, 10.583, 0.354, 4.479, True, 0.0,
|
|
27.625, 11.5, 8.0)
|
|
cache['pieces'] = {'A': core.piece_record(
|
|
'A', 1, 'sA', 1, 0, length_ft=27.625, width_ft=11.5, thickness_in=8.0,
|
|
features=[f])}
|
|
core.save(path, cache)
|
|
with open(path, 'rb') as handle:
|
|
raw = handle.read().decode('utf-8')
|
|
assert 'piece_length_ft' not in raw # stored once per piece, not per feature
|
|
assert '"cx"' not in raw # derived, not stored
|
|
back = core.load(path)
|
|
assert back['pieces']['A']['features'][0] == f
|
|
|
|
|
|
def test_a_measured_empty_piece_survives_the_round_trip(tmp_path):
|
|
path = str(tmp_path / 'index.json')
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 'sA', 1, 0, features=[]),
|
|
'B': core.piece_record('B', 2, 'sB', 1, 0, error='no concrete solid'),
|
|
}
|
|
core.save(path, cache)
|
|
back = core.load(path)
|
|
assert back['pieces']['A']['features'] == []
|
|
assert back['pieces']['B']['features'] is None
|
|
|
|
|
|
# ---------------------------------------------------------- MASTER, the one
|
|
# provenance signal the donor pool has. `TICKET_IS_REINFORCED` says somebody
|
|
# drew bars; MASTER says somebody reviewed them. It is typed BY HAND into the
|
|
# assembly's identity comment, so the rule that reads it has to be tolerant
|
|
# about case and about surrounding text -- and has to be predictable enough
|
|
# that a reader can say in advance what it will and will not accept.
|
|
|
|
|
|
def test_master_accepts_the_bare_token_in_any_case():
|
|
for text in ('MASTER', 'master', 'Master', 'mAsTeR'):
|
|
assert core.is_master_text(text) is True
|
|
|
|
|
|
def test_master_accepts_the_token_with_text_around_it():
|
|
# Every one of these is a shape that really occurs in the field, with the
|
|
# job-specific words replaced.
|
|
assert core.is_master_text('MASTER;reverse strands') is True
|
|
assert core.is_master_text('MASTER, bend bars to fit inside stirrups')
|
|
assert core.is_master_text('reverse strands; master') is True
|
|
assert core.is_master_text(' MASTER ') is True
|
|
assert core.is_master_text('MASTER-PANEL') is True
|
|
|
|
|
|
def test_master_rejects_an_empty_or_absent_field():
|
|
assert core.is_master_text(None) is False
|
|
assert core.is_master_text('') is False
|
|
assert core.is_master_text('deflect strand 1"') is False
|
|
|
|
|
|
def test_master_rejects_the_token_inside_another_word():
|
|
"""A whole-word rule, stated so it can be argued with rather than guessed.
|
|
|
|
`MASTERED` and `REMASTER` are not this plant's marker and are not treated
|
|
as one. The cost is that a plural or a typo is missed -- which is visible,
|
|
because the raw string is kept on the record as `master_note`.
|
|
"""
|
|
assert core.is_master_text('MASTERED') is False
|
|
assert core.is_master_text('REMASTER') is False
|
|
assert core.is_master_text('MASTERS') is False
|
|
|
|
|
|
def test_master_token_is_plant_knowledge_not_a_literal():
|
|
assert core.is_master_text('GEPRUEFT', 'GEPRUEFT') is True
|
|
assert core.is_master_text('MASTER', 'GEPRUEFT') is False
|
|
# An empty token switches the signal off rather than matching everything.
|
|
assert core.is_master_text('MASTER', '') is False
|
|
|
|
|
|
def test_a_piece_record_carries_the_boolean_and_the_raw_string():
|
|
rec = core.piece_record('PNL101', 1, 's', 1, 0, features=[],
|
|
master=True, master_note='MASTER;reverse strands')
|
|
assert rec['master'] is True
|
|
assert rec['master_note'] == 'MASTER;reverse strands'
|
|
|
|
|
|
def test_a_piece_record_defaults_to_not_a_master():
|
|
rec = core.piece_record('PNL102', 1, 's', 1, 0, features=[])
|
|
assert rec['master'] is False
|
|
assert rec['master_note'] is None
|
|
|
|
|
|
def test_the_signature_moves_when_a_piece_is_typed_up_as_master():
|
|
"""Without this the signal would never take effect.
|
|
|
|
Typing MASTER into the identity comment changes nothing else about the
|
|
piece -- not its members, not its flags -- so a stamp that ignored it
|
|
would leave the cached record standing and the new marking would do
|
|
nothing at all until something unrelated moved.
|
|
"""
|
|
before = core.signature(1, 0, [10, 11], master=False)
|
|
after = core.signature(1, 0, [10, 11], master=True)
|
|
assert before != after
|
|
|
|
|
|
def test_the_signature_moves_when_elevation_group_is_recorded():
|
|
"""TIER 1 (`elevation_group`) is cheap -- a single parameter read, the
|
|
same class of cost `master` already pays -- so it IS folded into the
|
|
stamp, unlike tier 2 (`elevation_band`, see below). Without this, MC
|
|
mapping `Tagged_Sheet` for the first time, or correcting a value on one
|
|
panel, would do nothing until something unrelated about the piece moved.
|
|
"""
|
|
before = core.signature(1, 0, [10, 11], elevation_group=None)
|
|
after = core.signature(1, 0, [10, 11], elevation_group='E3.16')
|
|
assert before != after
|
|
|
|
|
|
def test_the_signature_treats_case_and_whitespace_as_the_same_group():
|
|
"""`elevation_group_key`'s normalisation, at the seam that matters: two
|
|
spellings of the same sheet must not force a rebuild of a piece that has
|
|
not actually changed."""
|
|
assert (core.signature(1, 0, [10, 11], elevation_group='E3.16') ==
|
|
core.signature(1, 0, [10, 11], elevation_group=' e3.16 '))
|
|
|
|
|
|
def test_signature_is_five_pipe_joined_fields_of_flags_and_id_stamp():
|
|
"""The general shape, pinned so a future change to it is deliberate:
|
|
reinforced | on_hold | master | elevation_group | id-count | checksum
|
|
-- six segments now that tier 1 rejoined it."""
|
|
reinforced, on_hold, ids = 1, 0, [10, 11]
|
|
six_field = '{0}|{1}|{2}|{3}|{4}|{5}'.format(
|
|
1 if reinforced == 1 else 0, 1 if on_hold == 1 else 0, 0, '',
|
|
len(ids), sum(ids) % 1000000007)
|
|
assert core.signature(reinforced, on_hold, ids) == six_field
|
|
|
|
|
|
# --------------------------------------------------------- elevation_band
|
|
#
|
|
# Owner's ruling, 2026-09-02: elevation is a purely geometric fact -- same
|
|
# axis AND overlapping coordinate band, never a parameter. See
|
|
# `reinforcecore.elevation_band`'s module note for the full reasoning; these
|
|
# tests pin the definition itself, off plain numbers, no Revit involved.
|
|
|
|
def _basis(axis):
|
|
"""A clean unit basis vector along one global axis, for fixtures."""
|
|
return {'X': (1.0, 0.0, 0.0), 'Y': (0.0, 1.0, 0.0),
|
|
'Z': (0.0, 0.0, 1.0)}[axis]
|
|
|
|
|
|
def test_elevation_band_picks_the_smallest_dimension_as_thickness():
|
|
"""20 ft long, 10 ft wide, 0.667 ft (8 in) thick -- clean and
|
|
orthogonal. Local axis 2 (span 0.667) is thinnest; its basis (the
|
|
THIRD argument) points world +X, so the band lands on the X axis and
|
|
its range is read off the box at that axis.
|
|
"""
|
|
band = core.elevation_band(
|
|
[20.0, 10.0, 0.667],
|
|
_basis('Y'), _basis('Z'), _basis('X'),
|
|
[100.0, 0.0, 0.0], [100.667, 20.0, 10.0])
|
|
assert band == {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
|
|
|
|
def test_elevation_band_normalises_a_reversed_lo_hi_pair():
|
|
band = core.elevation_band(
|
|
[20.0, 10.0, 0.667],
|
|
_basis('Y'), _basis('Z'), _basis('X'),
|
|
[100.667, 20.0, 10.0], [100.0, 0.0, 0.0])
|
|
assert band == {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
|
|
|
|
def test_elevation_band_is_none_when_near_square():
|
|
"""Smallest and second-smallest spans are 8.0 and 8.5 -- ratio 1.0625,
|
|
under the 1.25 near-square ratio -- so which one is genuinely "the
|
|
thickness" cannot be told, and the band declines rather than guesses.
|
|
"""
|
|
band = core.elevation_band(
|
|
[20.0, 8.5, 8.0],
|
|
_basis('X'), _basis('Y'), _basis('Z'),
|
|
[0.0, 0.0, 0.0], [20.0, 8.5, 8.0],
|
|
near_square_ratio=1.25)
|
|
assert band is None
|
|
|
|
|
|
def test_elevation_band_is_none_when_not_cleanly_axis_aligned():
|
|
"""The thinnest axis's world direction is a 45-degree blend of X and Y
|
|
-- a skewed wall -- so no global axis is a clean match and the band
|
|
declines rather than forcing the nearest one."""
|
|
diag = (0.70710678, 0.70710678, 0.0)
|
|
band = core.elevation_band(
|
|
[20.0, 10.0, 0.667],
|
|
_basis('X'), _basis('Y'), diag,
|
|
[0.0, 0.0, 0.0], [10.0, 10.0, 10.0])
|
|
assert band is None
|
|
|
|
|
|
def test_elevation_band_declines_gracefully_on_bad_input():
|
|
"""Never raises: bad input is could-not-judge, the same as ambiguous or
|
|
off-axis geometry is."""
|
|
assert core.elevation_band(None, (1, 0, 0), (0, 1, 0), (0, 0, 1),
|
|
[0, 0, 0], [1, 1, 1]) is None
|
|
assert core.elevation_band([1.0, 2.0], (1, 0, 0), (0, 1, 0), (0, 0, 1),
|
|
[0, 0, 0], [1, 1, 1]) is None
|
|
assert core.elevation_band([1.0, 2.0, 0.5], (1, 0, 0), (0, 1, 0),
|
|
(0, 0, 1), None, [1, 1, 1]) is None
|
|
|
|
|
|
# ----------------------------------------------------------- same_elevation
|
|
|
|
def test_same_elevation_true_for_overlapping_bands_on_the_same_axis():
|
|
a = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
b = {'axis': 'X', 'min': 100.1, 'max': 100.767}
|
|
assert core.same_elevation(a, b) is True
|
|
|
|
|
|
def test_same_elevation_false_for_non_overlapping_bands_on_the_same_axis():
|
|
a = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
b = {'axis': 'X', 'min': 120.0, 'max': 120.667}
|
|
assert core.same_elevation(a, b) is False
|
|
|
|
|
|
def test_same_elevation_false_for_a_different_axis_even_when_close():
|
|
"""The owner's explicit ruling: perpendicular walls never match, however
|
|
numerically close the coordinates happen to be."""
|
|
a = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
b = {'axis': 'Y', 'min': 100.0, 'max': 100.667}
|
|
assert core.same_elevation(a, b) is False
|
|
|
|
|
|
def test_same_elevation_false_when_bands_only_abut():
|
|
"""Pinned: touching at exactly one point (no interior overlap) is NOT
|
|
the same elevation -- the signature of a stepped facade, not a shared
|
|
wall plane. See `same_elevation`'s own docstring for the reasoning."""
|
|
a = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
b = {'axis': 'X', 'min': 100.667, 'max': 101.334}
|
|
assert core.same_elevation(a, b) is False
|
|
|
|
|
|
def test_same_elevation_false_when_either_band_is_unknown():
|
|
"""UNKNOWN IS NEVER A MATCH -- two could-not-judge bands never agree
|
|
with each other, and a known band is never matched against nothing."""
|
|
a = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
assert core.same_elevation(a, None) is False
|
|
assert core.same_elevation(None, a) is False
|
|
assert core.same_elevation(None, None) is False
|
|
|
|
|
|
def test_same_elevation_never_raises_on_a_malformed_band():
|
|
assert core.same_elevation({'axis': 'X'},
|
|
{'axis': 'X', 'min': 1, 'max': 2}) is False
|
|
assert core.same_elevation('not a dict',
|
|
{'axis': 'X', 'min': 1, 'max': 2}) is False
|
|
|
|
|
|
# ----------------------------------------------------- elevation_group_key
|
|
#
|
|
# TIER 1. Owner's ruling, 2026-09-02 (the SAME day as the geometric ruling):
|
|
# MC's model already carries a grouping parameter (`Tagged_Sheet`, mapped
|
|
# through the role `panel.elevation_group`) -- two panels with the same
|
|
# non-empty value are the same elevation with no geometry involved. "对MC而
|
|
# 言，所有的墙应该都有一个参数叫Tagged_Sheet。如果俩panel这个参数一样，那么
|
|
# 他们就是同一个elevation，不用跑哪个几何引擎。"
|
|
|
|
def test_elevation_group_key_normalises_case_and_whitespace():
|
|
assert (core.elevation_group_key('E3.16') ==
|
|
core.elevation_group_key(' e3.16 '))
|
|
|
|
|
|
def test_elevation_group_key_is_none_for_none_and_for_blank():
|
|
"""Unmapped role and never-typed are the same could-not-judge answer."""
|
|
assert core.elevation_group_key(None) is None
|
|
assert core.elevation_group_key('') is None
|
|
assert core.elevation_group_key(' ') is None
|
|
|
|
|
|
def test_elevation_group_key_never_raises_on_a_non_string():
|
|
assert core.elevation_group_key(42) is None
|
|
assert core.elevation_group_key([1, 2]) is None
|
|
|
|
|
|
def test_two_different_groups_are_different_keys():
|
|
assert core.elevation_group_key('E3.16') != core.elevation_group_key(
|
|
'E3.22')
|
|
|
|
|
|
# --------------------------------------------------------- elevation_match
|
|
#
|
|
# The two-tier comparator itself: tier 1 (`elevation_group`) tried first,
|
|
# tier 2 (`elevation_band`) as the fallback. See the module note above
|
|
# `elevation_band` for the full ordering rule.
|
|
|
|
_BAND_X = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
_BAND_X_OVERLAP = {'axis': 'X', 'min': 100.1, 'max': 100.767}
|
|
_BAND_Y = {'axis': 'Y', 'min': 100.0, 'max': 100.667}
|
|
|
|
|
|
def test_elevation_match_by_group_needs_no_geometry_at_all():
|
|
"""Same non-empty group, WILDLY different (even absent) bands -- the
|
|
group still decides it, exactly the MC shortcut."""
|
|
target = {'elevation_group': 'E3.16', 'elevation_band': _BAND_X}
|
|
donor = {'elevation_group': ' e3.16 ', 'elevation_band': None}
|
|
assert core.elevation_match(target, donor) == core.ELEVATION_BY_GROUP
|
|
|
|
|
|
def test_elevation_match_falls_back_to_band_when_group_does_not_answer():
|
|
target = {'elevation_group': None, 'elevation_band': _BAND_X}
|
|
donor = {'elevation_group': None, 'elevation_band': _BAND_X_OVERLAP}
|
|
assert core.elevation_match(target, donor) == core.ELEVATION_BY_BAND
|
|
|
|
|
|
def test_elevation_match_falls_back_to_band_when_groups_differ():
|
|
"""A group mismatch does not veto the geometric answer -- it simply
|
|
fails to answer tier 1, and tier 2 gets to try."""
|
|
target = {'elevation_group': 'E3.16', 'elevation_band': _BAND_X}
|
|
donor = {'elevation_group': 'E3.22', 'elevation_band': _BAND_X_OVERLAP}
|
|
assert core.elevation_match(target, donor) == core.ELEVATION_BY_BAND
|
|
|
|
|
|
def test_elevation_match_is_none_when_neither_tier_answers():
|
|
target = {'elevation_group': None, 'elevation_band': _BAND_X}
|
|
donor = {'elevation_group': None, 'elevation_band': _BAND_Y}
|
|
assert core.elevation_match(target, donor) is None
|
|
|
|
|
|
def test_elevation_match_never_raises_on_missing_keys():
|
|
"""A pre-migration record with neither field at all -- could-not-judge,
|
|
not an exception."""
|
|
assert core.elevation_match({}, {}) is None
|
|
assert core.elevation_match(None, None) is None
|
|
assert core.elevation_match({'elevation_group': 'E3.16'}, {}) is None
|
|
|
|
|
|
# ------------------------------------------------------- piece_record: band
|
|
|
|
def test_a_piece_record_carries_its_elevation_band():
|
|
band = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
rec = core.piece_record('PNL425', 1, 's', 1, 0, features=[],
|
|
elevation_band=band)
|
|
assert rec['elevation_band'] == band
|
|
|
|
|
|
def test_a_piece_record_defaults_to_no_elevation_band():
|
|
"""None, not a missing key -- a record built with no argument still
|
|
reads the same could-not-judge value one built for an ambiguous or
|
|
off-axis panel would."""
|
|
rec = core.piece_record('PNL426', 1, 's', 1, 0, features=[])
|
|
assert rec['elevation_band'] is None
|
|
|
|
|
|
def test_elevation_band_survives_a_save_and_load_round_trip(tmp_path):
|
|
path = str(tmp_path / 'index.json')
|
|
band = {'axis': 'X', 'min': 100.0, 'max': 100.667}
|
|
cache = core.new_cache('doc')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[],
|
|
elevation_band=band),
|
|
}
|
|
core.save(path, cache)
|
|
back = core.load(path)
|
|
assert back['pieces']['A']['elevation_band'] == band
|
|
|
|
|
|
# ------------------------------------------------------ piece_record: group
|
|
|
|
def test_a_piece_record_carries_its_elevation_group():
|
|
rec = core.piece_record('PNL425', 1, 's', 1, 0, features=[],
|
|
elevation_group='E3.16')
|
|
assert rec['elevation_group'] == 'E3.16'
|
|
|
|
|
|
def test_a_piece_record_defaults_to_no_elevation_group():
|
|
rec = core.piece_record('PNL426', 1, 's', 1, 0, features=[])
|
|
assert rec['elevation_group'] is None
|
|
|
|
|
|
def test_elevation_group_survives_a_save_and_load_round_trip(tmp_path):
|
|
path = str(tmp_path / 'index.json')
|
|
cache = core.new_cache('doc')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[],
|
|
elevation_group='E3.16'),
|
|
}
|
|
core.save(path, cache)
|
|
back = core.load(path)
|
|
assert back['pieces']['A']['elevation_group'] == 'E3.16'
|
|
|
|
|
|
def test_masters_lists_only_the_reviewed_donors():
|
|
cache = core.new_cache('doc')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[], master=True),
|
|
'B': core.piece_record('B', 2, 's', 1, 0, features=[]),
|
|
'C': core.piece_record('C', 3, 's', 1, 0, error='unreadable',
|
|
master=True),
|
|
}
|
|
assert core.masters(cache) == ['A']
|
|
assert core.masters(cache, include_unjudged=True) == ['A', 'C']
|
|
|
|
|
|
def test_master_survives_a_save_and_load_round_trip(tmp_path):
|
|
path = str(tmp_path / 'index.json')
|
|
cache = core.new_cache('doc')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[],
|
|
master=True, master_note='MASTER'),
|
|
}
|
|
core.save(path, cache)
|
|
back = core.load(path)
|
|
assert back['pieces']['A']['master'] is True
|
|
assert back['pieces']['A']['master_note'] == 'MASTER'
|
|
|
|
|
|
def test_an_index_built_before_master_existed_is_not_read_as_no_masters(
|
|
tmp_path):
|
|
"""The whole answer to "what happens to yesterday's cache".
|
|
|
|
A file written under the previous schema is thrown away WHOLE. It is
|
|
never partially adopted, so not one piece from it can be served as a
|
|
record that was checked for MASTER and found not to be one. It costs a
|
|
rebuild, which is six seconds, and the reason is reported.
|
|
"""
|
|
path = str(tmp_path / 'index.json')
|
|
old = {'schema': core.SCHEMA - 1, 'doc_key': 'doc', 'built_at': 'y',
|
|
'pieces': {'A': {'name': 'A', 'sig': 's', 'features': []}}}
|
|
handle = open(path, 'wb')
|
|
handle.write(json.dumps(old).encode('utf-8'))
|
|
handle.close()
|
|
|
|
cache = core.load(path)
|
|
assert cache['pieces'] == {}
|
|
assert cache['_load_error']
|
|
# and nothing from it can be mistaken for a judged non-master
|
|
assert core.masters(cache) == []
|
|
|
|
|
|
def test_master_unknown_is_not_master_false():
|
|
"""The role unmapped is a third answer and the record keeps it.
|
|
|
|
`master=None` means the plant has not said which parameter carries the
|
|
marker, so nothing was read. It must not collapse into False, which is
|
|
the verdict of a field that WAS read.
|
|
"""
|
|
rec = core.piece_record('PNL103', 1, 's', 1, 0, features=[], master=None)
|
|
assert rec['master'] is None
|
|
|
|
cache = core.new_cache('doc')
|
|
cache['pieces'] = {'PNL103': rec}
|
|
assert core.masters(cache) == []
|
|
assert core.master_signal_read(cache) is False
|
|
|
|
|
|
def test_the_signal_reads_as_run_as_soon_as_one_piece_was_judged():
|
|
cache = core.new_cache('doc')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[], master=False),
|
|
}
|
|
assert core.master_signal_read(cache) is True
|
|
assert core.masters(cache) == []
|
|
|
|
|
|
# ------------------------------------------------------ the design number
|
|
#
|
|
# `design_number` decides who may be a template at all (owner ruling,
|
|
# 2026-09-01: a donor must be the same design as its target). Before that rule
|
|
# may be trusted, this has to answer whether the index records the field --
|
|
# and there are THREE answers, not two.
|
|
|
|
|
|
def test_a_recorded_design_number_reads_as_recorded():
|
|
assert core.design_status([
|
|
core.piece_record('A', 1, 's', 1, 0, design_number='PNL1.0'),
|
|
core.piece_record('B', 2, 's', 1, 0),
|
|
]) == core.DESIGN_RECORDED
|
|
|
|
|
|
def test_records_that_all_carry_an_empty_design_number_read_as_unread():
|
|
"""Measured on the real index, 2026-09-01: 773 pieces, every one of them
|
|
`design_number: null`, because the build ran with no knowledge layer. The
|
|
field was there and nobody had read it -- which is 'nobody looked', not
|
|
'every piece matches'."""
|
|
assert core.design_status([
|
|
core.piece_record('A', 1, 's', 1, 0),
|
|
core.piece_record('B', 2, 's', 1, 0),
|
|
]) == core.DESIGN_UNREAD
|
|
|
|
|
|
def test_records_written_before_the_field_read_as_not_recorded():
|
|
old = core.piece_record('A', 1, 's', 1, 0)
|
|
del old['design_number']
|
|
assert core.design_status([old]) == core.DESIGN_NOT_RECORDED
|
|
|
|
|
|
def test_nothing_at_all_is_not_recorded_never_recorded():
|
|
"""An empty sequence is 'nothing was looked at', which is the
|
|
could-not-judge answer and not a clean bill."""
|
|
assert core.design_status([]) == core.DESIGN_NOT_RECORDED
|
|
assert core.design_status(None) == core.DESIGN_NOT_RECORDED
|
|
|
|
|
|
def test_an_index_that_never_stamped_a_pattern_says_none_not_a_pattern():
|
|
"""None means 'this file does not say', and the caller then falls back the
|
|
way an unmapped nomenclature always has. It is not 'no pattern'."""
|
|
assert core.design_re_in(core.new_cache('doc')) is None
|
|
assert core.design_re_in({}) is None
|
|
assert core.design_re_in({'nomenclature': 'nonsense'}) is None
|
|
assert core.design_re_in({'nomenclature': {'design_number_re': ''}}) is None
|
|
|
|
|
|
def test_the_stamped_pattern_survives_a_save_and_load_round_trip(tmp_path):
|
|
"""The offline reader has no knowledge layer, so the pattern the button
|
|
reduced designs with has to be IN THE FILE or the two disagree."""
|
|
path = str(tmp_path / 'index.json')
|
|
cache = core.new_cache('doc')
|
|
cache['nomenclature'] = {'design_number_re': r'^([A-Za-z]+)'}
|
|
core.save(path, cache)
|
|
assert core.design_re_in(core.load(path)) == r'^([A-Za-z]+)'
|
|
|
|
|
|
def test_the_three_design_answers_are_three_different_strings():
|
|
assert len({core.DESIGN_RECORDED, core.DESIGN_UNREAD,
|
|
core.DESIGN_NOT_RECORDED}) == 3
|
|
|
|
|
|
# ------------------------------------------------- targets, and who may donate
|
|
#
|
|
# ELIGIBILITY FILTERS DONORS. IT MUST NOT DECIDE WHO CAN BE A TARGET.
|
|
#
|
|
# Up to schema 4 it decided both, and the consequence was invisible until
|
|
# somebody asked to run with Revit closed: the piece an engineer wants
|
|
# suggestions FOR is the one not yet reinforced, so it was the one piece the
|
|
# index never held. `plan(include=...)` measures it too. Nothing about the
|
|
# donor rule moves, and these tests are mostly about that: the pool must be
|
|
# exactly the same size afterwards.
|
|
|
|
def test_a_piece_that_may_not_donate_is_still_measured_as_a_target():
|
|
cache = core.new_cache('m')
|
|
live = _live(A=('sA', 1, 0), # a donor
|
|
T=('sT', 0, 0)) # not reinforced -- an ordinary target
|
|
plan = core.plan(cache, live, include=lambda name: name == 'T')
|
|
assert plan['build'] == ['A', 'T']
|
|
assert plan['donors'] == ['A']
|
|
assert plan['targets'] == ['T']
|
|
|
|
|
|
def test_without_a_predicate_the_index_is_donors_only_as_before():
|
|
cache = core.new_cache('m')
|
|
plan = core.plan(cache, _live(A=('sA', 1, 0), T=('sT', 0, 0)))
|
|
assert plan['build'] == ['A']
|
|
assert plan['targets'] == []
|
|
|
|
|
|
def test_widening_what_is_measured_never_widens_the_donor_pool():
|
|
"""The whole safety of the change, as arithmetic."""
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[]),
|
|
'T': core.piece_record('T', 2, 's', 0, 0, features=[]),
|
|
'H': core.piece_record('H', 3, 's', 1, 1, features=[]),
|
|
}
|
|
assert [r['name'] for r in core.donors(cache)] == ['A']
|
|
assert [r['name'] for r in core.targets(cache)] == ['H', 'T']
|
|
|
|
|
|
def test_a_target_typed_master_is_not_counted_as_a_reviewed_donor():
|
|
"""A MASTER that may not donate is not a template this answer could use."""
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[], master=True),
|
|
'T': core.piece_record('T', 2, 's', 0, 0, features=[], master=True),
|
|
}
|
|
assert core.masters(cache) == ['A']
|
|
|
|
|
|
def test_coverage_counts_donors_and_targets_apart():
|
|
cache = core.new_cache('m')
|
|
live = _live(A=('sA', 1, 0), B=('sB', 1, 0), T=('sT', 0, 0))
|
|
plan = core.plan(cache, live, include=lambda name: True)
|
|
cache = core.apply_plan(cache, plan, [
|
|
core.piece_record('A', 1, 'sA', 1, 0, features=[]),
|
|
core.piece_record('T', 3, 'sT', 0, 0, features=[]),
|
|
])
|
|
cov = core.coverage(cache, plan)
|
|
assert cov['eligible'] == 2 # A and B -- never T
|
|
assert cov['indexed'] == 1
|
|
assert cov['missing'] == ['B']
|
|
assert cov['targets'] == 1
|
|
assert cov['complete'] is False
|
|
|
|
|
|
def test_a_plan_without_the_donor_split_still_reads_as_it_did():
|
|
"""Back-compat: a hand-built plan from before the split."""
|
|
cache = core.new_cache('m')
|
|
cache['pieces'] = {'A': core.piece_record('A', 1, 's', 1, 0, features=[])}
|
|
cov = core.coverage(cache, {'build': [], 'fresh': ['A']})
|
|
assert cov['eligible'] == 1 and cov['complete'] is True
|
|
|
|
|
|
def test_a_predicate_that_raises_measures_the_piece_rather_than_dropping_it():
|
|
"""Could-not-tell may only ever resolve towards measuring, never towards
|
|
a silent exclusion -- `vibe.scope`'s standing rule, applied here."""
|
|
def boom(_name):
|
|
raise ValueError('no idea')
|
|
|
|
plan = core.plan(core.new_cache('m'), _live(T=('sT', 0, 0)), include=boom)
|
|
assert plan['build'] == ['T']
|
|
|
|
|
|
# ------------------------------------------- what must never reach the file
|
|
|
|
def test_a_load_error_is_never_written_into_the_index():
|
|
"""Measured on the real index, 2026-08-30: a healthy schema-4 file carried
|
|
`_load_error: schema 3, expected 4` for good, because `load()` stamps the
|
|
key on the empty cache it returns, `refresh` fills that same dict and
|
|
saves it. Every later press then reported a load failure that had not
|
|
happened and forced a rebuild it did not need."""
|
|
import os
|
|
import tempfile
|
|
folder = tempfile.mkdtemp()
|
|
path = os.path.join(folder, 'index.json')
|
|
cache = core.new_cache('m')
|
|
cache['_load_error'] = 'schema 4, expected 5'
|
|
cache['_anything_else'] = 'also transient'
|
|
cache['pieces'] = {'A': core.piece_record('A', 1, 's', 1, 0, features=[])}
|
|
core.save(path, cache)
|
|
|
|
raw = open(path, 'rb').read().decode('utf-8')
|
|
assert '_load_error' not in raw
|
|
assert '_anything_else' not in raw
|
|
assert core.load(path).get('_load_error') is None
|
|
|
|
|
|
def test_the_declared_scope_survives_a_round_trip():
|
|
"""The offline reader has no knowledge layer, so the scope has to travel
|
|
in the file itself."""
|
|
import os
|
|
import tempfile
|
|
path = os.path.join(tempfile.mkdtemp(), 'index.json')
|
|
cache = core.new_cache('m')
|
|
cache['scope'] = {'prefixes': ['PNL'], 'note': 'cladding only',
|
|
'declared_by': 'somebody'}
|
|
core.save(path, cache)
|
|
assert core.load(path)['scope']['prefixes'] == ['PNL']
|
|
|
|
|
|
def test_a_stray_load_error_in_a_healthy_file_is_not_a_load_error():
|
|
"""The leg that turned a cosmetic defect into a full re-scan per press.
|
|
|
|
Files written before `_pack` learned to strip transient keys carry the
|
|
message from the schema bump that CAUSED the rebuild, copied into the
|
|
rebuilt file. A reader that believes it rebuilds the index on every press
|
|
for ever. This file parsed and its schema matches; what some earlier load
|
|
thought is not a fact about it.
|
|
"""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
path = os.path.join(tempfile.mkdtemp(), 'index.json')
|
|
healthy = {'schema': core.SCHEMA, 'doc_key': 'm', 'built_at': 'x',
|
|
'scope': None, 'pieces': {},
|
|
'_load_error': 'schema 4, expected 5'}
|
|
handle = open(path, 'wb')
|
|
try:
|
|
handle.write(json.dumps(healthy).encode('utf-8'))
|
|
finally:
|
|
handle.close()
|
|
|
|
got = core.load(path)
|
|
assert got.get('_load_error') is None
|
|
assert got['built_at'] == 'x' # and the file was otherwise adopted
|
|
|
|
|
|
def test_a_real_schema_mismatch_still_discards_the_whole_file(tmp_path):
|
|
"""The third leg: a genuine mismatch must still be reported once, and
|
|
nothing may be adopted from it."""
|
|
import json
|
|
path = str(tmp_path / 'index.json')
|
|
old = {'schema': core.SCHEMA - 1, 'doc_key': 'm', 'built_at': 'x',
|
|
'pieces': {'A': {'name': 'A'}}}
|
|
handle = open(path, 'wb')
|
|
try:
|
|
handle.write(json.dumps(old).encode('utf-8'))
|
|
finally:
|
|
handle.close()
|
|
|
|
got = core.load(path)
|
|
assert got['_load_error']
|
|
assert got['pieces'] == {}
|
|
assert got['built_at'] is None
|
|
|
|
|
|
def test_the_plan_a_two_argument_caller_gets_matches_the_one_refresh_builds():
|
|
"""The button computes its own plan with two arguments, to size a
|
|
progress bar and to decide whether a refresh is needed at all. If that
|
|
plan were donors-only while `refresh`'s was not, every target in the file
|
|
would land in its `drop` list on every press -- and a caller that fed that
|
|
plan to `apply_plan` would evict them all."""
|
|
cache = core.new_cache('m')
|
|
cache['scope'] = {'prefixes': ['PNL'], 'note': None, 'declared_by': None}
|
|
cache['pieces'] = {
|
|
'PNL1': core.piece_record('PNL1', 1, 's1', 1, 0, features=[]),
|
|
'PNL2': core.piece_record('PNL2', 2, 's2', 0, 0, features=[]),
|
|
}
|
|
live = _live(PNL1=('s1', 1, 0), PNL2=('s2', 0, 0))
|
|
plan = core.plan(cache, live) # two arguments, as the button
|
|
assert plan['drop'] == []
|
|
assert plan['build'] == []
|
|
assert plan['targets'] == ['PNL2']
|
|
|
|
|
|
def test_an_index_with_no_scope_stamp_keeps_the_donors_only_behaviour():
|
|
"""A file that never recorded a scope has no targets in it to protect."""
|
|
cache = core.new_cache('m') # scope is None
|
|
plan = core.plan(cache, _live(A=('sA', 1, 0), T=('sT', 0, 0)))
|
|
assert plan['build'] == ['A']
|
|
|
|
|
|
# ==================================================================
|
|
# THE FEATURE REGISTRY (schema 6). The feature-id is an INDEX/ACCELERATOR
|
|
# only: `covers()` stays the sole ranker, the id never ranks and never becomes
|
|
# a tolerance bin. Three owner-locked decisions (2026-09-02): the grain is the
|
|
# index's EXISTING rounding (no new tolerance), the edge condition is the
|
|
# binary concrete-edge vs block-out-edge, and the id is a positive integer.
|
|
# These fixtures are neutral -- generic piece marks, no client or plant names.
|
|
# ==================================================================
|
|
|
|
# A strip (ligament) feature, in the shape `reinforcebind._record` produces:
|
|
# both axes, plus the derived `strip_ft` (width across) and `span_ft` (span
|
|
# along), `between`, and the outline flag. The registry reads exactly these.
|
|
def _strip(kind, x0, x1, y0, y1, between, outline_approximate=False, edge=None):
|
|
x0, x1, y0, y1 = float(x0), float(x1), float(y0), float(y1)
|
|
strip = (x1 - x0) if kind == 'strip_x' else (y1 - y0)
|
|
span = (y1 - y0) if kind == 'strip_x' else (x1 - x0)
|
|
return {'kind': kind, 'x0': x0, 'x1': x1, 'y0': y0, 'y1': y1,
|
|
'w': round(x1 - x0, 3), 'h': round(y1 - y0, 3),
|
|
'strip_ft': round(strip, 3), 'span_ft': round(span, 3),
|
|
'between': between, 'edge': edge,
|
|
'outline_approximate': outline_approximate}
|
|
|
|
|
|
# --------------------------------------------- descriptor: determinism + grain
|
|
|
|
def test_a_descriptor_is_deterministic_hashable_and_json_serialisable():
|
|
f = _strip('strip_x', 0.0, 3.0, 1.0, 2.0, 'void-void')
|
|
d1 = core.canonical_descriptor(f)
|
|
d2 = core.canonical_descriptor(dict(f))
|
|
assert d1 == d2
|
|
assert hash(d1) == hash(d2) # hashable
|
|
assert {d1: 'ok'}[d2] == 'ok' # usable as a dict key
|
|
assert json.loads(json.dumps(d1)) == list( # JSON-serialisable
|
|
[d1[0], list(d1[1]), d1[2], d1[3]])
|
|
|
|
|
|
def test_the_grain_is_the_existing_rounding_and_no_new_tolerance():
|
|
"""3.00 vs 3.02 -> DIFFERENT (covers() ranks them); 3.000 vs 3.0004 ->
|
|
SAME (the index's existing 3-dp / EXACT_TOL grain already collapsed them).
|
|
Driven by `canonical_descriptor`'s own `_f`, not a fresh bin -- the values
|
|
are passed raw here to prove the descriptor does the graining."""
|
|
def strip(width):
|
|
return {'kind': 'strip_x', 'strip_ft': width, 'span_ft': 1.0,
|
|
'between': 'void-void'}
|
|
assert (core.canonical_descriptor(strip(3.00)) !=
|
|
core.canonical_descriptor(strip(3.02)))
|
|
assert (core.canonical_descriptor(strip(3.000)) ==
|
|
core.canonical_descriptor(strip(3.0004)))
|
|
|
|
|
|
# ------------------------------------------------- descriptor: the binary edge
|
|
|
|
def test_the_specific_panel_edge_is_dropped_from_the_descriptor():
|
|
"""A VOID_EDGE strip against x0 and one against x1 -- same kind and dims --
|
|
share ONE descriptor: only concrete-vs-blockout is encoded, never the
|
|
x0/x1/y0/y1 sub-label."""
|
|
at_x0 = _strip('strip_x', 0.0, 2.0, 1.0, 3.0, 'void-edge', edge='x0')
|
|
at_x1 = _strip('strip_x', 5.0, 7.0, 1.0, 3.0, 'void-edge', edge='x1')
|
|
assert (core.canonical_descriptor(at_x0) ==
|
|
core.canonical_descriptor(at_x1))
|
|
|
|
|
|
def test_concrete_edge_and_block_out_edge_are_distinguished():
|
|
"""Same kind and dims, differing only in `between`: a VOID_EDGE (touches a
|
|
concrete panel edge) and a VOID_VOID (between two block-outs) land in
|
|
DIFFERENT descriptors."""
|
|
edge = _strip('strip_x', 0.0, 2.0, 1.0, 3.0, 'void-edge')
|
|
between = _strip('strip_x', 5.0, 7.0, 1.0, 3.0, 'void-void')
|
|
assert (core.canonical_descriptor(edge) !=
|
|
core.canonical_descriptor(between))
|
|
assert core.canonical_descriptor(edge)[2] == 'concrete_edge'
|
|
assert core.canonical_descriptor(between)[2] == 'block_out_edge'
|
|
|
|
|
|
def test_an_opening_is_block_out_edge_and_uses_its_own_dims():
|
|
op = core.feature('P1', 0, 1.0, 4.0, 2.0, 5.0, True, 0.0, 20.0, 10.0, 8.0)
|
|
d = core.canonical_descriptor(op)
|
|
assert d[0] == core.KIND_OPENING
|
|
assert d[1] == (3.0, 3.0) # (w, h)
|
|
assert d[2] == 'block_out_edge'
|
|
assert d[3] is False
|
|
|
|
|
|
# ------------------------------------------------------- descriptor: approx
|
|
|
|
def test_an_approximate_edge_feature_never_shares_a_descriptor_with_a_measured():
|
|
measured = _strip('strip_x', 0.0, 2.0, 1.0, 3.0, 'void-edge')
|
|
approx = _strip('strip_x', 0.0, 2.0, 1.0, 3.0, 'void-edge',
|
|
outline_approximate=True)
|
|
d_m = core.canonical_descriptor(measured)
|
|
d_a = core.canonical_descriptor(approx)
|
|
assert d_m != d_a
|
|
assert d_m[3] is False and d_a[3] is True
|
|
reg = core.Registry()
|
|
assert reg.get_or_add(d_m) != reg.get_or_add(d_a) # -> different ids
|
|
|
|
|
|
def test_outline_approximate_only_splits_edge_touching_features():
|
|
"""A VOID_VOID strip measures exactly even on an approximate-outline piece,
|
|
exactly as `covers()` treats it -- so the flag must NOT split it off. Only
|
|
edge-touching features (VOID_EDGE, opening) carry the approximation."""
|
|
plain = _strip('strip_x', 0.0, 2.0, 1.0, 3.0, 'void-void')
|
|
approx = _strip('strip_x', 0.0, 2.0, 1.0, 3.0, 'void-void',
|
|
outline_approximate=True)
|
|
assert (core.canonical_descriptor(plain) ==
|
|
core.canonical_descriptor(approx))
|
|
|
|
|
|
# ---------------------------------------------------- the ids themselves
|
|
|
|
def test_ids_are_positive_sequential_and_deduped_across_panels():
|
|
reg = core.Registry()
|
|
a = _strip('strip_x', 0.0, 3.0, 0.0, 2.0, 'void-void')
|
|
b = _strip('strip_y', 0.0, 2.0, 0.0, 3.0, 'void-void')
|
|
id_a = reg.get_or_add(core.canonical_descriptor(a))
|
|
id_b = reg.get_or_add(core.canonical_descriptor(b))
|
|
assert id_a == 1 and id_b == 2 # positive, sequential from 1
|
|
# DEDUP: an identical descriptor on another panel (different absolute
|
|
# position, same kind/dims/edge) shares the id.
|
|
elsewhere = _strip('strip_x', 10.0, 13.0, 4.0, 6.0, 'void-void')
|
|
assert reg.get_or_add(core.canonical_descriptor(elsewhere)) == id_a
|
|
|
|
|
|
def test_ids_are_stable_across_a_rebuild_and_never_renumbered():
|
|
a = {'kind': 'strip_x', 'strip_ft': 3.0, 'span_ft': 2.0,
|
|
'between': 'void-void'}
|
|
b = {'kind': 'opening', 'w': 3.0, 'h': 3.0}
|
|
c = {'kind': 'strip_y', 'strip_ft': 1.0, 'span_ft': 4.0,
|
|
'between': 'void-edge'}
|
|
cache = core.new_cache('m')
|
|
reg1 = core.register_features(cache, [{'name': 'P1', 'features': [a, b]}])
|
|
id_a = reg1.lookup(core.canonical_descriptor(a))
|
|
id_b = reg1.lookup(core.canonical_descriptor(b))
|
|
# A rebuild re-registers off the SAME persisted registry (cache['registry']
|
|
# was updated in place), and a NEW descriptor is appended, not renumbered.
|
|
reg2 = core.register_features(cache, [{'name': 'P1', 'features': [a, b, c]}])
|
|
assert reg2.lookup(core.canonical_descriptor(a)) == id_a
|
|
assert reg2.lookup(core.canonical_descriptor(b)) == id_b
|
|
assert reg2.lookup(core.canonical_descriptor(c)) == 3
|
|
|
|
|
|
def test_register_features_skips_a_never_measured_record():
|
|
cache = core.new_cache('m')
|
|
reg = core.register_features(cache, [
|
|
{'name': 'P1', 'features': None}, # could-not-judge
|
|
{'name': 'P2', 'features': [
|
|
{'kind': 'opening', 'w': 2.0, 'h': 2.0}]},
|
|
])
|
|
assert reg.serialize()['next_id'] == 2 # one descriptor, from P2
|
|
|
|
|
|
# --------------------------------------------------- lookup is read-only
|
|
|
|
def test_lookup_is_read_only_and_never_grows_the_vocabulary():
|
|
"""A target-style lookup of an ABSENT descriptor returns None and does not
|
|
add it or move next_id -- the target must never mutate the shared
|
|
vocabulary."""
|
|
reg = core.Registry()
|
|
reg.get_or_add(core.canonical_descriptor(
|
|
_strip('strip_x', 0.0, 3.0, 0.0, 2.0, 'void-void')))
|
|
before = reg.serialize()
|
|
absent = core.canonical_descriptor(
|
|
_strip('strip_y', 0.0, 2.0, 0.0, 9.0, 'void-edge'))
|
|
assert reg.lookup(absent) is None
|
|
assert reg.serialize() == before # unchanged, next_id and all
|
|
assert reg.serialize()['next_id'] == 2
|
|
|
|
|
|
# --------------------------------------------------- feature_ids_of projection
|
|
|
|
def test_feature_ids_of_is_a_correct_derived_projection():
|
|
a = {'kind': 'strip_x', 'strip_ft': 3.0, 'span_ft': 2.0,
|
|
'between': 'void-void'}
|
|
b = {'kind': 'opening', 'w': 3.0, 'h': 3.0}
|
|
cache = core.new_cache('m')
|
|
reg = core.register_features(cache, [{'name': 'P1', 'features': [a, b]}])
|
|
record = {'name': 'P1', 'features': [a, b]}
|
|
ids = core.feature_ids_of(record, reg)
|
|
assert ids == [reg.lookup(core.canonical_descriptor(a)),
|
|
reg.lookup(core.canonical_descriptor(b))]
|
|
assert ids == [1, 2]
|
|
# DERIVED, not stored: nothing wrote a feature-id array onto the record.
|
|
assert 'feature_ids' not in record
|
|
|
|
|
|
def test_feature_ids_of_projects_an_unregistered_feature_to_none():
|
|
"""A target feature no donor ever produced is an accelerator MISS (None),
|
|
which `covers()` then resolves the ordinary way -- never a fresh id, so the
|
|
read-only projection cannot grow the vocabulary."""
|
|
reg = core.Registry() # empty vocabulary
|
|
novel = {'kind': 'opening', 'w': 5.0, 'h': 5.0}
|
|
assert core.feature_ids_of({'features': [novel]}, reg) == [None]
|
|
assert reg.serialize()['next_id'] == 1 # not grown
|
|
|
|
|
|
def test_feature_ids_of_propagates_could_not_judge_as_none():
|
|
reg = core.Registry()
|
|
assert core.feature_ids_of({'features': None}, reg) is None
|
|
assert core.feature_ids_of({'features': []}, reg) == []
|
|
|
|
|
|
# ----------------------------------------------------- the newly-marked diff
|
|
|
|
def test_newly_reinforced_names_only_pieces_that_became_donors():
|
|
old = core.new_cache('m')
|
|
old['pieces'] = {
|
|
'A': core.piece_record('A', 1, 's', 1, 0, features=[]), # a donor
|
|
'H': core.piece_record('H', 2, 's', 1, 1, features=[]), # on hold
|
|
'T': core.piece_record('T', 3, 's', 0, 0, features=[]), # unreinforced
|
|
}
|
|
live = {
|
|
'A': ('s', 1, 0), # still a donor -- not "newly"
|
|
'H': ('s2', 1, 0), # taken off hold -> NEW
|
|
'T': ('s2', 1, 0), # newly reinforced -> NEW
|
|
'N': ('s', 1, 0), # brand new eligible piece -> NEW
|
|
'X': ('s', 0, 0), # still ineligible -> not
|
|
}
|
|
assert core.newly_reinforced(old, live) == {'H', 'T', 'N'}
|
|
|
|
|
|
def test_newly_reinforced_against_an_empty_old_cache_is_all_current_donors():
|
|
live = {'A': ('s', 1, 0), 'B': ('s', 0, 0), 'C': ('s', 1, 1)}
|
|
assert core.newly_reinforced(core.new_cache('m'), live) == {'A'}
|
|
assert core.newly_reinforced(None, live) == {'A'}
|
|
|
|
|
|
# ------------------------------------------ the snapshot update statistics
|
|
|
|
_OPEN = {'kind': 'opening', 'w': 2.0, 'h': 2.0}
|
|
_STRIP = {'kind': 'strip_x', 'strip_ft': 3.0, 'span_ft': 2.0,
|
|
'between': 'void-void'}
|
|
|
|
|
|
def _donor_rec(name, features, reinforced=1, on_hold=0):
|
|
# A plain donor record for the diff -- bare feature dicts (no x0/x1), so
|
|
# NOT built through `piece_record` (which sorts on absolute position); the
|
|
# registry helpers key off `canonical_descriptor`, which needs no position.
|
|
return {'name': name, 'reinforced': reinforced, 'on_hold': on_hold,
|
|
'features': features}
|
|
|
|
|
|
def test_snapshot_stats_splits_new_vs_known_ids_for_newly_marked_panels():
|
|
"""The owner's ask: this press marked P2; it has 2 features; one shape was
|
|
already in the vocabulary, one is new this press."""
|
|
old = core.new_cache('m')
|
|
old['pieces'] = {'P1': _donor_rec('P1', [_OPEN])} # already a donor
|
|
core.register_features(old, [old['pieces']['P1']]) # _OPEN -> id 1
|
|
|
|
new = core.new_cache('m')
|
|
new['registry'] = old['registry'] # ids persist a rebuild
|
|
p1 = _donor_rec('P1', [_OPEN])
|
|
p2 = _donor_rec('P2', [_OPEN, _STRIP]) # newly marked
|
|
new['pieces'] = {'P1': p1, 'P2': p2}
|
|
core.register_features(new, [p1, p2]) # _STRIP gets a fresh id
|
|
|
|
live = {'P1': ('s', 1, 0), 'P2': ('s', 1, 0)}
|
|
stats = core.snapshot_stats(old, new, live)
|
|
|
|
a_id = core.Registry.deserialize(old['registry']).lookup(
|
|
core.canonical_descriptor(_OPEN))
|
|
c_id = core.Registry.deserialize(new['registry']).lookup(
|
|
core.canonical_descriptor(_STRIP))
|
|
|
|
assert stats['first_snapshot'] is False
|
|
assert stats['newly'] == ['P2'] # P1 was already a donor
|
|
panel = stats['panels'][0]
|
|
assert panel['name'] == 'P2'
|
|
assert panel['feature_count'] == 2
|
|
assert panel['known_ids'] == [a_id] # _OPEN, seen before
|
|
assert panel['new_ids'] == [c_id] # _STRIP, new this press
|
|
assert stats['known_ids'] == [a_id]
|
|
assert stats['new_ids'] == [c_id]
|
|
|
|
|
|
def test_snapshot_stats_first_snapshot_is_flagged_and_everything_is_new():
|
|
"""An absent/empty old cache is the first snapshot: every reinforced piece
|
|
is new to the pool, and its feature-IDs are all new -- never a silent
|
|
empty delta."""
|
|
new = core.new_cache('m')
|
|
p1 = _donor_rec('P1', [_OPEN, _STRIP])
|
|
new['pieces'] = {'P1': p1}
|
|
core.register_features(new, [p1])
|
|
live = {'P1': ('s', 1, 0)}
|
|
|
|
for old in (core.new_cache('m'), None):
|
|
stats = core.snapshot_stats(old, new, live)
|
|
assert stats['first_snapshot'] is True
|
|
assert stats['newly'] == ['P1']
|
|
assert stats['panels'][0]['known_ids'] == [] # nothing was known
|
|
assert len(stats['panels'][0]['new_ids']) == 2
|
|
assert len(stats['new_ids']) == 2
|
|
|
|
|
|
def test_snapshot_stats_could_not_measure_panel_propagates_none():
|
|
"""A newly-marked piece whose geometry could not be read carries
|
|
features=None; the stat propagates could-not-judge rather than reporting
|
|
zero features."""
|
|
new = core.new_cache('m')
|
|
new['pieces'] = {'P9': _donor_rec('P9', None)} # could not judge
|
|
live = {'P9': ('s', 1, 0)}
|
|
stats = core.snapshot_stats(None, new, live)
|
|
panel = stats['panels'][0]
|
|
assert panel['name'] == 'P9'
|
|
assert panel['feature_count'] is None
|
|
assert panel['feature_ids'] is None
|
|
|
|
|
|
def test_snapshot_stats_no_new_donors_is_an_empty_diff_not_a_crash():
|
|
"""Nothing newly marked -> empty `newly`, empty pools. This is "nothing
|
|
changed", which the button distinguishes from "nothing is reinforced"."""
|
|
old = core.new_cache('m')
|
|
old['pieces'] = {'P1': _donor_rec('P1', [_OPEN])}
|
|
core.register_features(old, [old['pieces']['P1']])
|
|
new = core.new_cache('m')
|
|
new['registry'] = old['registry']
|
|
new['pieces'] = {'P1': _donor_rec('P1', [_OPEN])}
|
|
core.register_features(new, [new['pieces']['P1']])
|
|
live = {'P1': ('s', 1, 0)}
|
|
stats = core.snapshot_stats(old, new, live)
|
|
assert stats['newly'] == []
|
|
assert stats['panels'] == []
|
|
assert stats['new_ids'] == [] and stats['known_ids'] == []
|
|
|
|
|
|
# ---------------------------------------------- schema 6: seed + round-trip
|
|
|
|
def test_new_cache_seeds_an_empty_registry():
|
|
cache = core.new_cache('m')
|
|
reg = core.Registry.deserialize(cache['registry'])
|
|
assert reg.serialize()['next_id'] == 1
|
|
assert core.feature_ids_of({'features': []}, reg) == []
|
|
|
|
|
|
def test_the_registry_survives_a_save_and_load_round_trip(tmp_path):
|
|
path = str(tmp_path / 'index.json')
|
|
a = {'kind': 'strip_x', 'strip_ft': 3.0, 'span_ft': 2.0,
|
|
'between': 'void-void'}
|
|
b = {'kind': 'opening', 'w': 3.0, 'h': 3.0}
|
|
cache = core.new_cache('m')
|
|
core.register_features(cache, [{'name': 'P1', 'features': [a, b]}])
|
|
core.save(path, cache)
|
|
back = core.load(path)
|
|
reg = core.Registry.deserialize(back['registry'])
|
|
assert reg.lookup(core.canonical_descriptor(a)) == 1
|
|
assert reg.lookup(core.canonical_descriptor(b)) == 2
|
|
assert reg.serialize()['next_id'] == 3 # ids preserved verbatim
|
|
|
|
|
|
def test_a_pre_registry_schema_cache_is_discarded_to_a_loud_empty(tmp_path):
|
|
"""An older-schema file is thrown away WHOLE and comes back a loud EMPTY:
|
|
the stale vocabulary is never silently adopted, and an empty registry is a
|
|
could-not-judge / NO_DONORS condition, not a success."""
|
|
path = str(tmp_path / 'index.json')
|
|
old = {'schema': core.SCHEMA - 1, 'doc_key': 'm', 'built_at': 'y',
|
|
'registry': {'next_id': 9, 'by_id': {'1': '[]'}, 'by_key': {'x': 1}},
|
|
'pieces': {'A': {'name': 'A', 'features': []}}}
|
|
with open(path, 'wb') as handle:
|
|
handle.write(json.dumps(old).encode('utf-8'))
|
|
back = core.load(path)
|
|
assert back['_load_error'] # loud
|
|
assert back['pieces'] == {} # empty
|
|
# the stale next_id=9 vocabulary is NOT adopted
|
|
assert core.Registry.deserialize(back['registry']).serialize()['next_id'] == 1
|
|
|
|
|
|
def test_a_current_schema_cache_missing_a_registry_defaults_to_empty(tmp_path):
|
|
"""A hand-built same-schema cache with no registry key at all still loads
|
|
with a valid empty vocabulary rather than leaving a caller to guess."""
|
|
path = str(tmp_path / 'index.json')
|
|
healthy = {'schema': core.SCHEMA, 'doc_key': 'm', 'built_at': 'x',
|
|
'scope': None, 'pieces': {}} # no 'registry' key
|
|
with open(path, 'wb') as handle:
|
|
handle.write(json.dumps(healthy).encode('utf-8'))
|
|
back = core.load(path)
|
|
assert core.Registry.deserialize(
|
|
back['registry']).serialize()['next_id'] == 1
|