"""Explicit, bounded, read-only collectors for on-site industrial interfaces. No discovery scan, protocol writes, MQTT publication or arbitrary register function. """ import asyncio,ipaddress,json,math,re,secrets,struct,time from datetime import datetime,timezone FORMATS={'uint16':('H',1),'int16':('h',1),'uint32':('I',2),'int32':('i',2),'float32':('f',2),'float64':('d',4)} OBJECTS={'analogInput','analogValue','analogOutput','binaryInput','binaryValue','binaryOutput','multiStateInput','multiStateValue','multiStateOutput'} def integer(value,lo,hi):return type(value) is int and lo<=value<=hi def ip(value): address=ipaddress.ip_address(value) if not (address.is_private or address.is_loopback) or address.is_multicast or address.is_unspecified:raise ValueError('Une adresse locale précise est requise') return str(address) def validate(c): provider=c['provider'] if provider=='mqtt': ip(c.get('broker_host',''));topics=c.get('topics',[]) if not isinstance(topics,list) or not 1<=len(topics)<=10 or not all(isinstance(t,str) and 1<=len(t)<=160 and not any(x in t for x in ('#','+','\x00')) for t in topics):raise ValueError('Choisir 1 à 10 topics d’état exacts, sans joker') if not integer(c.get('broker_port',8883 if c.get('tls') else 1883),1,65535):raise ValueError('Port MQTT invalide') if not integer(c.get('listen_seconds',15),1,60):raise ValueError('Écoute MQTT limitée à 1–60 secondes') if c.get('broker_password') and not c.get('broker_username'):raise ValueError('Identifiant MQTT requis') return c points=c.get('points',[]) if not isinstance(points,list) or not 1<=len(points)<=40:raise ValueError('Définir entre 1 et 40 points explicitement documentés') if len({p.get('name') for p in points})!=len(points):raise ValueError('Chaque point doit avoir un nom distinct') for p in points: if not isinstance(p,dict) or not isinstance(p.get('name'),str) or not 1<=len(p['name'])<=120 or len(p.get('unit',''))>30:raise ValueError('Nom ou unité de point invalide') if provider=='modbus': if c.get('register_map_validated') is not True:raise ValueError('Valider la table de registres du modèle exact avant lecture') if c.get('transport','tcp')=='tcp': ip(c.get('host','')) if not integer(c.get('port',502),1,65535):raise ValueError('Port Modbus invalide') elif c['transport']=='rtu': if not re.fullmatch(r'/dev/[A-Za-z0-9_./-]+',str(c.get('serial_port',''))) or '..' in c['serial_port']:raise ValueError('Port série Linux local requis') if c.get('baudrate',9600) not in (1200,2400,4800,9600,19200,38400,57600,115200) or c.get('parity','N') not in ('N','E','O') or c.get('stopbits',1) not in (1,2):raise ValueError('Paramètres série invalides') else:raise ValueError('Transport Modbus inconnu') if not integer(c.get('device_id',1),1,247):raise ValueError('Adresse Modbus unicast entre 1 et 247 requise') for p in points: if p.get('function') not in ('holding','input') or p.get('type') not in FORMATS:raise ValueError('Seules les lectures 03/04 avec un type numérique documenté sont autorisées') count=FORMATS[p['type']][1] if not integer(p.get('address'),0,65536-count):raise ValueError('Adresse de registre invalide (base zéro)') if p.get('word_order','big') not in ('big','little') or p.get('byte_order','big') not in ('big','little'):raise ValueError('Ordre des octets/mots invalide') for key,default in [('scale',1),('offset',0)]: v=p.get(key,default) if not isinstance(v,(int,float)) or isinstance(v,bool) or not math.isfinite(v) or abs(v)>1e9:raise ValueError('Échelle ou décalage invalide') elif provider=='bacnet': interface=ipaddress.IPv4Interface(c.get('local_interface',''));ip(str(interface.ip));ip(c.get('device_address','')) if not integer(c.get('local_port',47808),1024,65535) or not integer(c.get('device_port',47808),1,65535) or not integer(c.get('local_device_id'),1,4194302):raise ValueError('Paramètres BACnet locaux invalides') for p in points: if p.get('object_type') not in OBJECTS or not integer(p.get('instance'),0,4194302):raise ValueError('Objet BACnet invalide') if set(p)-{'name','unit','object_type','instance'}:raise ValueError('Seule la propriété presentValue peut être lue') return c def decode(registers,p): fmt,count=FORMATS[p['type']] if len(registers)!=count or not all(integer(r,0,65535) for r in registers):raise ValueError('Réponse Modbus incomplète') words=[r.to_bytes(2,'big') for r in registers] if p.get('word_order','big')=='little':words.reverse() if p.get('byte_order','big')=='little':words=[w[::-1] for w in words] value=struct.unpack('>'+fmt,b''.join(words))[0]*p.get('scale',1)+p.get('offset',0) if not math.isfinite(value):raise ValueError('Mesure Modbus non finie') return value def collect_modbus(c): from pymodbus.client import ModbusTcpClient,ModbusSerialClient client=ModbusTcpClient(c['host'],port=c.get('port',502),timeout=2,retries=0) if c.get('transport','tcp')=='tcp' else ModbusSerialClient(c['serial_port'],baudrate=c.get('baudrate',9600),parity=c.get('parity','N'),stopbits=c.get('stopbits',1),timeout=2,retries=0) points=[];start=time.monotonic() try: if not client.connect():raise ValueError('Connexion Modbus indisponible') for p in c['points']: if time.monotonic()-start>45:raise ValueError('Temps de collecte dépassé') reader=client.read_holding_registers if p['function']=='holding' else client.read_input_registers response=reader(p['address'],count=FORMATS[p['type']][1],device_id=c.get('device_id',1)) if response.isError():raise ValueError('Lecture Modbus refusée') points.append({'name':p['name'],'value':decode(response.registers,p),'unit':p.get('unit',''),'observed_at':None}) finally:client.close() return points async def collect_bacnet(c): import BAC0 points=[] # No global discovery or write methods are called. BACnet stack runs only on site. async with asyncio.timeout(60): async with BAC0.start(ip=c['local_interface'],port=c.get('local_port',47808),deviceId=c['local_device_id'],ping=False) as bacnet: address=c['device_address']+':'+str(c.get('device_port',47808)) for p in c['points']: raw=await bacnet.read(address+' '+p['object_type']+' '+str(p['instance'])+' presentValue',timeout=3) if raw is None:raise ValueError('Point BACnet absent') if isinstance(raw,(int,float,bool)):value=raw else: try:value=float(raw) except (ValueError,TypeError):value=str(raw)[:200] if isinstance(value,float) and not math.isfinite(value):raise ValueError('Mesure BACnet non finie') points.append({'name':p['name'],'value':value,'unit':p.get('unit',''),'observed_at':None}) return points def mqtt_points(payload,topic,retained=False): if retained:return [] # A broker cache is not a fresh observation. if len(payload)>16000:return [] try:raw=json.loads(payload) except (ValueError,UnicodeDecodeError):return [] if not isinstance(raw,dict):return [] points=[] def walk(value,path,depth=0): if depth>5 or len(points)>=100 or len(path)>120:return if any(s in path.lower() for s in ('password','token','secret','email','address','serial','uuid')):return if isinstance(value,dict): for key,v in value.items():walk(v,path+'.'+str(key),depth+1) elif isinstance(value,(int,float,bool,str)) or value is None: if isinstance(value,str) and len(value)>200:return if isinstance(value,float) and not math.isfinite(value):return points.append({'name':path,'value':value,'unit':'','observed_at':None}) walk(raw,topic) return points def collect_mqtt(c): import paho.mqtt.client as mqtt points={};failure=[];connected=False client=mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,client_id='smarto-read-'+secrets.token_hex(8),clean_session=True) client.connect_timeout=5 if c.get('broker_username'):client.username_pw_set(c['broker_username'],c.get('broker_password')) if c.get('tls'):client.tls_set() # Verified TLS, never tls_insecure_set. def on_connect(cl,userdata,flags,reason_code,properties): nonlocal connected if reason_code.is_failure:failure.append(True);return connected=True for topic in c['topics']:cl.subscribe(topic,qos=0) def on_message(cl,userdata,message): if message.topic not in c['topics']:return for p in mqtt_points(message.payload,message.topic,bool(message.retain)): if len(points)<100 or p['name'] in points:points[p['name']]=p client.on_connect=on_connect;client.on_message=on_message try: client.connect(c['broker_host'],c.get('broker_port',8883 if c.get('tls') else 1883),keepalive=60) deadline=time.monotonic()+c.get('listen_seconds',15) while time.monotonic()