Pypulseq Sequence install draft

seq_install.py

import os
from sys import platform

class ScannerDefinition:
    # Are we able to install to this scanner? (i.e. are we on the same network with this scanner?)
    def can_install(self):
        pass
    
    # Install a sequence to this scanner
    def install(self, seq, **kwargs):
        pass


scanner_definitions = {}
scanner_targets = {}

def register_scanner(name, definition, groups=None):
    scanner_definitions[name] = definition
    scanner_targets[name] = [name]
    
    if groups is not None:
        for g in groups:
            if g not in scanner_targets:
                scanner_targets[g] = [name]
            else:
                scanner_targets[g].append(name)

def detect_scanner(target=None):
    if target is not None:
        if target not in scanner_targets:
            raise ValueError('Unknown scanner target')
        sd = {k:scanner_definitions[k] for k in scanner_targets[target]}
    else:
        sd = scanner_definitions
        
    for scanner, definition in sd.items():       
        if definition.can_install():
            return scanner, definition
    return None, None

# Implementation for seq.install(...) to be placed in Sequence
def sequence_install(seq, target=None, **kwargs):
    _, definition = detect_scanner(target)
    if definition is not None:
        if not definition.install(seq, **kwargs):
            raise RuntimeError('Sequence install failed')

# Implementation for Siemens scanners
class SiemensDefinition(ScannerDefinition):
    def __init__(self, ice_ip, pulseq_seq_path):
        self.ice_ip = ice_ip
        self.pulseq_seq_path = pulseq_seq_path
        
    def can_install(self):
        if platform == 'win32':
            # windows
            ping_command = 'ping -w 1000 -n 1'
        elif platform == 'linux' or platform == 'darwin':
            # unix-like
            ping_command = 'ping -q -n -W1 -c1'
        
        # Does the scanner respond to pings?
        status = os.system(f'{ping_command} {self.ice_ip}')
        
        if status != 0:
            return False
        
        # Does the pulseq seq path exist on the scanner? (i.e. is this the right scanner?)
        status = os.system(f'ssh -oBatchMode=yes -oStrictHostKeyChecking=no -oHostKeyAlgorithms=+ssh-rsa root@{self.ice_ip} "test -d {self.pulseq_seq_path}"')

        return status == 0
    
    def install(self, seq, name='external'):
        seq.write('external.seq.tmp')
        
        # Transfer file with temporary filename to scanner
        status = os.system(f'scp -oBatchMode=yes -oStrictHostKeyChecking=no -oHostKeyAlgorithms=+ssh-rsa external.seq.tmp root@{self.ice_ip}:{self.pulseq_seq_path}/external_tmp.seq')
        if status != 0:
            return False
        
        # Rename temporary file to final filename on scanner
        # TODO: Check status values?
        os.system(f'ssh -oBatchMode=yes -oStrictHostKeyChecking=no -oHostKeyAlgorithms=+ssh-rsa root@{self.ice_ip} "chmod a+rw {self.pulseq_seq_path}/external_tmp.seq"')
        os.system(f'ssh -oBatchMode=yes -oStrictHostKeyChecking=no -oHostKeyAlgorithms=+ssh-rsa root@{self.ice_ip} "rm -f {self.pulseq_seq_path}/{name}.seq"')
        os.system(f'ssh -oBatchMode=yes -oStrictHostKeyChecking=no -oHostKeyAlgorithms=+ssh-rsa root@{self.ice_ip} "mv {self.pulseq_seq_path}/external_tmp.seq {self.pulseq_seq_path}/{name}.seq"')

        return True

# Numaris 4, two different IPs
register_scanner('siemens_n4_2', SiemensDefinition('192.168.2.2', '/opt/medcom/MriCustomer/seq/pulseq'), ['siemens', 'siemens_n4'])
register_scanner('siemens_n4_3', SiemensDefinition('192.168.2.3', '/opt/medcom/MriCustomer/seq/pulseq'), ['siemens', 'siemens_n4'])
# Numaris X
register_scanner('siemens_nx', SiemensDefinition('192.168.2.2', '/opt/medcom/MriCustomer/CustomerSeq/pulseq'), ['siemens'])