227 lines
12 KiB
Python
227 lines
12 KiB
Python
"""Build data/services.json (stormwater, sewer, water + HOLD notes) and data/schedules.json (signs, studs,
|
|
road-marking symbols, water fittings) for Bushveld Joy Road.
|
|
|
|
Usage:
|
|
python3 -I extract_extras.py <Bushveld folder> <Bushveld-converted folder> <site-progress-test/data>
|
|
"""
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from dxf_reader import read_sections, get # noqa: E402
|
|
from extract_bushveld import read_xlsx_rows # noqa: E402
|
|
|
|
# Afrikaans site notes found in the sewer sheet → plain English for the HOLD flags
|
|
TRANSLATE = {
|
|
'Sal nog bevestig of ons die gaan insit': 'Still to be confirmed whether this will be installed.',
|
|
'Moet nie dadelik met die begin nie. Wil net seker maak Environmental is in plek.':
|
|
'Do not start yet — first make sure the environmental approval is in place.',
|
|
}
|
|
STUD_TYPES = { # anonymous dynamic-block colours → stud type (red=10, yellow=40, white=7)
|
|
frozenset({'10'}): 'Bidirectional red/red',
|
|
frozenset({'10', '40'}): 'Red/yellow',
|
|
frozenset({'10', '7'}): 'White/red',
|
|
}
|
|
|
|
|
|
def pdf_text(path, mode='-raw'):
|
|
return subprocess.run(['pdftotext', mode, path, '-'], capture_output=True, text=True).stdout
|
|
|
|
|
|
def seg_dist(px, py, ax, ay, bx, by):
|
|
dx, dy = bx - ax, by - ay
|
|
L2 = dx * dx + dy * dy
|
|
t = 0 if L2 == 0 else max(0, min(1, ((px - ax) * dx + (py - ay) * dy) / L2))
|
|
return math.hypot(px - ax - t * dx, py - ay - t * dy)
|
|
|
|
|
|
def blocks_of(sec):
|
|
out, blk = {}, None
|
|
for typ, t in sec['BLOCKS']:
|
|
if typ == 'BLOCK':
|
|
blk = get(t, 2); out[blk] = []
|
|
elif blk is not None and typ != 'ENDBLK':
|
|
out[blk].append((typ, t))
|
|
return out
|
|
|
|
|
|
def main(bv, conv, data_dir):
|
|
base = json.load(open(os.path.join(data_dir, 'bushveld.json')))
|
|
ox, oy = base['origin']['cadX'], base['origin']['cadY']
|
|
lo = lambda Y, X: (round(-Y - ox, 3), round(-X - oy, 3)) # Lo (Y, X) → local (e, n)
|
|
pdf = lambda pat: next(os.path.join(bv, 'PDF', f) for f in sorted(os.listdir(os.path.join(bv, 'PDF'))) if re.search(pat, f))
|
|
|
|
# ---------------- stormwater ----------------
|
|
runs, cur = [], []
|
|
for r in read_xlsx_rows(os.path.join(bv, 'SO DATA', 'SW SO REV1.xlsx')):
|
|
if r and isinstance(r[0], str) and 'PIPE DESIGN' in r[0]:
|
|
if cur: runs.append(cur)
|
|
cur = []
|
|
elif len(r) >= 6 and r[0] and str(r[0]).isdigit():
|
|
cur.append({'name': r[1], 'Y': float(r[2]), 'X': float(r[3]), 'ilIn': float(r[4]), 'ilOut': float(r[5])})
|
|
if cur: runs.append(cur)
|
|
|
|
# cover/ground level + depth per structure from the long-section tables (depth, IL, GL, name)
|
|
gl = {}
|
|
for f in sorted(os.listdir(os.path.join(bv, 'PDF'))):
|
|
if 'SW LS' in f:
|
|
for m in re.finditer(r'(\d+\.\d{3}) (\d{4}\.\d{3}) (\d{4}\.\d{2}) ((?:KI|MH|JB|OUT|DL)\d+-\d+)', pdf_text(os.path.join(bv, 'PDF', f), '-layout')):
|
|
gl.setdefault(m.group(4), float(m.group(3)))
|
|
|
|
# pipe-size labels placed along the pipes in plan (stormwater DWG)
|
|
sw_sec = read_sections(os.path.join(conv, 'C3092-12-M921-001 REV 1 - SW LS (SW1 -1) - 02-07-2026 BOUND.dxf'))
|
|
labels = []
|
|
for typ, t in blocks_of(sw_sec).get('SER SW', []):
|
|
if typ in ('TEXT', 'MTEXT') and 'X12' in (get(t, 8) or ''):
|
|
m = re.fullmatch(r'\s*(\d{3,4})ø\s*', get(t, 1) or '')
|
|
if m:
|
|
labels.append((int(m.group(1)), float(get(t, 10)) - ox, float(get(t, 20)) - oy))
|
|
|
|
structures, pipes = {}, []
|
|
for ri, run in enumerate(runs):
|
|
for s in run:
|
|
e, n = lo(s['Y'], s['X'])
|
|
st = structures.setdefault(s['name'], {'name': s['name'], 'e': e, 'n': n, 'Y': s['Y'], 'X': s['X'], 'il': s['ilOut'], 'gl': gl.get(s['name'])})
|
|
st['il'] = min(st['il'], s['ilIn'], s['ilOut'])
|
|
for a, b in zip(run, run[1:]):
|
|
ea, na = lo(a['Y'], a['X']); eb, nb = lo(b['Y'], b['X'])
|
|
L = math.hypot(eb - ea, nb - na)
|
|
near = sorted((seg_dist(x, y, ea, na, eb, nb), dia) for dia, x, y in labels)
|
|
dia = near[0][1] if near and near[0][0] < 6 else 450 # drawing note: 450 mm unless shown otherwise
|
|
pipes.append({'from': a['name'], 'to': b['name'], 'run': ri + 1, 'length': round(L, 2), 'dia': dia,
|
|
'ilFrom': a['ilOut'], 'ilTo': b['ilIn'], 'grade': round((a['ilOut'] - b['ilIn']) / L * 100, 3) if L else 0,
|
|
'diaSource': 'drawing label' if near and near[0][0] < 6 else 'default 450 (drawing note 3)'})
|
|
|
|
# ---------------- sewer ----------------
|
|
sr_coords = {}
|
|
for f in sorted(os.listdir(os.path.join(bv, 'PDF'))):
|
|
if 'SR LS' in f:
|
|
for m in re.finditer(r'\b(MH\d+-\d+)\s+(6\d{4}\.\d{3})\s+(28\d{5}\.\d{3})', pdf_text(os.path.join(bv, 'PDF', f), '-layout')):
|
|
sr_coords.setdefault(m.group(1), (float(m.group(2)), float(m.group(3))))
|
|
sr_runs, cur = [], []
|
|
for r in read_xlsx_rows(os.path.join(bv, 'SO DATA', 'SR SO.xlsx')):
|
|
if r and isinstance(r[0], str) and 'PIPE DESIGN' in r[0]:
|
|
if cur: sr_runs.append(cur)
|
|
cur = []
|
|
elif len(r) >= 6 and r[0] and str(r[0]).isdigit():
|
|
note = r[6] if len(r) > 6 and r[6] else None
|
|
cur.append({'name': r[1], 'ch': float(r[2]), 'ilIn': float(r[3]), 'ilOut': float(r[4]), 'dia': int(float(r[5])), 'note': note})
|
|
if cur: sr_runs.append(cur)
|
|
sewer_mh, sewer_pipes, holds = {}, [], []
|
|
for ri, run in enumerate(sr_runs):
|
|
hold_run = None
|
|
for s in run:
|
|
c = sr_coords.get(s['name'])
|
|
mh = sewer_mh.setdefault(f"S-{s['name']}", {'name': s['name'], 'id': f"S-{s['name']}", 'run': ri + 1, 'il': s['ilOut'],
|
|
'located': bool(c), **({'Y': c[0], 'X': c[1]} if c else {})})
|
|
if c:
|
|
mh['e'], mh['n'] = lo(*c)
|
|
if s['note']:
|
|
hold = {'id': mh['id'], 'name': s['name'], 'service': 'Sewer', 'note': s['note'],
|
|
'english': TRANSLATE.get(s['note'].strip(), s['note']),
|
|
'scope': 'whole line' if s is run[0] else 'from this manhole'}
|
|
holds.append(hold); mh['hold'] = hold['english']
|
|
if s is run[0]:
|
|
hold_run = hold['english']
|
|
elif hold_run:
|
|
mh['hold'] = hold_run
|
|
for a, b in zip(run, run[1:]):
|
|
sewer_pipes.append({'from': f"S-{a['name']}", 'to': f"S-{b['name']}", 'run': ri + 1, 'dia': b['dia'],
|
|
'length': round(b['ch'] - a['ch'], 2), 'ilFrom': a['ilOut'], 'ilTo': b['ilIn'],
|
|
'grade': round((a['ilOut'] - b['ilIn']) / (b['ch'] - a['ch']) * 100, 3) if b['ch'] > a['ch'] else 0,
|
|
'hold': sewer_mh[f"S-{a['name']}"].get('hold')})
|
|
|
|
# ---------------- water ----------------
|
|
wr_text = pdf_text(pdf(r'WR LO'), '-layout')
|
|
fittings = Counter()
|
|
for m in re.finditer(r'(\d+)\s*x\s*(\d{2,3})\s*mm\s*(?:Ø\s*)?([A-Z0-9°][A-Z0-9° /\-]{2,40}?)(?=\s{2,}|$)', wr_text, re.M):
|
|
item = re.sub(r'\s+', ' ', m.group(3)).strip()
|
|
fittings[(f"{m.group(2)}mm {item}")] += int(m.group(1))
|
|
water_pipes = Counter(re.findall(r'NEW\s+(\d{2,3})mm\s*Ø', wr_text))
|
|
|
|
services = {
|
|
'stormwater': {'structures': list(structures.values()), 'pipes': pipes,
|
|
'notes': ['450mm Class 100D unless shown otherwise', 'Class B bedding', 'Min. slope 1:150 (0.667%)',
|
|
'Min. cover 600mm in road, 1000mm elsewhere']},
|
|
'sewer': {'manholes': list(sewer_mh.values()), 'pipes': sewer_pipes,
|
|
'notes': ['160mm uPVC structured wall (400kPa heavy duty) unless shown', 'House connections 110mm uPVC', 'Class B bedding']},
|
|
'water': {'points': base['water'], 'pipeLabels': dict(water_pipes)},
|
|
'holds': holds,
|
|
}
|
|
json.dump(services, open(os.path.join(data_dir, 'services.json'), 'w'), separators=(',', ':'))
|
|
|
|
# ---------------- schedules: signs, studs, markings ----------------
|
|
sg = read_sections(os.path.join(conv, 'C3092-12-M910-002 REV 1 - RD SIGN & MARK LO - 02-07-2026 BOUND.dxf'))
|
|
sb = blocks_of(sg)
|
|
road = [(p['e'], p['n']) for p in base['pegs']]
|
|
near_road = lambda e, n: min(math.hypot(e - a, n - b) for a, b in road) < 40
|
|
# Signs and road markings: count the plan labels (one label per sign/marking), merging repeats at one spot.
|
|
# Symbols are not counted separately — each placed sign also carries a label.
|
|
signs, marks = [], []
|
|
seen = []
|
|
for typ, t in sb.get('RD SGN', []):
|
|
if typ not in ('MULTILEADER', 'MTEXT', 'TEXT'):
|
|
continue
|
|
lay = (get(t, 8) or '').split('$')[-1]
|
|
e, n = float(get(t, 10) or 0) - ox, float(get(t, 20) or 0) - oy
|
|
if not near_road(e, n) or 'DET' in lay:
|
|
continue
|
|
s = re.sub(r'\\[A-Za-z][^;]*;|[{}]', ' ', ' '.join(v for c, v in t if c in (1, 3, 304)))
|
|
for m in re.finditer(r'(?:(\d+)\s*[xX]\s*)?\b((?:RTM|RM|WM|GM|[RWG][A-Z]?)\d{1,3}(?:\.\d)?)\b(?:\s*[xX]\s*(\d+))?', s):
|
|
code, qty = m.group(2), int(m.group(1) or m.group(3) or 1)
|
|
if any(c == code and math.hypot(e - a, n - b) < 3 for c, a, b in seen):
|
|
continue
|
|
seen.append((code, e, n))
|
|
item = {'code': code, 'e': round(e, 2), 'n': round(n, 2), 'qty': qty}
|
|
(marks if re.match(r'(RTM|RM|WM|GM)\d', code) else signs).append(item)
|
|
|
|
studs = Counter()
|
|
stud_pts = []
|
|
for typ, t in sb.get('MKE-RDS', []):
|
|
if typ == 'INSERT' and (get(t, 8) or '').endswith('BJ-RM STUDS'):
|
|
cols = frozenset((get(x, 62) or '').strip() for ty, x in sb.get(get(t, 2), []) if ty == 'HATCH')
|
|
kind = STUD_TYPES.get(cols, 'Other (' + ','.join(sorted(cols)) + ')')
|
|
studs[kind] += 1
|
|
stud_pts.append({'e': round(float(get(t, 10)) - ox, 2), 'n': round(float(get(t, 20)) - oy, 2), 'type': kind})
|
|
markings = Counter()
|
|
for typ, t in sb.get('MKE-RDS', []):
|
|
lay = (get(t, 8) or '').split('$')[-1]
|
|
if typ == 'INSERT' and lay in ('BJ-RM W', 'BJ-RM ARROW W'):
|
|
markings['Arrows' if 'ARROW' in lay else 'Marking symbols (yield / stop / etc.)'] += 1
|
|
# painted marking areas from hatches on BJ marking layers
|
|
from dxf_reader import hatch_rings, ring_area
|
|
paint = defaultdict(float)
|
|
for typ, t in sb.get('MKE-RDS', []):
|
|
lay = (get(t, 8) or '').split('$')[-1]
|
|
if typ == 'HATCH' and lay.startswith('BJ-RM'):
|
|
paint['Yellow paint' if 'Y' in lay.split()[-1] else 'White paint'] += sum(abs(ring_area(r)) for r in hatch_rings(t))
|
|
|
|
schedules = {
|
|
'signs': signs, 'signCounts': dict(Counter(s['code'] for s in signs for _ in range(s['qty']))),
|
|
'markingLabels': marks, 'markingCounts': dict(Counter(m['code'] for m in marks for _ in range(m['qty']))),
|
|
'studs': stud_pts, 'studCounts': dict(studs),
|
|
'markings': dict(markings), 'paintAreas': {k: round(v, 1) for k, v in paint.items()},
|
|
'waterFittings': dict(fittings),
|
|
'notes': ['Road line markings: thermoplastic paint', 'Sign poles: D-section 2mm wall; poles > 3.6m are 100mm round',
|
|
'Sign boards: 1.4mm Chromadek steel plate'],
|
|
}
|
|
json.dump(schedules, open(os.path.join(data_dir, 'schedules.json'), 'w'), indent=1)
|
|
|
|
# summary
|
|
print(f"stormwater: {len(structures)} structures, {len(pipes)} pipes, {sum(p['length'] for p in pipes):.1f} m;"
|
|
f" GL found for {sum(1 for s in structures.values() if s['gl'])}; labelled dia {sum(1 for p in pipes if p['diaSource'] == 'drawing label')}")
|
|
print(' by dia:', dict(Counter(p['dia'] for p in pipes)))
|
|
print(f"sewer: {len(sewer_mh)} manholes ({sum(1 for m in sewer_mh.values() if m['located'])} located), {len(sewer_pipes)} pipes, holds {len(holds)}")
|
|
print('water fittings:', dict(fittings), 'pipes:', dict(water_pipes))
|
|
print('signs:', schedules['signCounts'])
|
|
print('marking labels:', schedules['markingCounts'])
|
|
print('studs:', dict(studs), '| markings:', dict(markings), '| paint m²:', schedules['paintAreas'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main(*sys.argv[1:4])
|