Merge pull request #110 from nunobrum/asc_editor

Asc editor

PyLTSpice/client_server/sim_client.py

@@ -26,6 +26,8 @@
 from collections import OrderedDict
 from dataclasses import dataclass
 import logging
+from typing import Union
+
 _logger = logging.getLogger("PyLTSpice.SimClient")
 
 
@@ -37,7 +39,7 @@ class SimClientInvalidRunId(LookupError):
 class JobInformation:
     """Contains information about pending simulation jobs"""
     run_number: int  # The run id that is returned by the Server and which identifies the server
-    file_dir : pathlib.Path
+    file_dir: pathlib.Path
 
 # class RunIterator(object):
 #
@@ -111,11 +113,10 @@ def __init__(self, host_address, port):
         self.stored_jobs = OrderedDict()  # This list keeps track of finished simulations that haven't yet been transferred.
         self.completed_jobs = 0
         self.minimum_time_between_server_calls = 0.2  # Minimum time between server calls
-        self._last_server_call = time.clock()
+        self._last_server_call = time.time()
 
     def __del__(self):
-        _logger.info("closing session ", self.session_id)
-        self.server.close_session(self.session_id)
+        self.close_session()
 
     def run(self, circuit):
         """
@@ -146,13 +147,13 @@ def run(self, circuit):
 
             run_id = self.server.run(self.session_id, circuit_name, zip_data)
             job_info = JobInformation(run_number=run_id, file_dir=circuit_path.parent)
-            self.started_jobs[job_info] = job_info
+            self.started_jobs[run_id] = job_info
             return run_id
         else:
             _logger.error(f"Circuit {circuit} doesn't exit")
             return -1
 
-    def get_runno_data(self, runno) -> str:
+    def get_runno_data(self, runno) -> Union[str, None]:
         """
         Returns the simulation output data inside a zip file name.
 
@@ -165,10 +166,12 @@ def get_runno_data(self, runno) -> str:
         job = self.stored_jobs.pop(runno)  # Removes it from stored jobs
         self.completed_jobs += 1
         if zip_filename != '':
-            store_path = job.file_dir
-            with open(store_path / zip_filename, 'wb') as f:
+            store_path = job.file_dir / zip_filename
+            with open(store_path, 'wb') as f:
                 f.write(zipdata.data)
-        return zip_filename
+            return store_path
+        else:
+            return None
 
     def __iter__(self):
         return self
@@ -182,28 +185,15 @@ def __next__(self):
                 # is added to the stored jobs
                 return runno
             else:
-                now = time.clock()
+                now = time.time()
                 delta = self.minimum_time_between_server_calls - (now - self._last_server_call)
                 if delta > 0:
                     time.sleep(delta)  # Go asleep for a sec
                 self._last_server_call = now
 
         # when there are no pending jobs left, exit the iterator    
         raise StopIteration
-
 
-if __name__ == "__main__":
-
-    server = SimClient('http://localhost', 9000)
-    print(server.session_id)
-    runid = server.run("../../tests/testfile.net")
-    print("Got Job id", runid)
-    for runid in server:  # Ma
-        zip_filename = server.get_runno_data(runid)
-        print(f"Received {zip_filename} from runid {runid}")
-        with zipfile.ZipFile(zip_filename, 'r') as zipf:  # Extract the contents of the zip file
-            print(zipf.namelist())  # Debug printing the contents of the zip file
-            zipf.extract(zipf.namelist()[0])  # Normally the raw file comes first
-
-    print("Finished")
-    server.server.stop_server()  # This will terminate the server
+    def close_session(self):
+        _logger.info("closing session ", self.session_id)
+        self.server.close_session(self.session_id)

PyLTSpice/client_server/sim_server.py

@@ -32,7 +32,7 @@
 
 class SimServer():
 
-    def __init__(self, simulator, parallel_sims=4, output_folder='./temp1', port=9000):
+    def __init__(self, simulator, parallel_sims=4, output_folder='./temp', port=9000):
         self.output_folder = output_folder
         self.simulation_manager = ServerSimRunner(parallel_sims=parallel_sims, timeout=5 * 60, verbose=True,
                                                   output_folder=output_folder, simulator=simulator)
@@ -114,6 +114,7 @@ def get_files(self, session_id, runno) -> Tuple[str, Binary]:
 
     def close_session(self, session_id):
         """Cleans all the pending sim_tasks with """
+        _logger.info("Closing session ", session_id)
         for runno in self.sessions[session_id]:
             self.simulation_manager.erase_files_of_runno(runno)
         return True  # Needs to return always something. None is not supported

PyLTSpice/client_server/srv_sim_runner.py

@@ -29,7 +29,7 @@
 from ..editor.base_editor import BaseEditor
 
 
-def zip_files(raw_filename: Path, log_filename:Path):
+def zip_files(raw_filename: Path, log_filename: Path):
     zip_filename = raw_filename.with_suffix('.zip')
     with zipfile.ZipFile(zip_filename, 'w', zipfile.ZIP_DEFLATED) as zip_file:
         zip_file.write(raw_filename)
@@ -67,6 +67,7 @@ def run(self) -> None:
                     self.completed_tasks.append({
                         'runno': task.runno,
                         'retcode': task.retcode,
+                        'circuit': task.netlist_file,
                         'raw': task.raw_file,
                         'log': task.log_file,
                         'zipfile': zip_filename,
@@ -97,9 +98,10 @@ def add_simulation(self, netlist: Union[str, Path, BaseEditor], *, timeout: floa
 
     def _erase_files_and_info(self, pos):
         task = self.completed_tasks[pos]
-        for filename in ('log', 'raw', 'zipfile'):
+        for filename in ('circuit', 'log', 'raw', 'zipfile'):
             f = task[filename]
             if f.exists():
+                _logger.info("deleting ", f)
                 f.unlink()
         del self.completed_tasks[pos]
 

PyLTSpice/editor/asc_editor.py

@@ -53,6 +53,9 @@ def circuit_file(self) -> Path:
         return self._asc_file_path
 
     def write_netlist(self, run_netlist_file: Union[str, Path]) -> None:
+        if isinstance(run_netlist_file, str):
+            run_netlist_file = Path(run_netlist_file)
+        run_netlist_file = run_netlist_file.with_suffix(".asc")
         with open(run_netlist_file, 'w') as asc_file:
             _logger.info(f"Writing ASC file {run_netlist_file}")
             asc_file.writelines(self._asc_file_lines)
@@ -117,7 +120,10 @@ def set_parameter(self, param: str, value: Union[str, int, float]) -> None:
         line_no, match = self._get_line_matching(".PARAM", param_regex)
         if match:
             _logger.debug(f"Parameter {param} found in ASC file, updating it")
-            value_str = format_eng(value)
+            if isinstance(value, (int, float)):
+                value_str = format_eng(value)
+            else:
+                value_str = value
             line: str = self._asc_file_lines[line_no]
             match = param_regex.search(line)  # repeating the search, so we update the correct start/stop parameter
             start, stop = match.span(param_regex.groupindex['replace'])

PyLTSpice/editor/spice_editor.py

@@ -790,6 +790,9 @@ def write_netlist(self, run_netlist_file: Union[str, Path]) -> None:
         :type run_netlist_file: Path or str
         :returns: Nothing
         """
+        if isinstance(run_netlist_file, str):
+            run_netlist_file = Path(run_netlist_file)
+        run_netlist_file = run_netlist_file.with_suffix('.net')
         with open(run_netlist_file, 'w', encoding=self.encoding) as f:
             lines = iter(self.netlist)
             for line in lines:

PyLTSpice/run_server.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+# -------------------------------------------------------------------------------
+#    ____        _   _____ ____        _
+#   |  _ \ _   _| | |_   _/ ___| _ __ (_) ___ ___
+#   | |_) | | | | |   | | \___ \| '_ \| |/ __/ _ \
+#   |  __/| |_| | |___| |  ___) | |_) | | (_|  __/
+#   |_|    \__, |_____|_| |____/| .__/|_|\___\___|
+#          |___/                |_|
+#
+# Name:        run_server.py
+# Purpose:     A Command Line Interface to run the LTSpice Server
+#
+# Author:      Nuno Brum (nuno.brum@gmail.com)
+#
+# Created:     10-08-2023
+# Licence:     refer to the LICENSE file
+# -------------------------------------------------------------------------------
+import sys
+import argparse
+from PyLTSpice.client_server.sim_server import SimServer
+import time
+import keyboard
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Run the LTSpice Server. This is a command line interface to the SimServer class."
+                    "The SimServer class is used to run simulations in parallel using a server-client architecture."
+                    "The server is a machine that runs the SimServer class and the client is a machine that runs the "
+                    "SimClient class."
+                    "The argument is the simulator to be used (LTSpice, NGSpice, XYCE, etc.)"
+    )
+    parser.add_argument('simulator', type=str, default="LTSpice",
+                        help="Simulator to be used (LTSpice, NGSpice, XYCE, etc.)")
+    parser.add_argument("-p", "--port", type=int, default=9000,
+                        help="Port to run the server. Default is 9000")
+    parser.add_argument("-o", "--output", type=str, default=".",
+                        help="Output folder for the results. Default is the current folder")
+    parser.add_argument("-l", "--parallel", type=int, default=4,
+                        help="Maximum number of parallel simulations. Default is 4")
+
+    if len(sys.argv) == 1:
+        parser.print_help(sys.stderr)
+        sys.exit(1)
+
+    args = parser.parse_args()
+    if args.parallel < 1:
+        args.parallel = 1
+
+    if args.simulator == "LTSpice":
+        from PyLTSpice.sim.ltspice_simulator import LTspice
+        simulator = LTspice
+    elif args.simulator == "NGSpice":
+        from PyLTSpice.sim.ngspice_simulator import NGSpice
+        simulator = NGSpice
+    elif args.simulator == "XYCE":
+        from PyLTSpice.sim.xyce_simulator import Xyce
+        simulator = Xyce
+    else:
+        raise ValueError(f"Simulator {args.simulator} is not supported")
+        exit(-1)
+
+    print("Starting Server")
+    server = SimServer(simulator, parallel_sims=args.parallel, output_folder=args.output, port=args.port)
+    print("Server Started. Press and hold 'q' to stop")
+    while server.running():
+        time.sleep(0.2)
+        # Check whether a key was pressed
+        if keyboard.is_pressed('q'):
+            server.stop_server()
+            break
+
+
+if __name__ == "__main__":
+    import logging
+    log1 = logging.getLogger("PyLTSpice.ServerSimRunner")
+    log2 = logging.getLogger("PyLTSpice.SimServer")
+    log3 = logging.getLogger("PyLTSpice.SimRunner")
+    log4 = logging.getLogger("PyLTSpice.RunTask")
+    log1.setLevel(logging.INFO)
+    log2.setLevel(logging.INFO)
+    log3.setLevel(logging.INFO)
+    log4.setLevel(logging.INFO)
+    main()

PyLTSpice/sim/sim_runner.py

@@ -177,10 +177,10 @@ def __init__(self, parallel_sims: int = 4, timeout=None, verbose=True, output_fo
         if simulator is None:
             from ..sim.ltspice_simulator import LTspice  # Used for defaults
             self.simulator = LTspice
-        elif issubclass(simulator, Simulator):
-            self.simulator = simulator
         elif isinstance(simulator, (str, Path)):
             self.simulator = Simulator.create_from(simulator)
+        elif issubclass(simulator, Simulator):
+            self.simulator = simulator
         else:
             raise TypeError("Invalid simulator type.")
 
@@ -263,7 +263,7 @@ def _to_output_folder(self, afile: Path, *, copy: bool, new_name: str = ''):
     def _run_file_name(self, netlist):
         if not isinstance(netlist, Path):
             netlist = Path(netlist)
-        return "%s_%i.net" % (netlist.stem, self.runno)
+        return "%s_%i%s" % (netlist.stem, self.runno, netlist.suffix)
 
     def create_netlist(self, asc_file: Union[str, Path]):
         """Creates a .net from an .asc using the LTSpice -netlist command line"""
@@ -287,7 +287,7 @@ def _prepare_sim(self, netlist: Union[str, Path, BaseEditor], run_filename: str)
                 run_filename = self._run_file_name(netlist.circuit_file)
 
             # Calculates the path where to store the new netlist.
-            run_netlist_file = self._on_output_folder(run_filename).with_suffix('.net')
+            run_netlist_file = self._on_output_folder(run_filename)
             netlist.write_netlist(run_netlist_file)
 
         elif isinstance(netlist, (Path, str)):
@@ -478,7 +478,7 @@ def file_cleanup(self):
         self.updated_stats()
         while len(self.workfiles):
             workfile = self.workfiles.pop(0)
-            if workfile.suffix == '.net':
+            if workfile.suffix == '.net' or workfile.suffix == '.asc':
                 # Delete the log file if exists
                 logfile = workfile.with_suffix('.log')
                 if logfile.exists():
@@ -497,8 +497,9 @@ def file_cleanup(self):
                     oprawfile.unlink()
 
             # Delete the file
-            _logger.info("Deleting..." + workfile.name)
-            workfile.unlink()
+            if workfile.exists():
+                _logger.info("Deleting..." + workfile.name)
+                workfile.unlink()
 
     def __iter__(self):
         return self

PyLTSpice/sim/sim_analysis.py → PyLTSpice/sim/tookit/failure_modes.py

@@ -9,49 +9,20 @@
 #   |_|    \__, |_____|_| |____/| .__/|_|\___\___|
 #          |___/                |_|
 #
-# Name:        sim_analysis.py
-# Purpose:     Classes to automate Monte-Carlo, FMEA or Worst Case Analysis
-#              be updated by user instructions
+# Name:        failure_modes.py
+# Purpose:     Class to automate FMEA
 #
 # Author:      Nuno Brum (nuno.brum@gmail.com)
 #
-# Created:     23-12-2016
+# Created:     10-08-2023
 # Licence:     refer to the LICENSE file
 # -------------------------------------------------------------------------------
 
 from collections import OrderedDict
-from typing import Iterable, Union, Optional
+from typing import Union, Optional, Iterable
 
-from .sim_runner import AnyRunner
-from ..editor.base_editor import BaseEditor, ComponentNotFoundError
-from ..sim.simulator import Simulator
-
-
-class SimAnalysis(object):
-    """
-    Base class for making Monte-Carlo, Extreme Value Analysis (EVA) or Failure Mode and Effects Analysis.
-    As a base class, a certain number of assertions must be made on the simulation results that will make the pass/fail.
-
-    Note: For the time being only measurements done with .MEAS are possible. At a later stage the parsing of RAW files
-    will be possible, although, it seems that the later solution is less computing intense.
-    """
-
-    def __init__(self, circuit_file: Union[str, BaseEditor], simulator: Optional[Simulator] = None,
-                 runner: Optional[AnyRunner] = None):
-        if isinstance(circuit_file, str):
-            from ..editor.spice_editor import SpiceEditor
-            self.editor = SpiceEditor(circuit_file)
-        else:
-            self.editor = circuit_file
-        if simulator is None:
-            from ..sim.ltspice_simulator import LTspice
-            self.simulator = LTspice()
-        else:
-            self.simulator = simulator
-        if runner is None:
-            from ..sim.sim_runner import SimRunner
-            self.runner = SimRunner(parallel_sims=1, timeout=None, verbose=False, output_folder=None)
-        self.simulations = OrderedDict()
+from editor.base_editor import BaseEditor, ComponentNotFoundError
+from .sim_analysis import SimAnalysis, AnyRunner, Simulator
 
 
 class FailureMode(SimAnalysis):