OMPython/OMPython/OMCSession.py at master · j-emils/OMPython · GitHub

History

1503 lines (1190 loc) · 56.7 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

# -*- coding: utf-8 -*-

"""

Definition of an OMC session.

"""

from __future__ import annotations

__license__ = """

This file is part of OpenModelica.

c/o Linköpings universitet, Department of Computer and Information Science,

SE-58183 Linköping, Sweden.

THIS PROGRAM IS PROVIDED UNDER THE TERMS OF THE BSD NEW LICENSE OR THE

GPL VERSION 3 LICENSE OR THE OSMC PUBLIC LICENSE (OSMC-PL) VERSION 1.2.

ANY USE, REPRODUCTION OR DISTRIBUTION OF THIS PROGRAM CONSTITUTES

RECIPIENT'S ACCEPTANCE OF THE OSMC PUBLIC LICENSE OR THE GPL VERSION 3,

ACCORDING TO RECIPIENTS CHOICE.

The OpenModelica software and the OSMC (Open Source Modelica Consortium)

Public License (OSMC-PL) are obtained from OSMC, either from the above

address, from the URLs: http://www.openmodelica.org or

http://www.ida.liu.se/projects/OpenModelica, and in the OpenModelica

distribution. GNU version 3 is obtained from:

http://www.gnu.org/copyleft/gpl.html. The New BSD License is obtained from:

http://www.opensource.org/licenses/BSD-3-Clause.

This program is distributed WITHOUT ANY WARRANTY; without even the implied

warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE, EXCEPT AS

EXPRESSLY SET FORTH IN THE BY RECIPIENT SELECTED SUBSIDIARY LICENSE

CONDITIONS OF OSMC-PL.

"""

import abc

import dataclasses

import io

import json

import logging

import os

import pathlib

import platform

import re

import shutil

import signal

import subprocess

import sys

import tempfile

import time

from typing import Any, Optional, Tuple

import uuid

import warnings

import zmq

import psutil

import pyparsing

# TODO: replace this with the new parser

from OMPython.OMTypedParser import om_parser_typed

from OMPython.OMParser import om_parser_basic

# define logger using the current module name as ID

logger = logging.getLogger(__name__)

class DummyPopen:

def __init__(self, pid):

self.pid = pid

self.process = psutil.Process(pid)

self.returncode = 0

def poll(self):

return None if self.process.is_running() else True

def kill(self):

return os.kill(self.pid, signal.SIGKILL)

def wait(self, timeout):

try:

self.process.wait(timeout=timeout)

except psutil.TimeoutExpired:

pass

class OMCSessionException(Exception):

pass

class OMCSessionCmd:

def __init__(self, session: OMCSessionZMQ, readonly: bool = False):

if not isinstance(session, OMCSessionZMQ):

raise OMCSessionException("Invalid session definition!")

self._session = session

self._readonly = readonly

self._omc_cache: dict[tuple[str, bool], Any] = {}

def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: bool = True):

if opt is None:

expression = question

elif isinstance(opt, list):

expression = f"{question}({','.join([str(x) for x in opt])})"

else:

raise OMCSessionException(f"Invalid definition of options for {repr(question)}: {repr(opt)}")

p = (expression, parsed)

if self._readonly and question != 'getErrorString':

# can use cache if readonly

if p in self._omc_cache:

return self._omc_cache[p]

try:

res = self._session.sendExpression(expression, parsed=parsed)

except OMCSessionException as ex:

raise OMCSessionException("OMC _ask() failed: %s (parsed=%s)", (expression, parsed)) from ex

# save response

self._omc_cache[p] = res

return res

# TODO: Open Modelica Compiler API functions. Would be nice to generate these.

def loadFile(self, filename):

return self._ask(question='loadFile', opt=[f'"{filename}"'])

def loadModel(self, className):

return self._ask(question='loadModel', opt=[className])

def isModel(self, className):

return self._ask(question='isModel', opt=[className])

def isPackage(self, className):

return self._ask(question='isPackage', opt=[className])

def isPrimitive(self, className):

return self._ask(question='isPrimitive', opt=[className])

def isConnector(self, className):

return self._ask(question='isConnector', opt=[className])

def isRecord(self, className):

return self._ask(question='isRecord', opt=[className])

def isBlock(self, className):

return self._ask(question='isBlock', opt=[className])

def isType(self, className):

return self._ask(question='isType', opt=[className])

def isFunction(self, className):

return self._ask(question='isFunction', opt=[className])

def isClass(self, className):

return self._ask(question='isClass', opt=[className])

def isParameter(self, className):

return self._ask(question='isParameter', opt=[className])

def isConstant(self, className):

return self._ask(question='isConstant', opt=[className])

def isProtected(self, className):

return self._ask(question='isProtected', opt=[className])

def getPackages(self, className="AllLoadedClasses"):

return self._ask(question='getPackages', opt=[className])

def getClassRestriction(self, className):

return self._ask(question='getClassRestriction', opt=[className])

def getDerivedClassModifierNames(self, className):

return self._ask(question='getDerivedClassModifierNames', opt=[className])

def getDerivedClassModifierValue(self, className, modifierName):

return self._ask(question='getDerivedClassModifierValue', opt=[className, modifierName])

def typeNameStrings(self, className):

return self._ask(question='typeNameStrings', opt=[className])

def getComponents(self, className):

return self._ask(question='getComponents', opt=[className])

def getClassComment(self, className):

try:

return self._ask(question='getClassComment', opt=[className])

except pyparsing.ParseException as ex:

logger.warning("Method 'getClassComment(%s)' failed; OMTypedParser error: %s",

className, ex.msg)

return 'No description available'

except OMCSessionException:

raise

def getNthComponent(self, className, comp_id):

""" returns with (type, name, description) """

return self._ask(question='getNthComponent', opt=[className, comp_id])

def getNthComponentAnnotation(self, className, comp_id):

return self._ask(question='getNthComponentAnnotation', opt=[className, comp_id])

def getImportCount(self, className):

return self._ask(question='getImportCount', opt=[className])

def getNthImport(self, className, importNumber):

# [Path, id, kind]

return self._ask(question='getNthImport', opt=[className, importNumber])

def getInheritanceCount(self, className):

return self._ask(question='getInheritanceCount', opt=[className])

def getNthInheritedClass(self, className, inheritanceDepth):

return self._ask(question='getNthInheritedClass', opt=[className, inheritanceDepth])

def getParameterNames(self, className):

try:

return self._ask(question='getParameterNames', opt=[className])

except KeyError as ex:

logger.warning('OMPython error: %s', ex)

# FIXME: OMC returns with a different structure for empty parameter set

return []

except OMCSessionException:

raise

def getParameterValue(self, className, parameterName):

try:

return self._ask(question='getParameterValue', opt=[className, parameterName])

except pyparsing.ParseException as ex:

logger.warning("Method 'getParameterValue(%s, %s)' failed; OMTypedParser error: %s",

className, parameterName, ex.msg)

return ""

except OMCSessionException:

raise

def getComponentModifierNames(self, className, componentName):

return self._ask(question='getComponentModifierNames', opt=[className, componentName])

def getComponentModifierValue(self, className, componentName):

return self._ask(question='getComponentModifierValue', opt=[className, componentName])

def getExtendsModifierNames(self, className, componentName):

return self._ask(question='getExtendsModifierNames', opt=[className, componentName])

def getExtendsModifierValue(self, className, extendsName, modifierName):

return self._ask(question='getExtendsModifierValue', opt=[className, extendsName, modifierName])

def getNthComponentModification(self, className, comp_id):

# FIXME: OMPython exception Results KeyError exception

# get {$Code(....)} field

# \{\$Code$(\S*\s*)*$\}

value = self._ask(question='getNthComponentModification', opt=[className, comp_id], parsed=False)

value = value.replace("{$Code(", "")

return value[:-3]

# return self.re_Code.findall(value)

# function getClassNames

# input TypeName class_ = $Code(AllLoadedClasses);

# input Boolean recursive = false;

# input Boolean qualified = false;

# input Boolean sort = false;

# input Boolean builtin = false "List also builtin classes if true";

# input Boolean showProtected = false "List also protected classes if true";

# output TypeName classNames[:];

# end getClassNames;

def getClassNames(self, className=None, recursive=False, qualified=False, sort=False, builtin=False,

showProtected=False):

opt = [className] if className else [] + [f'recursive={str(recursive).lower()}',

f'qualified={str(qualified).lower()}',

f'sort={str(sort).lower()}',

f'builtin={str(builtin).lower()}',

f'showProtected={str(showProtected).lower()}']

return self._ask(question='getClassNames', opt=opt)

class OMCPathReal(pathlib.PurePosixPath):

"""

Implementation of a basic (PurePosix)Path object which uses OMC as backend. The connection to OMC is provided via a

OMCSessionZMQ session object.

PurePosixPath is selected to cover usage of OMC in docker or via WSL. Usage of specialised function could result in

errors as well as usage on a Windows system due to slightly different definitions (PureWindowsPath).

"""

def __init__(self, *path, session: OMCSessionZMQ) -> None:

super().__init__(*path)

self._session = session

def with_segments(self, *pathsegments):

"""

Create a new OMCPath object with the given path segments.

The original definition of Path is overridden to ensure session is set.

"""

return type(self)(*pathsegments, session=self._session)

def is_file(self, *, follow_symlinks=True) -> bool:

"""

Check if the path is a regular file.

"""

return self._session.sendExpression(f'regularFileExists("{self.as_posix()}")')

def is_dir(self, *, follow_symlinks=True) -> bool:

"""

Check if the path is a directory.

"""

return self._session.sendExpression(f'directoryExists("{self.as_posix()}")')

def is_absolute(self):

"""

Check if the path is an absolute path considering the possibility that we are running locally on Windows. This

case needs special handling as the definition of is_absolute() differs.

"""

if isinstance(self._session, OMCProcessLocal) and platform.system() == 'Windows':

return pathlib.PureWindowsPath(self.as_posix()).is_absolute()

return super().is_absolute()

def read_text(self, encoding=None, errors=None, newline=None) -> str:

"""

Read the content of the file represented by this path as text.

The additional arguments `encoding`, `errors` and `newline` are only defined for compatibility with Path()

definition.

"""

return self._session.sendExpression(f'readFile("{self.as_posix()}")')

def write_text(self, data: str, encoding=None, errors=None, newline=None):

"""

Write text data to the file represented by this path.

The additional arguments `encoding`, `errors`, and `newline` are only defined for compatibility with Path()

definitions.

"""

if not isinstance(data, str):

raise TypeError(f"data must be str, not {data.__class__.__name__}")

data_omc = self._session.escape_str(data)

self._session.sendExpression(f'writeFile("{self.as_posix()}", "{data_omc}", false);')

return len(data)

def mkdir(self, mode=0o777, parents=False, exist_ok=False):

"""

Create a directory at the path represented by this OMCPath object.

The additional arguments `mode`, and `parents` are only defined for compatibility with Path() definitions.

"""

if self.is_dir() and not exist_ok:

raise FileExistsError(f"Directory {self.as_posix()} already exists!")

return self._session.sendExpression(f'mkdir("{self.as_posix()}")')

def cwd(self):

"""

Returns the current working directory as an OMCPath object.

"""

cwd_str = self._session.sendExpression('cd()')

return OMCPath(cwd_str, session=self._session)

def unlink(self, missing_ok: bool = False) -> None:

"""

Unlink (delete) the file or directory represented by this path.

"""

res = self._session.sendExpression(f'deleteFile("{self.as_posix()}")')

if not res and not missing_ok:

raise FileNotFoundError(f"Cannot delete file {self.as_posix()} - it does not exists!")

def resolve(self, strict: bool = False):

"""

Resolve the path to an absolute path. This is done based on available OMC functions.

"""

if strict and not (self.is_file() or self.is_dir()):

raise OMCSessionException(f"Path {self.as_posix()} does not exist!")

if self.is_file():

pathstr_resolved = self._omc_resolve(self.parent.as_posix())

omcpath_resolved = self._session.omcpath(pathstr_resolved) / self.name

elif self.is_dir():

pathstr_resolved = self._omc_resolve(self.as_posix())

omcpath_resolved = self._session.omcpath(pathstr_resolved)

else:

raise OMCSessionException(f"Path {self.as_posix()} is neither a file nor a directory!")

if not omcpath_resolved.is_file() and not omcpath_resolved.is_dir():

raise OMCSessionException(f"OMCPath resolve failed for {self.as_posix()} - path does not exist!")

return omcpath_resolved

def _omc_resolve(self, pathstr: str) -> str:

"""

Internal function to resolve the path of the OMCPath object using OMC functions *WITHOUT* changing the cwd

within OMC.

"""

expression = ('omcpath_cwd := cd(); '

f'omcpath_check := cd("{pathstr}"); ' # check requested pathstring

'cd(omcpath_cwd)')

try:

result = self._session.sendExpression(command=expression, parsed=False)

result_parts = result.split('\n')

pathstr_resolved = result_parts[1]

pathstr_resolved = pathstr_resolved[1:-1] # remove quotes

except OMCSessionException as ex:

raise OMCSessionException(f"OMCPath resolve failed for {pathstr}!") from ex

return pathstr_resolved

def absolute(self):

"""

Resolve the path to an absolute path. This is done by calling resolve() as it is the best we can do

using OMC functions.

"""

return self.resolve(strict=True)

def exists(self, follow_symlinks=True) -> bool:

"""

Semi replacement for pathlib.Path.exists().

"""

return self.is_file() or self.is_dir()

def size(self) -> int:

"""

Get the size of the file in bytes - this is an extra function and the best we can do using OMC.

"""

if not self.is_file():

raise OMCSessionException(f"Path {self.as_posix()} is not a file!")

res = self._session.sendExpression(f'stat("{self.as_posix()}")')

if res[0]:

return int(res[1])

raise OMCSessionException(f"Error reading file size for path {self.as_posix()}!")

def stat(self):

"""

The function stat() cannot be implemented using OMC.

"""

raise NotImplementedError("The function stat() cannot be implemented using OMC; "

"use size() to get the file size.")

if sys.version_info < (3, 12):

class OMCPathCompatibility(pathlib.Path):

"""

Compatibility class for OMCPath in Python < 3.12. This allows to run all code which uses OMCPath (mainly

ModelicaSystem) on these Python versions. There is one remaining limitation: only OMCProcessLocal will work as

OMCPathCompatibility is based on the standard pathlib.Path implementation.

"""

# modified copy of pathlib.Path.__new__() definition

def __new__(cls, *args, **kwargs):

logger.warning("Python < 3.12 - using a version of class OMCPath "

"based on pathlib.Path for local usage only.")

if cls is OMCPathCompatibility:

cls = OMCPathCompatibilityWindows if os.name == 'nt' else OMCPathCompatibilityPosix

self = cls._from_parts(args)

if not self._flavour.is_supported:

raise NotImplementedError("cannot instantiate %r on your system"

% (cls.__name__,))

return self

def size(self) -> int:

"""

Needed compatibility function to have the same interface as OMCPathReal

"""

return self.stat().st_size

class OMCPathCompatibilityPosix(pathlib.PosixPath, OMCPathCompatibility):

pass

class OMCPathCompatibilityWindows(pathlib.WindowsPath, OMCPathCompatibility):

pass

OMCPath = OMCPathCompatibility

else:

OMCPath = OMCPathReal

@dataclasses.dataclass

class OMCSessionRunData:

"""

Data class to store the command line data for running a model executable in the OMC environment.

All data should be defined for the environment, where OMC is running (local, docker or WSL)

"""

# cmd_path is the expected working directory

cmd_path: str

cmd_model_name: str

# command line arguments for the model executable

cmd_args: list[str]

# result file with the simulation output

cmd_result_path: str

# command prefix data (as list of strings); needed for docker or WSL

cmd_prefix: Optional[list[str]] = None

# cmd_model_executable is build out of cmd_path and cmd_model_name; this is mainly needed on Windows (add *.exe)

cmd_model_executable: Optional[str] = None

# additional library search path; this is mainly needed if OMCProcessLocal is run on Windows

cmd_library_path: Optional[str] = None

# command timeout

cmd_timeout: Optional[float] = 10.0

# working directory to be used on the *local* system

cmd_cwd_local: Optional[str] = None

def get_cmd(self) -> list[str]:

"""

Get the command line to run the model executable in the environment defined by the OMCProcess definition.

"""

if self.cmd_model_executable is None:

raise OMCSessionException("No model file defined for the model executable!")

cmdl = [] if self.cmd_prefix is None else self.cmd_prefix

cmdl += [self.cmd_model_executable] + self.cmd_args

return cmdl

class OMCSessionZMQ:

def __init__(

self,

timeout: float = 10.00,

omhome: Optional[str] = None,

omc_process: Optional[OMCProcess] = None,

) -> None:

"""

Initialisation for OMCSessionZMQ

Parameters

----------

timeout

omhome

omc_process

"""

self._timeout = timeout

if omc_process is None:

omc_process = OMCProcessLocal(omhome=omhome, timeout=timeout)

elif not isinstance(omc_process, OMCProcess):

raise OMCSessionException("Invalid definition of the OMC process!")

self.omc_process = omc_process

port = self.omc_process.get_port()

if not isinstance(port, str):

raise OMCSessionException(f"Invalid content for port: {port}")

# Create the ZeroMQ socket and connect to OMC server

context = zmq.Context.instance()

omc = context.socket(zmq.REQ)

omc.setsockopt(zmq.LINGER, 0) # Dismisses pending messages if closed

omc.setsockopt(zmq.IMMEDIATE, True) # Queue messages only to completed connections

omc.connect(port)

self.omc_zmq: Optional[zmq.Socket[bytes]] = omc

# variables to store compiled re expressions use in self.sendExpression()

self._re_log_entries: Optional[re.Pattern[str]] = None

self._re_log_raw: Optional[re.Pattern[str]] = None

def __del__(self):

if isinstance(self.omc_zmq, zmq.Socket):

try:

self.sendExpression("quit()")

except OMCSessionException:

pass

del self.omc_zmq

self.omc_zmq = None

@staticmethod

def escape_str(value: str) -> str:

"""

Escape a string such that it can be used as string within OMC expressions, i.e. escape all double quotes.

"""

return value.replace("\\", "\\\\").replace('"', '\\"')

def omcpath(self, *path) -> OMCPath:

"""

Create an OMCPath object based on the given path segments and the current OMC session.

"""

# fallback solution for Python < 3.12; a modified pathlib.Path object is used as OMCPath replacement

if sys.version_info < (3, 12):

if isinstance(self.omc_process, OMCProcessLocal):

# noinspection PyArgumentList

return OMCPath(*path)

else:

raise OMCSessionException("OMCPath is supported for Python < 3.12 only if OMCProcessLocal is used!")

else:

return OMCPath(*path, session=self)

def omcpath_tempdir(self, tempdir_base: Optional[OMCPath] = None) -> OMCPath:

"""

Get a temporary directory using OMC. It is our own implementation as non-local usage relies on OMC to run all

filesystem related access.

"""

names = [str(uuid.uuid4()) for _ in range(100)]

if tempdir_base is None:

# fallback solution for Python < 3.12; a modified pathlib.Path object is used as OMCPath replacement

if sys.version_info < (3, 12):

tempdir_str = tempfile.gettempdir()

else:

tempdir_str = self.sendExpression("getTempDirectoryPath()")

tempdir_base = self.omcpath(tempdir_str)

tempdir: Optional[OMCPath] = None

for name in names:

# create a unique temporary directory name

tempdir = tempdir_base / name

if tempdir.exists():

continue

tempdir.mkdir(parents=True, exist_ok=False)

break

if tempdir is None or not tempdir.is_dir():

raise OMCSessionException("Cannot create a temporary directory!")

return tempdir

def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:

"""

Modify data based on the selected OMCProcess implementation.

Needs to be implemented in the subclasses.

"""

return self.omc_process.omc_run_data_update(omc_run_data=omc_run_data)

@staticmethod

def run_model_executable(cmd_run_data: OMCSessionRunData) -> int:

"""

Run the command defined in cmd_run_data. This class is defined as static method such that there is no need to

keep instances of over classes around.

"""

my_env = os.environ.copy()

if isinstance(cmd_run_data.cmd_library_path, str):

my_env["PATH"] = cmd_run_data.cmd_library_path + os.pathsep + my_env["PATH"]

cmdl = cmd_run_data.get_cmd()

logger.debug("Run OM command %s in %s", repr(cmdl), cmd_run_data.cmd_path)

try:

cmdres = subprocess.run(

cmdl,

capture_output=True,

text=True,

env=my_env,

cwd=cmd_run_data.cmd_cwd_local,

timeout=cmd_run_data.cmd_timeout,

check=True,

)

stdout = cmdres.stdout.strip()

stderr = cmdres.stderr.strip()

returncode = cmdres.returncode

logger.debug("OM output for command %s:\n%s", repr(cmdl), stdout)

if stderr:

raise OMCSessionException(f"Error running model executable {repr(cmdl)}: {stderr}")

except subprocess.TimeoutExpired as ex:

raise OMCSessionException(f"Timeout running model executable {repr(cmdl)}") from ex

except subprocess.CalledProcessError as ex:

raise OMCSessionException(f"Error running model executable {repr(cmdl)}") from ex

return returncode

def execute(self, command: str):

warnings.warn("This function is depreciated and will be removed in future versions; "

"please use sendExpression() instead", DeprecationWarning, stacklevel=2)

return self.sendExpression(command, parsed=False)

def sendExpression(self, command: str, parsed: bool = True) -> Any:

"""

Send an expression to the OMC server and return the result.

"""

if self.omc_zmq is None:

raise OMCSessionException("No OMC running. Create a new instance of OMCProcess!")

logger.debug("sendExpression(%r, parsed=%r)", command, parsed)

attempts = 0

while True:

try:

self.omc_zmq.send_string(str(command), flags=zmq.NOBLOCK)

break

except zmq.error.Again:

pass

attempts += 1

if attempts >= 50:

raise OMCSessionException(f"No connection with OMC (timeout={self._timeout}). "

f"Log-file says: \n{self.omc_process.get_log()}")

time.sleep(self._timeout / 50.0)

if command == "quit()":

self.omc_zmq.close()

self.omc_zmq = None

return None

result = self.omc_zmq.recv_string()

if result.startswith('Error occurred building AST'):

raise OMCSessionException(f"OMC error: {result}")

if command == "getErrorString()":

# no error handling if 'getErrorString()' is called

if parsed:

logger.warning("Result of 'getErrorString()' cannot be parsed!")

return result

if command == "getMessagesStringInternal()":

# no error handling if 'getMessagesStringInternal()' is called

if parsed:

logger.warning("Result of 'getMessagesStringInternal()' cannot be parsed!")

return result

# always check for error

self.omc_zmq.send_string('getMessagesStringInternal()', flags=zmq.NOBLOCK)

error_raw = self.omc_zmq.recv_string()

# run error handling only if there is something to check

if error_raw != "{}\n":

if not self._re_log_entries:

self._re_log_entries = re.compile(pattern=r'record OpenModelica\.Scripting\.ErrorMessage'

'(.*?)'

r'end OpenModelica\.Scripting\.ErrorMessage;',

flags=re.MULTILINE | re.DOTALL)

if not self._re_log_raw:

self._re_log_raw = re.compile(

pattern=r"\s+message = \"(.*?)\",\n" # message

r"\s+kind = .OpenModelica.Scripting.ErrorKind.(.*?),\n" # kind

r"\s+level = .OpenModelica.Scripting.ErrorLevel.(.*?),\n" # level

r"\s+id = (.*?)" # id

"(,\n|\n)", # end marker

flags=re.MULTILINE | re.DOTALL)

# extract all ErrorMessage records

log_entries = self._re_log_entries.findall(string=error_raw)

for log_entry in reversed(log_entries):

log_raw = self._re_log_raw.findall(string=log_entry)

if len(log_raw) != 1 or len(log_raw[0]) != 5:

logger.warning("Invalid ErrorMessage record returned by 'getMessagesStringInternal()':"

f" {repr(log_entry)}!")

continue

log_message = log_raw[0][0].encode().decode('unicode_escape')

log_kind = log_raw[0][1]

log_level = log_raw[0][2]

log_id = log_raw[0][3]

msg = (f"[OMC log for 'sendExpression({command}, {parsed})']: "

f"[{log_kind}:{log_level}:{log_id}] {log_message}")

# response according to the used log level

# see: https://build.openmodelica.org/Documentation/OpenModelica.Scripting.ErrorLevel.html

if log_level == 'error':

raise OMCSessionException(msg)

if log_level == 'warning':

logger.warning(msg)

elif log_level == 'notification':

logger.info(msg)

else: # internal

logger.debug(msg)

if parsed is False:

return result

try:

return om_parser_typed(result)

except pyparsing.ParseException as ex:

logger.warning('OMTypedParser error: %s. Returning the basic parser result.', ex.msg)

try:

return om_parser_basic(result)

except (TypeError, UnboundLocalError) as ex:

raise OMCSessionException("Cannot parse OMC result") from ex

class OMCProcess(metaclass=abc.ABCMeta):

def __init__(

self,

timeout: float = 10.00,

**kwargs,

) -> None:

super().__init__(**kwargs)

# store variables

self._timeout = timeout

# omc process

self._omc_process: Optional[subprocess.Popen] = None

# omc ZMQ port to use

self._omc_port: Optional[str] = None

# generate a random string for this session

self._random_string = uuid.uuid4().hex

# omc port and log file

self._omc_filebase = f"openmodelica.{self._random_string}"

# get a temporary directory

self._temp_dir = pathlib.Path(tempfile.gettempdir())

# setup log file - this file must be closed in the destructor

logfile = self._temp_dir / (self._omc_filebase + ".log")

self._omc_loghandle: Optional[io.TextIOWrapper] = None

try:

self._omc_loghandle = open(file=logfile, mode="w+", encoding="utf-8")

except OSError as ex:

raise OMCSessionException(f"Cannot open log file {logfile}.") from ex

self._re_portfile_path = re.compile(pattern=r'\nDumped server port in file: (.*?)($|\n)',

flags=re.MULTILINE | re.DOTALL)

def __del__(self):

if self._omc_loghandle is not None:

try:

self._omc_loghandle.close()

except (OSError, IOError):

pass

self._omc_loghandle = None

if isinstance(self._omc_process, subprocess.Popen):

try:

self._omc_process.wait(timeout=2.0)

except subprocess.TimeoutExpired:

if self._omc_process:

logger.warning("OMC did not exit after being sent the quit() command; "

"killing the process with pid=%s", self._omc_process.pid)

self._omc_process.kill()

self._omc_process.wait()

finally:

self._omc_process = None

def get_port(self) -> Optional[str]:

"""

Get the port to connect to the OMC process.

"""

if not isinstance(self._omc_port, str):

raise OMCSessionException(f"Invalid port to connect to OMC process: {self._omc_port}")

return self._omc_port

def get_log(self) -> str:

"""

Get the log file content of the OMC session.

"""

if self._omc_loghandle is None:

raise OMCSessionException("Log file not available!")

self._omc_loghandle.seek(0)

log = self._omc_loghandle.read()

return log

def _get_portfile_path(self) -> Optional[pathlib.Path]:

omc_log = self.get_log()

portfile = self._re_portfile_path.findall(string=omc_log)

portfile_path = None

if portfile:

portfile_path = pathlib.Path(portfile[-1][0])

return portfile_path

@abc.abstractmethod

def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:

"""

Update the OMCSessionRunData object based on the selected OMCProcess implementation.

Needs to be implemented in the subclasses.

"""

raise NotImplementedError("This method must be implemented in subclasses!")

class OMCProcessPort(OMCProcess):

"""

OMCProcess implementation which uses a port to connect to an already running OMC server.

"""

def __init__(

self,

omc_port: str,

) -> None:

super().__init__()

self._omc_port = omc_port

def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:

"""

Update the OMCSessionRunData object based on the selected OMCProcess implementation.

"""

raise OMCSessionException("OMCProcessPort does not support omc_run_data_update()!")

class OMCProcessLocal(OMCProcess):

"""

OMCProcess implementation which runs the OMC server locally on the machine (Linux / Windows).

"""

def __init__(

self,

timeout: float = 10.00,

omhome: Optional[str | os.PathLike] = None,

) -> None:

super().__init__(timeout=timeout)

# where to find OpenModelica

self._omhome = self._omc_home_get(omhome=omhome)

# start up omc executable, which is waiting for the ZMQ connection

self._omc_process = self._omc_process_get()

# connect to the running omc instance using ZMQ

self._omc_port = self._omc_port_get()

@staticmethod

def _omc_home_get(omhome: Optional[str | os.PathLike] = None) -> pathlib.Path:

# use the provided path

if omhome is not None:

return pathlib.Path(omhome)

# check the environment variable

omhome = os.environ.get('OPENMODELICAHOME')

if omhome is not None:

return pathlib.Path(omhome)

# Get the path to the OMC executable, if not installed this will be None

path_to_omc = shutil.which("omc")

if path_to_omc is not None:

return pathlib.Path(path_to_omc).parents[1]

raise OMCSessionException("Cannot find OpenModelica executable, please install from openmodelica.org")

def _omc_process_get(self) -> subprocess.Popen:

my_env = os.environ.copy()

my_env["PATH"] = (self._omhome / "bin").as_posix() + os.pathsep + my_env["PATH"]

omc_command = [

(self._omhome / "bin" / "omc").as_posix(),

"--locale=C",

"--interactive=zmq",

f"-z={self._random_string}"]

omc_process = subprocess.Popen(omc_command,

stdout=self._omc_loghandle,

stderr=self._omc_loghandle,

env=my_env)

return omc_process

def _omc_port_get(self) -> str:

port = None

# See if the omc server is running

attempts = 0

while True:

omc_portfile_path = self._get_portfile_path()

if omc_portfile_path is not None and omc_portfile_path.is_file():

# Read the port file

with open(file=omc_portfile_path, mode='r', encoding="utf-8") as f_p:

port = f_p.readline()

break

if port is not None:

break

attempts += 1

if attempts == 80.0:

raise OMCSessionException(f"OMC Server did not start (timeout={self._timeout}). "

f"Could not open file {omc_portfile_path}. "

f"Log-file says:\n{self.get_log()}")

time.sleep(self._timeout / 80.0)

logger.info(f"Local OMC Server is up and running at ZMQ port {port} "

f"pid={self._omc_process.pid if isinstance(self._omc_process, subprocess.Popen) else '?'}")

return port

def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:

"""

Update the OMCSessionRunData object based on the selected OMCProcess implementation.

"""

# create a copy of the data

omc_run_data_copy = dataclasses.replace(omc_run_data)

# as this is the local implementation, pathlib.Path can be used

cmd_path = pathlib.Path(omc_run_data_copy.cmd_path)

if platform.system() == "Windows":

path_dll = ""

View remainder of file in raw view

OMPython/OMPython/OMCSession.py at master · j-emils/OMPython · GitHub

Expand file tree

Breadcrumbs

OMCSession.py

Latest commit

History

Breadcrumbs

OMCSession.py

File metadata and controls