awe/vna_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

removed symlinks to support windows

Browse Source
This commit is contained in:
Ayzen
2025-09-30 20:15:35 +03:00
parent 0ef860111e
commit c16cf2ba8a
11 changed files with 149 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@ -228,3 +228,8 @@ test*
processing_results*
plots*
# Runtime state files - managed by application
vna_system/binary_input/current_input.json
vna_system/calibration/current_calibration.json
vna_system/references/current_reference.json

1
vna_system/binary_input/current_input.bin
View File

@ -1 +0,0 @@
config_inputs/s11_start100_stop8800_points1000_bw1khz.bin

1
vna_system/calibration/current_calibration
View File

@ -1 +0,0 @@
s11_start100_stop8800_points1000_bw1khz/вфыввф

41
vna_system/core/acquisition/data_acquisition.py
View File

@ -1,8 +1,10 @@
import io
import json
import os
import struct
import threading
import time
from pathlib import Path
from typing import BinaryIO
import serial
@ -20,7 +22,8 @@ class VNADataAcquisition:
def __init__(self) -> None:
# Configuration
self.bin_log_path: str = cfg.BIN_INPUT_FILE_PATH
self.current_input_json = Path(cfg.BASE_DIR) / "binary_input" / "current_input.json"
self.config_inputs_dir = Path(cfg.BASE_DIR) / "binary_input" / "config_inputs"
self.baud: int = cfg.DEFAULT_BAUD_RATE
# Dependencies
@ -42,7 +45,32 @@ class VNADataAcquisition:
self._collected_rx_payloads: list[bytes] = []
self._meas_cmds_in_sweep: int = 0
logger.debug("VNADataAcquisition initialized", baud=self.baud, bin_log_path=self.bin_log_path)
logger.debug("VNADataAcquisition initialized", baud=self.baud)
def _get_current_bin_path(self) -> Path | None:
"""Get the path to the current binary input file from JSON config."""
if not self.current_input_json.exists():
logger.warning("Current input JSON not found", path=str(self.current_input_json))
return None
try:
with self.current_input_json.open("r", encoding="utf-8") as f:
current_info = json.load(f)
filename = current_info.get("filename")
if not filename:
logger.warning("Invalid current input JSON format")
return None
bin_path = self.config_inputs_dir / filename
if not bin_path.exists():
logger.warning("Current input binary file not found", path=str(bin_path))
return None
return bin_path
except Exception as exc: # noqa: BLE001
logger.error("Failed to load current input config", error=repr(exc))
return None
# --------------------------------------------------------------------- #
# Lifecycle
@ -165,12 +193,19 @@ class VNADataAcquisition:
logger.debug("Using port", device=port, baud=self.baud)
# Get current binary input file path
bin_path = self._get_current_bin_path()
if not bin_path:
logger.error("No current binary input configured, waiting...")
time.sleep(1.0)
continue
# Open serial + process one sweep from the binary log
with serial.Serial(port, self.baud) as ser:
self._drain_serial_input(ser)
# Open the log file each iteration to read the next sweep from start
with open(self.bin_log_path, "rb") as raw:
with open(bin_path, "rb") as raw:
buffered = io.BufferedReader(raw, buffer_size=cfg.SERIAL_BUFFER_SIZE)
self._process_sweep_data(buffered, ser)

4
vna_system/core/config.py
View File

@ -47,10 +47,6 @@ EXPECTED_POINTS_PER_SWEEP = 1000
SWEEP_BUFFER_MAX_SIZE = 100 # Maximum number of sweeps to store in circular buffer
SERIAL_BUFFER_SIZE = 512 * 1024
# -----------------------------------------------------------------------------
# Log file settings (binary input path, not to be confused with text logs)
# -----------------------------------------------------------------------------
BIN_INPUT_FILE_PATH = "./vna_system/binary_input/current_input.bin" # Symlink to current binary input
# -----------------------------------------------------------------------------
# Binary log format constants

2
vna_system/core/processors/configs/bscan_config.json
View File

@ -1,5 +1,5 @@
{
"open_air": false,
"open_air": true,
"axis": "real",
"data_limitation": "ph_only_1",
"cut": 0.816,

2
vna_system/core/processors/configs/magnitude_config.json
View File

@ -1,5 +1,5 @@
{
"y_min": -80,
"y_max": 40,
"show_phase": false
"show_phase": true
}

67
vna_system/core/settings/calibration_manager.py
View File

@ -86,7 +86,7 @@ class CalibrationManager:
Layout
------
<BASE_DIR>/calibration/
├─ current_calibration -> <preset_dir>/<calibration_name>
├─ current_calibration.json # Current active calibration metadata
├─ <preset_name>/
│ └─ <calibration_name>/
│ ├─ open.json / short.json / load.json / through.json
@ -97,7 +97,7 @@ class CalibrationManager:
def __init__(self, base_dir: Path | None = None) -> None:
self.base_dir = Path(base_dir or cfg.BASE_DIR)
self.calibration_dir = self.base_dir / "calibration"
self.current_calibration_symlink = self.calibration_dir / "current_calibration"
self.current_calibration_file = self.calibration_dir / "current_calibration.json"
self.calibration_dir.mkdir(parents=True, exist_ok=True)
self._current_working_set: CalibrationSet | None = None
@ -268,10 +268,10 @@ class CalibrationManager:
}
# ------------------------------------------------------------------ #
# Current calibration (symlink)
# Current calibration (JSON file)
# ------------------------------------------------------------------ #
def set_current_calibration(self, preset: ConfigPreset, calibration_name: str) -> None:
"""Point the `current_calibration` symlink to the chosen calibration dir."""
"""Save the current calibration selection to JSON file."""
preset_dir = self._get_preset_calibration_dir(preset)
calib_dir = preset_dir / calibration_name
if not calib_dir.exists():
@ -281,60 +281,59 @@ class CalibrationManager:
if not info.get("is_complete", False):
raise ValueError(f"Calibration {calibration_name} is incomplete")
# Refresh symlink
# Write current calibration info to JSON file
try:
if self.current_calibration_symlink.exists() or self.current_calibration_symlink.is_symlink():
self.current_calibration_symlink.unlink()
except Exception as exc: # noqa: BLE001
logger.warning("Failed to remove existing current_calibration link", error=repr(exc))
try:
# Create a relative link when possible to keep the tree portable
relative = calib_dir
try:
relative = calib_dir.relative_to(self.calibration_dir)
except ValueError:
pass
self.current_calibration_symlink.symlink_to(relative)
current_info = {
"preset_filename": preset.filename,
"calibration_name": calibration_name,
"preset_dir": preset.filename.replace(".bin", ""),
"updated_at": datetime.now().isoformat()
}
self._atomic_json_write(self.current_calibration_file, current_info)
logger.info("Current calibration set", preset=preset.filename, name=calibration_name)
except Exception as exc: # noqa: BLE001
logger.error("Failed to create current_calibration symlink", error=repr(exc))
logger.error("Failed to set current calibration", error=repr(exc))
raise
def get_current_calibration(self, current_preset: ConfigPreset) -> CalibrationSet | None:
"""
Resolve and load the calibration currently pointed to by the symlink.
Load the calibration currently saved in JSON file.
Returns None if the link doesn't exist, points to an invalid location,
Returns None if the file doesn't exist, is invalid,
or targets a different preset.
"""
if not self.current_calibration_symlink.exists():
if not self.current_calibration_file.exists():
return None
try:
target = self.current_calibration_symlink.resolve()
calibration_name = target.name
preset_dir_name = target.parent.name # <preset_name> (without .bin)
with self.current_calibration_file.open("r", encoding="utf-8") as f:
current_info = json.load(f)
expected_preset_name = current_preset.filename.replace(".bin", "")
if preset_dir_name != expected_preset_name:
calibration_name = current_info.get("calibration_name")
preset_filename = current_info.get("preset_filename")
if not calibration_name or not preset_filename:
logger.warning("Invalid current calibration file format")
return None
if preset_filename != current_preset.filename:
logger.warning(
"Current calibration preset mismatch",
expected=expected_preset_name,
actual=preset_dir_name,
expected=current_preset.filename,
actual=preset_filename,
)
raise RuntimeError("Current calibration belongs to a different preset")
return None
return self.load_calibration_set(current_preset, calibration_name)
except Exception as exc: # noqa: BLE001
logger.warning("Failed to resolve current calibration", error=repr(exc))
logger.warning("Failed to load current calibration", error=repr(exc))
return None
def clear_current_calibration(self) -> None:
"""Remove the current calibration symlink."""
if self.current_calibration_symlink.exists() or self.current_calibration_symlink.is_symlink():
"""Remove the current calibration JSON file."""
if self.current_calibration_file.exists():
try:
self.current_calibration_symlink.unlink()
self.current_calibration_file.unlink()
logger.info("Current calibration cleared")
except Exception as exc: # noqa: BLE001
logger.warning("Failed to clear current calibration", error=repr(exc))

54
vna_system/core/settings/preset_manager.py
View File

@ -1,3 +1,4 @@
import json
import re
from dataclasses import dataclass
from enum import Enum
@ -52,7 +53,7 @@ class PresetManager:
<BASE_DIR>/vna_system/binary_input/
├─ config_inputs/
│ └─ *.bin (preset files; configuration encoded in filename)
└─ current_input.bin -> config_inputs/<chosen>.bin
└─ current_input.json # Current active preset metadata
Filenames encode parameters, e.g.:
s11_start100_stop8800_points1000_bw1khz.bin
@ -69,7 +70,7 @@ class PresetManager:
def __init__(self, binary_input_dir: Path | None = None) -> None:
self.binary_input_dir = Path(binary_input_dir or (cfg.BASE_DIR / "vna_system" / "binary_input"))
self.config_inputs_dir = self.binary_input_dir / "config_inputs"
self.current_input_symlink = self.binary_input_dir / "current_input.bin"
self.current_input_file = self.binary_input_dir / "current_input.json"
self.config_inputs_dir.mkdir(parents=True, exist_ok=True)
logger.debug(
@ -184,43 +185,54 @@ class PresetManager:
def set_current_preset(self, preset: ConfigPreset) -> ConfigPreset:
"""
Select a preset by (re)pointing `current_input.bin` to the chosen file.
Select a preset by saving it to JSON file.
"""
src = self.config_inputs_dir / preset.filename
if not src.exists():
raise FileNotFoundError(f"Preset file not found: {preset.filename}")
# Remove any existing link/file
if self.current_input_symlink.exists() or self.current_input_symlink.is_symlink():
try:
self.current_input_symlink.unlink()
except Exception as exc: # noqa: BLE001
logger.warning("Failed to remove existing current_input.bin", error=repr(exc))
# Write current preset info to JSON file
current_info = {
"filename": preset.filename,
"mode": preset.mode.value,
"start_freq": preset.start_freq,
"stop_freq": preset.stop_freq,
"points": preset.points,
"bandwidth": preset.bandwidth
}
# Prefer a relative symlink for portability
try:
target = src.relative_to(self.binary_input_dir)
except ValueError:
target = src
with self.current_input_file.open("w", encoding="utf-8") as f:
json.dump(current_info, f, indent=2, ensure_ascii=False)
self.current_input_symlink.symlink_to(target)
logger.info("Current preset set", filename=preset.filename)
return preset
def get_current_preset(self) -> ConfigPreset | None:
"""
Resolve the `current_input.bin` symlink and parse the underlying preset.
Load current preset from JSON file.
Returns None if the symlink is missing or the filename cannot be parsed.
Returns None if the file is missing or invalid.
"""
if not self.current_input_symlink.exists():
if not self.current_input_file.exists():
return None
try:
target = self.current_input_symlink.resolve()
return self._parse_filename(target.name)
with self.current_input_file.open("r", encoding="utf-8") as f:
current_info = json.load(f)
filename = current_info.get("filename")
if not filename:
logger.warning("Invalid current preset file format")
return None
# Verify the file still exists
if not (self.config_inputs_dir / filename).exists():
logger.warning("Current preset file no longer exists", filename=filename)
return None
return self._parse_filename(filename)
except Exception as exc: # noqa: BLE001
logger.warning("Failed to resolve current preset", error=repr(exc))
logger.warning("Failed to load current preset", error=repr(exc))
return None
def preset_exists(self, preset: ConfigPreset) -> bool:

70
vna_system/core/settings/reference_manager.py
View File

@ -63,7 +63,7 @@ class ReferenceManager:
References are stored in the following structure:
references/
├── current_reference -> symlink to active reference
├── current_reference.json # Current active reference metadata
├── <preset_name>/
│ ├── <reference_name>/
│ │ ├── info.json # Reference metadata
@ -74,7 +74,7 @@ class ReferenceManager:
def __init__(self):
self.references_dir = Path(cfg.BASE_DIR) / "references"
self.current_reference_symlink = self.references_dir / "current_reference"
self.current_reference_file = self.references_dir / "current_reference.json"
# Ensure directory structure exists
self.references_dir.mkdir(exist_ok=True)
@ -219,17 +219,16 @@ class ReferenceManager:
raise ValueError(f"Reference '{name}' not found for preset {preset.filename}")
try:
# Remove existing symlink
if self.current_reference_symlink.exists() or self.current_reference_symlink.is_symlink():
self.current_reference_symlink.unlink()
# Write current reference info to JSON file
current_info = {
"reference_name": name,
"preset_filename": preset.filename,
"preset_dir": Path(preset.filename).stem,
"updated_at": datetime.now().isoformat()
}
# Create relative symlink for portability
try:
relative_target = reference_dir.relative_to(self.references_dir)
except ValueError:
relative_target = reference_dir
self.current_reference_symlink.symlink_to(relative_target)
with self.current_reference_file.open("w", encoding="utf-8") as f:
json.dump(current_info, f, indent=2, ensure_ascii=False)
logger.info("Set current reference", name=name, preset=preset.filename)
return True
@ -249,37 +248,42 @@ class ReferenceManager:
ReferenceInfo | None: Current reference info, or None if no reference is set
or if the current reference doesn't match the preset
"""
if not self.current_reference_symlink.exists():
if not self.current_reference_file.exists():
return None
try:
target_dir = self.current_reference_symlink.resolve()
with self.current_reference_file.open("r", encoding="utf-8") as f:
current_info = json.load(f)
if not target_dir.exists():
logger.warning("Current reference symlink points to non-existent directory")
reference_name = current_info.get("reference_name")
preset_filename = current_info.get("preset_filename")
if not reference_name or not preset_filename:
logger.warning("Invalid current reference file format")
return None
# Load reference info
info_file = target_dir / "info.json"
# Check if reference matches current preset
if preset_filename != preset.filename:
logger.debug(
"Current reference is for different preset",
reference_preset=preset_filename,
current_preset=preset.filename
)
return None
# Load full reference info
preset_dir = self._get_preset_dir(preset)
reference_dir = preset_dir / reference_name
info_file = reference_dir / "info.json"
if not info_file.exists():
logger.warning("Current reference missing info.json")
logger.warning("Current reference missing info.json", name=reference_name)
return None
with info_file.open("r", encoding="utf-8") as f:
info_data = json.load(f)
reference_info = ReferenceInfo.from_dict(info_data)
# Check if reference matches current preset
if reference_info.preset_filename != preset.filename:
logger.debug(
"Current reference is for different preset",
reference_preset=reference_info.preset_filename,
current_preset=preset.filename
)
return None
return reference_info
return ReferenceInfo.from_dict(info_data)
except Exception as exc:
logger.warning("Failed to get current reference", error=repr(exc))
@ -293,8 +297,8 @@ class ReferenceManager:
bool: True if cleared successfully
"""
try:
if self.current_reference_symlink.exists() or self.current_reference_symlink.is_symlink():
self.current_reference_symlink.unlink()
if self.current_reference_file.exists():
self.current_reference_file.unlink()
logger.info("Cleared current reference")
return True
except Exception as exc:

1
vna_system/references/current_reference
View File

@ -1 +0,0 @@
s11_start100_stop8800_points1000_bw1khz/asd